#!/usr/bin/env python3 """ LiDAR Processing Pipeline Script Processes E57 files through LAZ conversion, noise removal, and Potree generation """ import os import sys import subprocess import zipfile import shutil from pathlib import Path import argparse class LidarProcessor: def __init__(self, target_directory): self.target_dir = Path(target_directory) self.lastools_bin = Path("C:/tools/lastools/bin") self.potree_converter = Path("C:/tools/potreeconverter2/potreeconverter.exe") # Validate paths if not self.target_dir.exists(): raise ValueError(f"Target directory does not exist: {self.target_dir}") if not self.lastools_bin.exists(): raise ValueError(f"LAStools bin directory not found: {self.lastools_bin}") if not self.potree_converter.exists(): raise ValueError(f"Potree converter not found: {self.potree_converter}") # Define subdirectories self.potree_dir = self.target_dir / "potree" self.laz_dir = self.target_dir / "LAZ" self.ogc_dir = self.target_dir / "ogc_3d_tiles" self.source_dir = self.target_dir / "source" def create_subdirectories(self): """Step 1: Create required subdirectories""" print("Step 1: Creating subdirectories...") for directory in [self.potree_dir, self.laz_dir, self.ogc_dir, self.source_dir]: directory.mkdir(exist_ok=True) print(f" Created/verified: {directory}") def find_zip_file(self): """Find the zip file in the target directory""" zip_files = list(self.target_dir.glob("*.zip")) if not zip_files: raise FileNotFoundError(f"No zip file found in {self.target_dir}") if len(zip_files) > 1: print(f"Warning: Multiple zip files found. Using: {zip_files[0]}") return zip_files[0] def extract_e57_file(self, zip_path): """Step 2: Extract E57 file from zip""" print(f"\nStep 2: Extracting E57 file from {zip_path.name}...") with zipfile.ZipFile(zip_path, 'r') as zip_ref: # Find any E57 file in the zip e57_files = [f for f in zip_ref.namelist() if f.lower().endswith(".e57")] if not e57_files: raise FileNotFoundError("No E57 file found in zip") if len(e57_files) > 1: print(f"Warning: Multiple E57 files found. Using: {e57_files[0]}") e57_file = e57_files[0] print(f" Found E57 file: {e57_file}") # Create new filename based on parent directory name new_filename = f"{self.target_dir.name}_lidar_terrestrial.e57" target_path = self.source_dir / new_filename # Extract and rename source = zip_ref.open(e57_file) with open(target_path, 'wb') as target: shutil.copyfileobj(source, target) print(f" Extracted and renamed to: {target_path}") return target_path, zip_path.stem def convert_e57_to_laz(self, e57_path): """Step 3a: Convert E57 to LAZ files with split scans""" print("\nStep 3a: Converting E57 to LAZ with split scans...") # Create output directory for split scans split_scans_dir = self.source_dir / "e57_laz_scans_split" split_scans_dir.mkdir(exist_ok=True) # Run e572las64 e572las_exe = self.lastools_bin / "e572las64.exe" cmd = [ str(e572las_exe), "-i", str(e57_path), "-split_scans", "-odir", str(split_scans_dir), "-olaz" ] print(f" Running: {' '.join(cmd)}") result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: print(f" Error output: {result.stderr}") raise RuntimeError(f"e572las64 conversion failed with return code {result.returncode}") # Count output files laz_files = list(split_scans_dir.glob("*.laz")) print(f" Created {len(laz_files)} LAZ file(s)") return split_scans_dir def remove_noise(self, input_dir): """Step 3b: Remove noise from LAZ files""" print("\nStep 3b: Removing noise with lasnoise64...") # Create output directory denoised_dir = self.source_dir / "e57_laz_scans_dn" denoised_dir.mkdir(exist_ok=True) # Run lasnoise64 lasnoise_exe = self.lastools_bin / "lasnoise64.exe" cmd = [ str(lasnoise_exe), "-i", str(input_dir / "*.laz"), "-step", "0.1", "-cores", "10", "-odir", str(denoised_dir), "-olaz" ] print(f" Running: {' '.join(cmd)}") result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: print(f" Error output: {result.stderr}") raise RuntimeError(f"lasnoise64 failed with return code {result.returncode}") # Count output files laz_files = list(denoised_dir.glob("*.laz")) print(f" Processed {len(laz_files)} LAZ file(s)") return denoised_dir def remove_duplicates(self, input_dir): """Step 3c: Remove duplicate points""" print("\nStep 3c: Removing duplicates with lasduplicate64...") # Create output directory in LAZ folder deduplicated_dir = self.laz_dir / "e57_laz_scans_dn_dd" deduplicated_dir.mkdir(exist_ok=True) # Run lasduplicate64 lasduplicate_exe = self.lastools_bin / "lasduplicate64.exe" cmd = [ str(lasduplicate_exe), "-i", str(input_dir / "*.laz"), "-unique_xyz", "-odir", str(deduplicated_dir), "-olaz" ] print(f" Running: {' '.join(cmd)}") result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: print(f" Error output: {result.stderr}") raise RuntimeError(f"lasduplicate64 failed with return code {result.returncode}") # Count output files laz_files = list(deduplicated_dir.glob("*.laz")) print(f" Processed {len(laz_files)} LAZ file(s)") return deduplicated_dir def generate_potree(self, input_dir, output_name): """Step 4: Generate Potree visualization""" print("\nStep 4: Generating Potree visualization...") # Create output directory potree_output_dir = self.potree_dir / output_name # Run PotreeConverter cmd = [ str(self.potree_converter), str(input_dir), "-o", str(potree_output_dir), "--encoding", "BROTLI" ] print(f" Running: {' '.join(cmd)}") result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: print(f" Error output: {result.stderr}") raise RuntimeError(f"PotreeConverter failed with return code {result.returncode}") print(f" Potree output saved to: {potree_output_dir}") return True def cleanup(self, dirs_to_remove): """Step 5: Clean up temporary directories""" print("\nStep 5: Cleaning up temporary directories...") for directory in dirs_to_remove: if directory.exists(): shutil.rmtree(directory) print(f" Removed: {directory}") def process(self): """Execute the complete processing pipeline""" try: # Step 1: Create subdirectories self.create_subdirectories() # Step 2: Find and extract E57 file zip_file = self.find_zip_file() e57_path, zip_stem = self.extract_e57_file(zip_file) # Step 3: Process LAZ files split_scans_dir = self.convert_e57_to_laz(e57_path) denoised_dir = self.remove_noise(split_scans_dir) deduplicated_dir = self.remove_duplicates(denoised_dir) # Step 4: Generate Potree success = self.generate_potree(deduplicated_dir, zip_stem) # Step 5: Cleanup if successful if success: self.cleanup([split_scans_dir, denoised_dir]) print("\n✓ Processing completed successfully!") else: print("\n✗ Processing failed. Temporary directories preserved for debugging.") except Exception as e: print(f"\n✗ Error during processing: {e}") sys.exit(1) def main(): parser = argparse.ArgumentParser( description='Process LiDAR E57 files through LAZ conversion and Potree generation', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Example usage: python lidar_processing_pipeline.py C:/data/scan_project_001 Prerequisites: - LAStools installed at C:/tools/lastools/bin - PotreeConverter 2.0 at C:/tools/potreeconverter2/potreeconverter.exe - ZIP file containing any E57 file (will be renamed to _lidar_terrestrial.e57) """ ) parser.add_argument( 'directory', help='Path to the directory containing the ZIP file to process' ) args = parser.parse_args() print("=" * 70) print("LiDAR Processing Pipeline") print("=" * 70) processor = LidarProcessor(args.directory) processor.process() if __name__ == "__main__": main()