Suppression éclairage solaire, GPU accéléré, --file multi, tests unitaires
- Suppression de generate_solar (éclairage solaire) des visualisations - Accélération GPU de hillshade, slope, aspect, curvature, depressions, anomalies, roughness, texture GLCM, flow (sink filling) - Nettoyage mémoire GPU entre visualisations (gpu_cleanup) - Correction OOM texture GLCM: calcul entropie bin par bin au lieu d'un tableau 3D massif sur GPU - Correction bug: xp_minimum_filter manquant dans imports visualizations - Option --file accepte plusieurs noms complets sans extension - run.sh affiche l'aide si appelé sans arguments - Option --test pour exécuter les tests unitaires dans Docker - Filtre ReturnNumber>=1 intégré dans le pipeline PDAL (plus d'erreur SMRF) - 60 tests unitaires: GPU, visualisations, rendering, DTM, pipeline, CLI - Ajout pytest au Dockerfile Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
0
lidar_pipeline/tests/__init__.py
Normal file
0
lidar_pipeline/tests/__init__.py
Normal file
71
lidar_pipeline/tests/conftest.py
Normal file
71
lidar_pipeline/tests/conftest.py
Normal file
@ -0,0 +1,71 @@
|
||||
"""Shared fixtures for LiDAR pipeline tests."""
|
||||
|
||||
import numpy as np
|
||||
import rasterio
|
||||
from rasterio.transform import from_bounds
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tmp_output_dir(tmp_path):
|
||||
"""Create a temporary output directory."""
|
||||
out = tmp_path / "output"
|
||||
out.mkdir()
|
||||
return out
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def synthetic_dem(tmp_path):
|
||||
"""Create a small synthetic DEM GeoTIFF with realistic terrain features.
|
||||
|
||||
Includes:
|
||||
- A gaussian hill (tumulus)
|
||||
- A linear ridge (wall)
|
||||
- A depression (ditch)
|
||||
- Background noise
|
||||
"""
|
||||
size = 200
|
||||
x = np.linspace(0, 1000, size)
|
||||
y = np.linspace(0, 1000, size)
|
||||
X, Y = np.meshgrid(x, y)
|
||||
|
||||
# Base terrain with gentle slope
|
||||
dem = 100.0 + 0.01 * X + 0.005 * Y
|
||||
|
||||
# Gaussian hill (tumulus) at center
|
||||
dem += 5.0 * np.exp(-((X - 500)**2 + (Y - 500)**2) / (2 * 80**2))
|
||||
|
||||
# Linear ridge (wall) running SW-NE
|
||||
dist_to_wall = np.abs((X - 200) * 0.707 + (Y - 300) * 0.707) / np.sqrt(2)
|
||||
dem += 1.5 * np.exp(-dist_to_wall**2 / (2 * 10**2))
|
||||
|
||||
# Depression (ditch)
|
||||
dist_to_ditch = np.abs(X - 800)
|
||||
dem -= 2.0 * np.exp(-dist_to_ditch**2 / (2 * 15**2)) * (Y > 300) * (Y < 700)
|
||||
|
||||
# Small random noise
|
||||
rng = np.random.default_rng(42)
|
||||
dem += rng.normal(0, 0.05, dem.shape)
|
||||
|
||||
# Save as GeoTIFF (Lambert 93)
|
||||
dem_file = tmp_path / "test_dem.tif"
|
||||
transform = from_bounds(660000, 6700000, 661000, 6701000, size, size)
|
||||
|
||||
with rasterio.open(
|
||||
dem_file, 'w',
|
||||
driver='GTiff', height=size, width=size,
|
||||
count=1, dtype='float32',
|
||||
crs='EPSG:2154', transform=transform,
|
||||
compress='lzw'
|
||||
) as dst:
|
||||
dst.write(dem.astype('float32'), 1)
|
||||
|
||||
return dem_file
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def synthetic_dem_data(synthetic_dem):
|
||||
"""Return (array, transform, crs) from the synthetic DEM."""
|
||||
with rasterio.open(synthetic_dem) as src:
|
||||
return src.read(1), src.transform, src.crs
|
||||
86
lidar_pipeline/tests/test_cli.py
Normal file
86
lidar_pipeline/tests/test_cli.py
Normal file
@ -0,0 +1,86 @@
|
||||
"""Tests for CLI argument parsing."""
|
||||
|
||||
import pytest
|
||||
import sys
|
||||
|
||||
|
||||
class TestCLIParsing:
|
||||
def test_default_args(self):
|
||||
"""Default arguments are set correctly."""
|
||||
from lidar_pipeline.cli import main
|
||||
# We can't easily call main() without files, so test argparse directly
|
||||
import argparse
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("input", help="Input directory")
|
||||
parser.add_argument("-o", "--output", default="/data/output")
|
||||
parser.add_argument("-r", "--resolution", type=float, default=0.5)
|
||||
parser.add_argument("-w", "--workers", type=int, default=1)
|
||||
parser.add_argument("-f", "--force", action="store_true")
|
||||
parser.add_argument("--file", nargs="+", type=str, default=None)
|
||||
|
||||
args = parser.parse_args(["./input"])
|
||||
assert args.input == "./input"
|
||||
assert args.output == "/data/output"
|
||||
assert args.resolution == 0.5
|
||||
assert args.workers == 1
|
||||
assert args.force is False
|
||||
assert args.file is None
|
||||
|
||||
def test_file_flag_single(self):
|
||||
import argparse
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--file", nargs="+", type=str, default=None)
|
||||
args = parser.parse_args(["--file", "test_file"])
|
||||
assert args.file == ["test_file"]
|
||||
|
||||
def test_file_flag_multiple(self):
|
||||
import argparse
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--file", nargs="+", type=str, default=None)
|
||||
args = parser.parse_args(["--file", "file1", "file2", "file3"])
|
||||
assert args.file == ["file1", "file2", "file3"]
|
||||
|
||||
def test_force_flag(self):
|
||||
import argparse
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("-f", "--force", action="store_true")
|
||||
args = parser.parse_args(["-f"])
|
||||
assert args.force is True
|
||||
|
||||
def test_resolution_custom(self):
|
||||
import argparse
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("-r", "--resolution", type=float, default=0.5)
|
||||
args = parser.parse_args(["-r", "0.2"])
|
||||
assert args.resolution == 0.2
|
||||
|
||||
|
||||
class TestSetupLogging:
|
||||
def test_default_logging(self):
|
||||
"""Default logging shows messages without timestamps."""
|
||||
import logging
|
||||
from lidar_pipeline.cli import setup_logging
|
||||
logger = setup_logging(verbose=False, debug=False)
|
||||
assert logger.level == logging.INFO
|
||||
assert len(logger.handlers) == 1
|
||||
# Format should not include timestamps
|
||||
fmt = logger.handlers[0].formatter._fmt
|
||||
assert "%(asctime)" not in fmt
|
||||
|
||||
def test_verbose_logging(self):
|
||||
"""Verbose logging includes timestamps."""
|
||||
import logging
|
||||
from lidar_pipeline.cli import setup_logging
|
||||
logger = setup_logging(verbose=True, debug=False)
|
||||
fmt = logger.handlers[0].formatter._fmt
|
||||
assert "%(asctime)" in fmt
|
||||
|
||||
def test_debug_logging(self):
|
||||
"""Debug logging includes file:line info."""
|
||||
import logging
|
||||
from lidar_pipeline.cli import setup_logging
|
||||
logger = setup_logging(verbose=False, debug=True)
|
||||
assert logger.level == logging.DEBUG
|
||||
fmt = logger.handlers[0].formatter._fmt
|
||||
assert "%(filename)" in fmt
|
||||
assert "%(lineno)" in fmt
|
||||
41
lidar_pipeline/tests/test_dtm.py
Normal file
41
lidar_pipeline/tests/test_dtm.py
Normal file
@ -0,0 +1,41 @@
|
||||
"""Tests for DTM module."""
|
||||
|
||||
import json
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class TestSMRFPipeline:
|
||||
def test_pipeline_json_valid(self):
|
||||
"""create_smrf_pipeline produces valid JSON with expected stages."""
|
||||
from lidar_pipeline.dtm import create_smrf_pipeline
|
||||
result = create_smrf_pipeline("/data/input/test.laz", "/data/output/test_ground.las")
|
||||
pipeline = json.loads(result)
|
||||
|
||||
assert "pipeline" in pipeline
|
||||
stages = pipeline["pipeline"]
|
||||
|
||||
# Should have: reader, range filter (ReturnNumber), SMRF, range filter (Classification), writer
|
||||
stage_types = [s.get("type") if isinstance(s, dict) else None for s in stages]
|
||||
|
||||
# First stage is the filename string (reader)
|
||||
assert isinstance(stages[0], str)
|
||||
assert "test.laz" in stages[0]
|
||||
|
||||
# Must contain SMRF filter
|
||||
assert "filters.smrf" in stage_types
|
||||
|
||||
# Must contain ReturnNumber filter
|
||||
range_stages = [s for s in stages if isinstance(s, dict) and s.get("type") == "filters.range"]
|
||||
assert len(range_stages) >= 1
|
||||
# At least one should filter ReturnNumber
|
||||
assert any("ReturnNumber" in str(s.get("limits", "")) for s in range_stages)
|
||||
|
||||
def test_pipeline_output_path(self):
|
||||
"""Pipeline output path is set correctly."""
|
||||
from lidar_pipeline.dtm import create_smrf_pipeline
|
||||
result = create_smrf_pipeline("/input/a.laz", "/output/a_ground.las")
|
||||
pipeline = json.loads(result)
|
||||
# Last stage should be writer with correct output path
|
||||
writer = [s for s in pipeline["pipeline"] if isinstance(s, dict) and s.get("type") == "writers.las"][0]
|
||||
assert writer["filename"] == "/output/a_ground.las"
|
||||
93
lidar_pipeline/tests/test_gpu.py
Normal file
93
lidar_pipeline/tests/test_gpu.py
Normal file
@ -0,0 +1,93 @@
|
||||
"""Tests for GPU helper module."""
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
|
||||
def test_has_gpu_attribute():
|
||||
"""HAS_GPU must be a boolean."""
|
||||
from lidar_pipeline.gpu import HAS_GPU
|
||||
assert isinstance(HAS_GPU, bool)
|
||||
|
||||
|
||||
def test_to_gpu_returns_array():
|
||||
"""to_gpu returns a float64 array with correct values."""
|
||||
from lidar_pipeline.gpu import to_gpu, to_cpu, HAS_GPU
|
||||
arr = np.array([1.0, 2.0, 3.0])
|
||||
result = to_gpu(arr)
|
||||
# On GPU: cupy.ndarray, on CPU: numpy.ndarray
|
||||
assert result.dtype == np.float64
|
||||
# Always bring back to CPU for comparison
|
||||
np.testing.assert_array_equal(to_cpu(result), [1.0, 2.0, 3.0])
|
||||
|
||||
|
||||
def test_to_cpu_noop_numpy():
|
||||
"""to_cpu on a numpy array is a no-op."""
|
||||
from lidar_pipeline.gpu import to_cpu
|
||||
arr = np.array([1.0, 2.0])
|
||||
result = to_cpu(arr)
|
||||
assert result is arr
|
||||
|
||||
|
||||
def test_xp_gaussian_filter():
|
||||
"""xp_gaussian_filter blurs a point source correctly."""
|
||||
from lidar_pipeline.gpu import xp_gaussian_filter
|
||||
arr = np.zeros((50, 50), dtype=np.float64)
|
||||
arr[25, 25] = 1.0
|
||||
result = xp_gaussian_filter(arr, sigma=3)
|
||||
assert result.shape == (50, 50)
|
||||
# Center should still be the highest value
|
||||
center_val = float(np.asarray(result)[25, 25])
|
||||
corner_val = float(np.asarray(result)[0, 0])
|
||||
assert center_val > corner_val
|
||||
assert center_val > 0.01 # Not all energy is lost
|
||||
|
||||
|
||||
def test_xp_uniform_filter_cpu():
|
||||
"""xp_uniform_filter works on CPU arrays."""
|
||||
from lidar_pipeline.gpu import xp_uniform_filter
|
||||
arr = np.ones((50, 50), dtype=np.float64)
|
||||
arr[25, 25] = 100.0
|
||||
result = xp_uniform_filter(arr, size=5)
|
||||
# Mean should be close to 1 everywhere except near center
|
||||
assert result.shape == (50, 50)
|
||||
assert result[0, 0] == pytest.approx(1.0, abs=0.01)
|
||||
|
||||
|
||||
def test_xp_minimum_filter_cpu():
|
||||
"""xp_minimum_filter works on CPU arrays."""
|
||||
from lidar_pipeline.gpu import xp_minimum_filter
|
||||
arr = np.ones((50, 50), dtype=np.float64)
|
||||
arr[25, 25] = 0.0
|
||||
result = xp_minimum_filter(arr, size=3)
|
||||
assert result.shape == (50, 50)
|
||||
# Around the minimum, values should be 0
|
||||
assert result[25, 25] == 0.0
|
||||
assert result[24, 25] == 0.0
|
||||
|
||||
|
||||
def test_log_gpu_status(caplog):
|
||||
"""log_gpu_status emits a log message."""
|
||||
import logging
|
||||
from lidar_pipeline.gpu import log_gpu_status
|
||||
with caplog.at_level(logging.INFO, logger="lidar"):
|
||||
log_gpu_status()
|
||||
assert any("GPU" in r.message or "CPU" in r.message for r in caplog.records)
|
||||
|
||||
|
||||
@pytest.mark.skipif(
|
||||
not pytest.importorskip("cupy", reason="CuPy not available"),
|
||||
reason="Requires GPU + CuPy"
|
||||
)
|
||||
def test_to_gpu_roundtrip():
|
||||
"""to_gpu -> to_cpu preserves data when GPU is available."""
|
||||
import cupy as cp
|
||||
from lidar_pipeline.gpu import to_gpu, to_cpu, HAS_GPU
|
||||
if not HAS_GPU:
|
||||
pytest.skip("No GPU available")
|
||||
arr = np.array([1.0, 2.0, 3.0], dtype=np.float32)
|
||||
gpu_arr = to_gpu(arr)
|
||||
assert isinstance(gpu_arr, cp.ndarray)
|
||||
result = to_cpu(gpu_arr)
|
||||
assert isinstance(result, np.ndarray)
|
||||
np.testing.assert_array_almost_equal(result, [1.0, 2.0, 3.0])
|
||||
79
lidar_pipeline/tests/test_pipeline.py
Normal file
79
lidar_pipeline/tests/test_pipeline.py
Normal file
@ -0,0 +1,79 @@
|
||||
"""Tests for pipeline orchestration."""
|
||||
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class TestVizSteps:
|
||||
def test_viz_steps_not_empty(self):
|
||||
from lidar_pipeline.pipeline import VIZ_STEPS
|
||||
assert len(VIZ_STEPS) > 0
|
||||
|
||||
def test_viz_steps_have_callable_functions(self):
|
||||
from lidar_pipeline.pipeline import VIZ_STEPS
|
||||
for name, func in VIZ_STEPS:
|
||||
assert callable(func), f"VIZ_STEPS entry '{name}' is not callable"
|
||||
|
||||
def test_viz_steps_names_unique(self):
|
||||
from lidar_pipeline.pipeline import VIZ_STEPS
|
||||
names = [name for name, _ in VIZ_STEPS]
|
||||
assert len(names) == len(set(names)), "VIZ_STEPS has duplicate names"
|
||||
|
||||
def test_no_solar_in_viz_steps(self):
|
||||
"""Solar visualization was removed."""
|
||||
from lidar_pipeline.pipeline import VIZ_STEPS
|
||||
names = [name for name, _ in VIZ_STEPS]
|
||||
assert "solar" not in names
|
||||
|
||||
def test_expected_visualization_count(self):
|
||||
"""Should have 19 visualizations (18 terrain + ortho + topo - solar)."""
|
||||
from lidar_pipeline.pipeline import VIZ_STEPS
|
||||
assert len(VIZ_STEPS) == 19
|
||||
|
||||
def test_ortho_and_topo_present(self):
|
||||
from lidar_pipeline.pipeline import VIZ_STEPS
|
||||
names = [name for name, _ in VIZ_STEPS]
|
||||
assert "ortho" in names
|
||||
assert "topo" in names
|
||||
|
||||
|
||||
class TestLidarArchaeoPipeline:
|
||||
def test_init_creates_dirs(self, tmp_path):
|
||||
from lidar_pipeline.pipeline import LidarArchaeoPipeline
|
||||
input_dir = tmp_path / "input"
|
||||
input_dir.mkdir()
|
||||
output_dir = tmp_path / "output"
|
||||
|
||||
pipeline = LidarArchaeoPipeline(str(input_dir), str(output_dir))
|
||||
assert (tmp_path / "output").exists()
|
||||
assert (tmp_path / "output" / "DTM").exists()
|
||||
assert (tmp_path / "output" / "visualisations").exists()
|
||||
assert (tmp_path / "output" / "rapports").exists()
|
||||
|
||||
def test_init_raises_on_missing_input(self, tmp_path):
|
||||
from lidar_pipeline.pipeline import LidarArchaeoPipeline
|
||||
with pytest.raises(ValueError, match="introuvable"):
|
||||
LidarArchaeoPipeline("/nonexistent/path", str(tmp_path / "output"))
|
||||
|
||||
def test_find_laz_files_empty(self, tmp_path):
|
||||
from lidar_pipeline.pipeline import LidarArchaeoPipeline
|
||||
input_dir = tmp_path / "input"
|
||||
input_dir.mkdir()
|
||||
pipeline = LidarArchaeoPipeline(str(input_dir), str(tmp_path / "output"))
|
||||
files = pipeline.find_laz_files()
|
||||
assert files == []
|
||||
|
||||
def test_find_laz_files(self, tmp_path):
|
||||
from lidar_pipeline.pipeline import LidarArchaeoPipeline
|
||||
input_dir = tmp_path / "input"
|
||||
input_dir.mkdir()
|
||||
(input_dir / "test.laz").touch()
|
||||
(input_dir / "other.las").touch()
|
||||
(input_dir / "readme.txt").touch()
|
||||
|
||||
pipeline = LidarArchaeoPipeline(str(input_dir), str(tmp_path / "output"))
|
||||
files = pipeline.find_laz_files()
|
||||
names = [f.name for f in files]
|
||||
assert "test.laz" in names
|
||||
assert "other.las" in names
|
||||
assert "readme.txt" not in names
|
||||
98
lidar_pipeline/tests/test_rendering.py
Normal file
98
lidar_pipeline/tests/test_rendering.py
Normal file
@ -0,0 +1,98 @@
|
||||
"""Tests for rendering module (colormaps, tif_to_png)."""
|
||||
|
||||
import numpy as np
|
||||
import rasterio
|
||||
from rasterio.transform import from_bounds
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _make_test_tif(tmp_path, data=None, size=50):
|
||||
"""Create a small test GeoTIFF and return its path."""
|
||||
if data is None:
|
||||
rng = np.random.default_rng(42)
|
||||
data = rng.normal(0, 1, (size, size)).astype(np.float32)
|
||||
|
||||
transform = from_bounds(660000, 6700000, 661000, 6701000, size, size)
|
||||
tif_file = tmp_path / "test_vis.tif"
|
||||
with rasterio.open(
|
||||
tif_file, 'w', driver='GTiff', height=size, width=size,
|
||||
count=1, dtype='float32', crs='EPSG:2154', transform=transform,
|
||||
compress='lzw'
|
||||
) as dst:
|
||||
dst.write(data, 1)
|
||||
return tif_file
|
||||
|
||||
|
||||
class TestColormaps:
|
||||
def test_colormaps_dict_exists(self):
|
||||
from lidar_pipeline.rendering import COLORMAPS
|
||||
assert isinstance(COLORMAPS, dict)
|
||||
|
||||
def test_all_viz_steps_have_colormaps(self):
|
||||
"""Every VIZ_STEPS entry should have a corresponding COLORMAPS entry or render correctly."""
|
||||
from lidar_pipeline.pipeline import VIZ_STEPS
|
||||
from lidar_pipeline.rendering import COLORMAPS
|
||||
# Some viz names differ from colormap keys
|
||||
name_map = {
|
||||
'pos_open': 'positive_openness',
|
||||
'neg_open': 'negative_openness',
|
||||
}
|
||||
# IGN overlays (ortho, topo) are RGB images — no colormap needed
|
||||
skip = {'ortho', 'topo'}
|
||||
for name, _ in VIZ_STEPS:
|
||||
if name in skip:
|
||||
continue
|
||||
cmap_key = name_map.get(name, name)
|
||||
assert cmap_key in COLORMAPS, f"Missing colormap for: {name} (looked as {cmap_key})"
|
||||
|
||||
def test_colormap_has_required_keys(self):
|
||||
"""Each colormap entry must have cmap, title, legend, description."""
|
||||
from lidar_pipeline.rendering import COLORMAPS
|
||||
required = {'cmap', 'title', 'legend', 'description'}
|
||||
for name, entry in COLORMAPS.items():
|
||||
missing = required - set(entry.keys())
|
||||
assert not missing, f"Colormap '{name}' missing keys: {missing}"
|
||||
|
||||
|
||||
class TestTifToPng:
|
||||
def test_converts_tif_to_webp(self, tmp_path):
|
||||
from lidar_pipeline.rendering import tif_to_png
|
||||
tif_file = _make_test_tif(tmp_path)
|
||||
result = tif_to_png(tif_file, tmp_path, 5.0)
|
||||
assert result is not None
|
||||
assert result.exists()
|
||||
assert result.suffix == '.webp'
|
||||
|
||||
def test_removes_source_tif(self, tmp_path):
|
||||
from lidar_pipeline.rendering import tif_to_png
|
||||
tif_file = _make_test_tif(tmp_path)
|
||||
assert tif_file.exists()
|
||||
tif_to_png(tif_file, tmp_path, 5.0)
|
||||
assert not tif_file.exists(), "Source TIF should be deleted after conversion"
|
||||
|
||||
def test_webp_has_content(self, tmp_path):
|
||||
from lidar_pipeline.rendering import tif_to_png
|
||||
tif_file = _make_test_tif(tmp_path)
|
||||
result = tif_to_png(tif_file, tmp_path, 5.0)
|
||||
assert result.stat().st_size > 1000 # Must be a real image
|
||||
|
||||
|
||||
class TestApplyColormap:
|
||||
def test_symmetric_mode(self, tmp_path):
|
||||
from lidar_pipeline.rendering import COLORMAPS, tif_to_png
|
||||
# LRM uses symmetric mode
|
||||
data = np.random.default_rng(42).normal(0, 0.5, (50, 50)).astype(np.float32)
|
||||
tif_file = _make_test_tif(tmp_path, data)
|
||||
result = tif_to_png(tif_file, tmp_path, 5.0)
|
||||
assert result is not None
|
||||
assert result.exists()
|
||||
|
||||
def test_percentile_mode(self, tmp_path):
|
||||
from lidar_pipeline.rendering import tif_to_png
|
||||
# Most visualizations use percentile mode
|
||||
data = np.random.default_rng(42).normal(50, 10, (50, 50)).astype(np.float32)
|
||||
tif_file = _make_test_tif(tmp_path, data)
|
||||
result = tif_to_png(tif_file, tmp_path, 5.0)
|
||||
assert result is not None
|
||||
assert result.exists()
|
||||
216
lidar_pipeline/tests/test_visualizations.py
Normal file
216
lidar_pipeline/tests/test_visualizations.py
Normal file
@ -0,0 +1,216 @@
|
||||
"""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
|
||||
|
||||
|
||||
class TestAspect:
|
||||
def test_generates_tif(self, synthetic_dem, tmp_output_dir):
|
||||
from lidar_pipeline.visualizations import generate_aspect
|
||||
result = generate_aspect(synthetic_dem, "test", tmp_output_dir, 5.0)
|
||||
assert result is not None
|
||||
assert result.exists()
|
||||
|
||||
def test_aspect_values_0_360(self, synthetic_dem, tmp_output_dir):
|
||||
import rasterio
|
||||
from lidar_pipeline.visualizations import generate_aspect
|
||||
result = generate_aspect(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) <= 360
|
||||
|
||||
|
||||
class TestCurvature:
|
||||
def test_generates_tif(self, synthetic_dem, tmp_output_dir):
|
||||
from lidar_pipeline.visualizations import generate_curvature
|
||||
result = generate_curvature(synthetic_dem, "test", tmp_output_dir, 5.0)
|
||||
assert result is not None
|
||||
assert result.exists()
|
||||
|
||||
|
||||
# --- GPU-accelerated visualizations ---
|
||||
|
||||
class TestLRM:
|
||||
def test_generates_tif(self, synthetic_dem, tmp_output_dir):
|
||||
from lidar_pipeline.visualizations import generate_lrm
|
||||
result = generate_lrm(synthetic_dem, "test", tmp_output_dir, 5.0)
|
||||
assert result is not None
|
||||
assert result.exists()
|
||||
|
||||
def test_lrm_has_positive_negative(self, synthetic_dem, tmp_output_dir):
|
||||
import rasterio
|
||||
from lidar_pipeline.visualizations import generate_lrm
|
||||
result = generate_lrm(synthetic_dem, "test", tmp_output_dir, 5.0)
|
||||
with rasterio.open(result) as src:
|
||||
data = src.read(1)
|
||||
# LRM should have both positive and negative values
|
||||
assert np.nanmax(data) > 0
|
||||
assert np.nanmin(data) < 0
|
||||
|
||||
|
||||
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 TestTPI:
|
||||
def test_generates_tif(self, synthetic_dem, tmp_output_dir):
|
||||
from lidar_pipeline.visualizations import generate_tpi
|
||||
result = generate_tpi(synthetic_dem, "test", tmp_output_dir, 5.0)
|
||||
assert result is not None
|
||||
assert result.exists()
|
||||
|
||||
|
||||
class TestDepressions:
|
||||
def test_generates_tif(self, synthetic_dem, tmp_output_dir):
|
||||
from lidar_pipeline.visualizations import generate_depressions
|
||||
result = generate_depressions(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 TestAnomalies:
|
||||
def test_generates_tif(self, synthetic_dem, tmp_output_dir):
|
||||
from lidar_pipeline.visualizations import generate_anomalies
|
||||
result = generate_anomalies(synthetic_dem, "test", tmp_output_dir, 5.0)
|
||||
assert result is not None
|
||||
assert result.exists()
|
||||
|
||||
|
||||
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 TestTexture:
|
||||
def test_generates_tif(self, synthetic_dem, tmp_output_dir):
|
||||
from lidar_pipeline.visualizations import generate_texture
|
||||
result = generate_texture(synthetic_dem, "test", tmp_output_dir, 5.0)
|
||||
assert result is not None
|
||||
assert result.exists()
|
||||
|
||||
|
||||
class TestFlow:
|
||||
def test_generates_tif(self, synthetic_dem, tmp_output_dir):
|
||||
from lidar_pipeline.visualizations import generate_flow
|
||||
result = generate_flow(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
|
||||
result = generate_flow(synthetic_dem, "test", tmp_output_dir, 5.0)
|
||||
with rasterio.open(result) as src:
|
||||
data = src.read(1)
|
||||
# log1p(x) >= 0 for x >= 0
|
||||
valid = data[~np.isnan(data)]
|
||||
assert np.nanmin(valid) >= 0
|
||||
Reference in New Issue
Block a user