"""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 visualizations: 14 terrain products + point density + 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): """Without --only: only the displayed layers (oriented relief, point density) are produced; --only still allows the others.""" 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): """The map follows the ongoing render whatever the launcher.""" 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): """A tile finished during the debounce interval is picked up by a deferred pass, without waiting for the next tile.""" 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() # immediate pass p._last_index_rebuild = time.time() - 2.8 # debounce almost elapsed p._rebuild_index_incremental() # deferred (~0.2 s) p._rebuild_index_incremental() # already scheduled: no duplicate 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: """Edge stitching: pre-download of missing neighbors.""" def test_downloads_missing_ring_dedup(self, tmp_path, monkeypatch): """Missing neighbors are requested once each, the present one excluded.""" 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() # A neighbor already present must not be downloaded again. (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 neighbors - 1 already present assert (1000, 6779) not in calls assert sorted(set(calls)) == sorted(calls) # deduplicated def test_no_download_without_edge_buffer(self, tmp_path, monkeypatch): """Edge stitching disabled: no neighbor download.""" 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 must not be called") monkeypatch.setattr(fetch_ign, "fetch_tiles", boom) pipeline = LidarArchaeoPipeline(input_dir, tmp_path / "output", edge_buffer=0.0) pipeline._fetch_edge_neighbors([lazh]) class TestCleanupEdgeNeighborDuplicates: """Cleanup of duplicates between input/ and input/edge_neighbors/.""" def test_removes_true_duplicates_keeps_unique_and_part(self, tmp_path): from lidar_pipeline.pipeline import LidarArchaeoPipeline input_dir = tmp_path / "input" input_dir.mkdir() edge_dir = input_dir / "edge_neighbors" edge_dir.mkdir() dup_name = "LHD_FXX_1000_6779_PTS_LAMB93_IGN69.copc.laz" unique_name = "LHD_FXX_1001_6779_PTS_LAMB93_IGN69.copc.laz" part_name = "LHD_FXX_1002_6779_PTS_LAMB93_IGN69.copc.laz.part" mismatch_name = "LHD_FXX_1003_6779_PTS_LAMB93_IGN69.copc.laz" (input_dir / dup_name).write_bytes(b"authoritative") (edge_dir / dup_name).write_bytes(b"authoritative") # Truncated input/ copy (different size): the neighbor may be the only # sound copy, it must stay. (input_dir / mismatch_name).write_bytes(b"") (edge_dir / mismatch_name).write_bytes(b"complete") (edge_dir / unique_name).write_bytes(b"voisine-unique") (edge_dir / part_name).write_bytes(b"en-cours") pipeline = LidarArchaeoPipeline(input_dir, tmp_path / "output", edge_buffer=100.0) pipeline._cleanup_edge_neighbor_duplicates(edge_dir) assert not (edge_dir / dup_name).exists() assert (input_dir / dup_name).read_bytes() == b"authoritative" assert (edge_dir / unique_name).exists() assert (edge_dir / part_name).exists() assert (edge_dir / mismatch_name).read_bytes() == b"complete" def test_noop_when_edge_dir_missing(self, tmp_path): from lidar_pipeline.pipeline import LidarArchaeoPipeline input_dir = tmp_path / "input" input_dir.mkdir() pipeline = LidarArchaeoPipeline(input_dir, tmp_path / "output", edge_buffer=100.0) # Must not raise if edge_neighbors/ does not exist yet. pipeline._cleanup_edge_neighbor_duplicates(input_dir / "edge_neighbors") 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="not found"): LidarArchaeoPipeline("/nonexistent/path", str(tmp_path / "output")) def test_incremental_index_rebuild(self, tmp_path, monkeypatch): """Incremental mode: the index is rebuilt after a tile, with debounce.""" 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: debounced, deferred (not run immediately) assert len(calls) == 1 # --no-index: never an incremental rebuild 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): """LHD rows sorted north to south (decreasing row, increasing col).""" 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_other.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", # north row, lowest col "LHD_FXX_1056_6882_PTS_LAMB93_IGN69.copc.laz", # north row, highest col "LHD_FXX_1054_6880_PTS_LAMB93_IGN69.copc.laz", # south row "zz_other.laz", # non-LHD pattern: last ] class TestDtmMethodSidecar: """Classification method recorded next to the DTM (cache invalidation).""" 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') # No sidecar written → cache kept (considered 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" # The sidecar is a .txt file: it does not interfere with the .tif DTM lookup. 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" # Existing image, no force → skipped (no regeneration). pipeline.generate_all_visualizations(dtm, "tileA", resolution=0.5, vis_dir=vis_dir, force_images=False) assert calls == [] # Existing image, force_images=True → regenerated. 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): """IGN classes are encoded in the cache label (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: """Quality sidecar written even when the primary DTM is reused from the cache (ground classification skipped, no ground LAS available).""" @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 = [] # no visualization to compute (out of scope) # DTM already cached, compatible with the run config (no alignment, # no edge buffer, gap filling at the current version): the run must # take the "existing DTM" branch without reclassifying or regenerating. 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 expected: must not reclassify/regenerate the 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 class-2 points over 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 # Bounds: never below 2, never above 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): """Each pool process takes ONE slot when it starts: GPU (set_active_gpu) or forced CPU (-1); None or empty queue = free.""" 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): # 4th call: empty queue pipeline._init_worker_slot(q) assert calls == [("gpu", 1), ("cpu",)]