Add web map with zone generation API, side job queue, and restore historical DTM hole rendering

- webapp.py: FastAPI serving the continuous map (port 8973) with
  /api/preview, /api/generate and /api/status; tiles are downloaded
  from IGN and processed in a logged subprocess, tracked live in a
  side "File de génération" panel that survives page reloads
- fetch_ign.py: download missing 1 km LiDAR HD tiles from the IGN
  geoplateforme before processing
- index.py: tile thumbnails and 500 m subtiles are now invalidated by
  mtime so regenerating a tile refreshes its cached images; progress
  logging per tile
- dtm.py: back to the historical gap handling (small gaps filled by
  fillnodata only, larger holes left as nodata rendered black);
  lowest-return floor only via --bare-earth, IGN class selection via
  --ign-classes
- cli.py: positional input now optional (--rebuild-index works alone)
- docker-compose.yml: serve (GPU, port 8973) and process services;
  launch via docker compose only (documented in AGENTS.md/AGENTS.md)
- tests: 131 passing, incl. regressions for thumbnail staleness,
  --rebuild-index without input, and nodata rendering
This commit is contained in:
Antoine Jacquin
2026-08-31 18:07:14 +02:00
parent 35bd827790
commit 8ca65155db
19 changed files with 3676 additions and 771 deletions

View File

@ -70,4 +70,99 @@ class TestLidarArchaeoPipeline:
names = [f.name for f in files]
assert "test.laz" in names
assert "other.las" in names
assert "readme.txt" not in names
assert "readme.txt" not in names
class TestDtmMethodSidecar:
"""Méthode de classification enregistrée à côté du DTM (invalidation du cache)."""
def test_missing_sidecar_matches(self, tmp_path):
from lidar_pipeline.pipeline import LidarArchaeoPipeline
input_dir = tmp_path / "input"
input_dir.mkdir()
pipeline = LidarArchaeoPipeline(str(input_dir), str(tmp_path / "output"), ground_method='csf')
# Aucun sidecar écrit → cache conservé (considéré compatible).
assert pipeline._dtm_method_matches("tileA", "") is True
def test_matching_method(self, tmp_path):
from lidar_pipeline.pipeline import LidarArchaeoPipeline
input_dir = tmp_path / "input"
input_dir.mkdir()
pipeline = LidarArchaeoPipeline(str(input_dir), str(tmp_path / "output"), ground_method='csf')
pipeline._write_dtm_method("tileA", "")
assert pipeline._dtm_method_matches("tileA", "") is True
def test_different_method_invalidates_cache(self, tmp_path):
from lidar_pipeline.pipeline import LidarArchaeoPipeline
input_dir = tmp_path / "input"
input_dir.mkdir()
out = str(tmp_path / "output")
LidarArchaeoPipeline(str(input_dir), out, ground_method='ign')._write_dtm_method("tileA", "")
csf = LidarArchaeoPipeline(str(input_dir), out, ground_method='csf')
assert csf._dtm_method_matches("tileA", "") is False
def test_write_dtm_method(self, tmp_path):
from lidar_pipeline.pipeline import LidarArchaeoPipeline
input_dir = tmp_path / "input"
input_dir.mkdir()
pipeline = LidarArchaeoPipeline(str(input_dir), str(tmp_path / "output"), ground_method='smrf')
pipeline._write_dtm_method("tileA", "_r0p2")
sidecar = tmp_path / "output" / "DTM" / "tileA_dtm_r0p2_method.txt"
assert sidecar.exists()
assert sidecar.read_text(encoding="utf-8").strip() == "smrf"
assert pipeline._dtm_method_name("tileA", "_r0p2") == "smrf"
# Le sidecar est un fichier .txt : il ne gêne pas la recherche des DTM .tif.
dtm = tmp_path / "output" / "DTM" / "tileA_dtm_r0p2.tif"
dtm.touch()
assert [p.name for p in (tmp_path / "output" / "DTM").glob("*.tif")] == ["tileA_dtm_r0p2.tif"]
def test_force_images_regenerates_existing(self, tmp_path):
from lidar_pipeline.pipeline import LidarArchaeoPipeline
input_dir = tmp_path / "input"
input_dir.mkdir()
pipeline = LidarArchaeoPipeline(str(input_dir), str(tmp_path / "output"), output_format='avif')
calls = []
def fake_ortho(dem_file, basename, vis_dir, resolution):
calls.append(basename)
return vis_dir / f"{basename}_ortho.avif"
pipeline.viz_steps = [('ortho', fake_ortho)]
vis_dir = tmp_path / "output" / "visualisations" / "tileA"
vis_dir.mkdir(parents=True)
(vis_dir / "tileA_ortho.avif").touch()
dtm = tmp_path / "dtm.tif"
# Image existante, pas de force → ignorée (pas de régénération).
pipeline.generate_all_visualizations(dtm, "tileA", resolution=0.5, vis_dir=vis_dir, force_images=False)
assert calls == []
# Image existante, force_images=True → régénérée.
calls.clear()
pipeline.generate_all_visualizations(dtm, "tileA", resolution=0.5, vis_dir=vis_dir, force_images=True)
assert calls == ["tileA"]
class TestEffectiveGroundMethod:
def test_ign_label_encodes_classes(self):
"""Les classes IGN sont encodées dans l'étiquette de cache (reclassification)."""
from lidar_pipeline.pipeline import LidarArchaeoPipeline
import tempfile
with tempfile.TemporaryDirectory() as tmpdir:
p = LidarArchaeoPipeline(tmpdir, tmpdir, ground_method='ign',
ign_classes="sol,unclassified")
assert p._effective_ground_method() == "ign_1_2"
def test_ign_default_label(self):
from lidar_pipeline.pipeline import LidarArchaeoPipeline
import tempfile
with tempfile.TemporaryDirectory() as tmpdir:
p = LidarArchaeoPipeline(tmpdir, tmpdir, ground_method='ign')
assert p._effective_ground_method() == "ign"
def test_other_methods_unchanged(self):
from lidar_pipeline.pipeline import LidarArchaeoPipeline
import tempfile
with tempfile.TemporaryDirectory() as tmpdir:
p = LidarArchaeoPipeline(tmpdir, tmpdir, ground_method='smrf',
ign_classes="sol,unclassified")
assert p._effective_ground_method() == "smrf"