Skip ground classification when DTM already exists
If the DTM .tif exists and --force is not set, skip both ground classification and DTM generation entirely. Previously, the pipeline would spend 3+ minutes reclassifying ground even when the DTM was already present and would be reused anyway. Also includes: SharedDEM cache, enhanced WebP cartouche (compass rose, adaptive scale bar, enriched info bar), removed COG/viewer, UTF-8 fix for parallel workers, skip logic for DTM and PDF. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Jacquin Antoine
@ -8,6 +8,7 @@ Contains:
|
||||
|
||||
import logging
|
||||
import time
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
@ -238,7 +239,31 @@ def _apply_colormap(data, tif_file):
|
||||
return data, 'terrain', title, 'Altitude normalisée', '', False
|
||||
|
||||
|
||||
def tif_to_png(tif_file, vis_dir, resolution, keep_tif=False):
|
||||
def _nice_scale(extent_m):
|
||||
"""Choose a nice round scale distance that fits well in the image.
|
||||
|
||||
Returns (scale_m, label) where label is like '100 m' or '500 m' or '1 km'.
|
||||
"""
|
||||
nice_scales = [50, 100, 200, 500, 1000, 2000, 5000, 10000]
|
||||
# Pick the largest scale that is <= 15% of the extent
|
||||
for s in nice_scales:
|
||||
if s <= extent_m * 0.15:
|
||||
continue
|
||||
# s is the first one > 15% — take the one below
|
||||
break
|
||||
else:
|
||||
s = nice_scales[-1]
|
||||
# Actually pick the largest scale <= 20% of extent
|
||||
chosen = nice_scales[0]
|
||||
for s in nice_scales:
|
||||
if s <= extent_m * 0.20:
|
||||
chosen = s
|
||||
if chosen >= 1000:
|
||||
return chosen, f"{chosen // 1000} km"
|
||||
return chosen, f"{chosen} m"
|
||||
|
||||
|
||||
def tif_to_png(tif_file, vis_dir, resolution, keep_tif=False, source_info=None):
|
||||
"""Convert GeoTIFF to visualization WebP with GPS coordinates, legend, and scale bar.
|
||||
|
||||
Args:
|
||||
@ -440,20 +465,32 @@ def tif_to_png(tif_file, vis_dir, resolution, keep_tif=False):
|
||||
spine.set_color('black')
|
||||
spine.set_linewidth(0.8)
|
||||
|
||||
# North arrow
|
||||
north_ax = inset_axes(ax, width="4%", height="7%", loc='upper right',
|
||||
bbox_to_anchor=(-0.03, 0.12, 1, 1), bbox_transform=ax.transAxes)
|
||||
north_ax.set_xlim(0, 1)
|
||||
north_ax.set_ylim(0, 1)
|
||||
# North arrow — compass rose style
|
||||
north_ax = inset_axes(ax, width="5%", height="9%", loc='upper right',
|
||||
bbox_to_anchor=(-0.03, 0.08, 1, 1), bbox_transform=ax.transAxes)
|
||||
north_ax.set_xlim(-1.2, 1.2)
|
||||
north_ax.set_ylim(-0.5, 1.5)
|
||||
north_ax.axis('off')
|
||||
north_ax.plot([0.5, 0.5], [0.1, 0.65], color='black', linewidth=2.5, zorder=10)
|
||||
north_ax.add_patch(MplPolygon([[0.5, 0.2], [0.35, 0.4], [0.5, 0.65], [0.65, 0.4]],
|
||||
closed=True, facecolor='black', edgecolor='black', zorder=9))
|
||||
north_ax.text(0.5, 0.95, 'N', ha='center', va='top',
|
||||
fontsize=13, fontweight='bold', color='black', zorder=11)
|
||||
north_ax.set_aspect('equal')
|
||||
# N arrow
|
||||
north_ax.annotate('N', xy=(0, 1.3), fontsize=11, fontweight='bold',
|
||||
ha='center', va='bottom', color='#b22222')
|
||||
north_ax.plot([0, 0], [0.0, 1.0], color='#b22222', linewidth=2.0, zorder=10)
|
||||
north_ax.add_patch(MplPolygon([[0, 0.3], [-0.2, 0.7], [0, 1.0], [0.2, 0.7]],
|
||||
closed=True, facecolor='#b22222', edgecolor='#b22222', zorder=9))
|
||||
# Cardinal ticks
|
||||
for angle, label in [(90, ''), (0, 'E'), (180, 'O'), (270, 'S')]:
|
||||
rad = np.radians(angle)
|
||||
r_text = 1.25
|
||||
north_ax.plot([0.85*np.cos(rad), 1.05*np.cos(rad)],
|
||||
[0.85*np.sin(rad), 1.05*np.sin(rad)],
|
||||
color='#555555', linewidth=0.8, zorder=5)
|
||||
if label:
|
||||
north_ax.text(r_text*np.cos(rad), r_text*np.sin(rad), label,
|
||||
fontsize=7, ha='center', va='center', color='#555555')
|
||||
|
||||
# Bottom info bar — fixed position
|
||||
info_ax = fig.add_axes([data_left, 0.02, data_width_frac + cbar_width + 0.02, 0.07])
|
||||
# Bottom info bar — enriched with source, method, date
|
||||
info_ax = fig.add_axes([data_left, 0.015, data_width_frac + cbar_width + 0.02, 0.09])
|
||||
info_ax.axis('off')
|
||||
|
||||
extent_km_x = (max_x - min_x) / 1000
|
||||
@ -465,42 +502,73 @@ def tif_to_png(tif_file, vis_dir, resolution, keep_tif=False):
|
||||
alt_min = float(np.nanmin(valid_data)) if len(valid_data) > 0 else 0
|
||||
alt_max = float(np.nanmax(valid_data)) if len(valid_data) > 0 else 0
|
||||
|
||||
# Build info lines
|
||||
line1_parts = []
|
||||
if gps_coords:
|
||||
nw_lat, nw_lon = gps_coords['NW']
|
||||
se_lat, se_lon = gps_coords['SE']
|
||||
info_text = (
|
||||
f"GPS: {nw_lat:.5f}N {nw_lon:.5f}E - {se_lat:.5f}N {se_lon:.5f}E | "
|
||||
f"EPSG:2154 | Res: {resolution}m/px | "
|
||||
f"Emprise: {extent_km_x:.1f}x{extent_km_y:.1f}km"
|
||||
)
|
||||
line1_parts.append(f"GPS: {nw_lat:.5f}°N {nw_lon:.5f}°E — {se_lat:.5f}°N {se_lon:.5f}°E")
|
||||
else:
|
||||
info_text = (
|
||||
f"EPSG:2154 | X: {min_x:.0f}-{max_x:.0f} Y: {min_y:.0f}-{max_y:.0f} | "
|
||||
f"Res: {resolution}m/px | Emprise: {extent_km_x:.1f}x{extent_km_y:.1f}km"
|
||||
)
|
||||
line1_parts.append(f"X: {min_x:.0f}–{max_x:.0f} Y: {min_y:.0f}–{max_y:.0f}")
|
||||
line1_parts.append(f"EPSG:2154")
|
||||
line1_parts.append(f"Res: {resolution}m/px")
|
||||
line1_parts.append(f"Emprise: {extent_km_x:.1f}×{extent_km_y:.1f}km")
|
||||
if not is_rgb:
|
||||
line1_parts.append(f"Alt: {alt_min:.1f}–{alt_max:.1f}m")
|
||||
|
||||
info_ax.text(0.01, 0.5, info_text,
|
||||
transform=info_ax.transAxes, fontsize=8.5,
|
||||
line2_parts = []
|
||||
line2_parts.append("Source: LiDAR HD IGN")
|
||||
if source_info:
|
||||
if source_info.get('method'):
|
||||
line2_parts.append(f"Classif.: {source_info['method'].upper()}")
|
||||
if source_info.get('date'):
|
||||
line2_parts.append(f"Date: {source_info['date']}")
|
||||
else:
|
||||
line2_parts.append(datetime.now().strftime("Date: %Y-%m-%d"))
|
||||
|
||||
info_text_line1 = " | ".join(line1_parts)
|
||||
info_text_line2 = " | ".join(line2_parts)
|
||||
|
||||
info_ax.text(0.01, 0.7, info_text_line1,
|
||||
transform=info_ax.transAxes, fontsize=8,
|
||||
verticalalignment='center', family='monospace',
|
||||
bbox=dict(boxstyle='round,pad=0.3', facecolor='#f0f0f0',
|
||||
bbox=dict(boxstyle='round,pad=0.2', facecolor='#f0f0f0',
|
||||
edgecolor='#aaaaaa', alpha=0.95))
|
||||
info_ax.text(0.01, 0.2, info_text_line2,
|
||||
transform=info_ax.transAxes, fontsize=7.5,
|
||||
verticalalignment='center', family='monospace',
|
||||
color='#444444',
|
||||
bbox=dict(boxstyle='round,pad=0.2', facecolor='#f8f8f8',
|
||||
edgecolor='#cccccc', alpha=0.9))
|
||||
|
||||
# Scale bar
|
||||
scale_m = 100
|
||||
# Scale bar — adaptive with alternating black/white segments
|
||||
extent_m_x = max_x - min_x
|
||||
scale_m, scale_label = _nice_scale(extent_m_x)
|
||||
pixels_per_meter = 1.0 / pixel_size_x
|
||||
scale_px = int(scale_m * pixels_per_meter)
|
||||
scale_bar_frac = scale_px / width
|
||||
bar_left = 0.80
|
||||
bar_bottom = 0.15
|
||||
bar_width_frac = min(scale_bar_frac, 0.15)
|
||||
bar_height = 0.35
|
||||
n_segments = 5
|
||||
segment_px = scale_px / n_segments
|
||||
bar_bottom_y = 0.55
|
||||
bar_top_y = 0.85
|
||||
bar_height = bar_top_y - bar_bottom_y
|
||||
scale_start_x = 0.80
|
||||
|
||||
info_ax.add_patch(RectPatch((bar_left, bar_bottom), bar_width_frac, bar_height,
|
||||
facecolor='black', edgecolor='black', linewidth=0.5,
|
||||
transform=info_ax.transAxes, clip_on=False))
|
||||
info_ax.text(bar_left + bar_width_frac / 2, bar_bottom + bar_height + 0.12,
|
||||
f"{scale_m} m", ha='center', va='bottom', fontsize=9, fontweight='bold',
|
||||
transform=info_ax.transAxes)
|
||||
for seg_i in range(n_segments):
|
||||
color = 'black' if seg_i % 2 == 0 else 'white'
|
||||
seg_left = scale_start_x + seg_i * segment_px / width
|
||||
seg_width_frac = segment_px / width
|
||||
info_ax.add_patch(RectPatch((seg_left, bar_bottom_y), seg_width_frac, bar_height,
|
||||
facecolor=color, edgecolor='black', linewidth=0.5,
|
||||
transform=info_ax.transAxes, clip_on=False))
|
||||
info_ax.text(scale_start_x + scale_px / (2 * width), bar_top_y + 0.12,
|
||||
f"{scale_label}", ha='center', va='bottom', fontsize=8, fontweight='bold',
|
||||
transform=info_ax.transAxes)
|
||||
# Scale end ticks
|
||||
info_ax.plot([scale_start_x, scale_start_x], [bar_bottom_y - 0.05, bar_top_y + 0.05],
|
||||
color='black', linewidth=1, transform=info_ax.transAxes, clip_on=False)
|
||||
info_ax.plot([scale_start_x + scale_px / width, scale_start_x + scale_px / width],
|
||||
[bar_bottom_y - 0.05, bar_top_y + 0.05],
|
||||
color='black', linewidth=1, transform=info_ax.transAxes, clip_on=False)
|
||||
|
||||
fig.patch.set_facecolor('white')
|
||||
|
||||
@ -527,145 +595,6 @@ def tif_to_png(tif_file, vis_dir, resolution, keep_tif=False):
|
||||
return None
|
||||
|
||||
|
||||
def convert_to_cog(tif_file, cog_dir):
|
||||
"""Convert a GeoTIFF to Cloud Optimized GeoTIFF for web map serving.
|
||||
|
||||
Args:
|
||||
tif_file: Path to input GeoTIFF.
|
||||
cog_dir: Directory for output COG file.
|
||||
|
||||
Returns:
|
||||
Path to output COG file, or None on failure.
|
||||
"""
|
||||
cog_file = cog_dir / tif_file.name
|
||||
|
||||
if cog_file.exists():
|
||||
logger.debug(f" COG déjà existant: {cog_file.name}")
|
||||
return cog_file
|
||||
|
||||
try:
|
||||
from rio_cogeo.cogeo import cog_translate
|
||||
from rio_cogeo.profiles import cog_profiles
|
||||
|
||||
with rasterio.open(tif_file) as src:
|
||||
is_rgb = src.count >= 3
|
||||
|
||||
# Use deflate compression profile
|
||||
profile = dict(cog_profiles["deflate"]) # Make a mutable copy
|
||||
|
||||
if not is_rgb:
|
||||
# Single-band terrain data: keep float32 for continuous values
|
||||
profile.update(dtype="float32")
|
||||
|
||||
cog_translate(str(tif_file), str(cog_file), profile, quiet=True)
|
||||
logger.debug(f" COG créé: {cog_file.name}")
|
||||
return cog_file
|
||||
|
||||
except ImportError:
|
||||
logger.warning(" rio-cogeo non disponible — COG non généré")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.warning(f" Erreur conversion COG: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def generate_cog_metadata(vis_dir, basename):
|
||||
"""Generate metadata JSON for all visualization layers.
|
||||
|
||||
Scans the COG directory for files and reads their geographic bounds.
|
||||
Creates a metadata.json with WGS84 bounds and layer info for the web viewer.
|
||||
|
||||
Args:
|
||||
vis_dir: Directory containing COG files (vis_dir/basename/cog/).
|
||||
basename: Base name for the LiDAR tile.
|
||||
|
||||
Returns:
|
||||
Path to metadata.json, or None on failure.
|
||||
"""
|
||||
import json
|
||||
|
||||
cog_dir = vis_dir / basename / "cog"
|
||||
if not cog_dir.exists():
|
||||
return None
|
||||
|
||||
# Find the DTM to get geographic bounds
|
||||
# The COG files inherit bounds from the DTM, so we can read any COG
|
||||
cog_files = sorted(cog_dir.glob("*.tif"))
|
||||
if not cog_files:
|
||||
return None
|
||||
|
||||
# Read bounds from first COG file
|
||||
try:
|
||||
ref_cog = cog_files[0]
|
||||
with rasterio.open(ref_cog) as src:
|
||||
bounds = src.bounds
|
||||
crs = src.crs
|
||||
if crs and HAS_WARP:
|
||||
l93_xs = [bounds.left, bounds.right]
|
||||
l93_ys = [bounds.top, bounds.bottom]
|
||||
lons, lats = warp_transform(crs, 'EPSG:4326', l93_xs, l93_ys)
|
||||
bounds_wgs84 = {
|
||||
'west': float(min(lons)),
|
||||
'south': float(min(lats)),
|
||||
'east': float(max(lons)),
|
||||
'north': float(max(lats)),
|
||||
}
|
||||
else:
|
||||
# Fallback: use Lambert 93 bounds directly
|
||||
bounds_wgs84 = {
|
||||
'west': float(bounds.left),
|
||||
'south': float(bounds.bottom),
|
||||
'east': float(bounds.right),
|
||||
'north': float(bounds.top),
|
||||
}
|
||||
|
||||
bounds_l93 = {
|
||||
'min_x': float(bounds.left),
|
||||
'min_y': float(bounds.bottom),
|
||||
'max_x': float(bounds.right),
|
||||
'max_y': float(bounds.top),
|
||||
}
|
||||
resolution = float(abs(src.transform.a))
|
||||
except Exception as e:
|
||||
logger.warning(f" Erreur lecture bounds COG: {e}")
|
||||
return None
|
||||
|
||||
# Build layer list with titles
|
||||
layers = []
|
||||
for cog_file in cog_files:
|
||||
name = cog_file.stem.replace(basename + "_", "")
|
||||
# Find matching title from COLORMAPS or RGB_LEGENDS
|
||||
title = name.replace("_", " ").title()
|
||||
for key in sorted(COLORMAPS.keys(), key=len, reverse=True):
|
||||
if key in name:
|
||||
title = COLORMAPS[key]['title']
|
||||
break
|
||||
for key in RGB_LEGENDS:
|
||||
if key in name:
|
||||
title = RGB_LEGENDS[key]['title']
|
||||
break
|
||||
|
||||
layers.append({
|
||||
'name': name,
|
||||
'title': title,
|
||||
'file': cog_file.name,
|
||||
})
|
||||
|
||||
metadata = {
|
||||
'basename': basename,
|
||||
'bounds_l93': bounds_l93,
|
||||
'bounds_wgs84': bounds_wgs84,
|
||||
'resolution': resolution,
|
||||
'layers': layers,
|
||||
}
|
||||
|
||||
metadata_file = vis_dir / basename / "metadata.json"
|
||||
with open(metadata_file, 'w') as f:
|
||||
json.dump(metadata, f, indent=2, ensure_ascii=False)
|
||||
|
||||
logger.info(f" Métadonnées: {len(layers)} couches, bounds={bounds_wgs84}")
|
||||
return metadata_file
|
||||
|
||||
|
||||
def generate_pdf_report(basename, vis_dir, pdf_dir, resolution):
|
||||
"""Generate A3 PDF report for a LiDAR file with all visualizations.
|
||||
|
||||
Reference in New Issue
Block a user