Add multi-resolution support and remove PDF generation

- --resolution now accepts comma-separated values (e.g. 0.5,0.2)
- Additional resolutions get suffixed output dirs: basename_r0p2/
- DTM files are named basename_dtm_r0p2.tif for extra resolutions
- Ground classification is done once and shared across resolutions
- PDF report generation removed per user request
- Fix --file argument to accept full filenames with extensions

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Jacquin Antoine
2026-05-14 21:29:45 +02:00
parent 53b6369a1b
commit ac56ba8084
3 changed files with 103 additions and 66 deletions

View File

@ -97,9 +97,9 @@ Exemples:
) )
parser.add_argument( parser.add_argument(
"-r", "--resolution", "-r", "--resolution",
type=float, type=str,
default=0.5, default="0.5",
help="Résolution en mètres par pixel (défaut: 0.5)" help="Résolution en m/px, ou multiples séparées par virgules (défaut: 0.5, ex: 0.5,0.2)"
) )
parser.add_argument( parser.add_argument(
"-w", "--workers", "-w", "--workers",

View File

@ -402,7 +402,7 @@ def _repair_laz_with_laspy(input_laz, output_las):
return False return False
def create_dtm_fast(las_file, basename, dtm_dir, resolution, force=False): def create_dtm_fast(las_file, basename, dtm_dir, resolution, force=False, output_suffix=""):
"""Create DTM using fast binning method with gap filling. """Create DTM using fast binning method with gap filling.
Args: Args:
@ -411,11 +411,12 @@ def create_dtm_fast(las_file, basename, dtm_dir, resolution, force=False):
dtm_dir: Directory for output DTM GeoTIFF. dtm_dir: Directory for output DTM GeoTIFF.
resolution: Grid resolution in meters per pixel. resolution: Grid resolution in meters per pixel.
force: If True, regenerate even if DTM already exists. force: If True, regenerate even if DTM already exists.
output_suffix: Suffix for output filename (e.g. '_r0p2' for additional resolutions).
Returns: Returns:
Path to output DTM GeoTIFF, or None on failure. Path to output DTM GeoTIFF, or None on failure.
""" """
output_tif = dtm_dir / f"{basename}_dtm.tif" output_tif = dtm_dir / f"{basename}_dtm{output_suffix}.tif"
if output_tif.exists() and not force: if output_tif.exists() and not force:
logger.info(f" DTM déjà existant — fichier réutilisé: {output_tif.name}") logger.info(f" DTM déjà existant — fichier réutilisé: {output_tif.name}")

View File

@ -65,7 +65,7 @@ from .visualizations import (
) )
from .gpu import gpu_cleanup from .gpu import gpu_cleanup
from .ign import generate_ign_overlay from .ign import generate_ign_overlay
from .rendering import tif_to_png, generate_pdf_report from .rendering import tif_to_png
# Ordered list of visualization steps. # Ordered list of visualization steps.
@ -111,7 +111,14 @@ class LidarArchaeoPipeline:
def __init__(self, input_dir, output_dir, resolution=0.5, workers=1, force=False, ground_method='auto', force_classify=False, keep_tif=False, quality=85, only_viz=None, skip_viz=None): def __init__(self, input_dir, output_dir, resolution=0.5, workers=1, force=False, ground_method='auto', force_classify=False, keep_tif=False, quality=85, only_viz=None, skip_viz=None):
self.input_dir = Path(input_dir) self.input_dir = Path(input_dir)
self.output_dir = Path(output_dir) self.output_dir = Path(output_dir)
self.resolution = resolution # Accept single float or comma-separated string for multi-resolution
if isinstance(resolution, str):
self.resolutions = [float(r.strip()) for r in resolution.split(',')]
elif isinstance(resolution, (list, tuple)):
self.resolutions = [float(r) for r in resolution]
else:
self.resolutions = [float(resolution)]
self.resolution = self.resolutions[0] # Primary resolution (backward compat)
self.workers = workers self.workers = workers
self.force = force self.force = force
self.ground_method = ground_method self.ground_method = ground_method
@ -153,7 +160,10 @@ class LidarArchaeoPipeline:
logger.info("Pipeline initialisé") logger.info("Pipeline initialisé")
logger.info(f" Entrée : {self.input_dir}") logger.info(f" Entrée : {self.input_dir}")
logger.info(f" Sortie : {self.output_dir}") logger.info(f" Sortie : {self.output_dir}")
logger.info(f" Résolution : {resolution}m/px") if len(self.resolutions) > 1:
logger.info(f" Résolutions : {', '.join(f'{r}m/px' for r in self.resolutions)}")
else:
logger.info(f" Résolution : {self.resolution}m/px")
logger.info(f" Workers : {workers}") logger.info(f" Workers : {workers}")
logger.info(f" Force : {'OUI' if self.force else 'non (skip existing)'}") logger.info(f" Force : {'OUI' if self.force else 'non (skip existing)'}")
logger.info(f" Classification sol : {self.ground_method}") logger.info(f" Classification sol : {self.ground_method}")
@ -306,8 +316,22 @@ class LidarArchaeoPipeline:
return vis_results return vis_results
@staticmethod
def _res_suffix(resolution):
"""Return suffix for additional resolutions (empty string for primary)."""
if resolution == 0.5:
return "" # Default resolution — no suffix
res_str = f"{resolution}".replace('.', 'p')
return f"_r{res_str}"
def process_file(self, laz_file): def process_file(self, laz_file):
"""Process a single LAZ file through the full pipeline.""" """Process a single LAZ file through the full pipeline.
If self.resolutions has multiple entries, processes each resolution:
- Primary resolution uses current naming (no suffix)
- Additional resolutions use _r0p2 suffix in directories/filenames
- Ground classification is done once and shared across resolutions
"""
basename = _file_basename(laz_file) basename = _file_basename(laz_file)
_file_filter.basename = basename _file_filter.basename = basename
t_start = time.time() t_start = time.time()
@ -321,31 +345,34 @@ class LidarArchaeoPipeline:
if not validate_laz(laz_file): if not validate_laz(laz_file):
return False return False
# Skip ground classification + DTM if DTM already exists with matching resolution # Step 1: Ground classification (shared across all resolutions)
# --force only affects visualizations/PDF, not classification/DTM las_file = None
# Use --force-classification to force reclassification t_classif = 0
dtm_path = self.dtm_dir / f"{basename}_dtm.tif" for i, res in enumerate(self.resolutions):
res_suffix = self._res_suffix(res)
dtm_path = self.dtm_dir / f"{basename}_dtm{res_suffix}.tif"
if dtm_path.exists(): if dtm_path.exists():
# Check that existing DTM resolution matches requested resolution
import rasterio import rasterio
try: try:
with rasterio.open(dtm_path) as src: with rasterio.open(dtm_path) as src:
existing_res = abs(src.transform.a) existing_res = abs(src.transform.a)
if abs(existing_res - self.resolution) > 0.01: if abs(existing_res - res) > 0.01:
logger.info(f"[1/5] DTM existant à {existing_res}m/px — résolution demandée {self.resolution}m/px → régénération") logger.info(f" DTM{res_suffix} existant à {existing_res}m/px — résolution demandée {res}m/px → régénération")
dtm_path.unlink() dtm_path.unlink()
else: else:
logger.info(f"[1/5] Classification du sol — sautée (DTM existant à {existing_res}m/px)") if i == 0:
logger.info("[2/5] Génération DTM — sautée (DTM existant)") logger.info(f"[1/5] Classification du sol — sautée (DTM existant)")
dtm_file = dtm_path logger.info(f"[2/5] Génération DTM {res}m/px — sautée (DTM existant)")
t_classif = 0 else:
t_dtm = 0 logger.info(f" DTM {res}m/px déjà existant — ignoré")
continue
except Exception: except Exception:
logger.warning(f"Impossible de lire le DTM existant — régénération") logger.warning(f"Impossible de lire le DTM existant — régénération")
dtm_path.unlink() dtm_path.unlink()
if not dtm_path.exists(): # Need to classify/generate DTM for this resolution
# Step 1: Ground classification if las_file is None:
# First time: do ground classification
logger.info("[1/5] Classification du sol...") logger.info("[1/5] Classification du sol...")
t1 = time.time() t1 = time.time()
las_file = classify_ground(laz_file, self.temp_dir, method=self.ground_method, force=self.force_classify) las_file = classify_ground(laz_file, self.temp_dir, method=self.ground_method, force=self.force_classify)
@ -355,40 +382,49 @@ class LidarArchaeoPipeline:
return False return False
logger.info(f" ✓ Classification terminée ({t_classif:.1f}s)") logger.info(f" ✓ Classification terminée ({t_classif:.1f}s)")
# Step 2: Generate DTM # Generate DTM at this resolution
logger.info("[2/5] Génération DTM...") logger.info(f"{'[2/5]' if i == 0 else ' '} Génération DTM {res}m/px...")
t2 = time.time() t2 = time.time()
dtm_file = create_dtm_fast(las_file, basename, self.dtm_dir, self.resolution) dtm_file = create_dtm_fast(las_file, basename, self.dtm_dir, res, force=self.force, output_suffix=res_suffix)
t_dtm = time.time() - t2 t_dtm = time.time() - t2
if not dtm_file: if not dtm_file:
logger.error(f" ✗ Échec DTM ({t_dtm:.1f}s)") logger.error(f" ✗ Échec DTM {res}m/px ({t_dtm:.1f}s)")
return False if i == 0:
logger.info(f" ✓ DTM terminé ({t_dtm:.1f}s)") return False # Primary resolution failure is fatal
continue # Additional resolution failure is non-fatal
logger.info(f" ✓ DTM {res}m/px terminé ({t_dtm:.1f}s)")
# Process each resolution: visualizations + PDF
all_vis_results = {}
for res in self.resolutions:
res_suffix = self._res_suffix(res)
dtm_path = self.dtm_dir / f"{basename}_dtm{res_suffix}.tif"
if not dtm_path.exists():
logger.warning(f" DTM {res}m/px manquant — visualisations ignorées")
continue
# Step 3: Visualizations — use actual resolution from DTM
import rasterio import rasterio
with rasterio.open(dtm_file) as src: with rasterio.open(dtm_path) as src:
actual_res = abs(src.transform.a) actual_res = abs(src.transform.a)
if abs(actual_res - self.resolution) > 0.01:
logger.info(f" Résolution DTM: {actual_res}m/px (demandée: {self.resolution}m/px)")
self.generate_all_visualizations(dtm_file, basename, actual_res)
# Step 4: PDF report if len(self.resolutions) > 1:
t_pdf = 0 logger.info(f" --- Résolution {res}m/px ---")
file_vis_dir = self.vis_dir / basename
pdf_file = self.pdf_dir / f"{basename}_rapport.pdf" # For additional resolutions, use suffixed subdirectory and basename
if pdf_file.exists() and not self.force: if res_suffix:
logger.info(f"[4/5] Rapport PDF déjà existant — ignoré: {pdf_file.name}") vis_dir = self.vis_dir / f"{basename}{res_suffix}"
pdf_basename = f"{basename}{res_suffix}"
else: else:
logger.info("[4/5] Rapport PDF A3...") vis_dir = self.vis_dir / basename
t4 = time.time() pdf_basename = basename
generate_pdf_report(basename, file_vis_dir, self.pdf_dir, actual_res)
t_pdf = time.time() - t4 vis_dir.mkdir(exist_ok=True)
logger.info(f" ✓ Rapport PDF terminé ({t_pdf:.1f}s)")
self.generate_all_visualizations(dtm_path, basename, actual_res)
t_total = time.time() - t_start t_total = time.time() - t_start
logger.info(f"{basename} terminé en {t_total:.1f}s") logger.info(f"{basename} terminé en {t_total:.1f}s")
logger.debug(f" Détails: classification={t_classif:.1f}s, DTM={t_dtm:.1f}s, PDF={t_pdf:.1f}s")
_file_filter.basename = None _file_filter.basename = None
return True return True