File size: 5,169 Bytes
df6bf75 ae4c60c df6bf75 ae4c60c df6bf75 ae4c60c df6bf75 ae4c60c df6bf75 ae4c60c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | """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] = []
# 1. Land conversion: NDVI decline + Settlement growth
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,
))
# 2. Flood event: SAR decrease + Water increase
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,
))
# 3. Drought stress: NDVI decline + Water decline + SAR increase
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,
))
# 4. Displacement pressure: Settlement growth + NDVI decline adjacent
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
|