242 lines
8.4 KiB
Python
242 lines
8.4 KiB
Python
"""Qualité des données LiDAR par dalle : densité de points sol et date d'acquisition.
|
||
|
||
Un sidecar JSON par dalle (output/quality/{basename}.json) est écrit par le
|
||
pipeline, recopié dans l'inventaire (index_tiles.json, table `quality`) et
|
||
conservé sur les machines légères : l'encart qualité de l'export PDF le lit
|
||
depuis le disque local, amont joignable ou non.
|
||
|
||
numpy n'est importé que par le calcul : l'image légère (sans numpy) lit et
|
||
écrit les sidecars.
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
import tempfile
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
|
||
logger = logging.getLogger("lidar")
|
||
|
||
QUALITY_VERSION = 1
|
||
QUALITY_DIRNAME = "quality"
|
||
DENSITY_CELL_M = 50.0 # maille de la grille de densité (20 × 20 par dalle)
|
||
EMPTY_PIXEL_M = 0.2 # pixel du MNT : part sans point sol ≈ part interpolée
|
||
|
||
_GPS_EPOCH = datetime(1980, 1, 6, tzinfo=timezone.utc)
|
||
_GPS_ADJUSTED_OFFSET = 1e9 # temps GPS ajusté standard = secondes GPS − 1e9
|
||
|
||
|
||
def gps_adjusted_to_date(t):
|
||
"""Temps GPS ajusté standard → date ISO (UTC ; secondes intercalaires
|
||
négligées : précision au jour)."""
|
||
return (_GPS_EPOCH + timedelta(seconds=float(t) + _GPS_ADJUSTED_OFFSET)).date().isoformat()
|
||
|
||
|
||
def compute_quality(x, y, gps_time, bounds, gps_adjusted=True, header_date=None):
|
||
"""Mesure la qualité d'une dalle à partir de ses points sol.
|
||
|
||
Args:
|
||
x, y: coordonnées L93 des points sol.
|
||
gps_time: temps GPS des points (ou None).
|
||
bounds: emprise nominale (min_x, min_y, max_x, max_y) ; les points
|
||
hors emprise (bande de raccord) sont ignorés.
|
||
gps_adjusted: True si gps_time est en temps GPS ajusté standard.
|
||
header_date: date ISO de l'en-tête LAS (repli des dates).
|
||
|
||
Returns:
|
||
dict JSON-sérialisable (cf. spec, section 1).
|
||
"""
|
||
import numpy as np
|
||
|
||
min_x, min_y, max_x, max_y = bounds
|
||
x = np.asarray(x, dtype=np.float64)
|
||
y = np.asarray(y, dtype=np.float64)
|
||
inside = (x >= min_x) & (x < max_x) & (y >= min_y) & (y < max_y)
|
||
t = None
|
||
if gps_time is not None and gps_adjusted:
|
||
t = np.asarray(gps_time, dtype=np.float64)
|
||
t = t[inside] if len(t) == len(x) else None
|
||
x, y = x[inside], y[inside]
|
||
n = len(x)
|
||
width, height = max_x - min_x, max_y - min_y
|
||
|
||
nx = int(round(width / DENSITY_CELL_M))
|
||
ny = int(round(height / DENSITY_CELL_M))
|
||
cx = np.clip(((x - min_x) / DENSITY_CELL_M).astype(np.int64), 0, nx - 1)
|
||
cy = np.clip(((max_y - y) / DENSITY_CELL_M).astype(np.int64), 0, ny - 1)
|
||
counts = np.bincount(cy * nx + cx, minlength=nx * ny).reshape(ny, nx)
|
||
grid = np.round(counts / (DENSITY_CELL_M ** 2), 2)
|
||
|
||
px = int(round(width / EMPTY_PIXEL_M))
|
||
py = int(round(height / EMPTY_PIXEL_M))
|
||
occupied = np.zeros(px * py, dtype=bool)
|
||
if n:
|
||
ix = np.clip(((x - min_x) / EMPTY_PIXEL_M).astype(np.int64), 0, px - 1)
|
||
iy = np.clip(((max_y - y) / EMPTY_PIXEL_M).astype(np.int64), 0, py - 1)
|
||
occupied[iy * px + ix] = True
|
||
empty_fraction = 1.0 - float(occupied.sum()) / (px * py)
|
||
|
||
if t is not None and len(t):
|
||
start, end = gps_adjusted_to_date(t.min()), gps_adjusted_to_date(t.max())
|
||
source = "gps"
|
||
elif header_date:
|
||
start = end = header_date
|
||
source = "header"
|
||
else:
|
||
start = end = source = None
|
||
|
||
return {
|
||
"version": QUALITY_VERSION,
|
||
"ground_density": round(n / (width * height), 6),
|
||
"density_grid": grid.tolist(),
|
||
"empty_fraction": round(empty_fraction, 4),
|
||
"acq_start": start,
|
||
"acq_end": end,
|
||
"acq_source": source,
|
||
}
|
||
|
||
|
||
def quality_path(output_dir, basename):
|
||
"""Chemin du sidecar qualité d'une dalle."""
|
||
return Path(output_dir) / QUALITY_DIRNAME / f"{basename}.json"
|
||
|
||
|
||
def write_quality(output_dir, basename, data):
|
||
"""Écrit le sidecar (atomique) ; False si le contenu est déjà identique."""
|
||
path = quality_path(output_dir, basename)
|
||
payload = json.dumps(data, ensure_ascii=False, sort_keys=True)
|
||
try:
|
||
if path.read_text(encoding="utf-8") == payload:
|
||
return False
|
||
except OSError:
|
||
pass
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
fd, tmp = tempfile.mkstemp(dir=str(path.parent), suffix=".tmp")
|
||
try:
|
||
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
||
f.write(payload)
|
||
os.replace(tmp, path)
|
||
except OSError:
|
||
try:
|
||
os.unlink(tmp)
|
||
except OSError:
|
||
pass
|
||
raise
|
||
return True
|
||
|
||
|
||
def read_quality(output_dir, basename):
|
||
"""Sidecar d'une dalle, ou None (absent, illisible, autre version)."""
|
||
try:
|
||
data = json.loads(quality_path(output_dir, basename).read_text(encoding="utf-8"))
|
||
except (OSError, ValueError):
|
||
return None
|
||
if not isinstance(data, dict) or data.get("version") != QUALITY_VERSION:
|
||
return None
|
||
return data
|
||
|
||
|
||
def load_quality_table(output_dir):
|
||
"""Tous les sidecars valides : {basename: sidecar}."""
|
||
folder = Path(output_dir) / QUALITY_DIRNAME
|
||
table = {}
|
||
if not folder.is_dir():
|
||
return table
|
||
for f in sorted(folder.glob("*.json")):
|
||
data = read_quality(output_dir, f.stem)
|
||
if data is not None:
|
||
table[f.stem] = data
|
||
return table
|
||
|
||
|
||
_LAZ_EXTS = (".copc.laz", ".copc.las", ".laz", ".las")
|
||
|
||
|
||
def _laz_basename(path):
|
||
"""Nom de dalle sans extension LiDAR (.copc.laz compris)."""
|
||
name = Path(path).name
|
||
for ext in _LAZ_EXTS:
|
||
if name.lower().endswith(ext):
|
||
return name[:-len(ext)]
|
||
return Path(path).stem
|
||
|
||
|
||
def _cell_bounds_of(basename):
|
||
"""Emprise nominale 1 km d'une dalle LHD, ou None hors motif LHD."""
|
||
from .index import parse_basename_coords
|
||
coords = parse_basename_coords(basename)
|
||
if coords is None:
|
||
return None
|
||
col, row = coords
|
||
return (col * 1000.0, (row - 1) * 1000.0, (col + 1) * 1000.0, row * 1000.0)
|
||
|
||
|
||
def quality_from_las(las_path, bounds, codes=None):
|
||
"""Qualité mesurée depuis un LAS/LAZ ; None en cas d'échec (jamais d'exception).
|
||
|
||
codes : classes à retenir (retours ≥ 1) ; None = fichier déjà filtré sol.
|
||
"""
|
||
try:
|
||
import laspy
|
||
import numpy as np
|
||
las = laspy.read(str(las_path))
|
||
keep = np.ones(len(las.points), dtype=bool)
|
||
if codes is not None:
|
||
keep = ((np.asarray(las.return_number) >= 1)
|
||
& (np.asarray(las.number_of_returns) >= 1)
|
||
& np.isin(np.asarray(las.classification), np.asarray(sorted(codes))))
|
||
try:
|
||
gps = np.asarray(las.gps_time, dtype=np.float64)[keep]
|
||
except AttributeError:
|
||
gps = None
|
||
adjusted = las.header.global_encoding.gps_time_type == laspy.header.GpsTimeType.STANDARD
|
||
created = las.header.creation_date
|
||
return compute_quality(np.asarray(las.x)[keep], np.asarray(las.y)[keep], gps,
|
||
bounds, gps_adjusted=adjusted,
|
||
header_date=created.isoformat() if created else None)
|
||
except Exception as e: # noqa: BLE001 — qualité best-effort
|
||
logger.warning(f" Mesure de qualité impossible ({Path(las_path).name}) : {e}")
|
||
return None
|
||
|
||
|
||
def ensure_quality(las_path, basename, output_dir, codes=None):
|
||
"""Garantit le sidecar qualité d'une dalle (calcul si absent ou périmé).
|
||
|
||
Returns:
|
||
True si un sidecar valide existe à la sortie.
|
||
"""
|
||
if read_quality(output_dir, basename) is not None:
|
||
return True
|
||
bounds = _cell_bounds_of(basename)
|
||
if bounds is None:
|
||
return False
|
||
data = quality_from_las(las_path, bounds, codes)
|
||
if data is None:
|
||
return False
|
||
try:
|
||
write_quality(output_dir, basename, data)
|
||
except OSError as e:
|
||
logger.warning(f" Écriture du sidecar qualité impossible : {e}")
|
||
return False
|
||
return True
|
||
|
||
|
||
def backfill_quality(input_dir, output_dir, codes=(2,)):
|
||
"""Rattrapage : sidecar qualité de chaque LAZ LHD d'input_dir qui n'en a pas.
|
||
|
||
Returns:
|
||
Nombre de sidecars écrits.
|
||
"""
|
||
files = sorted(p for p in Path(input_dir).iterdir()
|
||
if p.is_file() and p.name.lower().endswith(_LAZ_EXTS))
|
||
written = 0
|
||
for i, laz in enumerate(files, 1):
|
||
base = _laz_basename(laz)
|
||
if read_quality(output_dir, base) is not None or _cell_bounds_of(base) is None:
|
||
continue
|
||
logger.info(f"Qualité [{i}/{len(files)}] {base}")
|
||
if ensure_quality(laz, base, output_dir, codes):
|
||
written += 1
|
||
return written
|