"""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): """17 visualisations : 14 terrain + densité de points + ortho + topo.""" from lidar_pipeline.pipeline import VIZ_STEPS assert len(VIZ_STEPS) == 17 def test_default_run_produces_only_panel_layers(self, tmp_path): """Sans --only : seules les couches affichées (relief orienté, densité de points) sont produites ; --only reste libre pour les autres.""" from lidar_pipeline.pipeline import LidarArchaeoPipeline p = LidarArchaeoPipeline(tmp_path, tmp_path / "out") assert [n for n, _ in p.viz_steps] == ["relief_oriente", "densite_sol"] p = LidarArchaeoPipeline(tmp_path, tmp_path / "out2", only_viz=["slope"]) assert [n for n, _ in p.viz_steps] == ["slope"] def test_incremental_index_on_by_default(self, tmp_path): """La carte suit le rendu en cours quel que soit le lanceur.""" from lidar_pipeline.pipeline import LidarArchaeoPipeline assert LidarArchaeoPipeline(tmp_path, tmp_path / "o").incremental_index assert not LidarArchaeoPipeline(tmp_path, tmp_path / "o2", no_index=True).incremental_index def test_debounced_tile_is_indexed_later(self, tmp_path, monkeypatch): """Une dalle terminée pendant l'anti-rebond est reprise par une passe différée, sans attendre la dalle suivante.""" import time import lidar_pipeline.index as index from lidar_pipeline.pipeline import LidarArchaeoPipeline calls = [] monkeypatch.setattr(index, "build_index", lambda *a, **k: calls.append(time.time())) p = LidarArchaeoPipeline(tmp_path, tmp_path / "o") p._rebuild_index_incremental() # passe immédiate p._last_index_rebuild = time.time() - 2.8 # anti-rebond presque écoulé p._rebuild_index_incremental() # différée (~0,2 s) p._rebuild_index_incremental() # déjà programmée : pas de doublon time.sleep(0.6) assert len(calls) == 2 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 TestQualityCacheHit: """Sidecar qualité écrit même quand le DTM primaire est réutilisé depuis le cache (classification sol sautée, pas de LAS sol disponible).""" @staticmethod def _write_las(path, x, y, cls, t): import laspy import numpy as np header = laspy.LasHeader(point_format=6, version="1.4") header.scales = [0.01, 0.01, 0.01] header.offsets = [652000.0, 6861000.0, 0.0] header.global_encoding.gps_time_type = laspy.header.GpsTimeType.STANDARD las = laspy.LasData(header) las.x = np.asarray(x, float); las.y = np.asarray(y, float) las.z = np.zeros(len(x)) las.classification = np.asarray(cls, np.uint8) las.return_number = np.ones(len(x), np.uint8) las.number_of_returns = np.ones(len(x), np.uint8) las.gps_time = np.asarray(t, float) las.write(str(path)) def test_cache_hit_writes_quality_sidecar(self, tmp_path, monkeypatch): import numpy as np import rasterio from rasterio.transform import from_bounds from lidar_pipeline.pipeline import LidarArchaeoPipeline import lidar_pipeline.pipeline as pipeline_mod from lidar_pipeline.dtm import GAP_FILL_TAG, GAP_FILL_VERSION from lidar_pipeline.quality import read_quality input_dir = tmp_path / "input" input_dir.mkdir() base = "LHD_FXX_0652_6862_PTS_LAMB93_IGN69" laz = input_dir / f"{base}.laz" import datetime epoch = datetime.datetime(1980, 1, 6, tzinfo=datetime.timezone.utc) gps = (datetime.datetime(2022, 6, 1, 12, tzinfo=datetime.timezone.utc) - epoch).total_seconds() - 1e9 self._write_las(laz, [652100, 652200, 652300], [6861100, 6861200, 6861300], [2, 2, 6], [gps, gps, gps]) pipeline = LidarArchaeoPipeline( str(input_dir), str(tmp_path / "output"), ground_method='ign', ign_classes="sol", strip_align=False, edge_buffer=0.0) pipeline.viz_steps = [] # aucune visualisation à calculer (hors périmètre) # DTM déjà en cache, compatible avec la config du run (pas de calage, # pas de raccord, comblement à la version courante) : le run doit # emprunter la branche « DTM existant » sans reclassifier ni régénérer. dtm_path = pipeline.dtm_dir / f"{base}_dtm.tif" with rasterio.open( dtm_path, "w", driver="GTiff", height=10, width=10, count=1, dtype="float32", crs="EPSG:2154", transform=from_bounds(652000.0, 6861995.0, 652005.0, 6862000.0, 10, 10), ) as dst: dst.write(np.zeros((10, 10), dtype="float32"), 1) dst.update_tags(**{GAP_FILL_TAG: GAP_FILL_VERSION}) def boom(*a, **k): raise AssertionError("cache hit attendu : ne doit pas reclassifier/régénérer le DTM") monkeypatch.setattr(pipeline_mod, "classify_ground", boom) monkeypatch.setattr(pipeline_mod, "create_dtm_fast", boom) assert pipeline.process_file(laz) is True data = read_quality(pipeline.output_dir, base) assert data is not None assert abs(data["ground_density"] - 2e-6) < 1e-9 # 2 points de classe 2 sur 1 km² assert data["acq_start"] == "2022-06-01" and data["acq_source"] == "gps" 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 def test_worker_slot_initializer(self, monkeypatch): """Chaque processus du pool prend UNE place à sa création : GPU (set_active_gpu) ou CPU forcé (-1) ; None ou file vide = libre.""" import queue from lidar_pipeline import gpu, pipeline calls = [] monkeypatch.setattr(gpu, "set_active_gpu", lambda i: calls.append(("gpu", i))) monkeypatch.setattr(gpu, "force_cpu", lambda: calls.append(("cpu",))) q = queue.Queue() for slot in (1, -1, None): q.put(slot) for _ in range(4): # 4e appel : file vide pipeline._init_worker_slot(q) assert calls == [("gpu", 1), ("cpu",)]