"""Tests de la pyramide de tuiles XYZ (schéma OpenStreetMap).""" import math from pathlib import Path # --------------------------------------------------------------------------- # Fixtures : dalles factices aux conventions du pipeline # --------------------------------------------------------------------------- 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): """Crée une dalle (image + vignettes) comme le ferait le pipeline.""" 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) de la tuile du niveau z contenant le centre d'une dalle.""" 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 # --------------------------------------------------------------------------- # Géométrie de la grille # --------------------------------------------------------------------------- def test_tile_bounds_3857_known_values(): """z0 = le monde entier ; z1/x1/y0 = quadrant nord-est.""" 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(): """Latitude du centre et résolution terrain (0,2 m/px ≈ z19 en 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 # Tuile du niveau natif à la latitude de la France métropolitaine 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 : deux fois plus fin pour le même (z, x, y) assert abs(target_resolution(z, y, scale=2) - res / 2) < 1e-9 def test_tile_bounds_l93_covers_cell(): """L'emprise L93 d'une tuile contient bien la dalle qu'elle recouvre.""" 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(): """Identité → coefficients neutres ; homothétie → facteur exact.""" 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] # Sortie deux fois plus grande que la source : Pillow échantillonne à 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(): """Quadrilatère dégénéré (dalle réduite à un point) → None, pas d'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(): """Plage de zooms servie ; @2x s'arrête un cran plus tôt (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) # --------------------------------------------------------------------------- # Index des sources # --------------------------------------------------------------------------- def test_source_index_and_layers(tmp_path): """L'index liste les couches présentes et leurs paliers (grossier → fin).""" from lidar_pipeline import tiles _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)] # vignette (3,9 m/px) → intermédiaire (1,56) → dalle (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_grid_bounds(tmp_path): """Emprise L93 et WGS84 de la grille disponible.""" 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): """Palier retenu : le plus grossier dont la résolution suffit à la tuile.""" 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 # vignette assert tiles._pick_tier(tiers, 2.0)[0].res == tiers[1][0].res # intermédiaire assert tiles._pick_tier(tiers, 0.2)[0].res == tiers[-1][0].res # dalle # Cible plus fine que tout ce qui existe : on garde le palier le plus fin assert tiles._pick_tier(tiers, 0.01)[0].res == tiers[-1][0].res def test_subtiles_preferred_over_full_dalle(tmp_path): """Les quadrants index_subtiles servent de palier fin (4× moins à décoder).""" 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("encodeur AVIF indisponible") tiers = tiles.source_index(tmp_path, force=True)["aspect"][(1054, 6882)] quads = next(t for t in tiers if len(t) == 4) # À résolution égale, les quadrants passent AVANT la dalle entière 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 # Emprises des quadrants : quatre demi-kilomètres jointifs 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)} # --------------------------------------------------------------------------- # Rendu et cache # --------------------------------------------------------------------------- def test_render_tile_paints_cell(tmp_path): """Une tuile au-dessus de la dalle est peinte ; ailleurs elle est vide.""" 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 # Tuile lointaine (autre continent) : aucune source assert tiles.render_tile(tmp_path, "aspect", z, 1, 1) is None def test_render_tile_scale2(tmp_path): """`scale=2` rend la même emprise en 512 px (convention @2x).""" 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): """Hors emprise des dalles, la tuile reste transparente (superposable).""" from lidar_pipeline import tiles _make_dalle(tmp_path, 1054, 6882, ["aspect"]) tiles.source_index(tmp_path, force=True) # Zoom où une tuile est bien plus grande que la dalle : les bords sont vides 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): """Le cache disque est réutilisé, puis invalidé par une dalle régénérée.""" 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 # Sans changement : la tuile en cache est resservie telle quelle time.sleep(0.02) assert tiles.get_tile(tmp_path, "aspect", z, x, y) == data assert cache.stat().st_mtime == first # Dalle régénérée (mtime plus récente) : la tuile est recalculée 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): """Tuile sans donnée : None + marqueur .empty mémorisé.""" 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): """Palier @2x en WebP : 512 px, chemin de cache distinct du 256 px.""" 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(): """La tuile de repli (zone vide) est entièrement transparente.""" 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): """Pré-chauffage : les tuiles de l'emprise sont calculées et mises en cache.""" 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): """La version globale suit la mtime la plus récente des dalles.""" 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): """Le cache d'images sources évince selon un budget en octets, pas un compte. Une dalle 5000² pèse ~75 Mo décodée : un cache « N entrées » ferait déborder la mémoire d'une petite machine (Raspberry Pi). """ from PIL import Image from lidar_pipeline import tiles tiles.clear_source_cache() # Budget volontairement minuscule : une seule image tient à la fois 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 # La dernière source utilisée est celle qui reste assert str(paths[-1]) == list(tiles._source_cache)[0][0] tiles.clear_source_cache() assert tiles._source_cache == {} def test_source_cache_keyed_by_mtime(tmp_path): """Une source réécrite n'est pas resservie depuis le 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` allège le PNG canonique (palette + alpha). Mesuré sur un rendu réaliste (rampe de couleur bruitée) : une dalle unie compresse déjà mieux en RGBA qu'en palette, elle ne prouverait rien. """ 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) # Le PNG palettisé reste un PNG lisible, à la bonne taille, avec alpha out = Image.open(io.BytesIO(palette)) assert out.size == (256, 256) assert out.convert("RGBA").getextrema()[3][1] == 255 # --------------------------------------------------------------------------- # Serveur de dalles amont (LIDAR_SOURCE_URL) # --------------------------------------------------------------------------- def _remote_payload(): """Charge utile /api/tiles telle que la sert la webapp du pipeline.""" 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): """Sans aucune donnée locale, l'index vient de l'amont (quadrants placés).""" 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)] # Trois paliers (vignette, intermédiaire, quadrant) × 4 quadrants chacun 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] # La date de référence vient du ?v= annoncé, sans rien télécharger 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): """Une source distante n'est rapatriée qu'au premier rendu qui en a besoin.""" 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, "aucune source rapatriée" r, g, b, a = img.getpixel((128, 128)) assert a == 255 and g > 150 and r < 80 # Les fichiers rapatriés atterrissent dans le cache local, au même chemin assert list(tmp_path.rglob("*.webp")) or list(tmp_path.rglob("*.avif")) # Second rendu : plus aucun téléchargement (cache local) 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): """Amont injoignable : l'index local reste servi, sans 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` : lecture cache seule — fresh / pending / empty, sans rendu.""" 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) # rendu (maintenance, warm…) data, state = tiles.cached_tile(tmp_path, "slope", 14, x, y) assert state == "fresh" and data # Hors données : état empty + marqueur posé, toujours sans rendu 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)