Recaler les lignes de balayage des passes et accélérer le rendu

Troisième passe du calage vertical : chaque ligne de balayage (décalage et
inclinaison due au roulis) est recalée contre le consensus des autres
faisceaux, à toutes les échelles, avec un profil d'étalonnage par faisceau
et par degré d'angle qui retire les écarts non linéaires en travers de la
fauchée. Les lignes sans recouvrement sont corrigées contre leur propre
faisceau. Efface les lignes en creux et la marche au bord de fauchée
mesurées sur LHD_FXX_0999_6882 (validé sur des blocs jamais vus). Calcul
vectorisé, CuPy si GPU ; la gigue par fenêtres de temps devient inutile
quand scan_angle existe.

Rendu plus rapide : encodage AVIF speed 9 (0,6 s au lieu de 4 s par dalle),
classification IGN par extraction directe laspy au lieu de PDAL (4,9 s au
lieu de 13,5 s), comblement des trous et gradients sur GPU, cache numba
persistant dans l'image.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

lidar_pipeline/tests/test_dtm.py

@@ -661,6 +661,13 @@ class TestStripAlignSidecar:
         assert not p._strip_align_matches("TILE", "_r0p2")
         # Calage désactivé + DTM calé : à régénérer
         assert not P(tmp_path, strip_align=False)._strip_align_matches("TILE", "_r0p2")
+        # Paramètres ligne à ligne différents : à régénérer
+        _write_strip_align_sidecar(p.dtm_dir, "TILE", "_r0p2", {})
+        data = json.loads(bad.read_text())
+        data["line_window"] = 99
+        bad.write_text(json.dumps(data))
+        assert not p._strip_align_matches("TILE", "_r0p2")
+
 
 
 class TestEdgeBuffer:
@@ -784,3 +791,159 @@ class TestEdgeBuffer:
         with rasterio.open(str(dtm)) as src:
             assert abs(src.bounds.left - 0.05) < 1e-6  # bornes de l'en-tête
         assert read_dtm_edge_buffer(dtm) == 0.0
