e0173e0b39
Les sous-tuiles voisines ne partageaient pas leurs bords (interpolation bilinéaire évaluée deux fois sur des fractions différentes), créant une ligne claire entre images au zoom moyen. Les points de bord sont désormais calculés une fois par dalle et partagés entre sous-tuiles adjacentes. Le cadre blanc autour des tuiles au dézoom venait du outline CSS dessiné par-dessus les bords d'images non jointives : supprimé (le survol garde son halo). La bascule overview ↔ vignettes oscillait d'un cran de zoom, d'où une hystérésis. Les seuils de paliers sont revus pour ne jamais étirer la vignette intermédiaire au-delà de 1:1.
861 lines
36 KiB
Python
861 lines
36 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 restreint au panneau (PANEL_VIZ =
|
||
# DEFAULT_LAYERS ; valeurs = noms d'étapes --only). Les couches hors
|
||
# panneau (svf, hillshade, wavelet, roughness) ne sont plus proposées.
|
||
assert 'id="genViz"' in content
|
||
assert 'value="slope"' in content
|
||
assert 'value="pos_open"' in content
|
||
assert 'value="aspect"' in content
|
||
assert 'value="roughness"' not in content
|
||
assert 'value="wavelet"' not in content
|
||
assert 'value="svf"' not in content
|
||
assert 'value="hillshade"' not in content
|
||
assert 'value="aniso_open"' 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
|
||
# Vidage de la file : les runs retirés sont mémorisés (par horodatage
|
||
# serveur) pour que le rechargement ne réadopte pas la « dernière
|
||
# génération » du serveur et ne refasse pas réapparaître les tâches finies.
|
||
assert "dismissedRuns" in app_js
|
||
assert "saveDismissed()" 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
|
||
# Préconfigurations de couches (onglet Couches) : le sélecteur applique
|
||
# des presets intégrés (couches visibles, niveaux de transparence —
|
||
# visualisations + fond de carte OSM —, ordre de pile) et 💾/✕ gèrent
|
||
# les presets personnels persistés en localStorage.
|
||
assert 'id="presetSelect"' in content
|
||
assert 'id="presetSave"' in content
|
||
assert "BUILTIN_PRESETS" in app_js and "applyPreset" in app_js
|
||
assert "presetDetail" in app_js and "lidarPresets_v1" in app_js
|
||
assert "osm: { on: true, opacity: 0.3, dark: true }" in app_js # preset porte le fond OSM
|
||
assert "p.osm" in app_js # l'application y inclut le fond
|
||
# Partage de la vue : bouton « Partager » + lien contenant vue, couches
|
||
# et transparences (hash #zoom,lat,lng&L=...&O=...) — le destinataire
|
||
# reçoit l'état exact, indépendamment de son localStorage local.
|
||
assert 'id="btnShare"' in content
|
||
assert "buildShareHash" in app_js and "parseSharedHash" 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_write_ui_assets(tmp_path):
|
||
"""_write_ui_assets écrit app.js/app.css dans le dossier cible (bâchés image)."""
|
||
from lidar_pipeline.index import _write_ui_assets, _APP_JS
|
||
|
||
dst = tmp_path / "ui"
|
||
_write_ui_assets(dst)
|
||
js = (dst / "app.js").read_text(encoding="utf-8")
|
||
assert js == _APP_JS
|
||
assert (dst / "app.css").read_text(encoding="utf-8").startswith("/*")
|
||
|
||
|
||
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
|
||
# L'URL porte un suffixe ?v= (invalidation cache) : chemin sans lui
|
||
assert (output_dir / viz['mid'].split('?')[0]).exists()
|
||
assert viz['mid'].split('?')[0].endswith("_mid.jpg")
|
||
|
||
|
||
def test_subtiles_cover_all_layers(tmp_path):
|
||
"""Toutes les couches d'une dalle 0,2 m sont sous-tuilées (plus de liste courte)."""
|
||
from lidar_pipeline.index import _CARTO_SUBTILED_VIZ, build_index
|
||
|
||
assert _CARTO_SUBTILED_VIZ == ()
|
||
|
||
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', 'svf', 'topo'), res_suffix='_r0p2')
|
||
|
||
assert build_index(output_dir) is not None
|
||
sub_dir = output_dir / "index_subtiles"
|
||
for viz in ('hillshade_multi', 'svf', 'topo'):
|
||
avif = sub_dir / f"LHD_FXX_1000_6881_PTS_LAMB93_IGN69_r0p2_{viz}_0_0.avif"
|
||
assert avif.exists(), viz
|
||
# L'index référence les sous-tuiles pour chaque couche, pas le repli dalle
|
||
data = json.loads((output_dir / "index_tiles.json").read_text(encoding='utf-8'))
|
||
subs = [t for t in data['tiles'] if t.get('sub_k') == 2]
|
||
assert len(subs) == 4
|
||
for t in subs:
|
||
for viz in ('hillshade_multi', 'svf', 'topo'):
|
||
assert t['viz'][viz]['full'].startswith("index_subtiles/"), viz
|
||
|
||
|
||
def test_panel_restricted_to_kept_layers():
|
||
"""Le panneau est restreint aux couches conservées (PANEL_VIZ).
|
||
|
||
Les couches allumées par défaut (DEFAULT_LAYERS) y figurent toutes — le
|
||
panneau peut en proposer davantage (ex. pente éteinte par défaut), et
|
||
chaque opacité par défaut vise une couche allumée.
|
||
"""
|
||
from lidar_pipeline.index import (PANEL_VIZ, DEFAULT_LAYERS, DEFAULT_OPACITY,
|
||
KEYWORD_TO_STEP)
|
||
from lidar_pipeline.pipeline import VIZ_STEPS
|
||
assert set(DEFAULT_LAYERS) <= set(PANEL_VIZ)
|
||
assert set(DEFAULT_OPACITY) <= set(DEFAULT_LAYERS)
|
||
# Chaque couche conservée correspond à une étape --only valide
|
||
steps = {name for name, _ in VIZ_STEPS}
|
||
for key in PANEL_VIZ:
|
||
assert KEYWORD_TO_STEP.get(key, key) in steps
|
||
|
||
|
||
def test_panel_selection_lists_kept_layers():
|
||
"""Le panneau propose uniquement les couches conservées (PANEL_VIZ).
|
||
|
||
Même sans données sur disque, le panneau affiche les couches conservées ;
|
||
une couche absente du panneau n'est plus proposée, même présente sur
|
||
disque.
|
||
"""
|
||
from lidar_pipeline.index import _panel_selection, PANEL_VIZ
|
||
ordered, default = _panel_selection({'aspect'})
|
||
assert set(ordered) == set(PANEL_VIZ)
|
||
assert default == 'aspect'
|
||
ordered, default = _panel_selection(set())
|
||
assert set(ordered) == set(PANEL_VIZ)
|
||
assert default # toujours une couche par défaut
|
||
# Une couche présente sur disque mais hors panneau est exclue
|
||
ordered, default = _panel_selection({'aniso_open'})
|
||
assert 'aniso_open' not in ordered
|
||
|
||
|
||
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'].split('?')[0]).exists(), v['thumb']
|
||
assert (output_dir / v['full'].split('?')[0]).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_urls_versioned_for_cache_busting(tmp_path):
|
||
"""Les URLs images changent quand une tuile est régénérée.
|
||
|
||
Les URLs versionnées (?v=) sont servies avec un cache immutable : un
|
||
recalcul DOIT changer l'URL pour forcer le rechargement, y compris en
|
||
direct pendant un run (--incremental-index). Chaque URL porte la mtime
|
||
de SON fichier (cf. test_urls_versioned_by_served_file_mtime).
|
||
"""
|
||
import os
|
||
import re
|
||
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
|
||
|
||
def read_urls():
|
||
data = json.loads((output_dir / "index_tiles.json").read_text(encoding='utf-8'))
|
||
viz = data['tiles'][0]['viz']['hillshade_multi']
|
||
return viz['thumb'], viz['mid'], viz['full']
|
||
|
||
thumb1, mid1, full1 = read_urls()
|
||
for url in (thumb1, mid1, full1):
|
||
assert re.search(r"\?v=\d+$", url), url # versionnée (mtime ms)
|
||
assert (output_dir / url.split('?')[0]).exists(), url
|
||
|
||
# Même source, rebuild → URLs identiques (cache navigateur conservé)
|
||
assert build_index(output_dir) is not None
|
||
assert read_urls() == (thumb1, mid1, full1)
|
||
|
||
# Régénération : mtime de la source plus récente → nouvelles URLs
|
||
src = tile_dir / "LHD_FXX_1000_6881_PTS_LAMB93_IGN69_hillshade_multi.webp"
|
||
m = src.stat().st_mtime
|
||
os.utime(src, (m + 10, m + 10))
|
||
assert build_index(output_dir) is not None
|
||
thumb2, mid2, full2 = read_urls()
|
||
assert (thumb2, mid2, full2) != (thumb1, mid1, full1)
|
||
assert thumb2.split('?')[0] == thumb1.split('?')[0] # même fichier, version neuve
|
||
|
||
|
||
def test_subtile_urls_versioned(tmp_path):
|
||
"""Les URLs des sous-tuiles 0,2 m sont aussi versionnées (?v=)."""
|
||
import re
|
||
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',),
|
||
res_suffix='_r0p2')
|
||
|
||
assert build_index(output_dir) is not None
|
||
data = json.loads((output_dir / "index_tiles.json").read_text(encoding='utf-8'))
|
||
subs = [t for t in data['tiles'] if t.get('sub_k') == 2]
|
||
assert subs
|
||
for t in subs:
|
||
for v in t['viz'].values():
|
||
for url in (v['thumb'], v['mid'], v['full']):
|
||
assert re.search(r"\?v=\d+$", url), url
|
||
assert (output_dir / url.split('?')[0]).exists(), url
|
||
|
||
|
||
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
|
||
|
||
def test_overview_generated_for_large_grids(tmp_path):
|
||
"""Grille >= 6 dalles : une mosaïque d'overview par couche + le payload."""
|
||
from lidar_pipeline.index import build_index
|
||
|
||
output_dir = tmp_path / "output"
|
||
vis_dir = output_dir / "visualisations"
|
||
vis_dir.mkdir(parents=True)
|
||
for col in (1000, 1001, 1002):
|
||
for row in (6881, 6882):
|
||
_make_fake_viz_dir(vis_dir, "t", col, row, ('hillshade_multi', 'svf'))
|
||
|
||
assert build_index(output_dir) is not None
|
||
data = json.loads((output_dir / "index_tiles.json").read_text(encoding='utf-8'))
|
||
ov = data['overview']
|
||
assert 'hillshade_multi' in ov and 'svf' in ov
|
||
items = ov['hillshade_multi']['items']
|
||
assert len(items) == 1 # une seule zone géographique
|
||
meta = items[0]
|
||
assert (output_dir / meta['url'].split('?')[0]).is_file()
|
||
assert meta['km_w'] == 3 and meta['km_h'] == 2
|
||
assert meta['w'] == 3 * meta['px_per_km']
|
||
assert meta['h'] == 2 * meta['px_per_km']
|
||
assert len(meta['corners']) == 4
|
||
# La coquille HTML embarque la constante pour l'interface
|
||
html = (output_dir / "index.html").read_text(encoding='utf-8')
|
||
assert "const OVERVIEW" in html
|
||
assert "index_overview/hillshade_multi_" in html
|
||
|
||
|
||
def test_overview_one_mosaic_per_zone(tmp_path):
|
||
"""Zones géographiques distantes : une mosaïque PAR ZONE (pas une grille
|
||
globale étirée sur des centaines de km aux coins introuvables)."""
|
||
from lidar_pipeline.index import build_index
|
||
|
||
output_dir = tmp_path / "output"
|
||
vis_dir = output_dir / "visualisations"
|
||
vis_dir.mkdir(parents=True)
|
||
# Deux clusters de 2×3 dalles, à ~400 km l'un de l'autre
|
||
for col in (1000, 1001, 1002):
|
||
for row in (6881, 6882):
|
||
_make_fake_viz_dir(vis_dir, "a", col, row, ('hillshade_multi',))
|
||
for col in (1400, 1401, 1402):
|
||
for row in (6481, 6482):
|
||
_make_fake_viz_dir(vis_dir, "b", col, row, ('hillshade_multi',))
|
||
|
||
assert build_index(output_dir) is not None
|
||
data = json.loads((output_dir / "index_tiles.json").read_text(encoding='utf-8'))
|
||
items = data['overview']['hillshade_multi']['items']
|
||
assert len(items) == 2
|
||
assert items[0]['km_w'] == 3 and items[0]['km_h'] == 2
|
||
assert items[1]['km_w'] == 3 and items[1]['km_h'] == 2
|
||
# Les deux mosaïques sont distinctes (coins différents)
|
||
assert items[0]['corners'] != items[1]['corners']
|
||
|
||
|
||
def test_overview_skipped_for_small_grids(tmp_path):
|
||
"""Grille < 6 dalles : pas d'overview, les vignettes suffisent."""
|
||
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',))
|
||
_make_fake_viz_dir(vis_dir, "b", 1001, 6881, ('hillshade_multi',))
|
||
|
||
assert build_index(output_dir) is not None
|
||
data = json.loads((output_dir / "index_tiles.json").read_text(encoding='utf-8'))
|
||
assert data['overview'] == {}
|
||
|
||
|
||
def test_subtile_corners_shared_edges_exact():
|
||
"""Bords partagés : deux sous-tuiles voisines reçoivent le MÊME point.
|
||
|
||
Sans partage, l'interpolation bilinéaire est évaluée deux fois sur des
|
||
fractions différentes et les bords ne coïncident plus à moins d'un pixel
|
||
— jointure cassée (ligne blanche) au zoom moyen.
|
||
"""
|
||
from lidar_pipeline.index import _subtile_corners
|
||
corners = [[10.0, 2.0], [10.0, 3.0], [11.0, 3.0], [11.0, 2.0]]
|
||
k = 4
|
||
a = _subtile_corners(corners, 0, 0, k)
|
||
b = _subtile_corners(corners, 1, 0, k)
|
||
assert a[1] == b[0] # bord est de (0,0) = bord ouest de (1,0)
|
||
assert a[2] == b[3]
|
||
c = _subtile_corners(corners, 0, 1, k)
|
||
assert a[3] == c[0] # bord nord de (0,0) = bord sud de (0,1)
|
||
d = _subtile_corners(corners, 3, 3, k)
|
||
assert d[2] == corners[2] # coin NE de la dalle partagé tel quel
|
||
|
||
|
||
def test_overview_corners_shared_between_tiles(tmp_path):
|
||
"""Le coin d'une dalle est aussi celui de sa voisine (même point).
|
||
|
||
Deux dalles adjacentes E-O partagent un bord vertical : le NE de la dalle
|
||
ouest = le NW de la dalle est. Sans partage, les interpolations bilinéaires
|
||
des sous-tuiles voisines ne coïncident plus à moins d'un pixel — jointure
|
||
cassée au zoom moyen.
|
||
"""
|
||
from lidar_pipeline.index import build_index
|
||
|
||
output_dir = tmp_path / "output"
|
||
vis_dir = output_dir / "visualisations"
|
||
vis_dir.mkdir(parents=True)
|
||
for col in (1000, 1001):
|
||
_make_fake_viz_dir(vis_dir, "a", col, 6881, ('hillshade_multi',))
|
||
|
||
assert build_index(output_dir) is not None
|
||
data = json.loads((output_dir / "index_tiles.json").read_text(encoding='utf-8'))
|
||
tiles = {t['col']: t for t in data['tiles']}
|
||
a_ne = tiles[1000]['corners'][2] # NE de la dalle ouest
|
||
b_nw = tiles[1001]['corners'][3] # NW de la dalle est
|
||
assert a_ne == b_nw, f"coins partagés divergents : {a_ne} != {b_nw}"
|
||
|
||
|
||
def test_urls_versioned_by_served_file_mtime(tmp_path):
|
||
"""?v= suit la mtime du fichier SERVI — prérequis du cache immutable.
|
||
|
||
Une vignette recalculée après coup (source inchangée) change de contenu :
|
||
son URL doit changer aussi, sans toucher celles des fichiers intacts.
|
||
"""
|
||
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',))
|
||
# 0,2 m : la dalle passe par le sous-tuilage (vignettes/mid/AVIF propres)
|
||
_make_fake_viz_dir(vis_dir, "b", 1001, 6881, ('hillshade_multi',),
|
||
res_suffix='_r0p2')
|
||
|
||
assert build_index(output_dir) is not None
|
||
data = json.loads((output_dir / "index_tiles.json").read_text(encoding='utf-8'))
|
||
import time
|
||
for t in data['tiles']:
|
||
for viz in t['viz'].values():
|
||
for tier in ('thumb', 'mid', 'full'):
|
||
url = viz.get(tier)
|
||
if not url or '?v=' not in url:
|
||
continue
|
||
path, v = url.split('?v=')
|
||
assert v.isdigit()
|
||
# La version est la mtime ms du fichier servi lui-même
|
||
# (tolérance 1 s : granularité du système de fichiers).
|
||
assert abs(int(v) - int((output_dir / path).stat().st_mtime * 1000)) < 1000
|
||
|
||
# Vignette 0,5 m recalculée après coup : seule SON URL change.
|
||
tiles05 = [t for t in data['tiles'] if t.get('resolution') == 0.5]
|
||
url_before = tiles05[0]['viz']['hillshade_multi']['thumb']
|
||
thumb_path = output_dir / url_before.split('?')[0]
|
||
time.sleep(0.05)
|
||
from PIL import Image as PILImage
|
||
PILImage.new('RGB', (30, 30), (200, 10, 10)).save(str(thumb_path), format='JPEG')
|
||
assert build_index(output_dir) is not None
|
||
data2 = json.loads((output_dir / "index_tiles.json").read_text(encoding='utf-8'))
|
||
tiles05b = [t for t in data2['tiles'] if t.get('resolution') == 0.5]
|
||
url_after = tiles05b[0]['viz']['hillshade_multi']['thumb']
|
||
assert url_after.split('?')[0] == url_before.split('?')[0]
|
||
assert url_after != url_before
|