Raccorder les bords de tuiles voisines et encadrer les rendus en cours
Le MNT de chaque dalle s'étend d'une bande de 100 m remplie avec les points sol des 8 tuiles voisines (option « Raccord des bords ») : les rendus à grand noyau (openness, SVF, LRM) deviennent continus d'une dalle à l'autre, les images restent recadrées sur le kilomètre exact. Pendant une génération, la carte encadre les dalles du run : orange pulsant en cours de rendu, rouge en échec — les coins WGS84 sont portés par /api/status pour toutes les dalles non terminées. La file de génération trie les dalles du nord au sud et passe à 10 workers GPU.
This commit is contained in:
Antoine Jacquin
@ -140,81 +140,6 @@ class TestInterpolateHoles:
|
||||
assert np.isnan(filled).all()
|
||||
|
||||
|
||||
class TestMinReturnGrid:
|
||||
def test_takes_lowest_return_per_cell(self, tmp_output_dir):
|
||||
"""_min_return_grid rasterise le point le plus bas par cellule (pas la moyenne)."""
|
||||
import laspy
|
||||
from lidar_pipeline.dtm import _min_return_grid
|
||||
out = tmp_output_dir / "pts.las"
|
||||
hdr = laspy.LasHeader(version='1.2', point_format=0)
|
||||
las = laspy.LasData(hdr)
|
||||
# Grille 2x2 sur [0,2]x[0,2]. La cellule (0,0) porte deux points z=5 et
|
||||
# z=2 (min=2, moyenne=3.5) ; (1,0) z=3 ; (0,1) z=4 ; (1,1) vide.
|
||||
las.x = [0.2, 0.5, 1.2, 0.3]
|
||||
las.y = [0.2, 0.3, 0.4, 1.5]
|
||||
las.z = [5.0, 2.0, 3.0, 4.0]
|
||||
las.write(str(out))
|
||||
|
||||
grid = _min_return_grid(out, 2, 2, (0.0, 0.0, 2.0, 2.0))
|
||||
assert grid.shape == (2, 2)
|
||||
assert int(np.isnan(grid).sum()) == 1
|
||||
# Une seule valeur par cellule, et la cellule (0,0) vaut le MIN (2.0).
|
||||
vals = sorted(float(v) for v in grid[~np.isnan(grid)])
|
||||
assert vals == [2.0, 3.0, 4.0]
|
||||
assert 3.5 not in vals
|
||||
|
||||
|
||||
class TestBareEarth:
|
||||
"""Le plancher « sol nu » ramène le DTM au retour le plus bas de chaque cellule."""
|
||||
|
||||
def _write_las(self, path, points):
|
||||
"""points: list of (x, y, z). Écrit un LAS 1.2 format 0 aux bornes [0,2]x[0,2]."""
|
||||
import laspy
|
||||
hdr = laspy.LasHeader(version='1.2', point_format=0)
|
||||
las = laspy.LasData(hdr)
|
||||
las.x = [p[0] for p in points]
|
||||
las.y = [p[1] for p in points]
|
||||
las.z = [p[2] for p in points]
|
||||
las.write(str(path))
|
||||
return path
|
||||
|
||||
def _make_clouds(self, tmp_output_dir):
|
||||
"""Grille 2x2 (res=1.0). Les points « coin » à 0.05/1.95 imposent l'étendue
|
||||
[0.05,1.95] (laspy re-déduit les bornes de l'en-tête depuis les points).
|
||||
Le sol (las_file) vaut z=10 partout. Le nuage complet (source_laz) a un
|
||||
retour plus bas dans les cellules (0,0) -> 2 et (1,0) -> 5 ; les deux autres
|
||||
cellules n'ont que z=10."""
|
||||
ground = [(0.5, 0.5, 10.0), (1.5, 0.5, 10.0), (0.5, 1.5, 10.0), (1.5, 1.5, 10.0),
|
||||
(0.05, 0.05, 10.0), (1.95, 1.95, 10.0)]
|
||||
source = list(ground) + [(0.3, 0.3, 2.0), (1.3, 0.3, 5.0)]
|
||||
las_file = self._write_las(tmp_output_dir / "ground.las", ground)
|
||||
source_laz = self._write_las(tmp_output_dir / "source.las", source)
|
||||
return las_file, source_laz
|
||||
|
||||
def _dtm_values(self, tmp_output_dir, bare_earth):
|
||||
from lidar_pipeline.dtm import create_dtm_fast
|
||||
import rasterio
|
||||
las_file, source_laz = self._make_clouds(tmp_output_dir)
|
||||
out = create_dtm_fast(las_file, "tile", tmp_output_dir, 1.0,
|
||||
force=True, source_laz=source_laz, bare_earth=bare_earth)
|
||||
assert out is not None
|
||||
with rasterio.open(str(out)) as src:
|
||||
arr = src.read(1).astype("float64")
|
||||
return arr[~np.isnan(arr)]
|
||||
|
||||
def test_bare_earth_pulls_dtm_to_lowest_return(self, tmp_output_dir):
|
||||
"""Avec bare_earth, le DTM descend aux retours les plus bas (2 et 5)."""
|
||||
vals = sorted(float(v) for v in self._dtm_values(tmp_output_dir, bare_earth=True))
|
||||
# Les cellules sans retour plus bas restent à 10 ; les deux autres descendent.
|
||||
assert vals == [2.0, 5.0, 10.0, 10.0]
|
||||
assert vals[0] == 2.0 and vals[1] == 5.0
|
||||
|
||||
def test_no_bare_earth_keeps_mean(self, tmp_output_dir):
|
||||
"""Sans bare_earth, le DTM garde la moyenne des points sol (10 partout)."""
|
||||
vals = sorted(float(v) for v in self._dtm_values(tmp_output_dir, bare_earth=False))
|
||||
assert all(v == 10.0 for v in vals)
|
||||
|
||||
|
||||
class TestDetectGroundMethod:
|
||||
def _make_mock_las(self, num_returns, z_values):
|
||||
"""Create a mock laspy object with specified NumberOfReturns and z."""
|
||||
@ -615,6 +540,69 @@ class TestStripVerticalOffsets:
|
||||
assert _strip_vertical_offsets(x, y, z, psid) == {}
|
||||
|
||||
|
||||
class TestStripJitterOffsets:
|
||||
"""Gigue verticale intra-faisceau par fenêtres de temps GPS."""
|
||||
|
||||
def _synthetic(self, bias_fn, n=800_000, extent=300.0, duration=20.0, seed=0):
|
||||
"""Deux faisceaux entrelacés ; le n° 1 porte un biais dépendant du temps."""
|
||||
rng = np.random.default_rng(seed)
|
||||
x = rng.uniform(0, extent, n)
|
||||
y = rng.uniform(0, extent, n)
|
||||
clean = 100 + 0.02 * x - 0.01 * y + rng.normal(0, 0.01, n)
|
||||
t = rng.uniform(0, duration, n)
|
||||
psid = rng.integers(0, 2, n).astype(np.uint16)
|
||||
return x, y, clean + bias_fn(t, psid), psid, t, clean
|
||||
|
||||
def test_recovers_time_varying_offset(self):
|
||||
"""Une oscillation lente ±4 cm du faisceau 1 est retirée du terrain vrai.
|
||||
|
||||
En recouvrement à deux, la référence hors-faisceau attribue une série
|
||||
à chaque faisceau (chacun absorbe sa part) : on vérifie le résidu
|
||||
contre le terrain synthétique propre, fenêtre par fenêtre.
|
||||
"""
|
||||
from lidar_pipeline.dtm import _strip_jitter_offsets, _apply_strip_jitter
|
||||
w = 2 * np.pi / 8.0
|
||||
x, y, z, psid, t, clean = self._synthetic(
|
||||
lambda tt, p: 0.04 * np.sin(w * tt) * (p == 1))
|
||||
jitter = _strip_jitter_offsets(x, y, z, psid, t)
|
||||
assert set(jitter) == {0, 1}
|
||||
resid = z - _apply_strip_jitter(psid, t, jitter) - clean
|
||||
assert np.sqrt(np.mean(resid ** 2)) < 0.012
|
||||
for lo in np.arange(0, 20.0, 2.0):
|
||||
m = (t >= lo) & (t < lo + 2.0)
|
||||
assert abs(resid[m].mean()) < 0.012, f"fenêtre {lo:.0f}-{lo + 2:.0f} s"
|
||||
|
||||
def test_tracks_step_offset(self):
|
||||
"""Un échelon −3 cm sur la seconde moitié du vol est suivi."""
|
||||
from lidar_pipeline.dtm import _strip_jitter_offsets, _apply_strip_jitter
|
||||
x, y, z, psid, t, clean = self._synthetic(
|
||||
lambda tt, p: np.where(tt >= 10.0, -0.03, 0.0) * (p == 1))
|
||||
jitter = _strip_jitter_offsets(x, y, z, psid, t)
|
||||
resid = z - _apply_strip_jitter(psid, t, jitter) - clean
|
||||
assert np.sqrt(np.mean(resid ** 2)) < 0.012
|
||||
for lo in (3.0, 6.0, 13.0, 16.0): # loin de la transition lissée
|
||||
m = (t >= lo) & (t < lo + 2.0)
|
||||
assert abs(resid[m].mean()) < 0.012, f"fenêtre {lo:.0f}-{lo + 2:.0f} s"
|
||||
|
||||
def test_apply_interpolates_linearly(self):
|
||||
"""Interpolation entre centres de fenêtres ; 0 hors faisceau connu."""
|
||||
from lidar_pipeline.dtm import _apply_strip_jitter
|
||||
jitter = {7: (np.array([10.0, 11.0]), np.array([0.0, 0.1]))}
|
||||
psid = np.array([7, 7, 7, 3], dtype=np.uint16)
|
||||
t = np.array([10.0, 10.5, 15.0, 10.5])
|
||||
np.testing.assert_allclose(
|
||||
_apply_strip_jitter(psid, t, jitter), [0.0, 0.05, 0.1, 0.0])
|
||||
|
||||
def test_requires_two_sources_and_time(self):
|
||||
"""Faisceau unique ou temps non fini : rien à corriger."""
|
||||
from lidar_pipeline.dtm import _strip_jitter_offsets
|
||||
x, y, z, psid, t, _clean = self._synthetic(lambda tt, p: 0.04 * np.sin(tt) * (p == 1))
|
||||
assert _strip_jitter_offsets(x, y, z, np.zeros_like(psid), t) == {}
|
||||
t_nan = t.copy()
|
||||
t_nan[0] = np.nan
|
||||
assert _strip_jitter_offsets(x, y, z, psid, t_nan) == {}
|
||||
|
||||
|
||||
class TestStripAlignSidecar:
|
||||
def test_sidecar_roundtrip_and_threshold(self, tmp_path):
|
||||
"""Sidecar records version/threshold/offsets and matches config."""
|
||||
@ -627,6 +615,22 @@ class TestStripAlignSidecar:
|
||||
assert data["version"] == STRIP_ALIGN_VERSION
|
||||
assert data["threshold"] == STRIP_ALIGN_THRESHOLD
|
||||
assert data["offsets"] == {"1049": 0.026, "1147": -0.026} # clés JSON en chaînes
|
||||
assert data["jitter"] == {}
|
||||
|
||||
def test_sidecar_records_jitter_series(self, tmp_path):
|
||||
"""Le sidecar consigne les séries de gigue et leurs paramètres."""
|
||||
from lidar_pipeline.dtm import (
|
||||
_write_strip_align_sidecar, STRIP_JITTER_BIN, STRIP_JITTER_SMOOTH)
|
||||
import json
|
||||
jitter = {11: (np.array([0.05, 0.15]), np.array([0.012, -0.008]))}
|
||||
_write_strip_align_sidecar(tmp_path, "T", "", {}, jitter)
|
||||
data = json.loads((tmp_path / "T_dtm_stripalign.json").read_text())
|
||||
assert data["jitter_bin"] == STRIP_JITTER_BIN
|
||||
assert data["jitter_smooth"] == STRIP_JITTER_SMOOTH
|
||||
entry = data["jitter"]["11"]
|
||||
assert entry["bins"] == 2
|
||||
assert entry["series_m"] == [0.012, -0.008]
|
||||
assert entry["max_m"] == 0.012
|
||||
|
||||
def test_pipeline_match_logic(self, tmp_path):
|
||||
"""_strip_align_matches invalidates legacy DTMs and config changes."""
|
||||
@ -651,5 +655,114 @@ class TestStripAlignSidecar:
|
||||
bad = p.dtm_dir / "TILE_dtm_r0p2_stripalign.json"
|
||||
bad.write_text(json.dumps({"version": 1, "threshold": 0.02, "offsets": {}}))
|
||||
assert not p._strip_align_matches("TILE", "_r0p2")
|
||||
# Paramètres de gigue différents : à régénérer
|
||||
bad.write_text(json.dumps({"version": 2, "threshold": 0.005, "offsets": {},
|
||||
"jitter_bin": 0.5, "jitter_smooth": 5}))
|
||||
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")
|
||||
|
||||
|
||||
class TestEdgeBuffer:
|
||||
"""Raccord des bords : MNT étendu par les points sol des tuiles voisines."""
|
||||
|
||||
BASENAME = "LHD_FXX_0638_6628_PTS_LAMB93_IGN69"
|
||||
# Grille LHD : (col, row) = coin nord-ouest → 0638_6628 couvre
|
||||
# X ∈ [638000, 639000], Y ∈ [6627000, 6628000] (bord nord = 6628 km).
|
||||
NOMINAL = (638000.0, 6627000.0, 639000.0, 6628000.0) # dalle 1 km
|
||||
|
||||
def _write_las(self, path, points, classification=None):
|
||||
import laspy
|
||||
hdr = laspy.LasHeader(version='1.2', point_format=0)
|
||||
las = laspy.LasData(hdr)
|
||||
las.x = [p[0] for p in points]
|
||||
las.y = [p[1] for p in points]
|
||||
las.z = [p[2] for p in points]
|
||||
if classification is not None:
|
||||
las.classification = classification
|
||||
las.write(str(path))
|
||||
return path
|
||||
|
||||
def _write_clouds(self, root, res=50.0):
|
||||
"""Tuile centrale à z=10 + voisine EST à z=20 (un point par maille res)."""
|
||||
import numpy as np
|
||||
input_dir = root / "input"
|
||||
input_dir.mkdir(exist_ok=True)
|
||||
min_x, min_y, max_x, max_y = self.NOMINAL
|
||||
xs = np.arange(min_x + res / 2, max_x, res)
|
||||
ys = np.arange(min_y + res / 2, max_y, res)
|
||||
gx, gy = np.meshgrid(xs, ys)
|
||||
main = list(zip(gx.ravel(), gy.ravel(), np.full(gx.size, 10.0)))
|
||||
nxs = xs + 1000.0
|
||||
ngx, ngy = np.meshgrid(nxs, ys)
|
||||
east = list(zip(ngx.ravel(), ngy.ravel(), np.full(ngx.size, 20.0)))
|
||||
ground = self._write_las(root / "ground.las", main)
|
||||
source = self._write_las(input_dir / f"{self.BASENAME}.copc.laz", main)
|
||||
self._write_las(input_dir / "LHD_FXX_0639_6628_PTS_LAMB93_IGN69.copc.laz",
|
||||
east, classification=[2] * len(east))
|
||||
return ground, source, input_dir
|
||||
|
||||
def test_neighbor_discovery(self, tmp_output_dir):
|
||||
from lidar_pipeline.dtm import _neighbor_laz_files
|
||||
(tmp_output_dir / f"{self.BASENAME}.copc.laz").touch()
|
||||
present = [(637, 6627), (639, 6629), (638, 6629)]
|
||||
for c, r in present:
|
||||
(tmp_output_dir / f"LHD_FXX_{c}_{r}_PTS_LAMB93_IGN69.copc.laz").touch()
|
||||
# Bruit non voisin : jamais retenu
|
||||
(tmp_output_dir / "LHD_FXX_0650_6700_PTS_LAMB93_IGN69.copc.laz").touch()
|
||||
found = _neighbor_laz_files(tmp_output_dir / f"{self.BASENAME}.copc.laz")
|
||||
assert {f.name for f in found} == {
|
||||
f"LHD_FXX_{c}_{r}_PTS_LAMB93_IGN69.copc.laz" for c, r in present}
|
||||
|
||||
def test_neighbor_discovery_non_lhd(self, tmp_output_dir):
|
||||
from lidar_pipeline.dtm import _neighbor_laz_files
|
||||
src = tmp_output_dir / "nuage_arbitraire.laz"
|
||||
src.touch()
|
||||
assert _neighbor_laz_files(src) == []
|
||||
|
||||
def test_buffered_dtm_extends_into_neighbor(self, tmp_output_dir):
|
||||
"""MNT 24x24 (dalle 20x20 + bande 100 m), bande EST remplie à z=20 par la voisine."""
|
||||
from lidar_pipeline.dtm import create_dtm_fast, read_dtm_edge_buffer, EDGE_BUFFER_TAG
|
||||
import rasterio
|
||||
ground, source, _ = self._write_clouds(tmp_output_dir)
|
||||
dtm = create_dtm_fast(ground, self.BASENAME, tmp_output_dir, 50.0,
|
||||
force=True, source_laz=source, strip_align=False,
|
||||
edge_buffer=100.0, neighbor_classes=[2])
|
||||
assert dtm is not None
|
||||
with rasterio.open(str(dtm)) as src:
|
||||
assert (src.width, src.height) == (24, 24)
|
||||
assert abs(src.bounds.left - 637900.0) < 1e-6
|
||||
assert abs(src.bounds.top - 6628100.0) < 1e-6
|
||||
assert src.tags().get(EDGE_BUFFER_TAG) == "100"
|
||||
arr = src.read(1)
|
||||
assert arr[12, 12] == 10.0 # cœur : tuile centrale
|
||||
assert arr[12, 23] == 20.0 # bande EST : points de la voisine
|
||||
assert np.isnan(arr[0, 0]) # bande OUEST sans voisine : vide
|
||||
assert read_dtm_edge_buffer(dtm) == 100.0
|
||||
|
||||
def test_unbuffered_dtm_has_no_tag(self, tmp_output_dir):
|
||||
from lidar_pipeline.dtm import create_dtm_fast, read_dtm_edge_buffer, EDGE_BUFFER_TAG
|
||||
import rasterio
|
||||
ground, source, _ = self._write_clouds(tmp_output_dir)
|
||||
dtm = create_dtm_fast(ground, self.BASENAME, tmp_output_dir, 50.0,
|
||||
force=True, source_laz=source, strip_align=False)
|
||||
assert dtm is not None
|
||||
with rasterio.open(str(dtm)) as src:
|
||||
assert EDGE_BUFFER_TAG not in src.tags()
|
||||
assert read_dtm_edge_buffer(dtm) == 0.0
|
||||
|
||||
def test_buffered_dtm_non_lhd_falls_back(self, tmp_output_dir):
|
||||
"""Nom hors pattern LHD : pas de tuile nominale, bornes d'en-tête conservées."""
|
||||
from lidar_pipeline.dtm import create_dtm_fast, read_dtm_edge_buffer
|
||||
import rasterio
|
||||
ground = self._write_las(tmp_output_dir / "ground.las",
|
||||
[(0.5, 0.5, 10.0), (0.05, 0.05, 10.0), (1.95, 1.95, 10.0)])
|
||||
source = self._write_las(tmp_output_dir / "nuage.laz",
|
||||
[(0.5, 0.5, 10.0), (0.05, 0.05, 10.0), (1.95, 1.95, 10.0)])
|
||||
dtm = create_dtm_fast(ground, "nuage", tmp_output_dir, 1.0,
|
||||
force=True, source_laz=source, strip_align=False,
|
||||
edge_buffer=100.0)
|
||||
assert dtm is not None
|
||||
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
|
||||
|
||||
@ -31,6 +31,51 @@ class TestVizSteps:
|
||||
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
|
||||
@ -95,6 +140,26 @@ class TestLidarArchaeoPipeline:
|
||||
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)."""
|
||||
|
||||
@ -174,4 +174,45 @@ class TestTifToCrop:
|
||||
out = tif_to_crop(tif_file, tmp_path, 5.0)
|
||||
assert out is not None and out.exists()
|
||||
img = PILImage.open(str(out))
|
||||
assert img.size == (40, 40)
|
||||
assert img.size == (40, 40)
|
||||
|
||||
class TestCoreTileWindow:
|
||||
"""Recadrage des TIF à bande de raccord sur la dalle nominale 1 km."""
|
||||
|
||||
def _write_tif(self, tmp_path, name, bounds, size):
|
||||
transform = from_bounds(*bounds, size, size)
|
||||
tif_file = tmp_path / name
|
||||
with rasterio.open(
|
||||
tif_file, 'w', driver='GTiff', height=size, width=size,
|
||||
count=1, dtype='float32', crs='EPSG:2154', transform=transform,
|
||||
) as dst:
|
||||
dst.write(np.zeros((size, size), dtype=np.float32), 1)
|
||||
return tif_file
|
||||
|
||||
def test_window_on_buffered_tif(self, tmp_path):
|
||||
from lidar_pipeline.rendering import _core_tile_window
|
||||
tif = self._write_tif(tmp_path, "LHD_FXX_0638_6628_PTS_LAMB93_IGN69_slope.tif",
|
||||
(637900, 6626900, 639100, 6628100), 1200)
|
||||
with rasterio.open(tif) as src:
|
||||
win = _core_tile_window(tif, src)
|
||||
assert win is not None
|
||||
assert (win.width, win.height) == (1000, 1000)
|
||||
assert (win.col_off, win.row_off) == (100, 100)
|
||||
wt = src.window_transform(win)
|
||||
assert abs(wt.c - 638000.0) < 1e-6
|
||||
assert abs(wt.f - 6628000.0) < 1e-6
|
||||
|
||||
def test_no_window_without_overflow(self, tmp_path):
|
||||
"""TIF historique sur les bornes d'en-tête (~999,99 m) : rien à recadrer."""
|
||||
from lidar_pipeline.rendering import _core_tile_window
|
||||
tif = self._write_tif(tmp_path, "LHD_FXX_0638_6628_PTS_LAMB93_IGN69_slope.tif",
|
||||
(638000, 6627000, 638999.99, 6627999.99), 5000)
|
||||
with rasterio.open(tif) as src:
|
||||
assert _core_tile_window(tif, src) is None
|
||||
|
||||
def test_no_window_for_non_lhd_name(self, tmp_path):
|
||||
from lidar_pipeline.rendering import _core_tile_window
|
||||
tif = self._write_tif(tmp_path, "test_vis.tif",
|
||||
(637900, 6626900, 639100, 6628100), 1200)
|
||||
with rasterio.open(tif) as src:
|
||||
assert _core_tile_window(tif, src) is None
|
||||
|
||||
@ -158,6 +158,17 @@ def test_build_command_resolutions_match_completeness():
|
||||
assert cmd[i + 1] == ",".join(str(r) for r in GENERATE_RESOLUTIONS)
|
||||
|
||||
|
||||
def test_resolve_request_orders_tiles_north_to_south():
|
||||
"""Les tuiles d'une demande partent du nord vers le sud (row décroissante)."""
|
||||
from types import SimpleNamespace
|
||||
from lidar_pipeline.webapp import _resolve_request
|
||||
req = SimpleNamespace(viz=[], all_missing=False, regenerate=False,
|
||||
tiles=[[1055, 6881], [1054, 6882], [1054, 6880], [1053, 6882]],
|
||||
ground_class="ign")
|
||||
tiles, _viz = _resolve_request(req)
|
||||
assert tiles == [(1053, 6882), (1054, 6882), (1055, 6881), (1054, 6880)]
|
||||
|
||||
|
||||
def test_job_state_persisted_across_reload(tmp_path, monkeypatch):
|
||||
"""L'état du job survit au redémarrage du serveur (file .generation.job.json)."""
|
||||
import lidar_pipeline.webapp as webapp
|
||||
@ -189,6 +200,27 @@ def test_laz_cells_parses_input(tmp_path):
|
||||
assert laz_cells(tmp_path) == [(1054, 6882), (1055, 6881), (1056, 6880)]
|
||||
|
||||
|
||||
def test_attach_frame_bounds_only_unfinished():
|
||||
"""Cadres de rendu : coins WGS84 sur les dalles non terminées seulement."""
|
||||
from lidar_pipeline.webapp import _attach_frame_bounds
|
||||
tiles = [
|
||||
{"short": "1054-6882", "name": "LHD_FXX_1054_6882_PTS_LAMB93_IGN69",
|
||||
"state": "running", "steps": []},
|
||||
{"short": "1054-6881", "name": "LHD_FXX_1054_6881_PTS_LAMB93_IGN69",
|
||||
"state": "done", "steps": []},
|
||||
{"short": "inconnu", "name": "pas_une_dalle", "state": "failed", "steps": []},
|
||||
]
|
||||
_attach_frame_bounds(tiles)
|
||||
assert tiles[0]["col"] == 1054 and tiles[0]["row"] == 6882
|
||||
assert len(tiles[0]["corners"]) == 4 # SW, SE, NE, NW
|
||||
lat = [c[0] for c in tiles[0]["corners"]]
|
||||
lon = [c[1] for c in tiles[0]["corners"]]
|
||||
assert abs(max(lat) - min(lat)) < 0.02 # ~1 km en degrés
|
||||
assert abs(max(lon) - min(lon)) < 0.03
|
||||
assert "corners" not in tiles[1] # terminée : pas de cadre
|
||||
assert "corners" not in tiles[2] # nom hors grille LHD
|
||||
|
||||
|
||||
def test_preview_all_missing(tmp_path, monkeypatch):
|
||||
"""all_missing: seulement les dalles présentes dans input/ et incomplètes."""
|
||||
import lidar_pipeline.webapp as webapp
|
||||
@ -913,7 +945,8 @@ def test_generate_forwards_to_remote(monkeypatch):
|
||||
{"tiles": [[1054, 6882]], "regenerate": False,
|
||||
"reclassify": False,
|
||||
"ground_class": "ign", "ign_classes": "sol",
|
||||
"bare_earth": False, "viz": ["aspect"],
|
||||
"edge_buffer": False,
|
||||
"viz": ["aspect"],
|
||||
"all_missing": False})]
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user