ed3e90ea89
- Web map: multi-select picker in the generation bar (aspect, slope, positive openness, anisotropic openness, wavelet) passed to the API; the layer panel is restricted to the same shortlist (PANEL_VIZ) and a refresh button rebuilds the index when new layers appear on disk; jobs started outside the UI are now adopted into the visible queue - Uniform colors across tiles: openness/anisotropic/sailore now store local z-scores, and all renderers use fixed ranges (0-3 sigma, SVF 0-1 physical, slope 0-30 deg) instead of per-tile percentile stretches - Ray-tracing falls back to CPU when VRAM is exhausted so openness and SVF no longer fail silently on shared GPUs - build_index merges visualizations available at only one resolution into the displayed tile so in-progress layers stay visible - 11 new tests (142 passing)
258 lines
10 KiB
Python
258 lines
10 KiB
Python
"""Tests for visualization functions.
|
|
|
|
Each test creates a small synthetic DEM and runs a visualization function,
|
|
checking that it produces a valid output file.
|
|
"""
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from pathlib import Path
|
|
|
|
|
|
# --- Core terrain visualizations (no GPU required) ---
|
|
|
|
class TestHillshade:
|
|
def test_generates_tif(self, synthetic_dem, tmp_output_dir):
|
|
from lidar_pipeline.visualizations import generate_hillshade
|
|
result = generate_hillshade(synthetic_dem, "test", tmp_output_dir, 5.0)
|
|
assert result is not None
|
|
assert result.exists()
|
|
assert result.suffix == ".tif"
|
|
|
|
def test_output_values_valid(self, synthetic_dem, tmp_output_dir):
|
|
import rasterio
|
|
from lidar_pipeline.visualizations import generate_hillshade
|
|
result = generate_hillshade(synthetic_dem, "test", tmp_output_dir, 5.0)
|
|
with rasterio.open(result) as src:
|
|
data = src.read(1)
|
|
assert data.shape[0] > 0
|
|
assert np.nanmin(data) >= 0
|
|
assert np.nanmax(data) <= 1
|
|
|
|
|
|
class TestSlope:
|
|
def test_generates_tif(self, synthetic_dem, tmp_output_dir):
|
|
from lidar_pipeline.visualizations import generate_slope
|
|
result = generate_slope(synthetic_dem, "test", tmp_output_dir, 5.0)
|
|
assert result is not None
|
|
assert result.exists()
|
|
|
|
def test_slope_values_degrees(self, synthetic_dem, tmp_output_dir):
|
|
import rasterio
|
|
from lidar_pipeline.visualizations import generate_slope
|
|
result = generate_slope(synthetic_dem, "test", tmp_output_dir, 5.0)
|
|
with rasterio.open(result) as src:
|
|
data = src.read(1)
|
|
assert np.nanmin(data) >= 0
|
|
assert np.nanmax(data) <= 90
|
|
|
|
|
|
# --- GPU-accelerated visualizations ---
|
|
|
|
class TestSVF:
|
|
def test_generates_tif(self, synthetic_dem, tmp_output_dir):
|
|
from lidar_pipeline.visualizations import generate_svf
|
|
result = generate_svf(synthetic_dem, "test", tmp_output_dir, 5.0)
|
|
assert result is not None
|
|
assert result.exists()
|
|
|
|
def test_svf_values_0_1(self, synthetic_dem, tmp_output_dir):
|
|
import rasterio
|
|
from lidar_pipeline.visualizations import generate_svf
|
|
result = generate_svf(synthetic_dem, "test", tmp_output_dir, 5.0)
|
|
with rasterio.open(result) as src:
|
|
data = src.read(1)
|
|
valid = data[~np.isnan(data)]
|
|
assert np.nanmin(valid) >= 0
|
|
assert np.nanmax(valid) <= 1
|
|
|
|
|
|
class TestOpenness:
|
|
def test_positive_generates_tif(self, synthetic_dem, tmp_output_dir):
|
|
from lidar_pipeline.visualizations import generate_openness
|
|
result = generate_openness(synthetic_dem, "test", tmp_output_dir, 5.0, positive=True)
|
|
assert result is not None
|
|
assert result.exists()
|
|
|
|
def test_negative_generates_tif(self, synthetic_dem, tmp_output_dir):
|
|
from lidar_pipeline.visualizations import generate_openness
|
|
result = generate_openness(synthetic_dem, "test", tmp_output_dir, 5.0, positive=False)
|
|
assert result is not None
|
|
assert result.exists()
|
|
|
|
|
|
class TestMSLRM:
|
|
def test_generates_tif(self, synthetic_dem, tmp_output_dir):
|
|
from lidar_pipeline.visualizations import generate_mslrm
|
|
result = generate_mslrm(synthetic_dem, "test", tmp_output_dir, 5.0)
|
|
assert result is not None
|
|
assert result.exists()
|
|
|
|
|
|
class TestSAILORE:
|
|
def test_generates_tif(self, synthetic_dem, tmp_output_dir):
|
|
from lidar_pipeline.visualizations import generate_sailore
|
|
result = generate_sailore(synthetic_dem, "test", tmp_output_dir, 5.0)
|
|
assert result is not None
|
|
assert result.exists()
|
|
|
|
|
|
class TestRoughness:
|
|
def test_generates_tif(self, synthetic_dem, tmp_output_dir):
|
|
from lidar_pipeline.visualizations import generate_roughness
|
|
result = generate_roughness(synthetic_dem, "test", tmp_output_dir, 5.0)
|
|
assert result is not None
|
|
assert result.exists()
|
|
|
|
def test_roughness_non_negative(self, synthetic_dem, tmp_output_dir):
|
|
import rasterio
|
|
from lidar_pipeline.visualizations import generate_roughness
|
|
result = generate_roughness(synthetic_dem, "test", tmp_output_dir, 5.0)
|
|
with rasterio.open(result) as src:
|
|
data = src.read(1)
|
|
# Standard deviation is always >= 0
|
|
assert np.nanmin(data) >= 0
|
|
|
|
|
|
class TestWavelet:
|
|
def test_generates_tif(self, synthetic_dem, tmp_output_dir):
|
|
from lidar_pipeline.visualizations import generate_wavelet
|
|
result = generate_wavelet(synthetic_dem, "test", tmp_output_dir, 5.0)
|
|
assert result is not None
|
|
assert result.exists()
|
|
|
|
|
|
class TestFlowAccumulation:
|
|
def test_generates_tif(self, synthetic_dem, tmp_output_dir):
|
|
from lidar_pipeline.visualizations import generate_flow_accumulation
|
|
result = generate_flow_accumulation(synthetic_dem, "test", tmp_output_dir, 5.0)
|
|
assert result is not None
|
|
assert result.exists()
|
|
|
|
def test_flow_log_values(self, synthetic_dem, tmp_output_dir):
|
|
import rasterio
|
|
from lidar_pipeline.visualizations import generate_flow_accumulation
|
|
result = generate_flow_accumulation(synthetic_dem, "test", tmp_output_dir, 5.0)
|
|
with rasterio.open(result) as src:
|
|
data = src.read(1)
|
|
# log10(x) >= 0 for x >= 1
|
|
valid = data[~np.isnan(data)]
|
|
assert np.nanmin(valid) >= 0
|
|
|
|
|
|
class TestRayTrace:
|
|
def test_rays_are_traced(self, synthetic_dem, tmp_output_dir):
|
|
"""Verify _ray_trace_horizons returns expected shapes."""
|
|
from lidar_pipeline.visualizations import _ray_trace_horizons, _prepare_dem_for_raycast
|
|
import rasterio
|
|
with rasterio.open(synthetic_dem) as src:
|
|
dem_np = src.read(1)
|
|
rows, cols = dem_np.shape
|
|
# Create a simple filled DEM for testing
|
|
import numpy as np
|
|
filled = np.nan_to_num(dem_np, nan=0)
|
|
# Test with numpy (no GPU)
|
|
pos, neg = _ray_trace_horizons(
|
|
filled, rows, cols, 5.0, n_dirs=4, max_dist=10, radii_m=[25, 50]
|
|
)
|
|
assert pos.shape == (4, 2, rows, cols)
|
|
assert neg.shape == (4, 2, rows, cols)
|
|
|
|
|
|
class TestNodataPreserved:
|
|
"""Nodata préservé dans les rendus (comportement historique).
|
|
|
|
Les trous du MNT sont comblés en amont (create_dtm_fast, tous modes) ;
|
|
si un nodata subsiste malgré tout, hillshade/slope/aspect le restituent
|
|
(rendu noir en carte) au lieu d'inventer des valeurs interpolées.
|
|
"""
|
|
|
|
@staticmethod
|
|
def _dem_with_hole(synthetic_dem, tmp_path):
|
|
import rasterio
|
|
with rasterio.open(synthetic_dem) as src:
|
|
arr = src.read(1).copy()
|
|
profile = src.profile.copy()
|
|
arr[80:120, 80:120] = np.nan
|
|
dem_hole = tmp_path / "dem_hole.tif"
|
|
profile.update(dtype='float32', nodata=float('nan'))
|
|
with rasterio.open(dem_hole, 'w', **profile) as dst:
|
|
dst.write(arr.astype('float32'), 1)
|
|
return dem_hole
|
|
|
|
def test_aspect_solo_preserves_nodata(self, synthetic_dem, tmp_path):
|
|
from lidar_pipeline.visualizations import generate_aspect
|
|
dem_hole = self._dem_with_hole(synthetic_dem, tmp_path)
|
|
out = generate_aspect(dem_hole, "solo", tmp_path, 5.0)
|
|
assert out is not None and out.exists()
|
|
import rasterio
|
|
with rasterio.open(out) as src:
|
|
data = src.read(1)
|
|
assert np.isnan(data[80:120, 80:120]).all(), "le trou doit rester en nodata"
|
|
# Le gradient au bord du trou propage NaN sur un anneau de 1 px :
|
|
# on vérifie une zone éloignée du trou
|
|
assert not np.isnan(data[0:40, 0:40]).any(), "NaN loin du trou"
|
|
|
|
def test_aspect_shared_preserves_nodata(self, synthetic_dem, tmp_path):
|
|
from lidar_pipeline.visualizations import SharedDEM, generate_aspect
|
|
dem_hole = self._dem_with_hole(synthetic_dem, tmp_path)
|
|
shared = SharedDEM(dem_hole, 5.0)
|
|
out = generate_aspect(dem_hole, "partage", tmp_path, 5.0, shared=shared)
|
|
assert out is not None and out.exists()
|
|
import rasterio
|
|
with rasterio.open(out) as src:
|
|
data = src.read(1)
|
|
assert np.isnan(data[80:120, 80:120]).all(), "le trou doit rester en nodata"
|
|
assert not np.isnan(data[0:40, 0:40]).any(), "NaN loin du trou"
|
|
|
|
def test_slope_and_hillshade_preserve_nodata(self, synthetic_dem, tmp_path):
|
|
from lidar_pipeline.visualizations import generate_slope, generate_hillshade
|
|
dem_hole = self._dem_with_hole(synthetic_dem, tmp_path)
|
|
import rasterio
|
|
for gen, name in ((generate_slope, "p"), (generate_hillshade, "h")):
|
|
out = gen(dem_hole, name, tmp_path, 5.0)
|
|
assert out is not None and out.exists()
|
|
with rasterio.open(out) as src:
|
|
data = src.read(1)
|
|
assert np.isnan(data[80:120, 80:120]).any(), f"{out.name} : trou disparu"
|
|
|
|
|
|
def test_ray_trace_horizons_cpu_fallback_on_oom(monkeypatch):
|
|
"""Sur OOM GPU, le ray-tracing désactive le GPU puis recommence sur CPU."""
|
|
import lidar_pipeline.visualizations as viz
|
|
import lidar_pipeline.gpu as gpu_mod
|
|
|
|
calls = {"n": 0}
|
|
disabled = []
|
|
|
|
def fake_core(dem, rows, cols, res, n_dirs, max_dist, radii_m=None):
|
|
calls["n"] += 1
|
|
if calls["n"] == 1:
|
|
raise RuntimeError("Out of memory allocating 600,000,000 bytes")
|
|
return ("pos", "neg")
|
|
|
|
monkeypatch.setattr(viz, "_ray_trace_horizons_core", fake_core)
|
|
monkeypatch.setattr(gpu_mod, "is_gpu_active", lambda: True)
|
|
monkeypatch.setattr(gpu_mod, "disable_gpu", lambda: disabled.append(True))
|
|
result = viz._ray_trace_horizons(None, 4, 4, 0.5, 8, 10)
|
|
assert result == ("pos", "neg")
|
|
assert calls["n"] == 2
|
|
assert disabled == [True]
|
|
|
|
|
|
def test_ray_trace_horizons_reraises_non_oom(monkeypatch):
|
|
"""Une erreur non-OOM n'est pas masquée par le repli CPU."""
|
|
import lidar_pipeline.visualizations as viz
|
|
import lidar_pipeline.gpu as gpu_mod
|
|
|
|
def fake_core(dem, rows, cols, res, n_dirs, max_dist, radii_m=None):
|
|
raise ValueError("autre erreur")
|
|
|
|
monkeypatch.setattr(viz, "_ray_trace_horizons_core", fake_core)
|
|
monkeypatch.setattr(gpu_mod, "is_gpu_active", lambda: True)
|
|
try:
|
|
viz._ray_trace_horizons(None, 4, 4, 0.5, 8, 10)
|
|
assert False, "ValueError attendue"
|
|
except ValueError:
|
|
pass
|