422f58d772
La webapp (carte + vignettes) et la génération de tuiles se déploient sur deux machines : image légère Dockerfile.webapp (FastAPI + Pillow AVIF natif + pyproj) sur Raspberry Pi, pipeline complet sur la machine de traitement. LIDAR_GENERATION_URL délègue /api/generate, /api/preview et /api/status ; /api/sync ramène les tuiles par rsync puis régénère vignettes et index localement. Token partagé optionnel (LIDAR_API_TOKEN/LIDAR_REMOTE_TOKEN). Retire du dépôt les journaux internes (.swival, audit-findings) et les données (data/, notebooks/). Doc : docs/DEPLOY_WEBAPP.md.
192 lines
8.3 KiB
Python
192 lines
8.3 KiB
Python
"""Tests for pipeline orchestration."""
|
|
|
|
import pytest
|
|
from pathlib import Path
|
|
|
|
|
|
class TestVizSteps:
|
|
def test_viz_steps_not_empty(self):
|
|
from lidar_pipeline.pipeline import VIZ_STEPS
|
|
assert len(VIZ_STEPS) > 0
|
|
|
|
def test_viz_steps_have_callable_functions(self):
|
|
from lidar_pipeline.pipeline import VIZ_STEPS
|
|
for name, func in VIZ_STEPS:
|
|
assert callable(func), f"VIZ_STEPS entry '{name}' is not callable"
|
|
|
|
def test_viz_steps_names_unique(self):
|
|
from lidar_pipeline.pipeline import VIZ_STEPS
|
|
names = [name for name, _ in VIZ_STEPS]
|
|
assert len(names) == len(set(names)), "VIZ_STEPS has duplicate names"
|
|
|
|
def test_expected_visualization_count(self):
|
|
"""Should have 15 visualizations (13 terrain + ortho + topo)."""
|
|
from lidar_pipeline.pipeline import VIZ_STEPS
|
|
assert len(VIZ_STEPS) == 15
|
|
|
|
def test_ortho_and_topo_present(self):
|
|
from lidar_pipeline.pipeline import VIZ_STEPS
|
|
names = [name for name, _ in VIZ_STEPS]
|
|
assert "ortho" in names
|
|
assert "topo" in names
|
|
|
|
|
|
class TestLidarArchaeoPipeline:
|
|
def test_init_creates_dirs(self, tmp_path):
|
|
from lidar_pipeline.pipeline import LidarArchaeoPipeline
|
|
input_dir = tmp_path / "input"
|
|
input_dir.mkdir()
|
|
output_dir = tmp_path / "output"
|
|
|
|
pipeline = LidarArchaeoPipeline(str(input_dir), str(output_dir))
|
|
assert (tmp_path / "output").exists()
|
|
assert (tmp_path / "output" / "DTM").exists()
|
|
assert (tmp_path / "output" / "visualisations").exists()
|
|
assert (tmp_path / "output" / "temp").exists()
|
|
|
|
def test_init_raises_on_missing_input(self, tmp_path):
|
|
from lidar_pipeline.pipeline import LidarArchaeoPipeline
|
|
with pytest.raises(ValueError, match="introuvable"):
|
|
LidarArchaeoPipeline("/nonexistent/path", str(tmp_path / "output"))
|
|
|
|
def test_incremental_index_rebuild(self, tmp_path, monkeypatch):
|
|
"""Mode incrémental : l'index est régénéré après une tuile, avec anti-rebond."""
|
|
from lidar_pipeline.pipeline import LidarArchaeoPipeline
|
|
import lidar_pipeline.index as index_mod
|
|
|
|
input_dir = tmp_path / "input"
|
|
input_dir.mkdir()
|
|
calls = []
|
|
monkeypatch.setattr(index_mod, "build_index",
|
|
lambda *a, **k: calls.append(a) or None)
|
|
|
|
pipeline = LidarArchaeoPipeline(str(input_dir), str(tmp_path / "output"),
|
|
incremental_index=True)
|
|
pipeline._rebuild_index_incremental()
|
|
pipeline._rebuild_index_incremental() # < 3 s : anti-rebond, ignoré
|
|
assert len(calls) == 1
|
|
|
|
# --no-index : jamais de rebuild incrémental
|
|
pipeline._last_index_rebuild = 0.0
|
|
pipeline.no_index = True
|
|
pipeline._rebuild_index_incremental()
|
|
assert len(calls) == 1
|
|
|
|
def test_find_laz_files_empty(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"))
|
|
files = pipeline.find_laz_files()
|
|
assert files == []
|
|
|
|
def test_find_laz_files(self, tmp_path):
|
|
from lidar_pipeline.pipeline import LidarArchaeoPipeline
|
|
input_dir = tmp_path / "input"
|
|
input_dir.mkdir()
|
|
(input_dir / "test.laz").touch()
|
|
(input_dir / "other.las").touch()
|
|
(input_dir / "readme.txt").touch()
|
|
|
|
pipeline = LidarArchaeoPipeline(str(input_dir), str(tmp_path / "output"))
|
|
files = pipeline.find_laz_files()
|
|
names = [f.name for f in files]
|
|
assert "test.laz" in names
|
|
assert "other.las" 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"
|