Porter la génération de tuiles dans lidar-maps et retirer l'ancienne webapp

La carte à tuiles XYZ devient la seule interface : l'API de génération
(preview, generate, status, stop, file, cell), le job unique + file
persistante, la délégation LIDAR_GENERATION_URL et les garde-fous
(token, LIDAR_REGEN_CIDR) passent de webapp.py à mapserve.py ; l'interface
gagne les boutons + Zone / ⤒ Compléter / régénération à la dalle, les
options de run et la progression cadre par cadre. mapserve sert aussi
l'inventaire /api/tiles et les dalles en statique versionné : le worker
image complète remplace la webapp sur le 8973 du générateur, le Pi
n'exécute plus que lidar-maps. index.py se réduit aux registres partagés
+ vignettes/sous-tuiles/inventaire (l'UI HTML/JS et les mosaïques
d'overview partent avec export.py et les compose/scripts de la webapp).
This commit is contained in:
Antoine Jacquin
2026-09-23 23:21:05 +02:00
parent 81defa85c9
commit 0a2ccee958
31 changed files with 1774 additions and 8465 deletions

View File

@ -1,276 +0,0 @@
"""Tests de l'export multi-dalles (mosaïque jointive image/PDF) et de son API."""
def _make_tile_dir(tmp_path, col, row, viz_keys, resolution=0.5, tile_px=16,
color=(255, 0, 0)):
"""Crée le dossier de visualisations d'une dalle (WebP lossless unis).
Le suffixe de résolution n'est que sur le DOSSIER : les fichiers gardent
le basename nu (cf. _expected_output_path dans le pipeline).
"""
from PIL import Image
base = f"LHD_FXX_{col:04d}_{row:04d}_PTS_LAMB93_IGN69"
suffix = "" if resolution == 0.5 else "_r" + str(resolution).replace(".", "p")
vis = tmp_path / "visualisations" / (base + suffix)
vis.mkdir(parents=True, exist_ok=True)
for key in viz_keys:
Image.new("RGB", (tile_px, tile_px), color).save(
str(vis / f"{base}_{key}.webp"), format="WEBP", lossless=True)
def _grid2x2(tmp_path):
"""Quatre dalles adjacentes de couleurs distinctes (row 20 au nord)."""
_make_tile_dir(tmp_path, 10, 20, ["slope"], color=(255, 0, 0)) # nord-ouest
_make_tile_dir(tmp_path, 11, 20, ["slope"], color=(0, 255, 0)) # nord-est
_make_tile_dir(tmp_path, 10, 19, ["slope"], color=(0, 0, 255)) # sud-ouest
_make_tile_dir(tmp_path, 11, 19, ["slope"], color=(255, 255, 0)) # sud-est
def test_export_places_tiles_and_is_seamless(tmp_path):
"""Grille 2×2 : placement exact (row haute au nord) et collage bord à bord.
Aucun trait ni espace entre les dalles : les pixels de part et d'autre de
chaque couture appartiennent aux dalles voisines, jamais au fond blanc.
"""
from PIL import Image
from lidar_pipeline.export import build_export
_grid2x2(tmp_path)
out = tmp_path / "exports"
res = build_export(tmp_path / "visualisations",
[(10, 19), (11, 19), (10, 20), (11, 20)],
["slope"], 0.5, "png", out)
with Image.open(str(res["file"])) as img:
# 2 dalles de 16 px + habillage → largeur = 32 exactement (le bandeau
# légende, comme bandeau/pied, n'élargit pas l'image)
assert img.width == 32
# Haut de la mosaïque : premier pixel rouge pur (dalles au-dessus des
# bandeaux texte blancs, insensible à la hauteur du bandeau légende)
def rgb(x, y):
return img.getpixel((x, y))
mosaic_top = next(y for y in range(img.height)
if rgb(4, y) == (255, 0, 0))
mid = mosaic_top + 16
# Nord : row 20 en haut (rouge à l'ouest, vert à l'est)
assert rgb(4, mosaic_top + 4) == (255, 0, 0)
assert rgb(28, mosaic_top + 4) == (0, 255, 0)
# Sud : row 19 en bas (bleu à l'ouest, jaune à l'est)
assert rgb(4, mosaic_top + 28) == (0, 0, 255)
assert rgb(28, mosaic_top + 28) == (255, 255, 0)
# Couture verticale (x=15|16) : couleurs des dalles, pas de blanc
assert rgb(15, mosaic_top + 8) == (255, 0, 0)
assert rgb(16, mosaic_top + 8) == (0, 255, 0)
assert rgb(15, mid) == (0, 0, 255)
assert rgb(16, mid) == (255, 255, 0)
# Couture horizontale (y=+15|+16) : couleurs des dalles, pas de blanc
assert rgb(4, mosaic_top + 15) == (255, 0, 0)
assert rgb(4, mosaic_top + 16) == (0, 0, 255)
def test_export_pdf_one_page_per_viz(tmp_path):
"""PDF multi-couches : une page par visualisation demandée."""
from lidar_pipeline.export import build_export
_make_tile_dir(tmp_path, 10, 20, ["slope", "aspect"])
res = build_export(tmp_path / "visualisations", [(10, 20)],
["slope", "aspect"], 0.5, "pdf", tmp_path / "exports")
assert res["file"].suffix == ".pdf"
data = res["file"].read_bytes()
assert data.startswith(b"%PDF")
pages = data.count(b"/Type /Page") - data.count(b"/Type /Pages")
assert pages == 2
def test_export_image_one_file_per_viz(tmp_path):
"""Image + plusieurs couches : un fichier par couche, chacun avec SA
légende (comme le PDF fait une page par couche)."""
from lidar_pipeline.export import build_export
_make_tile_dir(tmp_path, 10, 20, ["slope", "aspect"])
res = build_export(tmp_path / "visualisations", [(10, 20)],
["slope", "aspect"], 0.5, "png", tmp_path / "exports")
files = res["files"]
assert len(files) == 2
names = [e["file"].name for e in files]
assert any("_slope_" in n for n in names)
assert any("_aspect_" in n for n in names)
for e in files:
assert e["file"].is_file() and e["file"].suffix == ".png"
assert e["pages"] == 1
# Clés de compatibilité : premier fichier
assert res["file"] == files[0]["file"]
assert res["width"] == files[0]["width"] and res["height"] == files[0]["height"]
def test_export_missing_tile_or_viz(tmp_path):
"""Dalle absente à la résolution demandée ou visualisation manquante."""
from lidar_pipeline.export import build_export
_make_tile_dir(tmp_path, 10, 20, ["slope"])
try:
build_export(tmp_path / "visualisations", [(10, 20), (12, 20)],
["slope"], 0.5, "png", tmp_path / "exports")
assert False, "une ValueError était attendue"
except ValueError as e:
assert "12,20" in str(e)
try:
build_export(tmp_path / "visualisations", [(10, 20)],
["wavelet"], 0.5, "png", tmp_path / "exports")
assert False, "une ValueError était attendue"
except ValueError as e:
assert "wavelet" in str(e)
def test_export_includes_layer_legend(tmp_path):
"""Chaque export porte la légende de sa couche : dégradé de la colormap
(extrémités exactes) dans un bandeau sous le pied, en plus de l'habillage."""
from PIL import Image
from lidar_pipeline.export import build_export
_make_tile_dir(tmp_path, 10, 20, ["slope"], tile_px=256)
res = build_export(tmp_path / "visualisations", [(10, 20)],
["slope"], 0.5, "png", tmp_path / "exports")
with Image.open(str(res["file"])) as img:
# Bandeau légende présent : hauteur au-delà de mosaïque + bandeau + pied
assert img.height > 256 + 90 + 84
zone = [c for c in (img.getpixel((x, y))
for y in range(256 + 90 + 84, img.height)
for x in range(img.width))
if isinstance(c, tuple)]
# Extrémités du dégradé inferno de la pente (0° → 30°) : #000004 → #fcffa4
assert any(abs(c[0]) <= 1 and abs(c[1]) <= 1 and abs(c[2] - 4) <= 1
for c in zone)
assert any(abs(c[0] - 252) <= 1 and abs(c[1] - 255) <= 1
and abs(c[2] - 164) <= 1 for c in zone)
def test_export_legend_per_pdf_page():
"""PDF multi-couches : chaque page porte la légende de SA couche."""
from lidar_pipeline.export import _legend_layout
from lidar_pipeline.index import VIZ_LEGENDS
from PIL import Image
layout_slope = _legend_layout(VIZ_LEGENDS["slope"], Image.new("RGB", (600, 600)), 600)
layout_aspect = _legend_layout(VIZ_LEGENDS["aspect"], Image.new("RGB", (600, 600)), 600)
# Bandeaux distincts (dégradés/textes différents) et dimensionnés
assert layout_slope["height"] > 0 and layout_aspect["height"] > 0
assert layout_slope["has_bar"] and layout_aspect["has_bar"]
def test_viz_legends_registry_complete():
"""VIZ_LEGENDS couvre toutes les couches proposées (VIZ_LABELS)."""
from lidar_pipeline.index import VIZ_LABELS, VIZ_LEGENDS
assert set(VIZ_LABELS) == set(VIZ_LEGENDS)
for key, info in VIZ_LEGENDS.items():
assert info["title"], key
assert info["legend"], key
assert info["description"], key
if info.get("gradient") is not None:
assert len(info["gradient"]) >= 2, key
assert info["ticks"] and len(info["ticks"]) == 2, key
else: # couches photo/carte : pas de dégradé, pas de bornes
assert info["ticks"] is None, key
def test_viz_legends_gradients_match_colormaps():
"""Les dégradés de VIZ_LEGENDS restent calés sur les colormaps réelles
(un cmap changé dans rendering.py doit rafraîchir le dégradé)."""
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
from lidar_pipeline.index import VIZ_LEGENDS
from lidar_pipeline.rendering import COLORMAPS, RGB_LEGENDS
for key, info in VIZ_LEGENDS.items():
cmap_name = info.get("cmap")
if not cmap_name:
continue
ref = COLORMAPS.get(key) or RGB_LEGENDS.get(key)
assert ref is not None, key
assert ref["cmap"] == cmap_name, key
cm = plt.get_cmap(cmap_name)
stops = info["gradient"]
for i, hx in enumerate(stops):
r, g, b = (int(hx[j:j + 2], 16) for j in (1, 3, 5))
mr, mg, mb = (round(v * 255) for v in cm(i / (len(stops) - 1))[:3])
assert abs(r - mr) <= 1 and abs(g - mg) <= 1 and abs(b - mb) <= 1, key
def test_export_max_side_downsizes(tmp_path):
"""max_side réduit le plus grand côté de la mosaïque."""
from PIL import Image
from lidar_pipeline.export import build_export
_make_tile_dir(tmp_path, 10, 20, ["slope"], tile_px=64)
_make_tile_dir(tmp_path, 11, 20, ["slope"], tile_px=64)
res = build_export(tmp_path / "visualisations", [(10, 20), (11, 20)],
["slope"], 0.5, "png", tmp_path / "exports",
max_side=64)
with Image.open(str(res["file"])) as img:
assert img.width == 64 # mosaïque 128 px ramenée à 64
def test_export_endpoint(tmp_path, monkeypatch):
"""/api/export assemble et renvoie un fichier téléchargeable."""
import lidar_pipeline.webapp as webapp
_grid2x2(tmp_path)
monkeypatch.setattr(webapp, "OUTPUT_DIR", tmp_path)
req = webapp.ExportRequest(tiles=[[10, 20], [11, 20], [10, 19], [11, 19]],
viz=["slope"], resolution=0.5,
format="jpeg", max_side=0)
d = webapp.export_tiles(req)
files = d["fichiers"]
assert len(files) == 1
f = files[0]
assert f["url"].startswith("/api/export/file/")
assert f["nom"].endswith(".jpg")
assert (tmp_path / "exports" / f["nom"]).is_file()
assert f["taille"] > 0
def test_export_endpoint_multi_viz_image(tmp_path, monkeypatch):
"""/api/export en image avec 2 couches → 2 fichiers téléchargeables."""
import lidar_pipeline.webapp as webapp
_make_tile_dir(tmp_path, 10, 20, ["slope", "aspect"])
monkeypatch.setattr(webapp, "OUTPUT_DIR", tmp_path)
req = webapp.ExportRequest(tiles=[[10, 20]], viz=["slope", "aspect"],
resolution=0.5, format="jpeg", max_side=0)
d = webapp.export_tiles(req)
assert len(d["fichiers"]) == 2
for f in d["fichiers"]:
assert f["url"].startswith("/api/export/file/")
assert (tmp_path / "exports" / f["nom"]).is_file()
def test_export_endpoint_rejects_bad_requests(tmp_path, monkeypatch):
"""/api/export valide format, visualisations et tuiles."""
from fastapi import HTTPException
import lidar_pipeline.webapp as webapp
_grid2x2(tmp_path)
monkeypatch.setattr(webapp, "OUTPUT_DIR", tmp_path)
def expect_400(req):
try:
webapp.export_tiles(req)
assert False, "une HTTPException était attendue"
except HTTPException as e:
assert e.status_code == 400
expect_400(webapp.ExportRequest(tiles=[[10, 20]], viz=["slope"],
resolution=0.5, format="tiff", max_side=0))
expect_400(webapp.ExportRequest(tiles=[[10, 20]], viz=[],
resolution=0.5, format="png", max_side=0))
# Dalle inexistante à cette résolution
expect_400(webapp.ExportRequest(tiles=[[10, 20]], viz=["slope"],
resolution=0.2, format="png", max_side=0))
def test_export_file_route(tmp_path, monkeypatch):
"""Le fichier exporté est servi en pièce jointe ; traversée refusée."""
from fastapi import HTTPException
from fastapi.responses import FileResponse
import lidar_pipeline.webapp as webapp
exports = tmp_path / "exports"
exports.mkdir(parents=True)
(exports / "ok.png").write_bytes(b"x")
monkeypatch.setattr(webapp, "OUTPUT_DIR", tmp_path)
resp = webapp.export_file("ok.png")
assert isinstance(resp, FileResponse)
for bad in ("../visualisations/secret", "nope.png", "a/b.png"):
try:
webapp.export_file(bad)
assert False, f"une HTTPException était attendue pour {bad}"
except HTTPException as e:
assert e.status_code == 404

View File

@ -207,8 +207,8 @@ def test_collect_tile_metadata_fallback_date(tmp_path):
assert meta['generated'] is not None
def test_build_index_generates_html(tmp_path):
"""build_index génère index.html et les vignettes."""
def test_build_index_generates_inventory(tmp_path):
"""build_index génère l'inventaire index_tiles.json et les vignettes."""
from lidar_pipeline.index import build_index
output_dir = tmp_path / "output"
@ -221,105 +221,20 @@ def test_build_index_generates_html(tmp_path):
result = build_index(output_dir)
assert result is not None
html_path = Path(result)
assert html_path.exists()
assert html_path.name == "index.html"
inv_path = Path(result)
assert inv_path.exists()
assert inv_path.name == "index_tiles.json"
# L'ancienne interface HTML a été retirée : plus de coquille ni d'assets
assert not (output_dir / "index.html").exists()
assert not (output_dir / "assets").exists()
content = html_path.read_text(encoding='utf-8')
# Vérifie la présence des éléments clés
assert "Carte LiDAR" in content
assert "LHD_FXX_1000_6881" in content
assert "LHD_FXX_1001_6881" in content
# Vérifie que le JSON intégré est valide
assert "const TILES" in content
# Vérifie les assets de l'interface (CSS/JS séparés)
assets = output_dir / "assets"
assert (assets / "app.css").read_text(encoding='utf-8').startswith('/*')
app_js = (assets / "app.js").read_text(encoding='utf-8')
assert "Couches" in app_js or "layers" in app_js
# Barre de génération : sélecteur restreint au panneau (PANEL_VIZ =
# DEFAULT_LAYERS ; valeurs = noms d'étapes --only). Les couches hors
# panneau (svf, hillshade, wavelet, roughness) ne sont plus proposées.
assert 'id="genViz"' in content
assert 'value="slope"' in content
assert 'value="pos_open"' in content
assert 'value="aspect"' in content
assert 'value="roughness"' not in content
assert 'value="wavelet"' not in content
assert 'value="svf"' not in content
assert 'value="hillshade"' not in content
assert 'value="aniso_open"' not in content
assert "genViz" in app_js and "viz: selectedGenViz()" in app_js
assert app_js.count("selectedGenViz()") >= 3 # définition + preview + generate
# Adoption du dernier run au chargement (redémarrage du serveur) :
# un run terminé réapparaît dans la file avec ses tuiles, sans doublon.
assert "Dernière génération" in app_js
assert "t.started === s.started" in app_js
# Vidage de la file : les runs retirés sont mémorisés (par horodatage
# serveur) pour que le rechargement ne réadopte pas la « dernière
# génération » du serveur et ne refasse pas réapparaître les tâches finies.
assert "dismissedRuns" in app_js
assert "saveDismissed()" in app_js
# Bouton « Compléter » : visualisations manquantes des dalles présentes
# dans input/ (scope all_missing envoyé au preview et au generate).
# « + Zone » et « Compléter » ne vivent que dans l'onglet Génération,
# jamais dans la barre d'outils flottante de la carte.
assert 'id="btnComplete"' in content
map_tools = content.split('id="mapTools"', 1)[1].split('</nav>', 1)[0]
assert 'btnGenZone' not in map_tools and 'btnComplete' not in map_tools
tab_gen = content.split('id="tab-gen"', 1)[1].split('id="tab-tile"', 1)[0]
assert 'id="btnGenZone"' in tab_gen and 'id="btnComplete"' in tab_gen
assert "all_missing" in app_js and app_js.count("all_missing") >= 2
# Dessin de zone au 1er plan : pane dédié au-dessus des tuiles (marqueurs
# Leaflet, pane 600) pour rectangle et cellules fantômes.
assert "createPane('genDraw')" in app_js
assert app_js.count("pane: 'genDraw'") >= 3 # rectangle + 2 jeux de cellules
assert 'assets/app.css' in content
# Bouton d'actualisation des couches (détection des nouveautés sur disque) :
# le ↻ resynchronise puis reconstruit la carte (mode webapp distante).
assert 'id="layerRefresh"' in content
assert "/api/layers" in app_js and "/api/sync" in app_js
# Préconfigurations de couches (onglet Couches) : le sélecteur applique
# des presets intégrés (couches visibles, niveaux de transparence —
# visualisations + fond de carte OSM —, ordre de pile) et 💾/✕ gèrent
# les presets personnels persistés en localStorage.
assert 'id="presetSelect"' in content
assert 'id="presetSave"' in content
assert "BUILTIN_PRESETS" in app_js and "applyPreset" in app_js
assert "presetDetail" in app_js and "lidarPresets_v1" in app_js
assert "osm: { on: true, opacity: 0.3, dark: true }" in app_js # preset porte le fond OSM
assert "p.osm" in app_js # l'application y inclut le fond
# Partage de la vue : bouton « Partager » + lien contenant vue, couches
# et transparences (hash #zoom,lat,lng&L=...&O=...) — le destinataire
# reçoit l'état exact, indépendamment de son localStorage local.
assert 'id="btnShare"' in content
assert "buildShareHash" in app_js and "parseSharedHash" in app_js
assert 'assets/app.js' in content
# Autonomie hors ligne : Leaflet vendorisé localement, aucun CDN
assert 'assets/vendor/leaflet/leaflet.js' in content
assert 'assets/vendor/leaflet/leaflet.css' in content
assert 'unpkg.com' not in content
assert (assets / 'vendor' / 'leaflet' / 'leaflet.js').is_file()
# Fin d'un run de génération : synchronisation puis rechargement (le
# rechargement direct reste en secours si /api/sync est absent).
assert "syncThenReload" in app_js
# Tuiles en direct : la carte sonde /api/tiles pendant un run et fusionne
# les tuiles terminées sans recharger (mode incrémental du pipeline).
assert "/api/tiles" in app_js and "pollLiveTiles" in app_js
assert "registerTile" in app_js and "mergeTilesData" in app_js
# Garde anti-page périmée : un onglet ouvert depuis plus de 15 min
# (JS ancien après un déploiement) se recharge au retour au premier
# plan — l'ancienne interface n'est jamais affichée.
assert "reloadStalePage" in app_js and "visibilitychange" in app_js
# Export des données de carte (servi par /api/tiles)
data = json.loads((output_dir / "index_tiles.json").read_text(encoding='utf-8'))
# Inventaire servi par /api/tiles aux machines légères : tuiles avec coins,
# libellés de couches, compteur.
data = json.loads(inv_path.read_text(encoding='utf-8'))
assert data['tiles'] and data['tiles'][0]['corners']
assert 'aspect' in data['viz_meta']
assert data['viz_meta']['aspect']['label']
assert data['stats']['n_tiles'] == len(data['tiles'])
# Version de l'interface exportée : la webapp légère détecte une image
# périmée (assets bâchés au build, non reconstruite après évolution)
from lidar_pipeline.index import ui_version
assert data['stats']['ui_version'] == ui_version()
# Vérifie les vignettes générées
thumb_dir = output_dir / "index_thumbs"
assert thumb_dir.is_dir()
@ -327,27 +242,6 @@ def test_build_index_generates_html(tmp_path):
assert len(thumbs) >= 2 # au moins hillshade pour chaque tuile
def test_write_ui_assets(tmp_path):
"""_write_ui_assets écrit app.js/app.css dans le dossier cible (bâchés image).
app.js a __UI_VERSION__ substitué par la version réelle (ui_version) :
le JS compare cette version à celle du serveur de données et signale
une webapp périmée (bandeau, cf. checkUiVersion).
"""
from lidar_pipeline.index import _write_ui_assets, _APP_JS, ui_version
dst = tmp_path / "ui"
_write_ui_assets(dst)
v = ui_version()
assert v == ui_version() and len(v) == 10
js = (dst / "app.js").read_text(encoding="utf-8")
assert js == _APP_JS.replace("__UI_VERSION__", v)
assert v in js and "__UI_VERSION__" not in js
# Le placeholder et le check sont embarqués dans la source
assert '"__UI_VERSION__"' in _APP_JS and "checkUiVersion" in _APP_JS
assert (dst / "app.css").read_text(encoding="utf-8").startswith("/*")
def test_mid_thumbnails_generated(tmp_path):
"""Vignette intermédiaire 640 px générée et référencée dans l'index."""
from lidar_pipeline.index import build_index
@ -410,25 +304,6 @@ def test_panel_restricted_to_kept_layers():
assert KEYWORD_TO_STEP.get(key, key) in steps
def test_panel_selection_lists_kept_layers():
"""Le panneau propose uniquement les couches conservées (PANEL_VIZ).
Même sans données sur disque, le panneau affiche les couches conservées ;
une couche absente du panneau n'est plus proposée, même présente sur
disque.
"""
from lidar_pipeline.index import _panel_selection, PANEL_VIZ
ordered, default = _panel_selection({'aspect'})
assert set(ordered) == set(PANEL_VIZ)
assert default == 'aspect'
ordered, default = _panel_selection(set())
assert set(ordered) == set(PANEL_VIZ)
assert default # toujours une couche par défaut
# Une couche présente sur disque mais hors panneau est exclue
ordered, default = _panel_selection({'aniso_open'})
assert 'aniso_open' not in ordered
def test_build_index_merges_cross_resolution_viz(tmp_path):
"""Une couche produite seulement à 0,5 m reste visible sur la dalle 0,2 m.
@ -602,35 +477,6 @@ def test_build_index_empty_returns_none(tmp_path):
assert result is None
def test_build_index_embeds_valid_json(tmp_path):
"""Le JSON embarqué dans le HTML est valide et contient les tuiles."""
from lidar_pipeline.index import build_index
output_dir = tmp_path / "output"
vis_dir = output_dir / "visualisations"
vis_dir.mkdir(parents=True)
_make_fake_viz_dir(vis_dir, "a", 1000, 6881, ('hillshade_multi',))
build_index(output_dir)
content = (output_dir / "index.html").read_text(encoding='utf-8')
# Extrait le JSON entre "const TILES = " et la fin de déclaration
start = content.index("const TILES = ") + len("const TILES = ")
# Trouve le ; de fin de déclaration
depth = 0
end = start
for i, ch in enumerate(content[start:], start):
if ch in ('{', '['):
depth += 1
elif ch in ('}', ']'):
depth -= 1
if depth == 0:
end = i + 1
break
data = json.loads(content[start:end])
assert len(data) > 0
assert data[0]['col'] == 1000
assert data[0]['row'] == 6881
def test_attach_gps_bounds():
"""attach_gps_bounds ajoute des bounds GPS ordonnées (France métropolitaine)."""
@ -718,74 +564,7 @@ def test_approx_wgs84_to_l93_roundtrip():
x2, y2 = _approx_wgs84_to_l93(lon, lat)
assert abs(x2 - x) < 1e-6 and abs(y2 - y) < 1e-6
def test_overview_generated_for_large_grids(tmp_path):
"""Grille >= 6 dalles : une mosaïque d'overview par couche + le payload."""
from lidar_pipeline.index import build_index
output_dir = tmp_path / "output"
vis_dir = output_dir / "visualisations"
vis_dir.mkdir(parents=True)
for col in (1000, 1001, 1002):
for row in (6881, 6882):
_make_fake_viz_dir(vis_dir, "t", col, row, ('hillshade_multi', 'svf'))
assert build_index(output_dir) is not None
data = json.loads((output_dir / "index_tiles.json").read_text(encoding='utf-8'))
ov = data['overview']
assert 'hillshade_multi' in ov and 'svf' in ov
items = ov['hillshade_multi']['items']
assert len(items) == 1 # une seule zone géographique
meta = items[0]
assert (output_dir / meta['url'].split('?')[0]).is_file()
assert meta['km_w'] == 3 and meta['km_h'] == 2
assert meta['w'] == 3 * meta['px_per_km']
assert meta['h'] == 2 * meta['px_per_km']
assert len(meta['corners']) == 4
# La coquille HTML embarque la constante pour l'interface
html = (output_dir / "index.html").read_text(encoding='utf-8')
assert "const OVERVIEW" in html
assert "index_overview/hillshade_multi_" in html
def test_overview_one_mosaic_per_zone(tmp_path):
"""Zones géographiques distantes : une mosaïque PAR ZONE (pas une grille
globale étirée sur des centaines de km aux coins introuvables)."""
from lidar_pipeline.index import build_index
output_dir = tmp_path / "output"
vis_dir = output_dir / "visualisations"
vis_dir.mkdir(parents=True)
# Deux clusters de 2×3 dalles, à ~400 km l'un de l'autre
for col in (1000, 1001, 1002):
for row in (6881, 6882):
_make_fake_viz_dir(vis_dir, "a", col, row, ('hillshade_multi',))
for col in (1400, 1401, 1402):
for row in (6481, 6482):
_make_fake_viz_dir(vis_dir, "b", col, row, ('hillshade_multi',))
assert build_index(output_dir) is not None
data = json.loads((output_dir / "index_tiles.json").read_text(encoding='utf-8'))
items = data['overview']['hillshade_multi']['items']
assert len(items) == 2
assert items[0]['km_w'] == 3 and items[0]['km_h'] == 2
assert items[1]['km_w'] == 3 and items[1]['km_h'] == 2
# Les deux mosaïques sont distinctes (coins différents)
assert items[0]['corners'] != items[1]['corners']
def test_overview_skipped_for_small_grids(tmp_path):
"""Grille < 6 dalles : pas d'overview, les vignettes suffisent."""
from lidar_pipeline.index import build_index
output_dir = tmp_path / "output"
vis_dir = output_dir / "visualisations"
vis_dir.mkdir(parents=True)
_make_fake_viz_dir(vis_dir, "a", 1000, 6881, ('hillshade_multi',))
_make_fake_viz_dir(vis_dir, "b", 1001, 6881, ('hillshade_multi',))
assert build_index(output_dir) is not None
data = json.loads((output_dir / "index_tiles.json").read_text(encoding='utf-8'))
assert data['overview'] == {}
def test_subtile_corners_shared_edges_exact():

View File

@ -258,7 +258,7 @@ def test_ui_offers_reorder_and_blend_modes():
# Pile isolée + fusion par couche et fusion de la pile sur le fond
assert "createPane('lidarStack')" in _MAP_JS
assert "isolation = 'isolate'" in _MAP_JS
assert "map.createPane(pane, stack)" in _MAP_JS
assert "map.createPane(paneName, stack)" in _MAP_JS
assert "pane.style.mixBlendMode" in _MAP_JS
assert "STATE.stackBlend" in _MAP_JS and 'id="stackBlend"' in _MAP_HTML
# Le lien partagé transporte la fusion (4e champ, rétrocompatible)
@ -595,3 +595,231 @@ def test_background_scan_detects_and_fills(tmp_path, monkeypatch):
assert mapserve.background_scan(tmp_path) > 0
finally:
_reset_bg(mapserve)
# ---------------------------------------------------------------------------
# Génération de tuiles (portée de la webapp historique)
# ---------------------------------------------------------------------------
import types
from fastapi import HTTPException
import pytest
class _ClientRequest:
"""Requête minimale avec IP de connexion (restriction LIDAR_REGEN_CIDR)."""
def __init__(self, ip, xff=None):
self.client = types.SimpleNamespace(host=ip)
self.headers = {"x-forwarded-for": xff} if xff else {}
def _gen_setup(tmp_path, monkeypatch):
import lidar_pipeline.mapserve as mapserve
_make_dalle(tmp_path, 1054, 6882, ["aspect", "slope"])
monkeypatch.setattr(mapserve, "OUTPUT_DIR", tmp_path)
monkeypatch.setattr(mapserve, "INPUT_DIR", tmp_path / "input")
monkeypatch.setattr(mapserve, "GENERATION_URL", "")
monkeypatch.setattr(mapserve, "LOG_FILE", tmp_path / ".generation.log")
monkeypatch.setattr(mapserve, "JOB_FILE", tmp_path / ".generation.job.json")
monkeypatch.setattr(mapserve, "QUEUE_FILE", tmp_path / ".generation.queue.json")
mapserve._queue.clear()
mapserve._pipeline_import_cache.clear()
return mapserve
def test_bbox_to_cells_covers_1km():
"""Une bbox d'1 km² WGS84 sélectionne la cellule L93 correspondante."""
import lidar_pipeline.mapserve as mapserve
from lidar_pipeline.tiles import _transformer
x, y = 1054500, 6881500
w, n = _transformer("EPSG:2154", "EPSG:4326").transform(x - 200, y + 200)
e, s = _transformer("EPSG:2154", "EPSG:4326").transform(x + 200, y - 200)
cells = mapserve.bbox_to_cells(w, s, e, n)
assert (1054, 6882) in cells
# Cellule (col, row) : X ∈ [col, col+1] km, Y ∈ [row-1, row] km
assert all(1053 <= c <= 1055 and 6881 <= r <= 6883 for c, r in cells)
def test_preview_bbox_lists_cells_with_corners(tmp_path, monkeypatch):
"""Le preview retourne les cellules à traiter avec leurs coins GPS."""
mapserve = _gen_setup(tmp_path, monkeypatch)
from lidar_pipeline.tiles import _transformer
x, y = 1054500, 6881500
w, n = _transformer("EPSG:2154", "EPSG:4326").transform(x - 100, y + 100)
e, s = _transformer("EPSG:2154", "EPSG:4326").transform(x + 100, y - 100)
req = mapserve.PreviewRequest(bbox=[w, s, e, n], viz=["aspect"])
d = mapserve.preview(req)
assert d["capped"] is False
cells = {(c["col"], c["row"]) for c in d["cells"]}
assert (1054, 6882) in cells
first = d["cells"][0]
assert first["corners"] and len(first["corners"]) == 4
def test_preview_rejects_unknown_viz(tmp_path, monkeypatch):
mapserve = _gen_setup(tmp_path, monkeypatch)
req = mapserve.PreviewRequest(bbox=[0.0, 44.0, 0.01, 44.01],
viz=["inconnue"])
with pytest.raises(HTTPException) as e:
mapserve.preview(req)
assert e.value.status_code == 400
def test_preview_all_missing_without_laz(tmp_path, monkeypatch):
"""Passe globale sans dalle dans input/ : liste vide (rien à compléter).
L'erreur 400 « aucune tuile à compléter » ne concerne que le lancement
(/api/generate → _resolve_request) : le preview reste informatif.
"""
mapserve = _gen_setup(tmp_path, monkeypatch)
(tmp_path / "input").mkdir()
req = mapserve.PreviewRequest(all_missing=True, viz=["aspect"])
d = mapserve.preview(req)
assert d["count"] == 0 and d["cells"] == []
def test_generate_validates_request(tmp_path, monkeypatch):
mapserve = _gen_setup(tmp_path, monkeypatch)
for kwargs in ({"tiles": []},
{"tiles": [[1054]]},
{"tiles": [[1054, 6882]], "ground_class": "inconnue"},
{"tiles": [[c, 6882] for c in range(500)]}):
req = mapserve.GenerateRequest(**kwargs)
with pytest.raises(HTTPException) as e:
mapserve._resolve_request(req)
assert e.value.status_code == 400
# Tri nord → sud (row décroissante) puis col croissante
req = mapserve.GenerateRequest(tiles=[[1055, 6882], [1054, 6883],
[1054, 6882]])
tiles, _viz = mapserve._resolve_request(req)
assert tiles == [(1054, 6883), (1054, 6882), (1055, 6882)]
def test_generate_launches_and_queues(tmp_path, monkeypatch):
mapserve = _gen_setup(tmp_path, monkeypatch)
launched = []
monkeypatch.setattr(mapserve, "_launch_job",
lambda tiles, viz, req, qid=None:
launched.append((tiles, qid)) or ["fake", "cmd"])
req = mapserve.GenerateRequest(tiles=[[1054, 6882]], viz=["aspect"])
d = mapserve.generate(req)
assert d["demarré"] is True and d["tuiles"] == 1
# Run en cours : la demande part en file (jamais de coupure du travail)
import threading
with mapserve._job_lock:
mapserve._job["proc"] = types.SimpleNamespace(
poll=lambda: None, pid=4242, terminate=lambda: None, wait=lambda: None)
d = mapserve.generate(req)
assert d["demarré"] is False and d["en_file"] == 1
assert mapserve.generate(req)["en_file"] == 2
assert len(mapserve._queue_summary()) == 2
d = mapserve.queue_clear()
assert d["retirées"] == 2
with mapserve._job_lock:
mapserve._job["proc"] = None
mapserve._queue.clear()
def test_status_reports_rights_and_availability(tmp_path, monkeypatch):
"""/api/status annonce available (pipeline local) et regen_allowed (IP)."""
mapserve = _gen_setup(tmp_path, monkeypatch)
monkeypatch.setattr(mapserve, "_pipeline_import_cache", {"ok": True})
d = mapserve.generation_status(_ClientRequest("192.168.3.10"))
assert d["available"] is True and d["regen_allowed"] is True
assert d["running"] is False and d["tiles"] == []
# Client hors réseau local : génération interdite (boutons masqués)
d = mapserve.generation_status(_ClientRequest("203.0.113.7"))
assert d["regen_allowed"] is False
# X-Forwarded-For honoré seulement depuis un pair de confiance
d = mapserve.generation_status(
_ClientRequest("172.25.0.1", xff="203.0.113.7"))
assert d["regen_allowed"] is False
d = mapserve.generation_status(
_ClientRequest("172.25.0.1", xff="192.168.3.10"))
assert d["regen_allowed"] is True
# Image légère sans pipeline ni LIDAR_GENERATION_URL : indisponible
monkeypatch.setattr(mapserve, "_pipeline_import_cache", {"ok": False})
d = mapserve.generation_status(_ClientRequest("192.168.3.10"))
assert d["available"] is False
def test_status_proxies_to_worker(tmp_path, monkeypatch):
"""Webapp légère : /api/status relaie l'état du worker, 503 hors ligne."""
mapserve = _gen_setup(tmp_path, monkeypatch)
monkeypatch.setattr(mapserve, "GENERATION_URL", "http://worker:8973")
monkeypatch.setattr(mapserve, "_pipeline_import_cache", {"ok": False})
def fake_proxy(method, path, payload=None, timeout=30):
assert (method, path) == ("GET", "/api/status")
return {"running": True, "tiles": []}
monkeypatch.setattr(mapserve, "_proxy_generation", fake_proxy)
d = mapserve.generation_status(_ClientRequest("192.168.3.10"))
assert d["distant"] is True and d["running"] is True
assert d["available"] is True and d["regen_allowed"] is True
def offline(method, path, payload=None, timeout=30):
raise HTTPException(503, "machine de traitement injoignable")
monkeypatch.setattr(mapserve, "_proxy_generation", offline)
d = mapserve.generation_status(_ClientRequest("192.168.3.10"))
assert d["worker_offline"] is True and d["running"] is None
def test_generate_requires_lan(tmp_path, monkeypatch):
"""Un client hors LIDAR_REGEN_CIDR ne peut pas lancer de génération."""
mapserve = _gen_setup(tmp_path, monkeypatch)
with pytest.raises(HTTPException) as e:
mapserve._require_lan_for_generation(_ClientRequest("203.0.113.7"))
assert e.value.status_code == 403
def test_cell_at_point(tmp_path, monkeypatch):
"""/api/cell donne la dalle sous un point (sélection manuelle)."""
mapserve = _gen_setup(tmp_path, monkeypatch)
from lidar_pipeline.tiles import _transformer
lon, lat = _transformer("EPSG:2154", "EPSG:4326").transform(1054500, 6881500)
d = mapserve.cell_at_point(lat=lat, lng=lon)
assert (d["col"], d["row"]) == (1054, 6882)
assert d["corners"] and len(d["corners"]) == 4
def test_tiles_inventory_endpoint(tmp_path, monkeypatch):
"""/api/tiles sert index_tiles.json (machines légères LIDAR_SOURCE_URL)."""
mapserve = _gen_setup(tmp_path, monkeypatch)
payload = {"tiles": [{"col": 1054, "row": 6882,
"viz": {"aspect": {"full": "visualisations/x.avif?v=1"}}}],
"viz_meta": {}, "stats": {"n_tiles": 1}}
(tmp_path / "index_tiles.json").write_text(
__import__("json").dumps(payload), encoding="utf-8")
d = mapserve.tiles_inventory()
assert d["tiles"] == payload["tiles"] and d["stamp"] > 0
# ?stamp inchangé → réponse allégée
light = mapserve.tiles_inventory(stamp=d["stamp"])
assert light["tiles"] is None
def test_ui_offers_generation_controls():
"""L'interface embarque les contrôles de génération (+ Zone, options, run)."""
from lidar_pipeline.mapui import _MAP_CSS, _MAP_HTML, _MAP_JS
assert 'id="btnZone"' in _MAP_HTML and 'id="btnComplete"' in _MAP_HTML
assert 'id="gencard"' in _MAP_HTML and 'id="genGo"' in _MAP_HTML
assert 'id="genStop"' in _MAP_HTML and 'id="genQueueClear"' in _MAP_HTML
assert 'id="genViz"' in _MAP_HTML and 'id="genEdge"' in _MAP_HTML
# Boutons masqués par défaut : révélés seulement si /api/status autorise
assert "GEN.available && GEN.allowed" in _MAP_JS
# Dessin de zone + cadres de run + sondage d'état
assert "genSetDrawing" in _MAP_JS and "pane: 'genDraw'" in _MAP_JS
assert "gen-frame-running" in _MAP_CSS and "genDrawFrames" in _MAP_JS
assert "api/status" in _MAP_JS and "api/preview" in _MAP_JS
assert "api/generate" in _MAP_JS and "api/stop" in _MAP_JS
assert "all_missing: true" in _MAP_JS
# Fin de run : rafraîchissement des couches + relance de la maintenance
assert "genAfterRun" in _MAP_JS and "api/map/background" in _MAP_JS

View File

@ -405,7 +405,7 @@ def test_png_palette_option_shrinks_tiles(tmp_path, monkeypatch):
# ---------------------------------------------------------------------------
def _remote_payload():
"""Charge utile /api/tiles telle que la sert la webapp du pipeline."""
"""Charge utile /api/tiles telle que la sert mapserve (serveur de dalles)."""
base = "LHD_FXX_1054_6882_PTS_LAMB93_IGN69_r0p2"
tiles = []
for i in range(2):

File diff suppressed because it is too large Load Diff