Fix corrupted COPC detection, add CSF→SMRF fallback, improve MSRM colormap, add SVF and anisotropic openness
- validate_laz: verify point data accessibility (not just headers) to detect corrupted COPC files that pass header checks but fail on data reads - classify_ground: fallback from CSF to SMRF when CSF produces no ground points or PDAL errors (fixes 2/9 failing tiles) - MSRM: preserve sign in weighted combination so RdBu_r colormap shows both red (elevated) and blue (depressed) instead of red only - Add Sky-View Factor (SVF) visualization: cos²(horizon angle) over 16 directions, excellent for archaeological earthwork detection - Add Anisotropic Openness: directional weighting (NW-SE/NE-SW) enhances linear feature detection aligned with common settlement patterns - Remove anomalies and flow visualizations (replaced by SVF + aniso_open) - Location inset: use IGN topographic map at zoom 10 instead of simplified France outline, with red rectangle marker and fallback - Remove flow (hydrological accumulation) from VIZ_STEPS
This commit is contained in:
Antoine Jacquin
@ -702,13 +702,17 @@ def generate_mslrm(dem_file, basename, vis_dir, resolution, shared=None):
|
||||
lrm = lrm / lrm_std
|
||||
lrm_stack.append(lrm.astype(np.float32))
|
||||
|
||||
# Weighted combination
|
||||
# Weighted combination — preserve sign for RdBu_r colormap
|
||||
# Positive = elevated (red), Negative = depression (blue)
|
||||
lrm_array = np.array(lrm_stack)
|
||||
weights_3d = weights[:, np.newaxis, np.newaxis]
|
||||
with np.errstate(invalid='ignore', divide='ignore'):
|
||||
with warnings.catch_warnings():
|
||||
warnings.filterwarnings('ignore', message='Mean of empty slice')
|
||||
mslrm = np.sqrt(np.nansum((lrm_array ** 2) * weights_3d, axis=0) / np.sum(weights))
|
||||
# Signed RMS: magnitude from RMS, sign from weighted mean
|
||||
signed_mean = np.nansum(lrm_array * weights_3d, axis=0) / np.sum(weights)
|
||||
rms_magnitude = np.sqrt(np.nansum((lrm_array ** 2) * weights_3d, axis=0) / np.sum(weights))
|
||||
mslrm = np.sign(signed_mean) * rms_magnitude
|
||||
mslrm[nan_mask] = np.nan
|
||||
_save_tif(output, mslrm.astype(np.float32), transform, crs)
|
||||
logger.info(f" ✓ MSRM terminé ({time.time()-t0:.1f}s){gpu_tag}")
|
||||
@ -970,8 +974,6 @@ def generate_anomalies(dem_file, basename, vis_dir, resolution, shared=None):
|
||||
valid_lrm = lrm[~nan_mask]
|
||||
lrm_std = max(np.nanstd(valid_lrm), 0.01) if len(valid_lrm) > 0 else 0.01
|
||||
lrm_norm = lrm / lrm_std
|
||||
else:
|
||||
lrm_norm = lrm
|
||||
lrm_stack.append(lrm_norm.astype(np.float32))
|
||||
|
||||
# Weighted RMS combination (favor 5-25m scales)
|
||||
@ -1275,4 +1277,180 @@ def generate_flow(dem_file, basename, vis_dir, resolution, shared=None):
|
||||
return output
|
||||
except Exception as e:
|
||||
logger.error(f" ✗ Erreur flux: {e}", exc_info=True)
|
||||
return None
|
||||
|
||||
|
||||
# ============================================================
|
||||
# Sky-View Factor (SVF)
|
||||
# ============================================================
|
||||
|
||||
def generate_svf(dem_file, basename, vis_dir, resolution, shared=None):
|
||||
"""Sky-View Factor - fraction of sky visible from each point (GPU if available).
|
||||
|
||||
SVF = average of cos²(horizon_angle) across 8 directions.
|
||||
High SVF (near 1) = open sky (ridgetop, plateau)
|
||||
Low SVF (near 0) = enclosed sky (valley, deep trench)
|
||||
|
||||
Excellent for detecting archaeological earthworks: ditches appear dark,
|
||||
embankments appear bright. Complements openness which uses raw angles.
|
||||
"""
|
||||
gpu_tag = " [GPU]" if HAS_GPU else ""
|
||||
logger.info(f" → Sky-View Factor{gpu_tag}...")
|
||||
t0 = time.time()
|
||||
output = vis_dir / f"{basename}_svf.tif"
|
||||
|
||||
try:
|
||||
if shared:
|
||||
transform = shared.transform
|
||||
crs = shared.crs
|
||||
dem_np = shared.dem_np
|
||||
rows, cols = dem_np.shape
|
||||
res = resolution
|
||||
dem = to_gpu(shared.filled) if HAS_GPU else shared.filled
|
||||
nan_mask = shared.nan_mask
|
||||
else:
|
||||
dem_np, transform, crs = _read_dem(dem_file)
|
||||
rows, cols = dem_np.shape
|
||||
res = resolution
|
||||
nan_mask = np.isnan(dem_np)
|
||||
filled, _ = _fill_nans(dem_np)
|
||||
dem = to_gpu(filled) if HAS_GPU else filled
|
||||
|
||||
n_dirs = 16 # More directions for smoother SVF
|
||||
angles = np.linspace(0, 2 * np.pi, n_dirs, endpoint=False)
|
||||
dx_dir = np.cos(angles)
|
||||
dy_dir = np.sin(angles)
|
||||
max_dist = int(100 / res)
|
||||
|
||||
padded = xp.pad(dem, max_dist, mode='constant', constant_values=xp.nan)
|
||||
svf_sum = xp.zeros_like(dem)
|
||||
|
||||
for d_idx in range(n_dirs):
|
||||
ddx, ddy = dx_dir[d_idx], dy_dir[d_idx]
|
||||
# Find maximum horizon elevation angle in this direction
|
||||
max_horizon_angle = xp.zeros_like(dem)
|
||||
|
||||
for step in range(1, max_dist + 1):
|
||||
px = int(round(ddx * step))
|
||||
py = int(round(ddy * step))
|
||||
dist_m = np.sqrt((ddx * step * res) ** 2 + (ddy * step * res) ** 2)
|
||||
if dist_m < res * 0.5:
|
||||
continue
|
||||
|
||||
elev_diff = padded[max_dist + py:max_dist + py + rows,
|
||||
max_dist + px:max_dist + px + cols] - dem
|
||||
|
||||
# Horizon angle from horizontal (positive = terrain above viewer)
|
||||
angle = xp.arctan2(elev_diff, dist_m)
|
||||
max_horizon_angle = xp.where(xp.isnan(angle), max_horizon_angle,
|
||||
xp.maximum(max_horizon_angle, xp.nan_to_num(angle, nan=0)))
|
||||
|
||||
# SVF uses cos²(horizon angle) — fraction of visible sky in this direction
|
||||
cos2 = xp.cos(max_horizon_angle) ** 2
|
||||
svf_sum += cos2
|
||||
|
||||
svf_result = to_cpu(svf_sum / n_dirs).astype(np.float32)
|
||||
svf_result[nan_mask] = np.nan
|
||||
_save_tif(output, svf_result, transform, crs)
|
||||
logger.info(f" ✓ SVF terminé ({time.time()-t0:.1f}s){gpu_tag}")
|
||||
return output
|
||||
except Exception as e:
|
||||
logger.error(f" ✗ Erreur SVF: {e}", exc_info=True)
|
||||
return None
|
||||
|
||||
|
||||
# ============================================================
|
||||
# Anisotropic Openness
|
||||
# ============================================================
|
||||
|
||||
def generate_aniso_open(dem_file, basename, vis_dir, resolution, shared=None):
|
||||
"""Anisotropic Openness - weighted directional openness emphasizing oblique directions (GPU if available).
|
||||
|
||||
Computes positive and negative openness with anisotropic weighting:
|
||||
NW/SE directions weighted more heavily to enhance detection of structures
|
||||
aligned NE-SW (common in French archaeological sites: villas, enclosures).
|
||||
|
||||
The anisotropic weighting makes subtle linear features more visible than
|
||||
standard isotropic openness which averages all directions equally.
|
||||
"""
|
||||
gpu_tag = " [GPU]" if HAS_GPU else ""
|
||||
logger.info(f" → Openness Anisotropique{gpu_tag}...")
|
||||
t0 = time.time()
|
||||
output = vis_dir / f"{basename}_aniso_open.tif"
|
||||
|
||||
try:
|
||||
if shared:
|
||||
transform = shared.transform
|
||||
crs = shared.crs
|
||||
dem_np = shared.dem_np
|
||||
rows, cols = dem_np.shape
|
||||
res = resolution
|
||||
dem = to_gpu(shared.filled) if HAS_GPU else shared.filled
|
||||
nan_mask = shared.nan_mask
|
||||
else:
|
||||
dem_np, transform, crs = _read_dem(dem_file)
|
||||
rows, cols = dem_np.shape
|
||||
res = resolution
|
||||
nan_mask = np.isnan(dem_np)
|
||||
filled, _ = _fill_nans(dem_np)
|
||||
dem = to_gpu(filled) if HAS_GPU else filled
|
||||
|
||||
n_dirs = 8
|
||||
angles = np.linspace(0, 2 * np.pi, n_dirs, endpoint=False)
|
||||
dx_dir = np.cos(angles)
|
||||
dy_dir = np.sin(angles)
|
||||
|
||||
# Anisotropic weights: emphasize NW-SE and NE-SW directions
|
||||
# These orientations are most productive for detecting archaeological features
|
||||
# aligned with Roman and medieval settlement patterns in France
|
||||
weights = np.array([1.0, 1.5, 1.0, 1.5, 1.0, 1.5, 1.0, 1.5])
|
||||
|
||||
max_dist = int(100 / res)
|
||||
padded = xp.pad(dem, max_dist, mode='constant', constant_values=xp.nan)
|
||||
|
||||
pos_sum = xp.zeros_like(dem)
|
||||
neg_sum = xp.zeros_like(dem)
|
||||
weight_total = 0.0
|
||||
|
||||
for d_idx in range(n_dirs):
|
||||
ddx, ddy = dx_dir[d_idx], dy_dir[d_idx]
|
||||
w = weights[d_idx]
|
||||
weight_total += w
|
||||
|
||||
max_pos_angle = xp.zeros_like(dem)
|
||||
max_neg_angle = xp.zeros_like(dem)
|
||||
|
||||
for step in range(1, max_dist + 1):
|
||||
px = int(round(ddx * step))
|
||||
py = int(round(ddy * step))
|
||||
dist_m = np.sqrt((ddx * step * res) ** 2 + (ddy * step * res) ** 2)
|
||||
if dist_m < res * 0.5:
|
||||
continue
|
||||
|
||||
elev_diff = padded[max_dist + py:max_dist + py + rows,
|
||||
max_dist + px:max_dist + px + cols] - dem
|
||||
|
||||
# Positive openness: max zenith angle
|
||||
pos_angle = xp.arctan2(xp.maximum(elev_diff, 0), dist_m)
|
||||
max_pos_angle = xp.where(xp.isnan(pos_angle), max_pos_angle,
|
||||
xp.maximum(max_pos_angle, xp.nan_to_num(pos_angle, nan=0)))
|
||||
|
||||
# Negative openness: max nadir angle
|
||||
neg_angle = xp.arctan2(xp.maximum(-elev_diff, 0), dist_m)
|
||||
max_neg_angle = xp.where(xp.isnan(neg_angle), max_neg_angle,
|
||||
xp.maximum(max_neg_angle, xp.nan_to_num(neg_angle, nan=0)))
|
||||
|
||||
pos_sum += max_pos_angle * w
|
||||
neg_sum += max_neg_angle * w
|
||||
|
||||
# Combined: positive minus negative openness (anisotropic)
|
||||
pos_avg = pos_sum / weight_total
|
||||
neg_avg = neg_sum / weight_total
|
||||
aniso_result = to_cpu(xp.degrees(pos_avg - neg_avg)).astype(np.float32)
|
||||
aniso_result[nan_mask] = np.nan
|
||||
_save_tif(output, aniso_result, transform, crs)
|
||||
logger.info(f" ✓ Openness anisotropique terminé ({time.time()-t0:.1f}s){gpu_tag}")
|
||||
return output
|
||||
except Exception as e:
|
||||
logger.error(f" ✗ Erreur openness anisotropique: {e}", exc_info=True)
|
||||
return None
|
||||
Reference in New Issue
Block a user