| """Cross-EO-product compound signal detection.""" |
| from __future__ import annotations |
|
|
| import numpy as np |
|
|
| from app.models import CompoundSignal |
|
|
|
|
| def compute_overlap_pct(mask_a: np.ndarray, mask_b: np.ndarray) -> float: |
| """Compute overlap percentage: intersection / min(count_a, count_b) * 100.""" |
| intersection = np.sum(mask_a & mask_b) |
| min_count = min(np.sum(mask_a), np.sum(mask_b)) |
| if min_count == 0: |
| return 0.0 |
| return float(intersection / min_count * 100) |
|
|
|
|
| def _tag_confidence(n_products: int, overlap_pct: float) -> str: |
| if n_products >= 3 and overlap_pct > 20: |
| return "strong" |
| if n_products >= 2 and overlap_pct >= 10: |
| return "moderate" |
| return "weak" |
|
|
|
|
| def detect_compound_signals( |
| zscore_rasters: dict[str, np.ndarray], |
| pixel_area_ha: float, |
| threshold: float = 2.0, |
| ) -> list[CompoundSignal]: |
| """Test for compound signal patterns across EO product z-score rasters.""" |
| decline: dict[str, np.ndarray] = {} |
| increase: dict[str, np.ndarray] = {} |
| for product_id, z in zscore_rasters.items(): |
| decline[product_id] = z < -threshold |
| increase[product_id] = z > threshold |
|
|
| signals: list[CompoundSignal] = [] |
|
|
| |
| if "ndvi" in decline and "buildup" in increase: |
| overlap = compute_overlap_pct(decline["ndvi"], increase["buildup"]) |
| triggered = overlap > 10 |
| affected = float(np.sum(decline["ndvi"] & increase["buildup"])) * pixel_area_ha |
| signals.append(CompoundSignal( |
| name="land_conversion", triggered=triggered, |
| confidence=_tag_confidence(2, overlap) if triggered else "weak", |
| description=( |
| f"NDVI decline overlaps with settlement growth ({overlap:.0f}% overlap, " |
| f"{affected:.1f} ha affected). Suggests possible vegetation loss to urbanization." |
| ) if triggered else "No land conversion signal detected.", |
| indicators=["ndvi", "buildup"], overlap_pct=overlap, affected_ha=affected, |
| )) |
|
|
| |
| if "sar" in decline and "water" in increase: |
| overlap = compute_overlap_pct(decline["sar"], increase["water"]) |
| triggered = overlap > 10 |
| affected = float(np.sum(decline["sar"] & increase["water"])) * pixel_area_ha |
| signals.append(CompoundSignal( |
| name="flood_event", triggered=triggered, |
| confidence=_tag_confidence(2, overlap) if triggered else "weak", |
| description=( |
| f"SAR backscatter decrease coincides with water extent increase " |
| f"({overlap:.0f}% overlap, {affected:.1f} ha). Suggests potential flooding." |
| ) if triggered else "No flood signal detected.", |
| indicators=["sar", "water"], overlap_pct=overlap, affected_ha=affected, |
| )) |
|
|
| |
| if "ndvi" in decline and "water" in decline and "sar" in increase: |
| combined = decline["ndvi"] & decline["water"] & increase["sar"] |
| n_combined = int(np.sum(combined)) |
| min_single = min(np.sum(decline["ndvi"]), np.sum(decline["water"]), np.sum(increase["sar"])) |
| overlap = float(n_combined / min_single * 100) if min_single > 0 else 0.0 |
| triggered = overlap > 10 |
| affected = n_combined * pixel_area_ha |
| signals.append(CompoundSignal( |
| name="drought_stress", triggered=triggered, |
| confidence=_tag_confidence(3, overlap) if triggered else "weak", |
| description=( |
| f"NDVI decline, water decline, and SAR increase co-occur " |
| f"({overlap:.0f}% overlap, {affected:.1f} ha). Suggests possible drought." |
| ) if triggered else "No drought signal detected.", |
| indicators=["ndvi", "water", "sar"], overlap_pct=overlap, affected_ha=affected, |
| )) |
|
|
| |
| if "buildup" in increase and "ndvi" in decline: |
| from scipy.ndimage import binary_dilation |
| expanded_buildup = binary_dilation(increase["buildup"], iterations=1) |
| adjacent_decline = expanded_buildup & decline["ndvi"] & ~increase["buildup"] |
| n_adjacent = int(np.sum(adjacent_decline)) |
| n_buildup = int(np.sum(increase["buildup"])) |
| overlap = float(n_adjacent / max(n_buildup, 1) * 100) |
| triggered = overlap > 10 and n_adjacent > 0 |
| affected = n_adjacent * pixel_area_ha |
| signals.append(CompoundSignal( |
| name="displacement_pressure", triggered=triggered, |
| confidence=_tag_confidence(2, overlap) if triggered else "weak", |
| description=( |
| f"Settlement growth hotspots are adjacent to NDVI decline areas " |
| f"({affected:.1f} ha of surrounding vegetation loss). " |
| f"Suggests expansion into previously vegetated land." |
| ) if triggered else "No displacement pressure signal detected.", |
| indicators=["ndvi", "buildup"], overlap_pct=overlap, affected_ha=affected, |
| )) |
|
|
| return signals |
|
|
