238 lines
9.1 KiB
Python
238 lines
9.1 KiB
Python
"""Tests du téléchargement des dalles LiDAR HD de l'IGN (fetch_ign)."""
|
|
|
|
|
|
def test_parse_tile_specs():
|
|
"""Accepte 'col,row', 'col:row' et ignore les espaces."""
|
|
from lidar_pipeline.fetch_ign import parse_tile_specs
|
|
assert parse_tile_specs(["1055,6882"]) == [(1055, 6882)]
|
|
assert parse_tile_specs(["1055:6883", " 651 , 6630 "]) == [(1055, 6883), (651, 6630)]
|
|
|
|
|
|
def test_parse_tile_specs_rejects_invalid():
|
|
"""Les spécifications mal formées lèvent une erreur explicite."""
|
|
import pytest
|
|
from lidar_pipeline.fetch_ign import parse_tile_specs
|
|
with pytest.raises(ValueError):
|
|
parse_tile_specs(["1055"])
|
|
with pytest.raises(ValueError):
|
|
parse_tile_specs(["abc,def"])
|
|
with pytest.raises(ValueError):
|
|
parse_tile_specs(["1055,6882,9999"])
|
|
|
|
|
|
def test_tile_filename_pads_coordinates():
|
|
"""Le nom de fichier DALLE utilise des coordonnées à 4 chiffres."""
|
|
from lidar_pipeline.fetch_ign import tile_filename
|
|
assert tile_filename(1055, 6882) == "LHD_FXX_1055_6882_PTS_LAMB93_IGN69.copc.laz"
|
|
assert tile_filename(651, 6630) == "LHD_FXX_0651_6630_PTS_LAMB93_IGN69.copc.laz"
|
|
|
|
|
|
def test_match_feature_uses_coordonnees_nw():
|
|
"""La correspondance se fait sur lidarhd:coordonnees_NW (format 0816-6847)."""
|
|
from lidar_pipeline.fetch_ign import match_feature
|
|
features = [
|
|
{"id": "LHD_FXX_1054_6882_PTS_LAMB93_IGN69_NE",
|
|
"properties": {"lidarhd:coordonnees_NW": "1054-6882"},
|
|
"assets": {"data": {"href": "https://example.org/a.copc.laz"}}},
|
|
{"id": "LHD_FXX_1055_6882_PTS_LAMB93_IGN69_NE",
|
|
"properties": {"lidarhd:coordonnees_NW": "1055-6882"},
|
|
"assets": {"data": {"href": "https://example.org/b.copc.laz"}}},
|
|
]
|
|
assert match_feature(features, 1055, 6882) is features[1]
|
|
assert match_feature(features, 9999, 9999) is None
|
|
|
|
|
|
def test_find_tile_url_matches_and_returns_href(monkeypatch):
|
|
"""find_tile_url interroge le STAC et retourne l'href de l'asset data."""
|
|
from lidar_pipeline import fetch_ign
|
|
|
|
class FakeResponse:
|
|
def __init__(self, payload):
|
|
self._payload = payload.encode("utf-8")
|
|
|
|
def read(self):
|
|
return self._payload
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
payload = ('{"features": [{"properties": {"lidarhd:coordonnees_NW": "1055-6882"},'
|
|
'"assets": {"data": {"href": "https://data.geopf.fr/x.copc.laz"}}}]}')
|
|
captured = {}
|
|
|
|
def fake_urlopen(req, timeout=None):
|
|
captured["url"] = req.full_url
|
|
return FakeResponse(payload)
|
|
|
|
monkeypatch.setattr(fetch_ign.urllib.request, "urlopen", fake_urlopen)
|
|
url = fetch_ign.find_tile_url(1055, 6882)
|
|
assert url == "https://data.geopf.fr/x.copc.laz"
|
|
assert "api.stac.teledetection.fr" in captured["url"]
|
|
assert "bbox=" in captured["url"]
|
|
|
|
|
|
def test_fetch_tiles_skips_existing_and_generated(tmp_path, monkeypatch):
|
|
"""Pas de téléchargement si le LAZ existe ou si les visualisations existent."""
|
|
from lidar_pipeline import fetch_ign
|
|
|
|
input_dir = tmp_path / "input"
|
|
input_dir.mkdir()
|
|
output_dir = tmp_path / "output"
|
|
vis_dir = output_dir / "visualisations" / "LHD_FXX_1054_6882_PTS_LAMB93_IGN69"
|
|
vis_dir.mkdir(parents=True)
|
|
|
|
# 1054,6882 : visualisations déjà générées
|
|
# 1055,6882 : LAZ déjà présent
|
|
existing = input_dir / fetch_ign.tile_filename(1055, 6882)
|
|
existing.write_bytes(b"naze")
|
|
|
|
def fail_download(*args, **kwargs):
|
|
raise AssertionError("ne doit pas être appelé")
|
|
|
|
monkeypatch.setattr(fetch_ign, "find_tile_url", fail_download)
|
|
result = fetch_ign.fetch_tiles(input_dir, [(1054, 6882), (1055, 6882)],
|
|
output_dir=output_dir)
|
|
assert result == []
|
|
|
|
|
|
def _make_viz_dirs(output_dir, col, row, viz_keys, resolutions=(0.5, 0.2)):
|
|
"""Crée les dossiers de visualisations d'une tuile (une entrée par résolution).
|
|
|
|
Le suffixe de résolution n'est que sur le DOSSIER : les fichiers gardent
|
|
le basename nu (cf. _expected_output_path dans le pipeline).
|
|
"""
|
|
base = f"LHD_FXX_{col:04d}_{row:04d}_PTS_LAMB93_IGN69"
|
|
for res in resolutions:
|
|
suffix = "" if res == 0.5 else "_r" + str(res).replace('.', 'p')
|
|
vis = output_dir / "visualisations" / (base + suffix)
|
|
vis.mkdir(parents=True, exist_ok=True)
|
|
for key in viz_keys:
|
|
(vis / f"{base}_{key}.avif").write_bytes(b"x")
|
|
|
|
|
|
def test_fetch_tiles_downloads_incomplete_tile(tmp_path, monkeypatch):
|
|
"""Une tuile incomplète (visualisation manquante) est téléchargée, une complète non."""
|
|
from pathlib import Path
|
|
from lidar_pipeline import fetch_ign
|
|
|
|
input_dir = tmp_path / "input"
|
|
input_dir.mkdir()
|
|
output_dir = tmp_path / "output"
|
|
_make_viz_dirs(output_dir, 1054, 6882, ["aspect"]) # slope manquant
|
|
_make_viz_dirs(output_dir, 1055, 6882, ["aspect", "slope"]) # complète
|
|
|
|
calls = []
|
|
|
|
def fake_find_tile_url(col, row, timeout=20):
|
|
calls.append((col, row))
|
|
return "https://example.org/dalle.copc.laz"
|
|
|
|
def fake_download(url, dest, timeout=120, chunk=1024 * 1024):
|
|
Path(dest).write_bytes(b"laz")
|
|
return 3
|
|
|
|
monkeypatch.setattr(fetch_ign, "find_tile_url", fake_find_tile_url)
|
|
monkeypatch.setattr(fetch_ign, "download_file", fake_download)
|
|
|
|
result = fetch_ign.fetch_tiles(input_dir, [(1054, 6882), (1055, 6882)],
|
|
output_dir=output_dir,
|
|
only_viz=["aspect", "slope"],
|
|
resolutions=(0.5, 0.2))
|
|
# Seule la tuile incomplète est retéléchargée
|
|
assert calls == [(1054, 6882)]
|
|
assert [p.name for p in result] == [fetch_ign.tile_filename(1054, 6882)]
|
|
|
|
|
|
def test_fetch_tiles_promotes_edge_neighbor_copy(tmp_path, monkeypatch):
|
|
"""Une dalle déjà téléchargée comme voisine est déplacée dans input/ sans réseau."""
|
|
from lidar_pipeline import fetch_ign
|
|
|
|
input_dir = tmp_path / "input"
|
|
input_dir.mkdir()
|
|
edge_dir = input_dir / "edge_neighbors"
|
|
edge_dir.mkdir()
|
|
name = fetch_ign.tile_filename(1055, 6882)
|
|
neighbor_copy = edge_dir / name
|
|
neighbor_copy.write_bytes(b"laz-voisine")
|
|
|
|
def fail(*args, **kwargs):
|
|
raise AssertionError("ne doit pas être appelé (pas de réseau attendu)")
|
|
|
|
monkeypatch.setattr(fetch_ign, "find_tile_url", fail)
|
|
monkeypatch.setattr(fetch_ign, "download_file", fail)
|
|
|
|
result = fetch_ign.fetch_tiles(input_dir, [(1055, 6882)])
|
|
|
|
assert not neighbor_copy.exists()
|
|
assert (input_dir / name).read_bytes() == b"laz-voisine"
|
|
assert result == [] # promotion = "skip", comme le cas déjà-présent
|
|
|
|
|
|
def test_fetch_tiles_does_not_promote_when_called_on_edge_dir(tmp_path, monkeypatch):
|
|
"""Appel avec input_dir=edge_dir (cas _fetch_edge_neighbors) : jamais de promotion."""
|
|
from lidar_pipeline import fetch_ign
|
|
|
|
edge_dir = tmp_path / "input" / "edge_neighbors"
|
|
edge_dir.mkdir(parents=True)
|
|
inner_edge = edge_dir / "edge_neighbors"
|
|
name = fetch_ign.tile_filename(1055, 6882)
|
|
# Même si un dossier edge_neighbors/edge_neighbors/<name> existait par
|
|
# accident, il ne doit jamais être déplacé vers edge_dir/<name>.
|
|
inner_edge.mkdir()
|
|
(inner_edge / name).write_bytes(b"ne-doit-pas-bouger")
|
|
|
|
calls = []
|
|
|
|
def fake_find_tile_url(col, row, timeout=20):
|
|
calls.append((col, row))
|
|
return "https://example.org/dalle.copc.laz"
|
|
|
|
def fake_download(url, dest, timeout=120, chunk=1024 * 1024):
|
|
from pathlib import Path
|
|
Path(dest).write_bytes(b"laz")
|
|
return 3
|
|
|
|
monkeypatch.setattr(fetch_ign, "find_tile_url", fake_find_tile_url)
|
|
monkeypatch.setattr(fetch_ign, "download_file", fake_download)
|
|
|
|
result = fetch_ign.fetch_tiles(edge_dir, [(1055, 6882)])
|
|
|
|
assert calls == [(1055, 6882)] # téléchargée normalement, pas promue
|
|
assert (inner_edge / name).read_bytes() == b"ne-doit-pas-bouger"
|
|
assert (edge_dir / name).read_bytes() == b"laz"
|
|
assert len(result) == 1
|
|
|
|
|
|
def test_fetch_tiles_force_downloads_complete(tmp_path, monkeypatch):
|
|
"""force=True télécharge même une tuile complète (régénération avec --force)."""
|
|
from pathlib import Path
|
|
from lidar_pipeline import fetch_ign
|
|
|
|
input_dir = tmp_path / "input"
|
|
input_dir.mkdir()
|
|
output_dir = tmp_path / "output"
|
|
_make_viz_dirs(output_dir, 1054, 6882, ["aspect", "slope"])
|
|
|
|
calls = []
|
|
|
|
def fake_find_tile_url(col, row, timeout=20):
|
|
calls.append((col, row))
|
|
return "https://example.org/dalle.copc.laz"
|
|
|
|
def fake_download(url, dest, timeout=120, chunk=1024 * 1024):
|
|
Path(dest).write_bytes(b"laz")
|
|
return 3
|
|
|
|
monkeypatch.setattr(fetch_ign, "find_tile_url", fake_find_tile_url)
|
|
monkeypatch.setattr(fetch_ign, "download_file", fake_download)
|
|
|
|
result = fetch_ign.fetch_tiles(input_dir, [(1054, 6882)],
|
|
output_dir=output_dir,
|
|
only_viz=["aspect", "slope"],
|
|
resolutions=(0.5, 0.2), force=True)
|
|
assert calls == [(1054, 6882)]
|
|
assert len(result) == 1
|