fb892ea9f2
Comments, docstrings, logs, CLI help, map UI, legends, PDF sheet, scripts, compose files and AGENTS.md are now English. Data keys stay unchanged (relief_oriente, densite_sol, visualisations/, API JSON keys, link params). Wrong comments and help defaults found along the way are corrected. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
748 lines
32 KiB
Python
748 lines
32 KiB
Python
"""Tests for the XYZ tile pyramid (OpenStreetMap scheme)."""
|
||
|
||
import math
|
||
from pathlib import Path
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fixtures: fake tiles following the pipeline conventions
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _basename(col, row):
|
||
return f"LHD_FXX_{col:04d}_{row:04d}_PTS_LAMB93_IGN69"
|
||
|
||
|
||
def _make_dalle(output_dir, col, row, viz_keys, color=(200, 30, 30), px=64,
|
||
thumbs=True):
|
||
"""Create a tile (image + thumbnails) as the pipeline would."""
|
||
from PIL import Image
|
||
base = _basename(col, row)
|
||
vis = Path(output_dir) / "visualisations" / base
|
||
vis.mkdir(parents=True, exist_ok=True)
|
||
thumb_dir = Path(output_dir) / "index_thumbs"
|
||
thumb_dir.mkdir(parents=True, exist_ok=True)
|
||
for key in viz_keys:
|
||
Image.new("RGB", (px, px), color).save(
|
||
str(vis / f"{base}_{key}.webp"), format="WEBP", lossless=True)
|
||
if thumbs:
|
||
Image.new("RGB", (64, 64), color).save(
|
||
str(thumb_dir / f"{base}_{key}.jpg"), format="JPEG", quality=90)
|
||
Image.new("RGB", (64, 64), color).save(
|
||
str(thumb_dir / f"{base}_{key}_mid.jpg"), format="JPEG", quality=90)
|
||
return base
|
||
|
||
|
||
def _tile_of_cell(col, row, z):
|
||
"""Indices (x, y) of the level-z tile containing the centre of a source tile."""
|
||
from lidar_pipeline.tiles import _transformer
|
||
lon, lat = _transformer("EPSG:2154", "EPSG:4326").transform(
|
||
col * 1000 + 500, (row - 1) * 1000 + 500)
|
||
n = 2 ** z
|
||
x = int((lon + 180.0) / 360.0 * n)
|
||
rad = math.radians(lat)
|
||
y = int((1.0 - math.log(math.tan(rad) + 1 / math.cos(rad)) / math.pi) / 2.0 * n)
|
||
return x, y
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Grid geometry
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_tile_bounds_3857_known_values():
|
||
"""z0 = the whole world; z1/x1/y0 = north-east quadrant."""
|
||
from lidar_pipeline.tiles import ORIGIN, tile_bounds_3857
|
||
w, s, e, n = tile_bounds_3857(0, 0, 0)
|
||
assert (round(w), round(s), round(e), round(n)) == (
|
||
round(-ORIGIN), round(-ORIGIN), round(ORIGIN), round(ORIGIN))
|
||
w, s, e, n = tile_bounds_3857(1, 1, 0)
|
||
assert abs(w) < 1e-6 and abs(s) < 1e-6
|
||
assert abs(e - ORIGIN) < 1e-6 and abs(n - ORIGIN) < 1e-6
|
||
|
||
|
||
def test_tile_latitude_and_resolution():
|
||
"""Centre latitude and ground resolution (0.2 m/px ≈ z19 in France)."""
|
||
from lidar_pipeline.tiles import (TILE_MAX_NATIVE_Z, target_resolution,
|
||
tile_latitude)
|
||
assert abs(tile_latitude(0, 0)) < 1e-9
|
||
assert abs(tile_latitude(1, 0) - 66.51) < 0.05
|
||
# Native-level tile at the latitude of metropolitan France
|
||
z = TILE_MAX_NATIVE_Z
|
||
y = int((1.0 - math.log(math.tan(math.radians(47)) + 1 / math.cos(math.radians(47)))
|
||
/ math.pi) / 2.0 * 2 ** z)
|
||
res = target_resolution(z, y)
|
||
assert 0.15 < res < 0.25, res
|
||
# @2x: twice as fine for the same (z, x, y)
|
||
assert abs(target_resolution(z, y, scale=2) - res / 2) < 1e-9
|
||
|
||
|
||
def test_tile_bounds_l93_covers_cell():
|
||
"""The L93 extent of a tile does contain the source tile it covers."""
|
||
from lidar_pipeline.tiles import tile_bounds_l93
|
||
col, row, z = 1054, 6882, 14
|
||
x, y = _tile_of_cell(col, row, z)
|
||
min_x, min_y, max_x, max_y = tile_bounds_l93(z, x, y)
|
||
assert min_x < col * 1000 + 500 < max_x
|
||
assert min_y < (row - 1) * 1000 + 500 < max_y
|
||
|
||
|
||
def test_perspective_coeffs_identity_and_scale():
|
||
"""Identity → neutral coefficients; scaling → exact factor."""
|
||
from lidar_pipeline.tiles import perspective_coeffs
|
||
quad = [(0, 0), (256, 0), (256, 256), (0, 256)]
|
||
c = perspective_coeffs(quad, quad)
|
||
assert [round(v, 9) for v in c] == [1, 0, 0, 0, 1, 0, 0, 0]
|
||
# Output twice as large as the source: Pillow samples at x/2
|
||
c = perspective_coeffs([(0, 0), (512, 0), (512, 512), (0, 512)], quad)
|
||
assert abs(c[0] - 0.5) < 1e-9 and abs(c[4] - 0.5) < 1e-9
|
||
|
||
|
||
def test_perspective_coeffs_degenerate():
|
||
"""Degenerate quadrilateral (tile shrunk to a point) → None, no exception."""
|
||
from lidar_pipeline.tiles import perspective_coeffs
|
||
flat = [(0, 0), (0, 0), (0, 0), (0, 0)]
|
||
assert perspective_coeffs(flat, [(0, 0), (1, 0), (1, 1), (0, 1)]) is None
|
||
|
||
|
||
def test_zoom_supported_bounds():
|
||
"""Served zoom range; @2x stops one level earlier (512 px)."""
|
||
from lidar_pipeline.tiles import (TILE_MAX_NATIVE_Z, TILE_MIN_Z,
|
||
zoom_supported)
|
||
assert not zoom_supported(TILE_MIN_Z - 1)
|
||
assert zoom_supported(TILE_MIN_Z)
|
||
assert zoom_supported(TILE_MAX_NATIVE_Z)
|
||
assert not zoom_supported(TILE_MAX_NATIVE_Z + 1)
|
||
assert not zoom_supported(TILE_MAX_NATIVE_Z, scale=2)
|
||
assert zoom_supported(TILE_MAX_NATIVE_Z - 1, scale=2)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Source index
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_source_index_and_layers(tmp_path, monkeypatch):
|
||
"""The index lists the layers present and their tiers (coarse → fine)."""
|
||
from lidar_pipeline import index, tiles
|
||
monkeypatch.setattr(index, "PANEL_VIZ", None)
|
||
_make_dalle(tmp_path, 1054, 6882, ["aspect", "slope"])
|
||
layers = tiles.source_index(tmp_path, force=True)
|
||
assert set(layers) == {"aspect", "slope"}
|
||
tiers = layers["aspect"][(1054, 6882)]
|
||
# thumbnail (3.9 m/px) → intermediate (1.56) → tile (0.5)
|
||
assert [round(t[0].res, 2) for t in tiers] == [3.91, 1.56, 0.5]
|
||
assert tiles.available_layers(tmp_path) == ["slope", "aspect"] or \
|
||
set(tiles.available_layers(tmp_path)) == {"slope", "aspect"}
|
||
|
||
|
||
def test_available_layers_follow_panel(tmp_path, monkeypatch):
|
||
"""Layers on disk outside PANEL_VIZ: neither displayed nor served."""
|
||
from lidar_pipeline import index, tiles
|
||
_make_dalle(tmp_path, 1054, 6882, ["aspect", "relief_oriente"])
|
||
tiles.source_index(tmp_path, force=True)
|
||
monkeypatch.setattr(index, "PANEL_VIZ", ("relief_oriente",))
|
||
assert tiles.available_layers(tmp_path) == ["relief_oriente"]
|
||
|
||
|
||
def test_grid_bounds(tmp_path):
|
||
"""L93 and WGS84 extent of the available grid."""
|
||
from lidar_pipeline import tiles
|
||
_make_dalle(tmp_path, 1054, 6882, ["aspect"])
|
||
_make_dalle(tmp_path, 1055, 6883, ["aspect"])
|
||
tiles.source_index(tmp_path, force=True)
|
||
assert tiles.grid_bounds_l93(tmp_path) == (1054000.0, 6881000.0,
|
||
1056000.0, 6883000.0)
|
||
w, s, e, n = tiles.grid_bounds_wgs84(tmp_path)
|
||
assert 7.0 < w < 8.5 and 47.5 < s < 49.5 and e > w and n > s
|
||
|
||
|
||
def test_pick_tier_by_resolution(tmp_path):
|
||
"""Chosen tier: the coarsest whose resolution is sufficient for the tile."""
|
||
from lidar_pipeline import tiles
|
||
_make_dalle(tmp_path, 1054, 6882, ["aspect"])
|
||
tiers = tiles.source_index(tmp_path, force=True)["aspect"][(1054, 6882)]
|
||
assert tiles._pick_tier(tiers, 10.0)[0].res == tiers[0][0].res # thumbnail
|
||
assert tiles._pick_tier(tiers, 2.0)[0].res == tiers[1][0].res # intermediate
|
||
assert tiles._pick_tier(tiers, 0.2)[0].res == tiers[-1][0].res # tile
|
||
# Target finer than anything available: the finest tier is kept
|
||
assert tiles._pick_tier(tiers, 0.01)[0].res == tiers[-1][0].res
|
||
|
||
|
||
def test_sources_in_bbox_picks_tier(tmp_path):
|
||
from lidar_pipeline import tiles
|
||
_make_dalle(tmp_path, 1054, 6882, ["aspect"], px=64)
|
||
tiles.source_index(tmp_path, force=True)
|
||
bbox = (1054000.0, 6881000.0, 1055000.0, 6882000.0)
|
||
fine = tiles.sources_in_bbox(tmp_path, "aspect", bbox, 0.2)
|
||
coarse = tiles.sources_in_bbox(tmp_path, "aspect", bbox, 50.0)
|
||
assert fine and coarse and fine[0][0] == (1054, 6882)
|
||
assert min(s.res for _c, s in fine) <= min(s.res for _c, s in coarse)
|
||
assert tiles.sources_in_bbox(tmp_path, "aspect", (0, 0, 10, 10), 0.2) == []
|
||
img = tiles.load_source(fine[0][1])
|
||
assert img is not None and img.mode in ("RGB", "RGBA")
|
||
|
||
|
||
def test_subtiles_preferred_over_full_dalle(tmp_path):
|
||
"""index_subtiles quadrants serve as the fine tier (4× less to decode)."""
|
||
import pytest
|
||
from PIL import Image
|
||
from lidar_pipeline import tiles
|
||
base = _make_dalle(tmp_path, 1054, 6882, ["aspect"])
|
||
sub = tmp_path / "index_subtiles"
|
||
sub.mkdir(parents=True, exist_ok=True)
|
||
try:
|
||
for i in range(2):
|
||
for j in range(2):
|
||
Image.new("RGB", (32, 32), (10, 10, 10)).save(
|
||
str(sub / f"{base}_aspect_{i}_{j}.avif"), format="AVIF")
|
||
except Exception:
|
||
pytest.skip("AVIF encoder unavailable")
|
||
tiers = tiles.source_index(tmp_path, force=True)["aspect"][(1054, 6882)]
|
||
quads = next(t for t in tiers if len(t) == 4)
|
||
# At equal resolution, quadrants come BEFORE the whole tile
|
||
assert tiers.index(quads) < tiers.index(next(t for t in tiers if len(t) == 1
|
||
and t[0].path.suffix == ".webp"))
|
||
assert len(quads) == 4
|
||
# Quadrant extents: four adjoining half-kilometre squares
|
||
assert {q.bounds for q in quads} == {
|
||
(1054000.0, 6881000.0, 1054500.0, 6881500.0),
|
||
(1054500.0, 6881000.0, 1055000.0, 6881500.0),
|
||
(1054000.0, 6881500.0, 1054500.0, 6882000.0),
|
||
(1054500.0, 6881500.0, 1055000.0, 6882000.0)}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Rendering and cache
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_render_tile_paints_cell(tmp_path):
|
||
"""A tile over the source tile is painted; elsewhere it is empty."""
|
||
from lidar_pipeline import tiles
|
||
_make_dalle(tmp_path, 1054, 6882, ["aspect"], color=(200, 30, 30))
|
||
tiles.source_index(tmp_path, force=True)
|
||
z = 15
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
img = tiles.render_tile(tmp_path, "aspect", z, x, y)
|
||
assert img is not None and img.size == (256, 256)
|
||
r, g, b, a = img.getpixel((128, 128))
|
||
assert a == 255 and r > 150 and g < 90 and b < 90
|
||
# Distant tile (another continent): no source
|
||
assert tiles.render_tile(tmp_path, "aspect", z, 1, 1) is None
|
||
|
||
|
||
def test_render_tile_scale2(tmp_path):
|
||
"""`scale=2` renders the same extent at 512 px (@2x convention)."""
|
||
from lidar_pipeline import tiles
|
||
_make_dalle(tmp_path, 1054, 6882, ["aspect"])
|
||
tiles.source_index(tmp_path, force=True)
|
||
z = 15
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
img = tiles.render_tile(tmp_path, "aspect", z, x, y, scale=2)
|
||
assert img is not None and img.size == (512, 512)
|
||
|
||
|
||
def test_tile_edges_transparent_outside_data(tmp_path):
|
||
"""Outside the source tiles' extent, the tile stays transparent (overlayable)."""
|
||
from lidar_pipeline import tiles
|
||
_make_dalle(tmp_path, 1054, 6882, ["aspect"])
|
||
tiles.source_index(tmp_path, force=True)
|
||
# Zoom where a tile is much larger than the source tile: edges are empty
|
||
z = 11
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
img = tiles.render_tile(tmp_path, "aspect", z, x, y)
|
||
assert img is not None
|
||
assert img.getpixel((0, 0))[3] == 0
|
||
assert img.getpixel((255, 255))[3] == 0
|
||
|
||
|
||
def test_get_tile_cache_and_staleness(tmp_path, monkeypatch):
|
||
"""The disk cache is reused, then invalidated by a regenerated source tile."""
|
||
from lidar_pipeline import tiles as _t
|
||
monkeypatch.setattr(_t, "TILE_EVEN_LEVELS", False) # historical rule: everything is stored
|
||
monkeypatch.setattr(_t, "TILE_CACHE_MAX_Z", 99)
|
||
import os
|
||
import time
|
||
from lidar_pipeline import tiles
|
||
base = _make_dalle(tmp_path, 1054, 6882, ["aspect"])
|
||
tiles.source_index(tmp_path, force=True)
|
||
z = 15
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
data = tiles.get_tile(tmp_path, "aspect", z, x, y)
|
||
assert data and data[:8] == b"\x89PNG\r\n\x1a\n"
|
||
cache = tiles.tile_cache_path(tmp_path, "aspect", z, x, y)
|
||
assert cache.is_file()
|
||
first = cache.stat().st_mtime
|
||
|
||
# Unchanged: the cached tile is served again as-is
|
||
time.sleep(0.02)
|
||
assert tiles.get_tile(tmp_path, "aspect", z, x, y) == data
|
||
assert cache.stat().st_mtime == first
|
||
|
||
# Regenerated source tile (newer mtime): the tile is recomputed
|
||
src = tmp_path / "visualisations" / base / f"{base}_aspect.webp"
|
||
newer = first + 10
|
||
os.utime(src, (newer, newer))
|
||
for f in (tmp_path / "index_thumbs").iterdir():
|
||
os.utime(f, (newer, newer))
|
||
tiles.source_index(tmp_path, force=True)
|
||
tiles.get_tile(tmp_path, "aspect", z, x, y)
|
||
assert cache.stat().st_mtime > first
|
||
|
||
|
||
def test_get_tile_empty_marker(tmp_path, monkeypatch):
|
||
"""Tile without data: None + remembered .empty marker."""
|
||
from lidar_pipeline import tiles as _t
|
||
monkeypatch.setattr(_t, "TILE_EVEN_LEVELS", False) # historical rule: everything is stored
|
||
monkeypatch.setattr(_t, "TILE_CACHE_MAX_Z", 99)
|
||
from lidar_pipeline import tiles
|
||
_make_dalle(tmp_path, 1054, 6882, ["aspect"])
|
||
tiles.source_index(tmp_path, force=True)
|
||
assert tiles.get_tile(tmp_path, "aspect", 15, 1, 1) is None
|
||
assert tiles.empty_marker_exists(tmp_path, "aspect", 15, 1, 1)
|
||
|
||
|
||
def test_get_tile_webp_scale2(tmp_path):
|
||
"""@2x tier in WebP: 512 px, cache path distinct from the 256 px one."""
|
||
from PIL import Image
|
||
from lidar_pipeline import tiles
|
||
_make_dalle(tmp_path, 1054, 6882, ["aspect"])
|
||
tiles.source_index(tmp_path, force=True)
|
||
z = 15
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
data = tiles.get_tile(tmp_path, "aspect", z, x, y, scale=2, fmt="webp")
|
||
assert data and data[:4] == b"RIFF"
|
||
path = tiles.tile_cache_path(tmp_path, "aspect", z, x, y, 2, "webp")
|
||
assert path.name.endswith("@2x.webp") and path.is_file()
|
||
import io
|
||
assert Image.open(io.BytesIO(data)).size == (512, 512)
|
||
|
||
|
||
def test_transparent_tile_is_fully_transparent():
|
||
"""The fallback tile (empty area) is fully transparent."""
|
||
import io
|
||
from PIL import Image
|
||
from lidar_pipeline import tiles
|
||
img = Image.open(io.BytesIO(tiles.transparent_tile()))
|
||
assert img.size == (256, 256)
|
||
assert img.convert("RGBA").getextrema()[3] == (0, 0)
|
||
|
||
|
||
def test_tiles_in_bounds_and_warm(tmp_path):
|
||
"""Warm-up: the tiles of the extent are computed and cached."""
|
||
from lidar_pipeline import tiles
|
||
_make_dalle(tmp_path, 1054, 6882, ["aspect"])
|
||
tiles.source_index(tmp_path, force=True)
|
||
bounds = tiles.grid_bounds_wgs84(tmp_path)
|
||
assert len(tiles.tiles_in_bounds(bounds, 14)) >= 1
|
||
report = tiles.warm(tmp_path, ["aspect"], 12, 13)
|
||
assert report["rendues"] >= 1 and report["limite"] is False
|
||
assert any((tmp_path / tiles.TILE_DIRNAME / "aspect").rglob("*.png"))
|
||
|
||
|
||
def test_tiles_stamp_follows_sources(tmp_path):
|
||
"""The global version follows the most recent source-tile mtime."""
|
||
import os
|
||
from lidar_pipeline import tiles
|
||
base = _make_dalle(tmp_path, 1054, 6882, ["aspect"])
|
||
tiles.source_index(tmp_path, force=True)
|
||
first = tiles.tiles_stamp(tmp_path)
|
||
src = tmp_path / "visualisations" / base / f"{base}_aspect.webp"
|
||
os.utime(src, (first / 1000 + 60, first / 1000 + 60))
|
||
tiles.source_index(tmp_path, force=True)
|
||
assert tiles.tiles_stamp(tmp_path) > first
|
||
|
||
|
||
def test_source_cache_respects_memory_budget(tmp_path, monkeypatch):
|
||
"""The source image cache evicts by a byte budget, not by a count.
|
||
|
||
A decoded 5000² tile weighs ~100 MB (4 bytes/px): an "N entries" cache
|
||
would overflow the memory of a small machine (Raspberry Pi).
|
||
"""
|
||
from PIL import Image
|
||
from lidar_pipeline import tiles
|
||
tiles.clear_source_cache()
|
||
# Deliberately tiny budget: only one image fits at a time
|
||
monkeypatch.setattr(tiles, "SOURCE_CACHE_BYTES", 40 * 40 * 3 * 2 - 1)
|
||
paths = []
|
||
for i in range(3):
|
||
f = tmp_path / f"src{i}.png"
|
||
Image.new("RGB", (40, 40), (i * 40, 0, 0)).save(str(f))
|
||
paths.append(f)
|
||
for f in paths:
|
||
tiles._open_source(str(f), f.stat().st_mtime)
|
||
assert len(tiles._source_cache) == 1
|
||
# The last source used is the one that remains
|
||
assert str(paths[-1]) == list(tiles._source_cache)[0][0]
|
||
tiles.clear_source_cache()
|
||
assert tiles._source_cache == {}
|
||
|
||
|
||
def test_source_cache_holds_plain_decoded_images(tmp_path):
|
||
"""The cache holds pixels only: no open file, no decoder.
|
||
|
||
An open AVIF image keeps its decoder (libavif/dav1d buffers): ~43 MB
|
||
retained per 2500² quadrant instead of 25 — the Pi's container (1 GB)
|
||
was killed by the OOM killer when browsing at high zoom. The budget
|
||
counts 4 bytes/pixel: PIL stores RGB on 32 bits.
|
||
"""
|
||
from PIL import Image
|
||
from lidar_pipeline import tiles
|
||
tiles.clear_source_cache()
|
||
f = tmp_path / "src.png"
|
||
Image.new("RGB", (40, 30), (10, 20, 30)).save(str(f))
|
||
img = tiles._open_source(str(f), f.stat().st_mtime)
|
||
assert type(img) is Image.Image
|
||
assert getattr(img, "fp", None) is None
|
||
assert img.info["_bytes"] == 40 * 30 * 4
|
||
assert img.getpixel((0, 0))[:3] == (10, 20, 30)
|
||
tiles.clear_source_cache()
|
||
|
||
|
||
def test_source_cache_keyed_by_mtime(tmp_path):
|
||
"""A rewritten source is not served again from the cache."""
|
||
import os
|
||
from PIL import Image
|
||
from lidar_pipeline import tiles
|
||
tiles.clear_source_cache()
|
||
f = tmp_path / "src.png"
|
||
Image.new("RGB", (8, 8), (10, 10, 10)).save(str(f))
|
||
first = tiles._open_source(str(f), f.stat().st_mtime)
|
||
assert first.getpixel((0, 0))[:3] == (10, 10, 10)
|
||
Image.new("RGB", (8, 8), (200, 200, 200)).save(str(f))
|
||
os.utime(f, (f.stat().st_mtime + 5, f.stat().st_mtime + 5))
|
||
second = tiles._open_source(str(f), f.stat().st_mtime)
|
||
assert second.getpixel((0, 0))[:3] == (200, 200, 200)
|
||
tiles.clear_source_cache()
|
||
|
||
|
||
def test_png_palette_option_shrinks_tiles(tmp_path, monkeypatch):
|
||
"""`LIDAR_TILE_PNG_PALETTE=1` shrinks the canonical PNG (palette + alpha).
|
||
|
||
Measured on a realistic rendering (noisy color ramp): a flat-colored tile
|
||
already compresses better in RGBA than with a palette, it would prove
|
||
nothing.
|
||
"""
|
||
import io
|
||
import random
|
||
from PIL import Image
|
||
from lidar_pipeline import tiles
|
||
|
||
base = _basename(1054, 6882)
|
||
vis = tmp_path / "visualisations" / base
|
||
vis.mkdir(parents=True, exist_ok=True)
|
||
rng = random.Random(7)
|
||
img = Image.new("RGB", (256, 256))
|
||
px = img.load()
|
||
for j in range(256):
|
||
for i in range(256):
|
||
px[i, j] = (min(255, i + rng.randint(0, 12)),
|
||
min(255, j + rng.randint(0, 12)),
|
||
rng.randint(40, 90))
|
||
img.save(str(vis / f"{base}_aspect.webp"), format="WEBP", lossless=True)
|
||
tiles.source_index(tmp_path, force=True)
|
||
|
||
z = 15
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
rendered = tiles.render_tile(tmp_path, "aspect", z, x, y)
|
||
lossless = tiles._encode(rendered, "png")
|
||
monkeypatch.setattr(tiles, "PNG_PALETTE", True)
|
||
palette = tiles._encode(rendered, "png")
|
||
assert len(palette) < len(lossless)
|
||
# The palettized PNG remains a readable PNG, at the right size, with alpha
|
||
out = Image.open(io.BytesIO(palette))
|
||
assert out.size == (256, 256)
|
||
assert out.convert("RGBA").getextrema()[3][1] == 255
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Upstream tile server (LIDAR_SOURCE_URL)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _remote_payload():
|
||
"""/api/tiles payload as served by mapserve (tile server)."""
|
||
base = "LHD_FXX_1054_6882_PTS_LAMB93_IGN69_r0p2"
|
||
tiles = []
|
||
for i in range(2):
|
||
for j in range(2):
|
||
tiles.append({
|
||
"col": 1054, "row": 6882, "resolution": 0.2,
|
||
"dir_name": base, "sub_i": i, "sub_j": j, "sub_k": 2,
|
||
"viz": {"aspect": {
|
||
"thumb": f"index_subtiles/{base}_aspect_{i}_{j}_thumb160.webp?v=1700000000000",
|
||
"mid": f"index_subtiles/{base}_aspect_{i}_{j}_mid.webp?v=1700000000000",
|
||
"full": f"index_subtiles/{base}_aspect_{i}_{j}.avif?v=1700000000000",
|
||
}},
|
||
})
|
||
return {"tiles": tiles, "viz_meta": {"aspect": {"label": "Aspect"}}}
|
||
|
||
|
||
def test_remote_index_lists_tiles_without_local_data(tmp_path, monkeypatch):
|
||
"""Without any local data, the index comes from the upstream (quadrants placed)."""
|
||
from lidar_pipeline import tiles
|
||
monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://amont:8973")
|
||
monkeypatch.setattr(tiles, "_remote_payload", lambda force=False: _remote_payload())
|
||
layers = tiles.source_index(tmp_path, force=True)
|
||
assert set(layers) == {"aspect"}
|
||
tiers = layers["aspect"][(1054, 6882)]
|
||
# Three tiers (thumbnail, intermediate, quadrant) × 4 quadrants each
|
||
assert [len(t) for t in tiers] == [4, 4, 4]
|
||
assert [round(t[0].res, 2) for t in tiers] == [3.12, 0.78, 0.2]
|
||
fine = tiers[-1][0]
|
||
assert fine.url.startswith("http://amont:8973/index_subtiles/")
|
||
assert fine.path == tmp_path / fine.url.split("8973/")[1].split("?")[0]
|
||
# The reference date comes from the announced ?v=, without downloading anything
|
||
assert fine.mtime() == 1700000000.0
|
||
assert tiles.grid_bounds_l93(tmp_path) == (1054000.0, 6881000.0,
|
||
1055000.0, 6882000.0)
|
||
|
||
|
||
def test_remote_source_downloaded_on_demand(tmp_path, monkeypatch):
|
||
"""A remote source is only fetched by the first rendering that needs it."""
|
||
from PIL import Image
|
||
from lidar_pipeline import tiles
|
||
monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://amont:8973")
|
||
monkeypatch.setattr(tiles, "_remote_payload", lambda force=False: _remote_payload())
|
||
tiles.clear_source_cache()
|
||
|
||
import io
|
||
buf = io.BytesIO()
|
||
Image.new("RGB", (64, 64), (12, 200, 90)).save(buf, format="WEBP", lossless=True)
|
||
body = buf.getvalue()
|
||
fetched = []
|
||
|
||
def fake_fetch(url, dest):
|
||
fetched.append(url)
|
||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||
dest.write_bytes(body)
|
||
return True
|
||
|
||
monkeypatch.setattr(tiles, "_fetch_source", fake_fetch)
|
||
z = 15
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
img = tiles.render_tile(tmp_path, "aspect", z, x, y)
|
||
assert img is not None
|
||
assert fetched, "no source fetched"
|
||
r, g, b, a = img.getpixel((128, 128))
|
||
assert a == 255 and g > 150 and r < 80
|
||
# Fetched files land in the local cache, at the same path
|
||
assert list(tmp_path.rglob("*.webp")) or list(tmp_path.rglob("*.avif"))
|
||
# Second rendering: no more downloads (local cache)
|
||
before = len(fetched)
|
||
tiles.render_tile(tmp_path, "aspect", z, x, y)
|
||
assert len(fetched) == before
|
||
tiles.clear_source_cache()
|
||
|
||
|
||
def test_remote_index_failure_keeps_local(tmp_path, monkeypatch):
|
||
"""Upstream unreachable: the local index is still served, without exception."""
|
||
from lidar_pipeline import tiles
|
||
_make_dalle(tmp_path, 1054, 6882, ["slope"])
|
||
monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://amont:8973")
|
||
monkeypatch.setattr(tiles, "_remote_payload", lambda force=False: None)
|
||
layers = tiles.source_index(tmp_path, force=True)
|
||
assert set(layers) == {"slope"}
|
||
|
||
|
||
def test_cached_tile_states(tmp_path):
|
||
"""`cached_tile`: cache-only read — fresh / pending / empty, no rendering."""
|
||
from lidar_pipeline import tiles
|
||
_make_dalle(tmp_path, 1054, 6882, ["slope"])
|
||
x, y = _tile_of_cell(1054, 6882, 14)
|
||
data, state = tiles.cached_tile(tmp_path, "slope", 14, x, y)
|
||
assert state == "pending" and data is None
|
||
assert not tiles.tile_cache_path(tmp_path, "slope", 14, x, y).exists()
|
||
tiles.get_tile(tmp_path, "slope", 14, x, y) # rendering (maintenance, warm…)
|
||
data, state = tiles.cached_tile(tmp_path, "slope", 14, x, y)
|
||
assert state == "fresh" and data
|
||
# Outside the data: empty state + marker set, still without rendering
|
||
data, state = tiles.cached_tile(tmp_path, "slope", 14, 0, 0)
|
||
assert state == "empty" and data is None
|
||
assert tiles.empty_marker_exists(tmp_path, "slope", 14, 0, 0)
|
||
|
||
|
||
def test_storage_policy_even_levels_up_to_max(monkeypatch):
|
||
"""Reduced storage: even standard levels ≤ LIDAR_TILE_CACHE_MAX_Z (an
|
||
@2x tile at level z equals a 256 px tile at z+1)."""
|
||
from lidar_pipeline import tiles
|
||
monkeypatch.setattr(tiles, "TILE_EVEN_LEVELS", True)
|
||
monkeypatch.setattr(tiles, "TILE_CACHE_MAX_Z", 16)
|
||
assert tiles.zoom_cached(16, 1) and not tiles.zoom_cached(15, 1)
|
||
assert tiles.zoom_cached(15, 2) and not tiles.zoom_cached(16, 2) # @2x z15 = standard z16
|
||
assert not tiles.zoom_cached(18, 1) and not tiles.zoom_cached(17, 2) # fine level: on the fly
|
||
|
||
|
||
def test_unstored_level_rendered_on_the_fly_without_disk(tmp_path, monkeypatch):
|
||
"""Unstored level: tile rendered, nothing written to disk, second
|
||
request served by the memory cache."""
|
||
from lidar_pipeline import tiles
|
||
monkeypatch.setattr(tiles, "TILE_EVEN_LEVELS", True)
|
||
monkeypatch.setattr(tiles, "TILE_CACHE_MAX_Z", 16)
|
||
_make_dalle(tmp_path, 1054, 6882, ["aspect"])
|
||
tiles.source_index(tmp_path, force=True)
|
||
z = 17
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
calls = []
|
||
real = tiles.render_tile
|
||
monkeypatch.setattr(tiles, "render_tile", lambda *a, **k: calls.append(1) or real(*a, **k))
|
||
data = tiles.get_tile(tmp_path, "aspect", z, x, y, 1, "png")
|
||
assert data and data[:4] == b"\x89PNG"
|
||
assert not tiles.tile_cache_path(tmp_path, "aspect", z, x, y, 1, "png").exists()
|
||
assert tiles.get_tile(tmp_path, "aspect", z, x, y, 1, "png") == data and len(calls) == 1
|
||
|
||
|
||
def test_remote_payload_failure_not_retried_each_call(monkeypatch):
|
||
"""Upstream unreachable with no known inventory: the failure is remembered (TTL).
|
||
|
||
Otherwise every call (several per /api/map/meta) pays the connection
|
||
timeout again: the interface no longer loaded when the worker was down.
|
||
"""
|
||
import urllib.request
|
||
from lidar_pipeline import tiles
|
||
monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://amont:8973")
|
||
monkeypatch.setattr(tiles, "_remote_cache",
|
||
{"payload": None, "at": 0.0, "index": None, "root": None})
|
||
calls = []
|
||
|
||
def boom(*a, **k):
|
||
calls.append(1)
|
||
raise OSError("host unreachable")
|
||
|
||
monkeypatch.setattr(urllib.request, "urlopen", boom)
|
||
for _ in range(3):
|
||
assert tiles._remote_payload() is None
|
||
assert len(calls) == 1
|
||
|
||
|
||
def test_fetch_source_offline_breaker(tmp_path, monkeypatch):
|
||
"""Failed upstream source: the following ones fail without waiting for the
|
||
network timeout during the suspension (the maintenance falls back to local
|
||
rendering)."""
|
||
import urllib.request
|
||
from lidar_pipeline import tiles
|
||
monkeypatch.setattr(tiles, "_SOURCE_OFFLINE", {"until": 0.0})
|
||
calls = []
|
||
|
||
def boom(*a, **k):
|
||
calls.append(1)
|
||
raise OSError("host unreachable")
|
||
|
||
monkeypatch.setattr(urllib.request, "urlopen", boom)
|
||
assert tiles._fetch_source("http://amont/a", tmp_path / "a.avif") is False
|
||
assert tiles._fetch_source("http://amont/b", tmp_path / "b.avif") is False
|
||
assert len(calls) == 1
|
||
|
||
|
||
def test_fetched_source_dated_to_upstream_version(tmp_path, monkeypatch):
|
||
"""A fetched source carries the upstream version date: when the upstream
|
||
goes down, the local index finds the same dates and already-rendered tiles
|
||
stay fresh (no whole pyramid to redo)."""
|
||
from lidar_pipeline import tiles
|
||
|
||
def fake_fetch(url, dest):
|
||
dest.write_bytes(b"x")
|
||
return True
|
||
|
||
monkeypatch.setattr(tiles, "_fetch_source", fake_fetch)
|
||
src = tiles._Source(tmp_path / "q.avif", (0, 0, 1, 1), 0.2,
|
||
url="http://amont/q.avif", version=1700000000000)
|
||
assert src.ensure() is True
|
||
assert (tmp_path / "q.avif").stat().st_mtime == 1700000000.0
|
||
|
||
|
||
def test_tile_refreshed_when_older_dalle_appears(tmp_path):
|
||
"""Source tile entering the inventory AFTER an XYZ tile covering it was
|
||
rendered, but with an older version date (written before, inventoried
|
||
after): the XYZ tile must become stale — otherwise a permanent hole at
|
||
that level, on disk, in memory and in the browser (unchanged stamp)."""
|
||
import io
|
||
import os
|
||
from PIL import Image
|
||
from lidar_pipeline import tiles
|
||
tiles.clear_source_cache()
|
||
tiles._mem_tiles.clear()
|
||
z = 10
|
||
assert _tile_of_cell(1054, 6882, z) == _tile_of_cell(1055, 6882, z)
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
_make_dalle(tmp_path, 1054, 6882, ["slope"], color=(200, 30, 30))
|
||
tiles.source_index(tmp_path, force=True)
|
||
first = tiles.get_tile(tmp_path, "slope", z, x, y)
|
||
assert first is not None
|
||
stamp = tiles.tiles_stamp(tmp_path)
|
||
|
||
_make_dalle(tmp_path, 1055, 6882, ["slope"], color=(30, 200, 30))
|
||
for f in tmp_path.rglob("*1055_6882*"):
|
||
os.utime(f, (1_000_000, 1_000_000)) # version older than the XYZ tile
|
||
tiles.source_index(tmp_path, force=True)
|
||
assert tiles.cached_tile(tmp_path, "slope", z, x, y)[1] == "pending"
|
||
second = tiles.get_tile(tmp_path, "slope", z, x, y)
|
||
img = Image.open(io.BytesIO(second)).convert("RGBA")
|
||
assert any(g > 150 and r < 80 and a == 255
|
||
for r, g, b, a in img.getdata()), "new source tile missing from the XYZ tile"
|
||
assert tiles.tiles_stamp(tmp_path) > stamp
|
||
# Persistent registry: a restart invalidates nothing again
|
||
tiles._index_cache.clear()
|
||
tiles.source_index(tmp_path, force=True)
|
||
assert tiles.cached_tile(tmp_path, "slope", z, x, y)[1] == "fresh"
|
||
tiles.clear_source_cache()
|
||
|
||
|
||
def test_existing_cache_without_registry_is_refreshed_once(tmp_path):
|
||
"""Upgrade: tile cache present but no registry — the existing tiles
|
||
(possibly with holes) are made stale exactly once."""
|
||
from lidar_pipeline import tiles
|
||
tiles._seen_cache.clear()
|
||
tiles._mem_tiles.clear()
|
||
z = 10
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
_make_dalle(tmp_path, 1054, 6882, ["slope"])
|
||
tiles.source_index(tmp_path, force=True)
|
||
assert tiles.get_tile(tmp_path, "slope", z, x, y) is not None
|
||
(tmp_path / tiles.TILE_DIRNAME / tiles._SEEN_FILE).unlink() # previous version
|
||
tiles._seen_cache.clear()
|
||
tiles.source_index(tmp_path, force=True)
|
||
assert tiles.cached_tile(tmp_path, "slope", z, x, y)[1] == "pending"
|
||
tiles.get_tile(tmp_path, "slope", z, x, y)
|
||
tiles._seen_cache.clear()
|
||
tiles.source_index(tmp_path, force=True)
|
||
assert tiles.cached_tile(tmp_path, "slope", z, x, y)[1] == "fresh"
|
||
|
||
|
||
def test_webp_subtiles_indexed_and_rendered_nearest(tmp_path):
|
||
"""Density layer: lossless .webp quadrants recognised as the fine tier,
|
||
and rendered with nearest neighbour (16 exact greys even when upscaled)."""
|
||
from PIL import Image
|
||
from lidar_pipeline import tiles
|
||
base = _make_dalle(tmp_path, 1054, 6882, ["densite_sol"])
|
||
sub = tmp_path / "index_subtiles"
|
||
sub.mkdir(parents=True, exist_ok=True)
|
||
for i in range(2):
|
||
for j in range(2):
|
||
im = Image.new("L", (8, 8), 0)
|
||
im.paste(255, (0, 0, 4, 8)) # half white, half black
|
||
im.save(str(sub / f"{base}_densite_sol_{i}_{j}.webp"), format="WEBP", lossless=True)
|
||
tiers = tiles.source_index(tmp_path, force=True)["densite_sol"][(1054, 6882)]
|
||
quads = next(t for t in tiers if len(t) == 4)
|
||
assert all(q.path.suffix == ".webp" for q in quads)
|
||
assert "densite_sol" in tiles.NEAREST_LAYERS
|
||
# z17: target resolution (~0.8 m) reached by the quadrants (0.5 m here)
|
||
x, y = _tile_of_cell(1054, 6882, 17)
|
||
img = tiles.render_tile(tmp_path, "densite_sol", 17, x, y)
|
||
assert img is not None
|
||
grays = {p[0] for p in img.getdata() if p[3] == 255}
|
||
assert grays <= {0, 255} # no invented intermediate grey
|
||
|
||
|
||
def test_remote_quality_persisted_locally(tmp_path, monkeypatch):
|
||
"""The upstream quality table is copied into local sidecars (self-sufficient Pi)."""
|
||
from lidar_pipeline import tiles
|
||
from lidar_pipeline.quality import read_quality
|
||
payload = _remote_payload()
|
||
base = "LHD_FXX_1054_6882_PTS_LAMB93_IGN69"
|
||
payload["quality"] = {base: {"version": 1, "ground_density": 3.0},
|
||
"../evasion": {"version": 1}}
|
||
monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://amont")
|
||
monkeypatch.setattr(tiles, "_remote_payload", lambda force=False: payload)
|
||
monkeypatch.setattr(tiles, "_remote_cache", dict(tiles._remote_cache, index=None, built=None))
|
||
tiles._remote_index(tmp_path, force=True)
|
||
assert read_quality(tmp_path, base) == {"version": 1, "ground_density": 3.0}
|
||
assert not (tmp_path / "evasion.json").exists()
|
||
assert list((tmp_path / "quality").iterdir()) == [tmp_path / "quality" / f"{base}.json"]
|