31645a42e8
PANEL_VIZ ne contient plus que relief_oriente et pilote désormais tout : le pipeline sans --only ne produit que cette couche, la génération lancée depuis la carte aussi, et la carte ne liste ni ne sert en tuiles (panneau, XYZ, TileJSON, WMTS, JOSM) les autres visualisations présentes sur disque. Les autres visualisations restent calculables explicitement avec --only. Corrige au passage _panel_viz_steps, qui ne retenait que les couches dont le nom de fichier diffère du nom d'étape : la génération depuis la carte ne produisait que l'openness positive. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
289 lines
13 KiB
Python
289 lines
13 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 16 visualizations (14 terrain + ortho + topo)."""
|
|
from lidar_pipeline.pipeline import VIZ_STEPS
|
|
assert len(VIZ_STEPS) == 16
|
|
|
|
def test_default_run_produces_only_relief(self, tmp_path):
|
|
"""Sans --only : seule la couche affichée (relief orienté) est produite ;
|
|
--only reste libre pour les autres visualisations."""
|
|
from lidar_pipeline.pipeline import LidarArchaeoPipeline
|
|
p = LidarArchaeoPipeline(tmp_path, tmp_path / "out")
|
|
assert [n for n, _ in p.viz_steps] == ["relief_oriente"]
|
|
p = LidarArchaeoPipeline(tmp_path, tmp_path / "out2", only_viz=["slope"])
|
|
assert [n for n, _ in p.viz_steps] == ["slope"]
|
|
|
|
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 TestFetchEdgeNeighbors:
|
|
"""Raccord des bords : pré-téléchargement des voisines manquantes."""
|
|
|
|
def test_downloads_missing_ring_dedup(self, tmp_path, monkeypatch):
|
|
"""Les 8 voisines manquantes sont demandées une seule fois, présente exclue."""
|
|
import lidar_pipeline.fetch_ign as fetch_ign
|
|
from lidar_pipeline.pipeline import LidarArchaeoPipeline
|
|
input_dir = tmp_path / "input"
|
|
input_dir.mkdir()
|
|
lazh = input_dir / "LHD_FXX_0999_6778_PTS_LAMB93_IGN69.copc.laz"
|
|
lazh.touch()
|
|
# Une voisine déjà présente ne doit pas être retéléchargée.
|
|
(input_dir / "LHD_FXX_1000_6779_PTS_LAMB93_IGN69.copc.laz").touch()
|
|
calls = []
|
|
|
|
def fake_fetch_tiles(input_dir_, specs, **kwargs):
|
|
calls.extend(specs)
|
|
return [input_dir / "fake" for _ in specs]
|
|
|
|
monkeypatch.setattr(fetch_ign, "fetch_tiles", fake_fetch_tiles)
|
|
pipeline = LidarArchaeoPipeline(input_dir, tmp_path / "output",
|
|
edge_buffer=100.0)
|
|
pipeline._fetch_edge_neighbors([lazh])
|
|
assert len(calls) == 7 # 8 voisines - 1 déjà présente
|
|
assert (1000, 6779) not in calls
|
|
assert sorted(set(calls)) == sorted(calls) # dédupliqué
|
|
|
|
def test_no_download_without_edge_buffer(self, tmp_path, monkeypatch):
|
|
"""Raccord désactivé : aucun téléchargement de voisines."""
|
|
import lidar_pipeline.fetch_ign as fetch_ign
|
|
from lidar_pipeline.pipeline import LidarArchaeoPipeline
|
|
input_dir = tmp_path / "input"
|
|
input_dir.mkdir()
|
|
lazh = input_dir / "LHD_FXX_0999_6778_PTS_LAMB93_IGN69.copc.laz"
|
|
lazh.touch()
|
|
|
|
def boom(*args, **kwargs):
|
|
raise AssertionError("fetch_tiles ne doit pas être appelé")
|
|
|
|
monkeypatch.setattr(fetch_ign, "fetch_tiles", boom)
|
|
pipeline = LidarArchaeoPipeline(input_dir, tmp_path / "output",
|
|
edge_buffer=0.0)
|
|
pipeline._fetch_edge_neighbors([lazh])
|
|
|
|
|
|
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
|
|
|
|
def test_find_laz_files_sorted_north_to_south(self, tmp_path):
|
|
"""Lignes LHD triées du nord au sud (row décroissante, col croissante)."""
|
|
from lidar_pipeline.pipeline import LidarArchaeoPipeline
|
|
input_dir = tmp_path / "input"
|
|
input_dir.mkdir()
|
|
for name in ("LHD_FXX_1054_6880_PTS_LAMB93_IGN69.copc.laz",
|
|
"LHD_FXX_1056_6882_PTS_LAMB93_IGN69.copc.laz",
|
|
"LHD_FXX_1053_6882_PTS_LAMB93_IGN69.copc.laz",
|
|
"zz_autre.laz"):
|
|
(input_dir / name).touch()
|
|
|
|
pipeline = LidarArchaeoPipeline(str(input_dir), str(tmp_path / "output"))
|
|
names = [f.name for f in pipeline.find_laz_files()]
|
|
assert names == [
|
|
"LHD_FXX_1053_6882_PTS_LAMB93_IGN69.copc.laz", # ligne nord, col mini
|
|
"LHD_FXX_1056_6882_PTS_LAMB93_IGN69.copc.laz", # ligne nord, col maxi
|
|
"LHD_FXX_1054_6880_PTS_LAMB93_IGN69.copc.laz", # ligne sud
|
|
"zz_autre.laz", # hors pattern : en fin
|
|
]
|
|
|
|
|
|
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"
|
|
|
|
|
|
class TestResolveWorkers:
|
|
def test_auto_scales_with_cpus(self):
|
|
from lidar_pipeline.pipeline import resolve_workers
|
|
import os
|
|
w = resolve_workers('auto')
|
|
assert w == max(2, min((os.cpu_count() or 4) - 2, 16))
|
|
|
|
def test_auto_bounded(self):
|
|
from lidar_pipeline.pipeline import resolve_workers
|
|
# Bornes : jamais sous 2, jamais au-dessus de 16
|
|
assert 2 <= resolve_workers('auto') <= 16
|
|
|
|
def test_explicit_int(self):
|
|
from lidar_pipeline.pipeline import resolve_workers
|
|
assert resolve_workers('4') == 4
|
|
assert resolve_workers(7) == 7
|
|
|
|
def test_invalid_falls_back_to_one(self):
|
|
from lidar_pipeline.pipeline import resolve_workers
|
|
assert resolve_workers('abc') == 1
|
|
assert resolve_workers(None) == 1
|