+
+
+def _synthetic_beam(n_lines=240, spacing=0.4, seed=0, offsets=None, tilts=None):
+    """Faisceau synthétique : lignes de balayage (dents de scie de scan_angle)
+    sur un terrain en pente traversé par un fossé ; offsets verticaux par ligne."""
+    rng = np.random.default_rng(seed)
+    xs, ys, zs, ts, angs, ids = [], [], [], [], [], []
+    x_line = np.arange(0, 100, 0.12)
+    for k in range(n_lines):
+        y = k * spacing + rng.normal(0, 0.02, x_line.size)
+        z = 100 + 0.03 * x_line + 0.05 * y
+        z = z - 0.5 * (np.abs(x_line - 41) < 1.0)          # fossé perpendiculaire aux lignes
+        u = np.linspace(-1, 1, x_line.size)
+        z = z + offsets[k] + (0 if tilts is None else tilts[k]) * u + rng.normal(0, 0.01, x_line.size)
+        xs.append(x_line); ys.append(y); zs.append(z); ids.append(np.full(x_line.size, k))
+        ts.append(k * 0.0067 + np.linspace(0, 0.006, x_line.size))
+        angs.append(np.linspace(-3300, 3300, x_line.size))
+    return (np.concatenate(xs), np.concatenate(ys), np.concatenate(zs),
+            np.concatenate(ts), np.concatenate(angs), np.concatenate(ids))
+
+
+class TestScanLineAlignment:
+    """3ᵉ passe du calage : décalage vertical entre lignes de balayage successives."""
+
+    def test_line_ids_follow_sawtooth_not_vegetation_gaps(self):
+        from lidar_pipeline.dtm import _scan_line_ids
+        t = np.arange(50) * 0.0001
+        ang = np.tile(np.linspace(-3000, 3000, 10), 5)
+        keep = np.ones(50, bool); keep[13:17] = False        # trou de végétation dans la ligne 2
+        ids = _scan_line_ids(t[keep], ang[keep])
+        assert ids.max() == 4
+        t2 = t.copy(); t2[30:] += 1.0                          # fin de passe : nouvelle ligne
+        assert _scan_line_ids(t2, np.zeros(50)).max() == 1
+
+    def test_group_median_matches_numpy(self):
+        from lidar_pipeline.dtm import _group_median
+        rng = np.random.default_rng(3)
+        g = rng.integers(0, 20, 2000)
+        v = rng.normal(size=2000); v[::17] = np.nan
+        med = _group_median(v, g, 20, 1)
+        for k in range(20):
+            vals = v[(g == k) & np.isfinite(v)]
+            assert med[k] == pytest.approx(np.median(vals))
+
+    def test_removes_alternating_line_offsets_and_keeps_ditch(self):
+        from lidar_pipeline.dtm import _scan_line_corrections_beam
+        n = 240
+        rng = np.random.default_rng(1)
+        offsets = 0.015 * (-1.0) ** np.arange(n) + rng.normal(0, 0.006, n)
+        x, y, z, t, ang, ids = _synthetic_beam(n, offsets=offsets)
+        corr, per_line, _ = _scan_line_corrections_beam(x, y, z, t, ang)
+        assert len(per_line) == n
+        from scipy.ndimage import gaussian_filter1d
+        hp = lambda v: v - gaussian_filter1d(v, 3, mode="nearest")
+        before, after = hp(offsets)[10:-10].std(), hp(offsets - per_line)[10:-10].std()
+        assert after < 0.2 * before, f"{before*1000:.1f} → {after*1000:.1f} mm"
+        ditch = np.abs(x - 41) < 0.8
+        depth = lambda zz: np.median(zz[~ditch & (np.abs(x - 41) < 4)]) - np.median(zz[ditch])
+        assert depth(z - corr) == pytest.approx(depth(z), abs=0.01)
+
+    def test_removes_alternating_line_tilts(self):
+        """Roulis : lignes basculées alternativement (un bout haut, l'autre bas)."""
+        from scipy.ndimage import gaussian_filter1d
+        from lidar_pipeline.dtm import _scan_line_corrections_beam
+        n = 240
+        rng = np.random.default_rng(4)
+        tilts = 0.02 * (-1.0) ** np.arange(n) + rng.normal(0, 0.008, n)
+        x, y, z, t, ang, ids = _synthetic_beam(n, offsets=np.zeros(n), tilts=tilts)
+        corr, _, per_tilt = _scan_line_corrections_beam(x, y, z, t, ang)
+        hp = lambda v: v - gaussian_filter1d(v, 3, mode="nearest")
+        before, after = hp(tilts)[10:-10].std(), hp(tilts - per_tilt)[10:-10].std()
+        assert after < 0.2 * before, f"{before*1000:.1f} → {after*1000:.1f} mm"
+
+    def test_joint_adjustment_fixes_all_scales_against_other_beam(self):
+        """Deux faisceaux superposés : le faisceau fautif (dérive lente, roulis,
+        ligne isolée à −8 cm) est recalé sur l'autre à toutes les échelles,
+        sans dérive de l'altitude d'ensemble."""
+        from lidar_pipeline.dtm import _joint_line_corrections
+        n = 240
+        k = np.arange(n)
+        bad_off = 0.03 * np.sin(2 * np.pi * k / 120)          # dérive lente (non vue sur sa propre surface)
+        bad_off[100] -= 0.08                                   # ligne isolée très décalée
+        bad_tilt = 0.02 * np.cos(2 * np.pi * k / 60)           # roulis lent
+        xa, ya, za, ta, aa, _ = _synthetic_beam(n, seed=1, offsets=bad_off, tilts=bad_tilt)
+        xb, yb, zb, tb, ab, _ = _synthetic_beam(n, seed=2, offsets=np.zeros(n))
+        tb = tb + 1000.0
+        x = np.r_[xa, xb]; y = np.r_[ya, yb]; z = np.r_[za, zb]; t = np.r_[ta, tb]
+        ang = np.r_[aa, ab]; psid = np.r_[np.full(len(za), 1), np.full(len(zb), 2)]
+        corr, gl, touched, iters = _joint_line_corrections(x, y, z, psid, t, ang)
+        truth = np.r_[bad_off[np.repeat(k, len(za) // n)] + bad_tilt[np.repeat(k, len(za) // n)]
+                      * np.tile(np.linspace(-1, 1, len(za) // n), n), np.zeros(len(zb))]
+        # Sans vérité terrain, l'écart est partagé entre les faisceaux : c'est
+        # l'écart ENTRE faisceaux (points homologues, même géométrie) qui doit
+        # disparaître.
+        na = len(za)
+        before = truth[:na] - truth[:na].mean()
+        rel = (truth[:na] - corr[:na]) - (0.0 - corr[na:])
+        after = rel - rel.mean()
+        assert np.std(after) < 0.25 * np.std(before), \
+            f"{np.std(before)*1000:.1f} → {np.std(after)*1000:.1f} mm"
+        assert abs(np.mean(corr)) < 0.002                     # pas de dérive d'ensemble
+        assert touched.mean() > 0.9 and iters <= 8
+
+    def test_joint_adjustment_removes_static_angle_profile(self):
+        """Étalonnage en arc selon l'angle (même pour toutes les lignes) :
+        non linéaire, invisible pour décalage + inclinaison, retiré par le
+        profil par faisceau et classe d'angle."""
+        from lidar_pipeline.dtm import _joint_line_corrections
+        n = 240
+        xa, ya, za, ta, aa, _ = _synthetic_beam(n, seed=5, offsets=np.zeros(n))
+        uu = aa / 3300.0
+        prof = 0.02 * (uu ** 2 - np.mean(uu ** 2))
+        za = za + prof
+        xb, yb, zb, tb, ab, _ = _synthetic_beam(n, seed=6, offsets=np.zeros(n))
+        x = np.r_[xa, xb]; y = np.r_[ya, yb]; z = np.r_[za, zb]; t = np.r_[ta, tb + 1000.0]
+        ang = np.r_[aa, ab]; psid = np.r_[np.full(len(za), 1), np.full(len(zb), 2)]
+        corr, _, _, _ = _joint_line_corrections(x, y, z, psid, t, ang)
+        na = len(za)
+        rel = (prof - corr[:na]) - (0.0 - corr[na:])
+        assert np.std(rel - rel.mean()) < 0.3 * np.std(prof), \
+            f"{np.std(prof)*1000:.1f} → {np.std(rel - rel.mean())*1000:.1f} mm"
+
+    def test_clean_beam_left_untouched(self):
+        from lidar_pipeline.dtm import _scan_line_corrections
+        x, y, z, t, ang, ids = _synthetic_beam(240, offsets=np.zeros(240))
+        corr, stats = _scan_line_corrections(x, y, z, np.full(len(z), 7), t, ang)
+        assert stats == {} and not corr.any()
+
+
+class TestIgnDirectExtraction:
+    """Classification IGN : extraction directe par laspy (PDAL en secours)."""
+
+    def test_keeps_requested_classes_and_valid_returns(self, tmp_path):
+        import laspy
+        from lidar_pipeline.dtm import _extract_ign_ground
+        n = 1000
+        rng = np.random.default_rng(0)
+        hdr = laspy.LasHeader(point_format=6, version="1.4")
+        hdr.scales = np.array([0.01, 0.01, 0.001]); hdr.offsets = np.array([1000.0, 6800000.0, 0.0])
+        las = laspy.LasData(hdr)
+        las.x = 1000 + rng.uniform(0, 100, n); las.y = 6800000 + rng.uniform(0, 100, n)
+        las.z = rng.uniform(100, 110, n)
+        cls = rng.choice([1, 2, 3, 6, 9], n); las.classification = cls
+        rn = np.ones(n, dtype=np.uint8); rn[:10] = 0; las.return_number = rn
+        las.number_of_returns = np.ones(n, dtype=np.uint8)
+        las.point_source_id = np.full(n, 42, dtype=np.uint16)
+        src = tmp_path / "t.las"; las.write(str(src))
+        out = tmp_path / "g.las"
+        assert _extract_ign_ground(src, out, [2])
+        g = laspy.read(str(out))
+        expected = (cls == 2) & (rn >= 1)
+        assert len(g.points) == int(expected.sum())
+        assert set(np.unique(np.asarray(g.classification))) == {2}
+        assert g.header.point_format.id == 6 and np.all(np.asarray(g.point_source_id) == 42)
+        assert _extract_ign_ground(src, tmp_path / "e.las", [1, 2])
+        assert len(laspy.read(str(tmp_path / "e.las")).points) == int(((np.isin(cls, [1, 2])) & (rn >= 1)).sum())