178 lines
7.7 KiB
Python
178 lines
7.7 KiB
Python
"""Tests de la mesure de qualité des dalles (densité sol, date d'acquisition)."""
|
||
|
||
BOUNDS = (652000.0, 6861000.0, 653000.0, 6862000.0)
|
||
|
||
|
||
def _gps_seconds(year, month, day):
|
||
"""Temps GPS ajusté standard (secondes GPS − 1e9) d'un midi UTC."""
|
||
from datetime import datetime, timezone
|
||
epoch = datetime(1980, 1, 6, tzinfo=timezone.utc)
|
||
return (datetime(year, month, day, 12, tzinfo=timezone.utc) - epoch).total_seconds() - 1e9
|
||
|
||
|
||
def test_gps_adjusted_to_date():
|
||
from lidar_pipeline.quality import gps_adjusted_to_date
|
||
assert gps_adjusted_to_date(_gps_seconds(2023, 3, 15)) == "2023-03-15"
|
||
|
||
|
||
def test_compute_quality_density_grid_and_dates():
|
||
import numpy as np
|
||
from lidar_pipeline.quality import compute_quality
|
||
# 1 point par m² sur la moitié nord, rien au sud ; 2 dates de vol.
|
||
xs, ys = np.meshgrid(np.arange(652000.5, 653000.0, 1.0),
|
||
np.arange(6861500.5, 6862000.0, 1.0))
|
||
x, y = xs.ravel(), ys.ravel()
|
||
t = np.where(x < 652500, _gps_seconds(2023, 3, 15), _gps_seconds(2023, 3, 17))
|
||
q = compute_quality(x, y, t, BOUNDS)
|
||
assert q["version"] == 1
|
||
assert abs(q["ground_density"] - 0.5) < 1e-6
|
||
grid = q["density_grid"]
|
||
assert len(grid) == 20 and all(len(r) == 20 for r in grid)
|
||
assert grid[0][0] == 1.0 # ligne 0 = nord : couverte
|
||
assert grid[19][0] == 0.0 # sud : vide
|
||
assert q["acq_start"] == "2023-03-15" and q["acq_end"] == "2023-03-17"
|
||
assert q["acq_source"] == "gps"
|
||
# 1 point par m² : 1 pixel 0,2 m sur 25 occupé au nord, aucun au sud.
|
||
assert abs(q["empty_fraction"] - (1 - 0.5 / 25)) < 1e-3
|
||
|
||
|
||
def test_compute_quality_ignores_points_outside_bounds():
|
||
import numpy as np
|
||
from lidar_pipeline.quality import compute_quality
|
||
x = np.array([652500.0, 660000.0]); y = np.array([6861500.0, 6861500.0])
|
||
q = compute_quality(x, y, None, BOUNDS, header_date="2024-05-03")
|
||
assert abs(q["ground_density"] - 1e-6) < 1e-9
|
||
assert q["acq_start"] == q["acq_end"] == "2024-05-03"
|
||
assert q["acq_source"] == "header"
|
||
|
||
|
||
def test_compute_quality_week_time_falls_back_to_header():
|
||
import numpy as np
|
||
from lidar_pipeline.quality import compute_quality
|
||
x = np.array([652500.0]); y = np.array([6861500.0])
|
||
q = compute_quality(x, y, np.array([345600.0]), BOUNDS, gps_adjusted=False,
|
||
header_date="2024-05-03")
|
||
assert q["acq_source"] == "header" and q["acq_start"] == "2024-05-03"
|
||
|
||
|
||
def test_compute_quality_empty_tile():
|
||
import numpy as np
|
||
from lidar_pipeline.quality import compute_quality
|
||
q = compute_quality(np.array([]), np.array([]), np.array([]), BOUNDS)
|
||
assert q["ground_density"] == 0.0 and q["empty_fraction"] == 1.0
|
||
assert q["acq_start"] is None and q["acq_end"] is None and q["acq_source"] is None
|
||
|
||
|
||
def test_sidecar_roundtrip_and_table(tmp_path):
|
||
from lidar_pipeline.quality import (load_quality_table, quality_path,
|
||
read_quality, write_quality)
|
||
base = "LHD_FXX_0652_6862_PTS_LAMB93_IGN69"
|
||
data = {"version": 1, "ground_density": 4.2}
|
||
assert write_quality(tmp_path, base, data) is True
|
||
assert write_quality(tmp_path, base, data) is False # contenu identique : pas réécrit
|
||
assert quality_path(tmp_path, base).is_file()
|
||
assert read_quality(tmp_path, base) == data
|
||
assert load_quality_table(tmp_path) == {base: data}
|
||
|
||
|
||
def test_read_quality_rejects_other_version(tmp_path):
|
||
from lidar_pipeline.quality import read_quality, write_quality
|
||
base = "LHD_FXX_0652_6862_PTS_LAMB93_IGN69"
|
||
write_quality(tmp_path, base, {"version": 0})
|
||
assert read_quality(tmp_path, base) is None
|
||
|
||
|
||
def test_quality_module_imports_without_numpy():
|
||
"""L'image légère n'a pas numpy : le module ne doit pas l'importer au chargement."""
|
||
import subprocess, sys
|
||
code = ("import sys; sys.modules['numpy'] = None; "
|
||
"import lidar_pipeline.quality as q; print(q.QUALITY_VERSION)")
|
||
r = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True)
|
||
assert r.returncode == 0, r.stderr
|
||
|
||
|
||
def _write_las(path, x, y, cls, t, adjusted=True, creation=None):
|
||
"""LAS minimal (format 6 : gps_time) pour les tests."""
|
||
import laspy
|
||
import numpy as np
|
||
header = laspy.LasHeader(point_format=6, version="1.4")
|
||
header.scales = [0.01, 0.01, 0.01]
|
||
header.offsets = [652000.0, 6861000.0, 0.0]
|
||
if adjusted:
|
||
header.global_encoding.gps_time_type = laspy.header.GpsTimeType.STANDARD
|
||
if creation is not None:
|
||
header.creation_date = creation
|
||
las = laspy.LasData(header)
|
||
las.x = np.asarray(x, float); las.y = np.asarray(y, float)
|
||
las.z = np.zeros(len(x))
|
||
las.classification = np.asarray(cls, np.uint8)
|
||
las.return_number = np.ones(len(x), np.uint8)
|
||
las.number_of_returns = np.ones(len(x), np.uint8)
|
||
las.gps_time = np.asarray(t, float)
|
||
las.write(str(path))
|
||
|
||
|
||
def test_quality_from_las_filters_classes(tmp_path):
|
||
from lidar_pipeline.quality import quality_from_las
|
||
p = tmp_path / "LHD_FXX_0652_6862_PTS_LAMB93_IGN69.laz"
|
||
t = _gps_seconds(2022, 6, 1)
|
||
_write_las(p, [652100, 652200, 652300], [6861100, 6861200, 6861300],
|
||
[2, 2, 5], [t, t, t])
|
||
q = quality_from_las(p, BOUNDS, codes=(2,))
|
||
assert abs(q["ground_density"] - 2e-6) < 1e-9
|
||
assert q["acq_start"] == "2022-06-01" and q["acq_source"] == "gps"
|
||
|
||
|
||
def test_quality_from_las_week_time_uses_header_date(tmp_path):
|
||
from datetime import date
|
||
from lidar_pipeline.quality import quality_from_las
|
||
p = tmp_path / "x.las"
|
||
_write_las(p, [652100], [6861100], [2], [1000.0], adjusted=False,
|
||
creation=date(2021, 9, 2))
|
||
q = quality_from_las(p, BOUNDS)
|
||
assert q["acq_source"] == "header" and q["acq_start"] == "2021-09-02"
|
||
|
||
|
||
def test_quality_from_las_unreadable_returns_none(tmp_path):
|
||
from lidar_pipeline.quality import quality_from_las
|
||
p = tmp_path / "bad.laz"; p.write_bytes(b"pas un LAS")
|
||
assert quality_from_las(p, BOUNDS) is None
|
||
|
||
|
||
def test_ensure_quality_skips_existing_and_unparsable(tmp_path):
|
||
from lidar_pipeline.quality import ensure_quality, read_quality, write_quality
|
||
base = "LHD_FXX_0652_6862_PTS_LAMB93_IGN69"
|
||
p = tmp_path / f"{base}_ground.las"
|
||
_write_las(p, [652100], [6861100], [2], [_gps_seconds(2022, 6, 1)])
|
||
assert ensure_quality(p, base, tmp_path) is True
|
||
assert read_quality(tmp_path, base)["acq_start"] == "2022-06-01"
|
||
write_quality(tmp_path, base, dict(read_quality(tmp_path, base), ground_density=99.0))
|
||
assert ensure_quality(p, base, tmp_path) is True
|
||
assert read_quality(tmp_path, base)["ground_density"] == 99.0 # non recalculé
|
||
assert ensure_quality(p, "nom_libre", tmp_path) is False
|
||
|
||
|
||
def test_backfill_quality_reads_input_laz(tmp_path):
|
||
from lidar_pipeline.quality import backfill_quality, read_quality
|
||
inp = tmp_path / "input"; inp.mkdir()
|
||
out = tmp_path / "output"
|
||
base = "LHD_FXX_0652_6862_PTS_LAMB93_IGN69"
|
||
_write_las(inp / f"{base}.copc.laz", [652100, 652200], [6861100, 6861200],
|
||
[2, 6], [_gps_seconds(2022, 6, 1)] * 2)
|
||
assert backfill_quality(inp, out) == 1
|
||
assert abs(read_quality(out, base)["ground_density"] - 1e-6) < 1e-9
|
||
assert backfill_quality(inp, out) == 0 # déjà à jour
|
||
|
||
|
||
def test_cli_quality_backfill(tmp_path):
|
||
import subprocess, sys
|
||
inp = tmp_path / "input"; inp.mkdir()
|
||
out = tmp_path / "output"
|
||
base = "LHD_FXX_0652_6862_PTS_LAMB93_IGN69"
|
||
_write_las(inp / f"{base}.laz", [652100], [6861100], [2], [_gps_seconds(2022, 6, 1)])
|
||
r = subprocess.run([sys.executable, "-m", "lidar_pipeline", str(inp), "-o", str(out),
|
||
"--quality-backfill"], capture_output=True, text=True, timeout=180)
|
||
assert r.returncode == 0, r.stderr
|
||
assert (out / "quality" / f"{base}.json").is_file()
|
||
assert not (out / "DTM").exists()
|