68272cabf0
- URLs images versionnées (?v=mtime) : une tuile recalculée change d'URL et force le rechargement navigateur (cache heuristique contourné), y compris en plein run via /api/tiles ; veille permanente 15 s sur la carte - simulation locale du mode deux machines (docker-compose.local-2m.yml) : worker GPU :8974 + webapp légère :8973 au cache séparé output-webapp/ - override webapp pour le Pi 5 (192.168.3.3) : volume /srv/lidar/output, rsync vers le worker, labels Traefik (proxy/websecure/myresolver) - purge 0,5 m : worker/process et politique générale passés à 0,2 m seul - intègre le travail parallèle non commité : export mosaïque multi-dalles (export.py + /api/export), sous-tuilage intégral des couches, légendes VIZ_LEGENDS, docs et tests associés (213 tests verts)
277 lines
12 KiB
Python
277 lines
12 KiB
Python
"""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
|