190 lines
8.0 KiB
Python
190 lines
8.0 KiB
Python
from datetime import date, datetime
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from damage_engine.detectors import flood
|
|
from damage_engine.schemas import EVENT, NO_DATA, NO_EVENT, Severity
|
|
from damage_engine.sources.fields import Field
|
|
|
|
RUN_DAY = date(2025, 6, 12)
|
|
HOURS = 24 * flood.LOOKBACK_DAYS # 240 hourly records in the fetch window
|
|
|
|
|
|
def _rain(station=3, rain=None):
|
|
"""Hourly rain over the whole fetch window; `rain` = 240 values (default dry)."""
|
|
start, end = flood.fetch_window(RUN_DAY)
|
|
ts = pd.date_range(start, end, freq="h")
|
|
values = np.zeros(len(ts)) if rain is None else np.asarray(rain, dtype=float)
|
|
return pd.DataFrame({"CodiceStazione": station, "Timestamp": ts, flood.COL_RAIN: values})
|
|
|
|
|
|
def _with_day(day_values, before=0.0):
|
|
"""240 values: `before` mm/h in the 9 days before, then the 24 event-day values."""
|
|
return [before] * (HOURS - 24) + list(day_values)
|
|
|
|
|
|
def _field(station=3, bbch=None):
|
|
return Field(id=f"ST{station}", name=f"field {station}", station_id=station, bbch=bbch)
|
|
|
|
|
|
def _one(df, bbch=None, **kw):
|
|
[a] = flood.detect(df, [_field(bbch=bbch)], RUN_DAY, {3: "Lagnasco"}, **kw)
|
|
return a
|
|
|
|
|
|
def test_windows():
|
|
assert flood.event_window(RUN_DAY) == (datetime(2025, 6, 11, 7), datetime(2025, 6, 12, 7))
|
|
start, end = flood.fetch_window(RUN_DAY)
|
|
assert (start, end) == (datetime(2025, 6, 2, 8), datetime(2025, 6, 12, 7))
|
|
assert len(_rain()) == HOURS
|
|
|
|
|
|
def test_downpour_in_one_hour():
|
|
day = [0.0] * 24
|
|
day[10] = 35.0
|
|
a = _one(_rain(rain=_with_day(day)))
|
|
assert a.status == EVENT and a.details["criteria_met"] == ["downpour_1h"]
|
|
assert a.severity == Severity.LIGHT and a.duration_h == 1
|
|
assert a.start == datetime(2025, 6, 11, 18)
|
|
|
|
|
|
def test_contract_operator_at_least_30mm():
|
|
day = [0.0] * 24
|
|
day[0] = 30.0
|
|
assert _one(_rain(rain=_with_day(day))).status == EVENT
|
|
day[0] = 29.9
|
|
assert _one(_rain(rain=_with_day(day))).status == NO_EVENT
|
|
|
|
|
|
def test_severity_grows_with_ratio():
|
|
day = [0.0] * 24
|
|
day[5] = 95.0 # 95 / 30 = 3.2 -> extreme
|
|
a = _one(_rain(rain=_with_day(day)))
|
|
assert a.severity == Severity.EXTREME and a.severity_score == 1.0
|
|
|
|
|
|
def test_intense_rain_over_72_hours():
|
|
rain = [0.0] * (HOURS - 72) + [1.2] * 72 # 86.4 mm in 72 h, no downpour
|
|
a = _one(_rain(rain=rain))
|
|
assert a.status == EVENT and a.details["criteria_met"] == ["intense_72h"]
|
|
assert a.details["max_72h_mm"] == 86.4 and a.severity == Severity.LIGHT
|
|
|
|
|
|
def test_dry_day_after_heavy_rain_is_not_repeated():
|
|
rain = [0.0] * (HOURS - 96) + [1.2] * 72 + [0.0] * 24
|
|
a = _one(_rain(rain=rain))
|
|
assert a.status == NO_EVENT and a.details["max_72h_mm"] is None
|
|
assert a.details["rain_72h_to_run_mm"] > 0
|
|
|
|
|
|
def test_light_rain_is_no_event():
|
|
a = _one(_rain(rain=_with_day([0.5] * 24)))
|
|
assert a.status == NO_EVENT and "12.0 mm in the last 24 h" in a.summary
|
|
|
|
|
|
def test_prolonged_rain_only_with_reference_average():
|
|
rain = [0.375] * HOURS # 90 mm in 10 days, 27 mm in 72 h
|
|
a = _one(_rain(rain=rain))
|
|
assert a.status == NO_EVENT
|
|
assert any("No 5-year reference average" in n for n in a.data_notes)
|
|
a = _one(_rain(rain=rain), reference_10d_mm={3: 50.0}) # 90 > 1.5 x 50 and >= 80
|
|
assert a.status == EVENT and a.details["criteria_met"] == ["prolonged_10d"]
|
|
assert a.details["rain_trigger_ratio"] == round(90 / 80, 3) # trigger = max(75, 80)
|
|
assert not any("reference average" in n for n in a.data_notes)
|
|
a = _one(_rain(rain=rain), reference_10d_mm={3: 70.0}) # 90 < 1.5 x 70
|
|
assert a.status == NO_EVENT
|
|
|
|
|
|
def test_missing_hours_are_noted_and_counted_as_zero():
|
|
rain = _with_day([0.0] * 24)
|
|
rain[3], rain[HOURS - 1] = np.nan, np.nan
|
|
a = _one(_rain(rain=rain).drop(index=10)) # a missing row too
|
|
assert a.status == NO_EVENT
|
|
assert any("1 of 24 hours of the event day" in n for n in a.data_notes)
|
|
assert any("2 of 216 hours" in n for n in a.data_notes)
|
|
|
|
|
|
def test_no_records_is_no_data():
|
|
a = _one(_rain(station=5))
|
|
assert a.status == NO_DATA and a.data_notes
|
|
rain = _with_day([np.nan] * 24, before=0.1)
|
|
a = _one(_rain(rain=rain))
|
|
assert a.status == NO_DATA and "No rain data for the event day" in a.data_notes[0]
|
|
|
|
|
|
def _at(i):
|
|
"""Timestamp (ISO, minutes) of record i of the fetch window (record 0 = 2025-06-02 08:00)."""
|
|
start, _ = flood.fetch_window(RUN_DAY)
|
|
return (start + i * flood.HOUR).isoformat(timespec="minutes")
|
|
|
|
|
|
def test_last_heavy_72h_found_days_later_for_claims():
|
|
rain = [0.0] * HOURS
|
|
rain[150], rain[160], rain[180] = 30.0, 30.0, 30.0 # 90 mm in 31 h, 2-4 days ago
|
|
a = _one(_rain(rain=rain))
|
|
assert a.status == NO_EVENT # the event day is dry: no new event
|
|
heavy = a.details["last_heavy_72h"]
|
|
assert heavy == {"mm": 90.0, "first_rain": _at(150), "last_rain": _at(180),
|
|
"reached_at": _at(180), "partial": False}
|
|
assert heavy["first_rain"] == "2025-06-08T14:00" and heavy["last_rain"] == "2025-06-09T20:00"
|
|
|
|
|
|
def test_last_heavy_72h_is_the_latest_not_the_wettest():
|
|
rain = [0.0] * HOURS
|
|
rain[150], rain[230] = 100.0, 85.0 # both end inside the last 4 days
|
|
d = _one(_rain(rain=rain)).details
|
|
assert d["last_heavy_72h"]["mm"] == 85.0 and d["last_heavy_72h"]["reached_at"] == _at(230)
|
|
assert d["wettest_72h"] == {"mm": 100.0, "first_rain": _at(150), "last_rain": _at(150),
|
|
"partial": False}
|
|
|
|
|
|
def test_no_heavy_72h_gives_the_wettest_window():
|
|
d = _one(_rain(rain=_with_day([0.5] * 24))).details
|
|
assert d["last_heavy_72h"] is None
|
|
assert d["wettest_72h"]["mm"] == 12.0 and d["wettest_72h"]["first_rain"] == _at(HOURS - 24)
|
|
d = _one(_rain(rain=_with_day([0.0] * 24))).details
|
|
assert d["last_heavy_72h"] is None and d["wettest_72h"] is None # no rain at all
|
|
|
|
|
|
def test_heavy_rain_watch_counts_only_the_rain_of_the_last_4_days():
|
|
first = HOURS - 24 * flood.CLAIM_SEARCH_DAYS # first record of the watch period
|
|
assert flood.CLAIM_SEARCH_DAYS == 4 and first == 144 and _at(first) == "2025-06-08T08:00"
|
|
rain = [0.0] * HOURS
|
|
rain[first - 1] = 85.0 # 07:00 four days ago: just before the watch
|
|
d = _one(_rain(rain=rain)).details
|
|
assert d["last_heavy_72h"] is None and d["wettest_72h"] is None
|
|
rain[first] = 85.0 # first hour of the watch
|
|
heavy = _one(_rain(rain=rain)).details["last_heavy_72h"]
|
|
assert heavy["mm"] == 85.0 and heavy["first_rain"] == _at(first) # the earlier 85 mm left out
|
|
assert heavy["reached_at"] == _at(first) and heavy["partial"] is True # rain just before it
|
|
rain[first - 1] = 0.0
|
|
assert _one(_rain(rain=rain)).details["last_heavy_72h"]["partial"] is False
|
|
|
|
|
|
def test_heavy_rain_watch_storm_partly_before_the_4_days():
|
|
rain = [0.0] * HOURS
|
|
for i in range(110, 150): # 2 mm/h, 40 h; 6 h inside the watch
|
|
rain[i] = 2.0
|
|
d = _one(_rain(rain=rain)).details
|
|
assert d["last_heavy_72h"] is None # 80 mm in all, only 12 mm counted
|
|
assert d["wettest_72h"] == {"mm": 12.0, "first_rain": _at(144), "last_rain": _at(149),
|
|
"partial": True}
|
|
|
|
|
|
def test_last_heavy_72h_uses_the_contract_operator_at_least_80mm():
|
|
rain = [0.0] * (HOURS - 90) + [1.1] * 70 + [3.0] + [0.0] * 19 # 80 mm (float noise)
|
|
assert _one(_rain(rain=rain)).details["last_heavy_72h"]["mm"] == 80.0
|
|
rain[HOURS - 20] = 2.9 # 79.9 mm
|
|
assert _one(_rain(rain=rain)).details["last_heavy_72h"] is None
|
|
|
|
|
|
def test_cover_window_starts_at_fruit_set():
|
|
assert flood.in_cover_window(RUN_DAY, 65) is False # flowering: excluded
|
|
assert flood.in_cover_window(RUN_DAY, 72) is True
|
|
assert flood.in_cover_window(RUN_DAY, None) is None
|
|
assert flood.in_cover_window(date(2025, 11, 2), 85) is False
|
|
day = [0.0] * 24
|
|
day[0] = 40.0
|
|
assert _one(_rain(rain=_with_day(day)), bbch=72).details["in_cover_window"] is True
|