23969c9e14
index.py rewritten as a continuous Leaflet map: rotated L93 tiles, stackable visualization layers (per-layer opacity, drag-reorder persisted in localStorage), tile info panel, live rebuild after each tile during a run. Leaflet is vendored in assets/vendor/ so the map works fully offline; georeferencing falls back rasterio -> pyproj -> affine so the lightweight webapp (no GDAL) is supported. docker-compose.worker.yml adds the tile generator service (full image + GPU) that remote webapps call via LIDAR_GENERATION_URL, plus a one-shot process profile. webapp.py gains LIDAR_AUTO_SYNC_SECONDS periodic cache refresh and LIDAR_REGEN_CIDR restricting generation to the local network. run.sh --serve-webapp now mounts ~/.ssh read-only so the rsync sync works. docs/DEPLOY_WEBAPP.md completed for Raspberry Pi deployment: prerequisites, git clone install, SSH key setup, first sync, update procedure and troubleshooting.
561 lines
23 KiB
Python
561 lines
23 KiB
Python
"""Tests pour la carte globale interactive (index.py)."""
|
||
|
||
import json
|
||
from pathlib import Path
|
||
|
||
|
||
def test_parse_basename_coords_valid():
|
||
"""Parse les coordonnées d'un basename LHD valide."""
|
||
from lidar_pipeline.index import parse_basename_coords
|
||
assert parse_basename_coords("LHD_FXX_1000_6881_PTS_LAMB93_IGN69") == (1000, 6881)
|
||
assert parse_basename_coords("LHD_FXX_1049_6895_PTS_LAMB93_IGN69") == (1049, 6895)
|
||
|
||
|
||
def test_parse_basename_coords_with_res_suffix():
|
||
"""Les noms de dossier avec suffixe résolution sont aussi parsables."""
|
||
from lidar_pipeline.index import parse_basename_coords
|
||
assert parse_basename_coords("LHD_FXX_1000_6881_PTS_LAMB93_IGN69_r0p2") == (1000, 6881)
|
||
|
||
|
||
def test_parse_basename_coords_invalid():
|
||
"""Les noms non-LHD retournent None."""
|
||
from lidar_pipeline.index import parse_basename_coords
|
||
assert parse_basename_coords("random_dir") is None
|
||
assert parse_basename_coords("DTM") is None
|
||
assert parse_basename_coords("") is None
|
||
|
||
|
||
def test_strip_res_suffix_primary():
|
||
"""Dossier sans suffixe = résolution primaire (0.5)."""
|
||
from lidar_pipeline.index import _strip_res_suffix
|
||
base, res = _strip_res_suffix("LHD_FXX_1000_6881_PTS_LAMB93_IGN69")
|
||
assert base == "LHD_FXX_1000_6881_PTS_LAMB93_IGN69"
|
||
assert res == 0.5
|
||
|
||
|
||
def test_strip_res_suffix_multi():
|
||
"""Dossier avec suffixe _r0p2 = résolution 0.2."""
|
||
from lidar_pipeline.index import _strip_res_suffix
|
||
base, res = _strip_res_suffix("LHD_FXX_1000_6881_PTS_LAMB93_IGN69_r0p2")
|
||
assert base == "LHD_FXX_1000_6881_PTS_LAMB93_IGN69"
|
||
assert res == 0.2
|
||
|
||
|
||
def test_compute_bbox():
|
||
"""Calcule la bounding box d'un ensemble de tuiles."""
|
||
from lidar_pipeline.index import compute_bbox
|
||
tiles = [
|
||
{'col': 1000, 'row': 6881},
|
||
{'col': 1001, 'row': 6882},
|
||
{'col': 1002, 'row': 6880},
|
||
]
|
||
bbox = compute_bbox(tiles)
|
||
assert bbox == {'min_col': 1000, 'max_col': 1002, 'min_row': 6880, 'max_row': 6882}
|
||
|
||
|
||
def test_compute_bbox_empty():
|
||
"""Aucune tuile → None."""
|
||
from lidar_pipeline.index import compute_bbox
|
||
assert compute_bbox([]) is None
|
||
|
||
|
||
def _make_fake_viz_dir(vis_dir, basename, col, row, viz_keys=('hillshade_multi', 'svf'), ext='webp', res_suffix=''):
|
||
"""Crée un faux dossier de visualisations avec de petites images.
|
||
|
||
Le suffixe de résolution apparaît seulement dans le nom du dossier (miroir
|
||
du pipeline : les fichiers restent préfixés par le basename nu).
|
||
"""
|
||
from PIL import Image as PILImage
|
||
import numpy as np
|
||
|
||
dir_name = f"LHD_FXX_{col}_{row}_PTS_LAMB93_IGN69{res_suffix}"
|
||
tile_dir = Path(vis_dir) / dir_name
|
||
tile_dir.mkdir(parents=True, exist_ok=True)
|
||
for v in viz_keys:
|
||
arr = np.random.randint(0, 255, (50, 50, 3), dtype=np.uint8)
|
||
img = PILImage.fromarray(arr)
|
||
fname = f"LHD_FXX_{col}_{row}_PTS_LAMB93_IGN69_{v}.{ext}"
|
||
img.save(str(tile_dir / fname), format='WEBP', quality=80)
|
||
return tile_dir
|
||
|
||
|
||
def test_scan_tiles(tmp_path):
|
||
"""scan_tiles détecte les dossiers de tuiles et leurs visualisations."""
|
||
from lidar_pipeline.index import scan_tiles
|
||
|
||
vis_dir = tmp_path / "visualisations"
|
||
vis_dir.mkdir()
|
||
_make_fake_viz_dir(vis_dir, "a", 1000, 6881, ('hillshade_multi', 'svf'))
|
||
_make_fake_viz_dir(vis_dir, "b", 1001, 6881, ('hillshade_multi',))
|
||
|
||
tiles = scan_tiles(vis_dir)
|
||
assert len(tiles) == 2
|
||
names = sorted(t['dir_name'] for t in tiles)
|
||
assert "LHD_FXX_1000_6881_PTS_LAMB93_IGN69" in names
|
||
assert "LHD_FXX_1001_6881_PTS_LAMB93_IGN69" in names
|
||
# Vérifie que les viz sont détectées
|
||
t0 = next(t for t in tiles if t['col'] == 1000)
|
||
assert 'hillshade_multi' in t0['viz']
|
||
assert 'svf' in t0['viz']
|
||
|
||
|
||
def test_scan_tiles_ignores_non_lhd(tmp_path):
|
||
"""Les dossiers non-LHD (ex: temp, DTM) sont ignorés."""
|
||
from lidar_pipeline.index import scan_tiles
|
||
|
||
vis_dir = tmp_path / "visualisations"
|
||
vis_dir.mkdir()
|
||
(vis_dir / "random_folder").mkdir()
|
||
_make_fake_viz_dir(vis_dir, "a", 1000, 6881)
|
||
|
||
tiles = scan_tiles(vis_dir)
|
||
assert len(tiles) == 1
|
||
assert tiles[0]['col'] == 1000
|
||
|
||
|
||
def test_scan_tiles_multi_resolution(tmp_path):
|
||
"""Les dossiers avec suffixe résolution sont correctement décodés."""
|
||
from lidar_pipeline.index import scan_tiles
|
||
|
||
vis_dir = tmp_path / "visualisations"
|
||
vis_dir.mkdir()
|
||
_make_fake_viz_dir(vis_dir, "a", 1000, 6881, res_suffix='')
|
||
_make_fake_viz_dir(vis_dir, "a", 1000, 6881, res_suffix='_r0p2')
|
||
|
||
tiles = scan_tiles(vis_dir)
|
||
assert len(tiles) == 2
|
||
resolutions = sorted(t['resolution'] for t in tiles)
|
||
assert resolutions == [0.2, 0.5]
|
||
|
||
|
||
def test_res_suffix_str():
|
||
"""Le suffixe de résolution reflète le nommage du pipeline (miroir)."""
|
||
from lidar_pipeline.index import _res_suffix_str
|
||
assert _res_suffix_str(0.5) == ''
|
||
assert _res_suffix_str(0.2) == '_r0p2'
|
||
|
||
|
||
def test_collect_tile_metadata(tmp_path):
|
||
"""Les métadonnées lisent la méthode DTM et les dates/tailles des viz."""
|
||
import os
|
||
from datetime import datetime
|
||
from lidar_pipeline.index import _collect_tile_metadata
|
||
|
||
basename = "LHD_FXX_1000_6881_PTS_LAMB93_IGN69"
|
||
tile_dir = tmp_path / "visualisations" / basename
|
||
tile_dir.mkdir(parents=True)
|
||
viz_file = tile_dir / f"{basename}_hillshade_multi.webp"
|
||
viz_file.write_bytes(b"fake")
|
||
|
||
dtm_dir = tmp_path / "DTM"
|
||
dtm_dir.mkdir()
|
||
method_file = dtm_dir / f"{basename}_dtm_method.txt"
|
||
method_file.write_text("ign", encoding="utf-8")
|
||
# Dates déterministes : method.txt plus ancien que la viz
|
||
os.utime(method_file, (1600000000, 1600000000))
|
||
os.utime(viz_file, (1700000000, 1700000000))
|
||
fmt = lambda ts: datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M')
|
||
|
||
tile = {
|
||
'basename': basename, 'resolution': 0.5,
|
||
'dir_path': str(tile_dir),
|
||
'viz': {'hillshade_multi': {'filename': viz_file.name, 'ext': 'webp'}},
|
||
}
|
||
meta = _collect_tile_metadata(tile, dtm_dir)
|
||
assert meta['method'] == 'ign'
|
||
assert meta['generated'] == fmt(1600000000)
|
||
assert meta['viz']['hillshade_multi']['size'] == 4
|
||
assert meta['viz']['hillshade_multi']['date'] == fmt(1700000000)
|
||
|
||
|
||
def test_collect_tile_metadata_resolution_suffix(tmp_path):
|
||
"""Une tuile 0,2 m lit son sidecar _dtm_r0p2_method.txt dédié."""
|
||
from lidar_pipeline.index import _collect_tile_metadata
|
||
|
||
basename = "LHD_FXX_1000_6881_PTS_LAMB93_IGN69"
|
||
tile_dir = tmp_path / "visualisations" / (basename + "_r0p2")
|
||
tile_dir.mkdir(parents=True)
|
||
dtm_dir = tmp_path / "DTM"
|
||
dtm_dir.mkdir()
|
||
(dtm_dir / f"{basename}_dtm_r0p2_method.txt").write_text("smrf", encoding="utf-8")
|
||
|
||
tile = {'basename': basename, 'resolution': 0.2,
|
||
'dir_path': str(tile_dir),
|
||
'viz': {}}
|
||
meta = _collect_tile_metadata(tile, dtm_dir)
|
||
assert meta['method'] == 'smrf'
|
||
# La date vient du sidecar (écrit juste après la création du DTM)
|
||
assert meta['generated'] is not None
|
||
assert meta['viz'] == {}
|
||
|
||
|
||
def test_collect_tile_metadata_fallback_date(tmp_path):
|
||
"""Sans sidecar DTM, la date de génération remonte au plus ancien fichier viz."""
|
||
from lidar_pipeline.index import _collect_tile_metadata
|
||
|
||
basename = "LHD_FXX_1000_6881_PTS_LAMB93_IGN69"
|
||
tile_dir = tmp_path / "visualisations" / basename
|
||
tile_dir.mkdir(parents=True)
|
||
f = tile_dir / f"{basename}_svf.webp"
|
||
f.write_bytes(b"x")
|
||
|
||
tile = {'basename': basename, 'resolution': 0.5,
|
||
'dir_path': str(tile_dir),
|
||
'viz': {'svf': {'filename': f.name, 'ext': 'webp'}}}
|
||
meta = _collect_tile_metadata(tile, tmp_path / "DTM")
|
||
assert meta['method'] is None
|
||
assert meta['generated'] is not None
|
||
|
||
|
||
def test_build_index_generates_html(tmp_path):
|
||
"""build_index génère index.html et les vignettes."""
|
||
from lidar_pipeline.index import build_index
|
||
|
||
output_dir = tmp_path / "output"
|
||
vis_dir = output_dir / "visualisations"
|
||
vis_dir.mkdir(parents=True)
|
||
_make_fake_viz_dir(vis_dir, "a", 1000, 6881,
|
||
('aspect', 'slope', 'positive_openness',
|
||
'wavelet', 'hillshade_multi', 'svf'))
|
||
_make_fake_viz_dir(vis_dir, "b", 1001, 6881, ('hillshade_multi',))
|
||
|
||
result = build_index(output_dir)
|
||
assert result is not None
|
||
html_path = Path(result)
|
||
assert html_path.exists()
|
||
assert html_path.name == "index.html"
|
||
|
||
content = html_path.read_text(encoding='utf-8')
|
||
# Vérifie la présence des éléments clés
|
||
assert "Carte LiDAR" in content
|
||
assert "LHD_FXX_1000_6881" in content
|
||
assert "LHD_FXX_1001_6881" in content
|
||
# Vérifie que le JSON intégré est valide
|
||
assert "const TILES" in content
|
||
# Vérifie les assets de l'interface (CSS/JS séparés)
|
||
assets = output_dir / "assets"
|
||
assert (assets / "app.css").read_text(encoding='utf-8').startswith('/*')
|
||
app_js = (assets / "app.js").read_text(encoding='utf-8')
|
||
assert "Couches" in app_js or "layers" in app_js
|
||
# Barre de génération : sélecteur limité aux couches proposées dans le
|
||
# panneau de visualisation (aspect + slope + openness + ondelette,
|
||
# valeurs = noms d'étapes --only). La sélection est envoyée à /api/preview
|
||
# (détection des tuiles incomplètes) comme à /api/generate.
|
||
# Les couches hors panneau (svf, hillshade) n'y figurent pas.
|
||
assert 'id="genViz"' in content
|
||
assert 'value="wavelet"' in content
|
||
assert 'value="aniso_open"' not in content
|
||
assert 'value="pos_open"' in content
|
||
assert 'value="slope"' in content
|
||
assert 'value="svf"' not in content
|
||
assert 'value="hillshade"' not in content
|
||
assert "genViz" in app_js and "viz: selectedGenViz()" in app_js
|
||
assert app_js.count("selectedGenViz()") >= 3 # définition + preview + generate
|
||
# Adoption du dernier run au chargement (redémarrage du serveur) :
|
||
# un run terminé réapparaît dans la file avec ses tuiles, sans doublon.
|
||
assert "Dernière génération" in app_js
|
||
assert "t.started === s.started" in app_js
|
||
# Bouton « Compléter » : visualisations manquantes des dalles présentes
|
||
# dans input/ (scope all_missing envoyé au preview et au generate).
|
||
# « ＋ Zone » et « Compléter » ne vivent que dans l'onglet Génération,
|
||
# jamais dans la barre d'outils flottante de la carte.
|
||
assert 'id="btnComplete"' in content
|
||
map_tools = content.split('id="mapTools"', 1)[1].split('</nav>', 1)[0]
|
||
assert 'btnGenZone' not in map_tools and 'btnComplete' not in map_tools
|
||
tab_gen = content.split('id="tab-gen"', 1)[1].split('id="tab-tile"', 1)[0]
|
||
assert 'id="btnGenZone"' in tab_gen and 'id="btnComplete"' in tab_gen
|
||
assert "all_missing" in app_js and app_js.count("all_missing") >= 2
|
||
# Dessin de zone au 1er plan : pane dédié au-dessus des tuiles (marqueurs
|
||
# Leaflet, pane 600) pour rectangle et cellules fantômes.
|
||
assert "createPane('genDraw')" in app_js
|
||
assert app_js.count("pane: 'genDraw'") >= 3 # rectangle + 2 jeux de cellules
|
||
assert 'assets/app.css' in content
|
||
# Bouton d'actualisation des couches (détection des nouveautés sur disque) :
|
||
# le ↻ resynchronise puis reconstruit la carte (mode webapp distante).
|
||
assert 'id="layerRefresh"' in content
|
||
assert "/api/layers" in app_js and "/api/sync" in app_js
|
||
assert 'assets/app.js' in content
|
||
# Autonomie hors ligne : Leaflet vendorisé localement, aucun CDN
|
||
assert 'assets/vendor/leaflet/leaflet.js' in content
|
||
assert 'assets/vendor/leaflet/leaflet.css' in content
|
||
assert 'unpkg.com' not in content
|
||
assert (assets / 'vendor' / 'leaflet' / 'leaflet.js').is_file()
|
||
# Fin d'un run de génération : synchronisation puis rechargement (le
|
||
# rechargement direct reste en secours si /api/sync est absent).
|
||
assert "syncThenReload" in app_js
|
||
# Tuiles en direct : la carte sonde /api/tiles pendant un run et fusionne
|
||
# les tuiles terminées sans recharger (mode incrémental du pipeline).
|
||
assert "/api/tiles" in app_js and "pollLiveTiles" in app_js
|
||
assert "registerTile" in app_js and "mergeTilesData" in app_js
|
||
# Export des données de carte (servi par /api/tiles)
|
||
data = json.loads((output_dir / "index_tiles.json").read_text(encoding='utf-8'))
|
||
assert data['tiles'] and data['tiles'][0]['corners']
|
||
assert 'aspect' in data['viz_meta']
|
||
assert data['stats']['n_tiles'] == len(data['tiles'])
|
||
# Vérifie les vignettes générées
|
||
thumb_dir = output_dir / "index_thumbs"
|
||
assert thumb_dir.is_dir()
|
||
thumbs = list(thumb_dir.glob("*.jpg"))
|
||
assert len(thumbs) >= 2 # au moins hillshade pour chaque tuile
|
||
|
||
|
||
def test_mid_thumbnails_generated(tmp_path):
|
||
"""Vignette intermédiaire 640 px générée et référencée dans l'index."""
|
||
from lidar_pipeline.index import build_index
|
||
|
||
output_dir = tmp_path / "output"
|
||
vis_dir = output_dir / "visualisations"
|
||
vis_dir.mkdir(parents=True)
|
||
_make_fake_viz_dir(vis_dir, "a", 1000, 6881, ('hillshade_multi',))
|
||
|
||
assert build_index(output_dir) is not None
|
||
data = json.loads((output_dir / "index_tiles.json").read_text(encoding='utf-8'))
|
||
viz = data['tiles'][0]['viz']['hillshade_multi']
|
||
assert 'mid' in viz
|
||
assert (output_dir / viz['mid']).exists()
|
||
assert viz['mid'].endswith("_mid.jpg")
|
||
|
||
|
||
def test_subtiled_viz_covers_generation_choices():
|
||
"""Les viz générables depuis la carte sont sous-tuilées (affichage fluide)."""
|
||
from lidar_pipeline.index import _CARTO_SUBTILED_VIZ
|
||
for key in ('aspect', 'hillshade_multi', 'slope', 'positive_openness',
|
||
'wavelet'):
|
||
assert key in _CARTO_SUBTILED_VIZ
|
||
|
||
|
||
def test_panel_restricted_to_requested_layers():
|
||
"""Le panneau et le sélecteur ne proposent que la base aspect + les 3 couches demandées."""
|
||
from lidar_pipeline.index import PANEL_VIZ, KEYWORD_TO_STEP
|
||
from lidar_pipeline.pipeline import VIZ_STEPS
|
||
assert PANEL_VIZ == ('aspect', 'slope', 'positive_openness', 'wavelet')
|
||
# Chaque couche du panneau correspond à une étape --only valide du pipeline
|
||
steps = {name for name, _ in VIZ_STEPS}
|
||
for key in PANEL_VIZ:
|
||
assert KEYWORD_TO_STEP.get(key, key) in steps
|
||
|
||
|
||
def test_build_index_merges_cross_resolution_viz(tmp_path):
|
||
"""Une couche produite seulement à 0,5 m reste visible sur la dalle 0,2 m.
|
||
|
||
Le dernier run peut être interrompu entre les deux passes de résolution :
|
||
la fusion inter-résolutions doit pointer vignettes et URLs vers le dossier
|
||
d'origine de chaque fichier (dir_name), pas vers dir_path de la dalle
|
||
affichée — sinon les vignettes échouent et la couche disparaît.
|
||
"""
|
||
from lidar_pipeline.index import build_index
|
||
|
||
output_dir = tmp_path / "output"
|
||
vis_dir = output_dir / "visualisations"
|
||
vis_dir.mkdir(parents=True)
|
||
base = "LHD_FXX_1000_6881_PTS_LAMB93_IGN69"
|
||
# Passe 0,5 m : aspect + slope ; passe 0,2 m : aspect seulement
|
||
_make_fake_viz_dir(vis_dir, base, 1000, 6881, ('aspect', 'slope'))
|
||
_make_fake_viz_dir(vis_dir, base, 1000, 6881, ('aspect',), res_suffix='_r0p2')
|
||
|
||
assert build_index(output_dir) is not None
|
||
data = json.loads((output_dir / "index_tiles.json").read_text(encoding='utf-8'))
|
||
# Une seule position affichée (résolution la plus fine), deux couches
|
||
tiles = [t for t in data['tiles'] if (t['col'], t['row']) == (1000, 6881)]
|
||
keys = set()
|
||
for t in tiles:
|
||
keys.update(t['viz'].keys())
|
||
assert {'aspect', 'slope'} <= keys
|
||
# Vignette slope générée + chaque thumb/full existe réellement sur disque
|
||
assert (output_dir / "index_thumbs" / f"{base}_r0p2_slope.jpg").exists()
|
||
for t in tiles:
|
||
for v in t['viz'].values():
|
||
assert (output_dir / v['thumb']).exists(), v['thumb']
|
||
assert (output_dir / v['full']).exists(), v['full']
|
||
|
||
|
||
def test_build_index_regenerates_stale_thumbnails(tmp_path):
|
||
"""Une tuile recalculée (source plus récente) régénère sa vignette."""
|
||
import os
|
||
import time
|
||
import numpy as np
|
||
from PIL import Image as PILImage
|
||
from lidar_pipeline.index import build_index
|
||
|
||
output_dir = tmp_path / "output"
|
||
vis_dir = output_dir / "visualisations"
|
||
vis_dir.mkdir(parents=True)
|
||
tile_dir = _make_fake_viz_dir(vis_dir, "a", 1000, 6881, ('hillshade_multi',))
|
||
|
||
assert build_index(output_dir) is not None
|
||
thumb_path = output_dir / "index_thumbs" / "LHD_FXX_1000_6881_PTS_LAMB93_IGN69_hillshade_multi.jpg"
|
||
assert thumb_path.exists()
|
||
m1 = thumb_path.stat().st_mtime
|
||
|
||
# Recalcul de la tuile : source réécrite avec une mtime plus récente
|
||
src = tile_dir / "LHD_FXX_1000_6881_PTS_LAMB93_IGN69_hillshade_multi.webp"
|
||
arr = np.random.randint(0, 255, (50, 50, 3), dtype=np.uint8)
|
||
PILImage.fromarray(arr).save(str(src), format='WEBP', quality=80)
|
||
os.utime(src, (m1 + 5, m1 + 5))
|
||
|
||
assert build_index(output_dir) is not None
|
||
m2 = thumb_path.stat().st_mtime
|
||
assert m2 > m1 # vignette régénérée
|
||
|
||
# Source non modifiée depuis → pas de régénération inutile
|
||
os.utime(src, (time.time() - 10, time.time() - 10))
|
||
assert build_index(output_dir) is not None
|
||
assert thumb_path.stat().st_mtime == m2
|
||
|
||
|
||
def test_build_subtiles_regenerates_stale_crops(tmp_path):
|
||
"""Une dalle 0,2 m recalculée régénère ses sous-tuiles (par visualisation)."""
|
||
import os
|
||
from lidar_pipeline.index import build_index
|
||
|
||
output_dir = tmp_path / "output"
|
||
vis_dir = output_dir / "visualisations"
|
||
vis_dir.mkdir(parents=True)
|
||
tile_dir = _make_fake_viz_dir(vis_dir, "a", 1000, 6881,
|
||
('hillshade_multi', 'aspect'), res_suffix='_r0p2')
|
||
|
||
assert build_index(output_dir) is not None
|
||
sub_dir = output_dir / "index_subtiles"
|
||
hill_avif = sub_dir / "LHD_FXX_1000_6881_PTS_LAMB93_IGN69_r0p2_hillshade_multi_0_0.avif"
|
||
aspect_avif = sub_dir / "LHD_FXX_1000_6881_PTS_LAMB93_IGN69_r0p2_aspect_0_0.avif"
|
||
assert hill_avif.exists() and aspect_avif.exists()
|
||
hill_mid = sub_dir / "LHD_FXX_1000_6881_PTS_LAMB93_IGN69_r0p2_hillshade_multi_0_0_mid.webp"
|
||
assert hill_mid.exists()
|
||
m_hill_1 = hill_avif.stat().st_mtime
|
||
m_aspect_1 = aspect_avif.stat().st_mtime
|
||
|
||
# Recalcul : seule la source hillshade est plus récente
|
||
src = tile_dir / "LHD_FXX_1000_6881_PTS_LAMB93_IGN69_hillshade_multi.webp"
|
||
os.utime(src, (m_hill_1 + 5, m_hill_1 + 5))
|
||
|
||
assert build_index(output_dir) is not None
|
||
assert hill_avif.stat().st_mtime > m_hill_1 # sous-tuiles hillshade régénérées
|
||
assert aspect_avif.stat().st_mtime == m_aspect_1 # aspect intact
|
||
|
||
|
||
def test_build_index_empty_returns_none(tmp_path):
|
||
"""Aucune tuile → build_index retourne None sans crash."""
|
||
from lidar_pipeline.index import build_index
|
||
|
||
output_dir = tmp_path / "output"
|
||
(output_dir / "visualisations").mkdir(parents=True)
|
||
result = build_index(output_dir)
|
||
assert result is None
|
||
|
||
|
||
def test_build_index_embeds_valid_json(tmp_path):
|
||
"""Le JSON embarqué dans le HTML est valide et contient les tuiles."""
|
||
from lidar_pipeline.index import build_index
|
||
|
||
output_dir = tmp_path / "output"
|
||
vis_dir = output_dir / "visualisations"
|
||
vis_dir.mkdir(parents=True)
|
||
_make_fake_viz_dir(vis_dir, "a", 1000, 6881, ('hillshade_multi',))
|
||
|
||
build_index(output_dir)
|
||
content = (output_dir / "index.html").read_text(encoding='utf-8')
|
||
# Extrait le JSON entre "const TILES = " et la fin de déclaration
|
||
start = content.index("const TILES = ") + len("const TILES = ")
|
||
# Trouve le ; de fin de déclaration
|
||
depth = 0
|
||
end = start
|
||
for i, ch in enumerate(content[start:], start):
|
||
if ch in ('{', '['):
|
||
depth += 1
|
||
elif ch in ('}', ']'):
|
||
depth -= 1
|
||
if depth == 0:
|
||
end = i + 1
|
||
break
|
||
data = json.loads(content[start:end])
|
||
assert len(data) > 0
|
||
assert data[0]['col'] == 1000
|
||
assert data[0]['row'] == 6881
|
||
|
||
|
||
def test_attach_gps_bounds():
|
||
"""attach_gps_bounds ajoute des bounds GPS ordonnées (France métropolitaine)."""
|
||
from lidar_pipeline.index import attach_gps_bounds
|
||
tiles = [{'col': 1000, 'row': 6881}, {'col': 1042, 'row': 6900}]
|
||
attach_gps_bounds(tiles)
|
||
for t in tiles:
|
||
assert 'bounds' in t
|
||
(lat_s, lon_w), (lat_n, lon_e) = t['bounds']
|
||
assert lat_n > lat_s
|
||
assert lon_e > lon_w
|
||
# France métropolitaine
|
||
assert 41 < lat_s < 51
|
||
assert -5 < lon_w < 10
|
||
|
||
|
||
def test_attach_gps_bounds_row_is_north_edge():
|
||
"""Le numéro de ligne du fichier = bord NORD (convention LiDAR HD IGN).
|
||
|
||
Vérifié sur les bounds des DTM : X ∈ [col, col+1] km, Y ∈ [row-1, row] km.
|
||
La régression historique plaçait Y ∈ [row, row+1] (1 km trop au nord).
|
||
"""
|
||
from rasterio.warp import transform as warp_transform
|
||
from lidar_pipeline.index import attach_gps_bounds
|
||
|
||
col, row = 1054, 6882
|
||
tiles = [{'col': col, 'row': row}]
|
||
attach_gps_bounds(tiles)
|
||
corners = tiles[0]['corners']
|
||
|
||
# Référence exacte de la vraie cellule : SW, SE, NE, NW
|
||
xs = [col * 1000, (col + 1) * 1000, (col + 1) * 1000, col * 1000]
|
||
ys = [(row - 1) * 1000, (row - 1) * 1000, row * 1000, row * 1000]
|
||
lons, lats = warp_transform('EPSG:2154', 'EPSG:4326', xs, ys)
|
||
for k in range(4):
|
||
assert abs(corners[k][0] - lats[k]) < 1e-9
|
||
assert abs(corners[k][1] - lons[k]) < 1e-9
|
||
|
||
# L'ancienne convention (row = bord sud) serait décalée d'environ 1 km
|
||
lat_n = max(c[0] for c in corners)
|
||
assert abs(lat_n - max(lats)) < 1e-9 # bord nord = Y = row×1000
|
||
|
||
|
||
|
||
def test_pick_display_viz_prefers_hillshade():
|
||
"""Le choix de viz par défaut privilégie hillshade_multi."""
|
||
from lidar_pipeline.index import _pick_display_viz
|
||
assert _pick_display_viz(['svf', 'hillshade_multi', 'slope']) == 'hillshade_multi'
|
||
assert _pick_display_viz(['svf', 'slope']) == 'svf'
|
||
assert _pick_display_viz(['topo']) == 'topo'
|
||
|
||
|
||
def test_subdivision_k():
|
||
"""0,5 m/px (2000 px) reste entier ; 0,2 m/px (5000 px) est découpé en 2×2."""
|
||
from lidar_pipeline.index import _subdivision_k
|
||
assert _subdivision_k(0.5) == 1
|
||
assert _subdivision_k(0.2) == 2
|
||
assert _subdivision_k(1.0) == 1
|
||
|
||
|
||
def test_subtile_corners_grid():
|
||
"""Les sous-tuiles reconstruisent exactement la grille de la dalle."""
|
||
from lidar_pipeline.index import _subtile_corners
|
||
corners = [[10.0, 2.0], [10.0, 3.0], [11.0, 3.0], [11.0, 2.0]] # SW SE NE NW
|
||
k = 2
|
||
sw_quad = _subtile_corners(corners, 0, 0, k) # quadrant sud-ouest
|
||
ne_quad = _subtile_corners(corners, 1, 1, k) # quadrant nord-est
|
||
# Le quadrant SW partage le coin SW de la dalle
|
||
assert sw_quad[0] == corners[0]
|
||
# Le quadrant NE partage le coin NE de la dalle
|
||
assert ne_quad[2] == corners[2]
|
||
# Le quadrant SW a son coin NE au centre de la dalle
|
||
assert sw_quad[2] == [10.5, 2.5]
|
||
# Adjacence : bord est du SW = bord ouest du SE (0,0)-(1,0)
|
||
se_quad = _subtile_corners(corners, 1, 0, k)
|
||
assert sw_quad[1] == se_quad[0]
|
||
assert sw_quad[2] == se_quad[3]
|
||
|
||
|
||
def test_approx_wgs84_to_l93_roundtrip():
|
||
"""L'approximation affine WGS84→L93 est l'inverse exacte de L93→WGS84."""
|
||
from lidar_pipeline.index import _approx_l93_to_wgs84, _approx_wgs84_to_l93
|
||
for x, y in ((1054000.0, 6882000.0), (700000.0, 6600000.0), (950123.0, 6410456.0)):
|
||
lon, lat = _approx_l93_to_wgs84(x, y)
|
||
x2, y2 = _approx_wgs84_to_l93(lon, lat)
|
||
assert abs(x2 - x) < 1e-6 and abs(y2 - y) < 1e-6
|