"""Command-line interface for the LiDAR archaeological pipeline. Handles argument parsing, logging configuration, and entry point. """ import argparse import logging import sys from .pipeline import LidarArchaeoPipeline from .gpu import log_gpu_status logger = logging.getLogger("lidar") def setup_logging(verbose=False, debug=False): """Configure the 'lidar' logger. Args: verbose: If True, include timestamps and level names. debug: If True, set level to DEBUG and add file:line info. """ if debug: level = logging.DEBUG fmt = "%(asctime)s.%(msecs)03d %(levelname)-5s [%(filename)s:%(lineno)d] %(message)s" elif verbose: level = logging.INFO fmt = "%(asctime)s %(levelname)-5s %(message)s" else: level = logging.INFO fmt = "%(message)s" handler = logging.StreamHandler(sys.stdout) handler.setFormatter(logging.Formatter(fmt, datefmt="%H:%M:%S")) logger.setLevel(level) logger.handlers.clear() logger.addHandler(handler) # Also configure the root logger so worker processes log properly root_logger = logging.getLogger() root_logger.setLevel(level) if not root_logger.handlers: root_handler = logging.StreamHandler(sys.stdout) root_handler.setFormatter(logging.Formatter(fmt, datefmt="%H:%M:%S")) root_logger.addHandler(root_handler) return logger def main(): """Entry point for the LiDAR archaeological pipeline.""" parser = argparse.ArgumentParser( description="Pipeline LiDAR pour détection archéologique", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="""\ Exemples: Traitement standard: python -m lidar_pipeline /data/input -o /data/output Haute résolution avec accélération GPU: python -m lidar_pipeline /data/input -o /data/output -r 0.2 -g Mode verbeux (timestamps): python -m lidar_pipeline /data/input -o /data/output -v Mode debug (détails internes): python -m lidar_pipeline /data/input -o /data/output --debug Forcer la régénération de tous les fichiers: python -m lidar_pipeline /data/input -o /data/output --force Traiter un seul fichier (pour tests): python -m lidar_pipeline /data/input -o /data/output --file LHD_FXX_1000_6881_PTS_LAMB93_IGN69.copc Traitement parallèle (4 workers): python -m lidar_pipeline /data/input -o /data/output -w 4 """ ) parser.add_argument( "input", help="Dossier contenant les fichiers LAZ/LAS" ) parser.add_argument( "-o", "--output", default="/data/output", help="Dossier de sortie (défaut: /data/output)" ) parser.add_argument( "-r", "--resolution", type=float, default=0.5, help="Résolution en mètres par pixel (défaut: 0.5)" ) parser.add_argument( "-w", "--workers", type=int, default=1, help="Nombre de workers pour traitement parallèle (défaut: 1)" ) parser.add_argument( "-f", "--force", action="store_true", help="Régénérer tous les fichiers même si les WebP existent déjà" ) parser.add_argument( "--file", nargs="+", type=str, default=None, help="Traiter un ou plusieurs fichiers LAZ/LAS (nom complet sans extension, ex: LHD_FXX_1000_6882_PTS_LAMB93_IGN69.copc)" ) parser.add_argument( "-v", "--verbose", action="store_true", help="Mode verbeux : affiche les timestamps et niveaux" ) parser.add_argument( "--debug", action="store_true", help="Mode debug : affiche les détails internes (fichier:ligne)" ) args = parser.parse_args() # Configure logging before any other output setup_logging(verbose=args.verbose, debug=args.debug) logger.info("=" * 60) logger.info("Pipeline LiDAR Archéologique") logger.info("=" * 60) log_gpu_status() try: pipeline = LidarArchaeoPipeline( input_dir=args.input, output_dir=args.output, resolution=args.resolution, workers=args.workers, force=args.force ) # If --file is specified, process only matching files if args.file: from pathlib import Path input_dir = Path(args.input) # Each pattern is the full filename without extension (e.g. LHD_FXX_1000_6882_PTS_LAMB93_IGN69.copc) selected_files = [] for pattern in args.file: matches = list(input_dir.glob(f"{pattern}.laz")) + list(input_dir.glob(f"{pattern}.las")) # Remove duplicates matches = list(dict.fromkeys(matches)) if not matches: logger.warning(f"Aucun fichier trouvé pour: {pattern}") continue selected_files.extend(matches) # Remove duplicates across patterns seen = set() unique_files = [] for f in selected_files: if f not in seen: seen.add(f) unique_files.append(f) if not unique_files: logger.error("Aucun fichier trouvé pour les motifs spécifiés") sys.exit(1) logger.info(f"Traitement de {len(unique_files)} fichier(s) sélectionné(s)") for laz_file in unique_files: logger.info(f" → {laz_file.name}") for laz_file in unique_files: pipeline.process_file(laz_file) else: pipeline.process_all() except Exception as e: logger.error(f"Erreur fatale: {e}", exc_info=True) sys.exit(1)