diff --git a/Dockerfile b/Dockerfile index e3c2e52..3c18100 100644 --- a/Dockerfile +++ b/Dockerfile @@ -25,11 +25,13 @@ RUN pip3 install --no-cache-dir \ rasterio \ 'laspy[laspy]' \ scikit-image \ + scikit-learn \ tqdm -# Copy script +# Copy scripts COPY process_lidar.py /usr/local/bin/ -RUN chmod +x /usr/local/bin/process_lidar.py +COPY semantic_classifier.py /usr/local/bin/ +RUN chmod +x /usr/local/bin/process_lidar.py /usr/local/bin/semantic_classifier.py # Create directories with correct permissions RUN mkdir -p /data/output /data/input && \ diff --git a/process_lidar.py b/process_lidar.py index 2801035..08b048e 100755 --- a/process_lidar.py +++ b/process_lidar.py @@ -320,6 +320,134 @@ class LidarArchaeoPipeline: print(f" ✗ Erreur slope: {e}") return None + def generate_aspect(self, dem_file, basename): + """Generate aspect (orientation of slopes) map""" + print(f" → Aspect (Orientation)...") + + output = self.vis_dir / f"{basename}_aspect.tif" + + try: + with rasterio.open(dem_file) as src: + dem = src.read(1) + transform = src.transform + crs = src.crs + + # Calculate aspect (direction of slope) + dy, dx = np.gradient(dem) + aspect = np.arctan2(-dy, dx) * 180 / np.pi + aspect = np.mod(aspect, 360) # Convert to 0-360 + + # Save + with rasterio.open( + output, + 'w', + driver='GTiff', + height=aspect.shape[0], + width=aspect.shape[1], + count=1, + dtype='float32', + crs=crs, + transform=transform, + compress='lzw' + ) as dst: + dst.write(aspect.astype('float32'), 1) + + return output + except Exception as e: + print(f" ✗ Erreur aspect: {e}") + return None + + def generate_curvature(self, dem_file, basename): + """Generate curvature (terrain concavity/convexity) map""" + print(f" → Courbure (Curvature)...") + + output = self.vis_dir / f"{basename}_curvature.tif" + + try: + with rasterio.open(dem_file) as src: + dem = src.read(1) + transform = src.transform + crs = src.crs + + # Calculate curvature using Laplacian + dz_dx = np.gradient(dem, axis=1) + dz_dy = np.gradient(dem, axis=0) + d2z_dx2 = np.gradient(dz_dx, axis=1) + d2z_dy2 = np.gradient(dz_dy, axis=0) + + # Mean curvature + curvature = (d2z_dx2 + d2z_dy2) / 2 + + # Save + with rasterio.open( + output, + 'w', + driver='GTiff', + height=curvature.shape[0], + width=curvature.shape[1], + count=1, + dtype='float32', + crs=crs, + transform=transform, + compress='lzw' + ) as dst: + dst.write(curvature.astype('float32'), 1) + + return output + except Exception as e: + print(f" ✗ Erreur curvature: {e}") + return None + + def generate_solar(self, dem_file, basename): + """Generate solar irradiance simulation""" + print(f" → Éclairage Solaire (Solar Irradiance)...") + + output = self.vis_dir / f"{basename}_solar.tif" + + try: + with rasterio.open(dem_file) as src: + dem = src.read(1) + transform = src.transform + crs = src.crs + + # Calculate gradients + dy, dx = np.gradient(dem) + + # Calculate slope and aspect + slope = np.arctan(np.sqrt(dx**2 + dy**2)) + aspect = np.arctan2(-dy, dx) + + # Solar irradiance (morning sun - azimuth 90, altitude 30) + az_rad = np.radians(90) + alt_rad = np.radians(30) + + # Solar radiation formula + solar = np.sin(alt_rad) * np.sin(slope) + \ + np.cos(alt_rad) * np.cos(slope) * np.cos(az_rad - aspect) + + # Clip negative values (shadows) + solar = np.clip(solar, 0, 1) + + # Save + with rasterio.open( + output, + 'w', + driver='GTiff', + height=solar.shape[0], + width=solar.shape[1], + count=1, + dtype='float32', + crs=crs, + transform=transform, + compress='lzw' + ) as dst: + dst.write(solar.astype('float32'), 1) + + return output + except Exception as e: + print(f" ✗ Erreur solar: {e}") + return None + def generate_lrm(self, dem_file, basename): """Local Relief Model - deviation from local mean""" print(f" → Local Relief Model...") @@ -479,6 +607,29 @@ class LidarArchaeoPipeline: title = "Pente (Slope)" legend_label = f"Pente (°)\nMin: {vmin:.1f}° | Max: {vmax:.1f}°" description = "Orange/Clair = Forte pente (murs, talus) | Foncé = Faible pente" + elif 'aspect' in str(tif_file): + cmap = 'hsv' # Cyclic colormap for directions + # Aspect is 0-360, normalize to 0-1 + vmin, vmax = 0, 360 + data = np.clip((data - vmin) / (vmax - vmin), 0, 1) + title = "Aspect (Orientation)" + legend_label = "Orientation (° du Nord)\nN=0°, E=90°, S=180°, O=270°" + description = "Couleur = Direction de la pente (utile pour orientation bâtiments)" + elif 'curvature' in str(tif_file): + cmap = 'RdYlBu_r' # Diverging for positive/negative curvature + vmax = max(abs(np.percentile(valid_data, 5)), abs(np.percentile(valid_data, 95)), 0.001) + vmin, vmax = -vmax, vmax + data = np.clip((data - vmin) / (vmax - vmin), 0, 1) + title = "Courbure (Curvature)" + legend_label = f"Courbure\nRouge = Convexe (bosse)\nBleu = Concave (creux)" + description = "⭐ Excellent pour fossés, levées, terrasses, talus" + elif 'solar' in str(tif_file): + cmap = 'YlOrBr' # Sun-like colormap + vmin, vmax = 0, 1 + data = np.clip(data, vmin, vmax) + title = "Éclairage Solaire (Matin)" + legend_label = "Irradiance Solaire\nFoncé = Ombre | Clair = Éclairé" + description = "Simulation soleil matin - révèle structures orientées Est" elif 'svf' in str(tif_file): cmap = 'viridis' vmin, vmax = np.percentile(valid_data, (5, 95)) @@ -564,20 +715,23 @@ class LidarArchaeoPipeline: vis_results = {} - # Generate rasters + # Generate rasters (existing + new) vis_results['hillshade'] = self.generate_hillshade(dtm_file, basename) vis_results['slope'] = self.generate_slope(dtm_file, basename) + vis_results['aspect'] = self.generate_aspect(dtm_file, basename) + vis_results['curvature'] = self.generate_curvature(dtm_file, basename) + vis_results['solar'] = self.generate_solar(dtm_file, basename) vis_results['svf'] = self.generate_svf(dtm_file, basename) vis_results['lrm'] = self.generate_lrm(dtm_file, basename) vis_results['pos_open'] = self.generate_openness(dtm_file, basename, positive=True) vis_results['neg_open'] = self.generate_openness(dtm_file, basename, positive=False) - # Convert to PNG - print(f"\n Conversion images PNG:") + # Convert to JPEG + print(f"\n Conversion images JPEG:") for name, tif_file in vis_results.items(): - png_file = self.tif_to_png(tif_file) - if png_file: - print(f" ✓ {png_file.name}") + jpg_file = self.tif_to_png(tif_file) + if jpg_file: + print(f" ✓ {jpg_file.name}") return vis_results @@ -586,6 +740,9 @@ class LidarArchaeoPipeline: patterns = { 'Hillshade': '*_hillshade_multi.jpg', 'Pente': '*_slope.jpg', + 'Aspect': '*_aspect.jpg', + 'Courbure': '*_curvature.jpg', + 'Éclairage': '*_solar.jpg', 'Sky-View Factor': '*_svf.jpg', 'Local Relief': '*_lrm.jpg', 'Pos. Openness': '*_positive_openness.jpg', @@ -601,39 +758,71 @@ class LidarArchaeoPipeline: if len(images) < 2: return None - # 2x3 grid with white background - fig, axes = plt.subplots(2, 3, figsize=(20, 14), facecolor='white') + # 3x3 grid for 9 visualizations + fig, axes = plt.subplots(3, 3, figsize=(24, 18), facecolor='white') axes = axes.flatten() for idx, (name, img_path) in enumerate(images.items()): - if idx >= 6: + if idx >= 9: break img = plt.imread(img_path) axes[idx].imshow(img) - axes[idx].set_title(name, fontsize=12, fontweight='bold') + axes[idx].set_title(name, fontsize=11, fontweight='bold') axes[idx].axis('off') # Hide unused subplots - for idx in range(len(images), 6): + for idx in range(len(images), 9): axes[idx].axis('off') # Title fig.suptitle( f"Analyse Archéologique LiDAR - {basename}", - fontsize=16, + fontsize=18, fontweight='bold', y=0.98 ) plt.tight_layout() output = self.report_dir / f"{basename}_overview.jpg" - plt.savefig(output, dpi=150, bbox_inches='tight', facecolor='white', format='jpg') + plt.savefig(output, dpi=120, bbox_inches='tight', facecolor='white', format='jpg') plt.close() return output # ============ Complete Pipeline ============ + def run_semantic_classification(self, dtm_file, basename): + """Run semantic classification on DTM""" + print(f"\n[4/4] Classification Sémantique Automatique...") + + try: + # Import semantic classifier + from semantic_classifier import ArchaeoSemanticClassifier + + # Create output subdirectory for semantic results + semantic_dir = self.output_dir / "semantic" + semantic_dir.mkdir(exist_ok=True) + + # Run classification + classifier = ArchaeoSemanticClassifier(dtm_file, semantic_dir) + results = classifier.process(basename) + + print(f" ✓ Classification sémantique terminée") + print(f" → Carte: {results['tif'].name}") + print(f" → Visualisation: {results['jpg'].name}") + stats_file = Path(results['tif']).parent / f"{Path(results['tif']).stem.replace('_semantic', '')}_statistics.json" + print(f" → Statistiques: {stats_file.name}") + + return results + except ImportError as e: + print(f" ✗ Module non disponible: {e}") + return None + except Exception as e: + print(f" ✗ Erreur classification: {e}") + import traceback + traceback.print_exc() + return None + def process_file(self, laz_file): """Process a single LAZ file""" basename = laz_file.stem @@ -642,21 +831,21 @@ class LidarArchaeoPipeline: print(f"{'='*60}") # Step 1: Ground classification - print(f"\n[1/3] Classification du sol...") + print(f"\n[1/4] Classification du sol...") las_file = self.classify_ground(laz_file) if not las_file: print(f" ✗ Échec classification") return False # Step 2: Generate DTM - print(f"\n[2/3] Génération DTM...") + print(f"\n[2/4] Génération DTM...") dtm_file = self.create_dtm_fast(las_file, basename) if not dtm_file: print(f" ✗ Échec DTM") return False # Step 3: Visualizations - print(f"\n[3/3] Visualisations archéologiques...") + print(f"\n[3/4] Visualisations archéologiques...") vis = self.generate_all_visualizations(dtm_file, basename) # Overview @@ -664,6 +853,20 @@ class LidarArchaeoPipeline: if overview: print(f"\n ✓ Vue synthétique: {overview}") + # Step 4: Semantic Classification (NEW!) + semantic_results = self.run_semantic_classification(dtm_file, basename) + + print(f"\n✓ {basename} traité avec succès !") + return True + + # Overview + overview = self.create_overview(basename) + if overview: + print(f"\n ✓ Vue synthétique: {overview}") + + # Step 4: Semantic Classification (NEW!) + semantic_results = self.run_semantic_classification(dtm_file, basename) + print(f"\n✓ {basename} traité avec succès !") return True diff --git a/requirements.txt b/requirements.txt index ef862ef..0a00f3b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,3 +6,4 @@ rasterio>=1.3 laspy[laspy]>=2.5 scikit-image>=0.21 tqdm>=4.65 +scikit-learn>=1.3 diff --git a/semantic_classifier.py b/semantic_classifier.py new file mode 100644 index 0000000..ab8ee94 --- /dev/null +++ b/semantic_classifier.py @@ -0,0 +1,293 @@ +#!/usr/bin/env python3 +""" +Module de Classification Sémantique Simplifié pour LiDAR Archéologique +Approche robuste avec K-Means pour classification automatique +""" + +import numpy as np +import rasterio +from rasterio.transform import from_bounds +from sklearn.cluster import KMeans +from scipy import ndimage +import json +from pathlib import Path +import matplotlib.pyplot as plt +import matplotlib.colors as mcolors + + +class ArchaeoSemanticClassifier: + """Classification sémantique automatique robuste""" + + def __init__(self, dtm_file, output_dir): + self.dtm_file = Path(dtm_file) + self.output_dir = Path(output_dir) + self.output_dir.mkdir(parents=True, exist_ok=True) + + # Load DTM + with rasterio.open(self.dtm_file) as src: + self.dem = src.read(1) + self.transform = src.transform + self.crs = src.crs + self.height, self.width = self.dem.shape + + print(f"✓ Classifieur initialisé (DTM: {self.width}x{self.height})") + + def extract_features_simple(self): + """Extraction simplifiée des caractéristiques""" + print(" → Extraction caractéristiques...") + + # Calculate gradients + dy, dx = np.gradient(self.dem) + + # Slope + slope = np.arctan(np.sqrt(dx**2 + dy**2)) * 180 / np.pi + + # Aspect + aspect = np.arctan2(-dy, dx) * 180 / np.pi + aspect = np.mod(aspect, 360) + + # Curvature + dz_dx = np.gradient(dx, axis=1) + dz_dy = np.gradient(dy, axis=0) + curvature = (dz_dx + dz_dy) / 2 + + # Local Relief + from scipy.ndimage import uniform_filter + local_mean = uniform_filter(self.dem, size=int(15/0.5)) + local_relief = self.dem - local_mean + + return { + 'elevation': self.dem, + 'slope': slope, + 'aspect': aspect, + 'curvature': curvature, + 'local_relief': local_relief + } + + def classify_kmeans(self, n_clusters=6): + """Classification K-Means robuste""" + print(" → Classification K-Means...") + + features = self.extract_features_simple() + + # Normalize each feature to 0-1 + normalized_features = {} + for name, data in features.items(): + min_val = np.percentile(data, 2) + max_val = np.percentile(data, 98) + normalized = np.clip((data - min_val) / (max_val - min_val + 1e-6), 0, 1) + normalized_features[name] = normalized + + # Stack features for clustering + feature_stack = np.stack([ + normalized_features['elevation'].flatten(), + normalized_features['slope'].flatten(), + normalized_features['curvature'].flatten(), + normalized_features['local_relief'].flatten() + ], axis=1) + + # Remove NaN values + valid_mask = ~np.isnan(feature_stack).any(axis=1) + feature_stack = feature_stack[valid_mask] + + # K-Means clustering with random_state for reproducibility + kmeans = KMeans(n_clusters=n_clusters, random_state=42, n_init=10, max_iter=300) + labels_flat = kmeans.fit_predict(feature_stack) + + # Create full resolution labels + full_labels = np.zeros(self.height * self.width, dtype=int) + full_indices = np.where(valid_mask)[0] + full_labels[full_indices] = labels_flat + 1 # +1 to shift from 0-based + full_labels = full_labels.reshape(self.height, self.width) + + # Interpret clusters + self._interpret_clusters(kmeans.cluster_centers_, features) + + return full_labels + + def _interpret_clusters(self, centers, features): + """Interprète les clusters selon les centroïdes""" + interpretations = {} + + # Centers are in normalized feature space + # Features order: elevation, slope, curvature, local_relief + for i, center in enumerate(centers): + elev = center[0] + slope = center[1] + curve = center[2] + relief = center[3] + + # Interpret based on feature values + if relief > 0.7: + category = "ÉLEVÉE (Tumulus possible)" + elif relief < -0.3: + category = "ENFONCÉ (Fossé, cavité)" + elif abs(curve) > 0.5: + if curve > 0: + category = "CONVEX (Bosse, monticule)" + else: + category = "CONCAVE (Creux, dépression)" + elif slope > 0.6: + category = "PENTE FORTE (Talus, mur)" + elif slope < 0.2 and elev < 0.3: + category = "PLAT (Zone plane)" + else: + category = "TOPOGRAPHIE MIXTE" + + interpretations[i + 1] = category # +1 because labels are 1-indexed + + print(f" Interprétation des {len(centers)} clusters :") + for i, interp in interpretations.items(): + print(f" Classe {i}: {interp}") + + def generate_semantic_map(self, labels): + """Génère une carte sémantique colorée""" + print(" → Génération carte sémantique...") + + # Create color map for semantic classes + # 0: Background, 1-6: Different semantic classes + colors = { + 0: [200, 200, 200], # Gray: Background/Unknown + 1: [139, 69, 19], # Brown: Linear/Walls + 2: [128, 0, 128], # Purple: Circular/Mounds + 3: [255, 140, 0], # Orange: Elevated + 4: [0, 200, 255], # Cyan: Depressed + 5: [220, 220, 0], # Yellow: Slope + 6: [0, 128, 0] # Green: Vegetation/Natural + } + + # Create RGB image + rgb = np.zeros((self.height, self.width, 3), dtype=np.uint8) + + for class_id, color in colors.items(): + mask = labels == class_id + for c in range(3): + rgb[:, :, c][mask] = color[c] + + return rgb + + def process(self, basename): + """Pipeline complet de classification sémantique""" + print(f"\n{'='*60}") + print(f" CLASSIFICATION SÉMANTIQUE - {basename}") + print(f"{'='*60}") + + # Run K-Means classification + labels = self.classify_kmeans(n_clusters=6) + + # Generate semantic map + rgb = self.generate_semantic_map(labels) + + # Save semantic classification + output_tif = self.output_dir / f"{basename}_semantic.tif" + with rasterio.open( + output_tif, + 'w', + driver='GTiff', + height=self.height, + width=self.width, + count=1, + dtype='uint8', + crs=self.crs, + transform=self.transform, + compress='lzw' + ) as dst: + dst.write(labels, 1) + + # Save visualization + output_jpg = self.output_dir / f"{basename}_semantic.jpg" + plt.figure(figsize=(16, 12), facecolor='white') + plt.imshow(rgb) + + # Create legend + legend_elements = [ + plt.Rectangle((0, 0), 1, 1, facecolor=np.array(c)/255, edgecolor='black', label=label) + for label, c in [ + ("Inconnu/Fond", [200, 200, 200]), + ("Linéaire (murs)", [139, 69, 19]), + ("Circulaire (tumulus)", [128, 0, 128]), + ("Élevé (monticules)", [255, 140, 0]), + ("Enfoncé (fossés)", [0, 200, 255]), + ("Pente forte (talus)", [220, 220, 0]), + ("Naturel", [0, 128, 0]) + ] + ] + + plt.legend(handles=legend_elements, loc='upper right', fontsize=11) + plt.title(f"Classification Sémantique LiDAR - {basename}\n", + fontsize=14, fontweight='bold', pad=15) + plt.axis('off') + plt.tight_layout() + plt.savefig(output_jpg, dpi=150, bbox_inches='tight', format='jpg') + plt.close() + + # Generate statistics + stats = self._generate_stats(labels, basename) + + print(f"\n✓ Classification terminée !") + print(f" • Carte sémantique: {output_tif.name}") + print(f" • Visualisation: {output_jpg.name}") + print(f" • Statistiques: {self.output_dir / f'{basename}_statistics.json'}") + + return { + 'labels': labels, + 'tif': output_tif, + 'jpg': output_jpg, + 'stats': stats + } + + def _generate_stats(self, labels, basename): + """Génère les statistiques de classification""" + print(" → Génération statistiques...") + + total_pixels = labels.size + stats = {} + class_names = { + 0: "Inconnu/Fond", + 1: "Linéaire", + 2: "Circulaire", + 3: "Élevée", + 4: "Enfoncée", + 5: "Pente forte", + 6: "Naturel" + } + + for class_id in range(7): + count = np.sum(labels == class_id) + percentage = (count / total_pixels) * 100 + if count > 0: + stats[class_id] = { + 'name': class_names[class_id], + 'count': int(count), + 'percentage': float(percentage) + } + + # Save as JSON + stats_file = self.output_dir / f"{basename}_statistics.json" + with open(stats_file, 'w') as f: + json.dump(stats, f, indent=2, default=lambda x: float(x) if isinstance(x, (np.floating, np.integer)) else x) + + # Print summary + print(f"\n 📊 Statistiques :") + for class_id, info in stats.items(): + print(f" {info['name']}: {info['count']:.0f} px ({info['percentage']:.1f}%)") + + return stats + + +def main(): + import sys + + if len(sys.argv) < 2: + print("Usage: python semantic_classifier.py [output_dir]") + sys.exit(1) + + dtm_file = sys.argv[1] + output_dir = sys.argv[2] if len(sys.argv) > 2 else "semantic_output" + + classifier = ArchaeoSemanticClassifier(dtm_file, output_dir) + classifier.process(Path(dtm_file).stem) + + +if __name__ == "__main__": + main()