155 lines
7.4 KiB
Python
155 lines
7.4 KiB
Python
from datetime import date, datetime
|
|
|
|
import pandas as pd
|
|
|
|
from damage_engine.detectors import brina, freezing
|
|
from damage_engine.schemas import EVENT, NO_DATA, NO_EVENT, Severity
|
|
from damage_engine.sources import derived
|
|
from damage_engine.sources.fields import Field
|
|
|
|
RUN_DAY = date(2025, 3, 12)
|
|
# the 24 h checked for RUN_DAY: index 0-10 = day hours 07:00-17:00 of 11 March,
|
|
# index 11-23 = Brina's night window 18:00 (11 March) -> 06:00 (12 March)
|
|
HOURS = pd.date_range("2025-03-11 07:00", "2025-03-12 06:00", freq="h")
|
|
DAY, NIGHT = 11, 13
|
|
WARM_NIGHT = [3.0] * NIGHT
|
|
COLD_DAY = [1, 0.5, 0, -0.5, -1.5, -2.5, -1, 0, 1, 2, 3] + WARM_NIGHT # 4 h below 0 °C, 10:00-13:00
|
|
|
|
|
|
def _day(station, temps, rh=80.0, wind=None, solar=None, ts=None):
|
|
ts = HOURS[:len(temps)] if ts is None else ts
|
|
df = pd.DataFrame({"CodiceStazione": station, "Timestamp": ts, freezing.COL_TEMP: temps,
|
|
freezing.COL_RH: rh, freezing.COL_RH_MAX: rh})
|
|
if wind is not None:
|
|
df[freezing.COL_WIND] = wind
|
|
if solar is not None:
|
|
df[freezing.COL_SOLAR] = solar
|
|
return df
|
|
|
|
|
|
def _field(station, bbch=None, fid=None):
|
|
return Field(id=fid or f"ST{station}", name=f"field {station}", station_id=station, bbch=bbch)
|
|
|
|
|
|
def test_windows_match_brina():
|
|
assert freezing.day_window(RUN_DAY) == (datetime(2025, 3, 11, 6), datetime(2025, 3, 12, 6))
|
|
b_start, b_end = brina.night_window(RUN_DAY)
|
|
assert freezing.night_start(RUN_DAY) == b_start and freezing.day_window(RUN_DAY)[1] == b_end
|
|
|
|
|
|
def test_cold_day_is_freezing_fallback_severity():
|
|
[a] = freezing.detect(_day(3, COLD_DAY), [_field(3)], RUN_DAY, {3: "Lagnasco"})
|
|
assert a.status == EVENT and a.block == "freezing" and a.station_name == "Lagnasco"
|
|
assert a.duration_h == 4 # 0 °C hours do not count ("below 0 °C")
|
|
assert a.start == datetime(2025, 3, 11, 10) and a.end == datetime(2025, 3, 11, 13)
|
|
assert a.details["day_min_temp_c"] == -2.5
|
|
assert a.severity == Severity.MODERATE # fallback: -2 > -2.5 >= -4
|
|
assert a.details["severity_method"] == "fallback_min_bud_temp"
|
|
assert a.details["in_cover_window"] is None # stage unknown
|
|
|
|
|
|
def test_exactly_zero_is_not_freezing():
|
|
[a] = freezing.detect(_day(3, [0.0] * 24), [_field(3)], RUN_DAY)
|
|
assert a.status == NO_EVENT and a.severity == Severity.NONE
|
|
|
|
|
|
def test_sub_zero_night_is_left_to_brina():
|
|
[a] = freezing.detect(_day(3, [5.0] * DAY + [-3.0] * NIGHT), [_field(3)], RUN_DAY)
|
|
assert a.status == NO_EVENT
|
|
assert a.details["day_min_temp_c"] == 5.0
|
|
assert a.details["sunlit_hours_in_night_window"] == 0
|
|
|
|
|
|
def test_sunlit_hours_of_the_night_window_are_freezing():
|
|
solar = [400.0] * DAY + [200, 50, 5] + [0.0] * 9 + [30] # 18:00, 19:00 and 06:00 sunlit
|
|
temps = [5.0] * DAY + [-1.0] * NIGHT
|
|
[a] = freezing.detect(_day(566, temps, solar=solar), [_field(566)], RUN_DAY)
|
|
assert a.status == EVENT and a.duration_h == 3
|
|
assert a.details["sunlit_hours_in_night_window"] == 3 and a.details["daylight_hours"] == 14
|
|
assert a.start == datetime(2025, 3, 11, 18) and a.end == datetime(2025, 3, 12, 6)
|
|
|
|
|
|
def test_every_hour_goes_to_brina_or_freezing_never_both():
|
|
solar = [400.0] * DAY + [200, 50, 5] + [0.0] * 9 + [30]
|
|
for rh, wind in ((95.0, 0.5), (40.0, 5.0)): # brina all night: calm + humid, windy + dry
|
|
df = _day(566, [-3.0] * 24, rh=rh, wind=[wind] * 24, solar=solar)
|
|
frz = freezing.day_hours(df, RUN_DAY)
|
|
frz_hours = set(frz.loc[frz["freezing"], "Timestamp"])
|
|
night = df[df["Timestamp"] >= brina.night_window(RUN_DAY)[0]]
|
|
br = brina.brina_hours(night)
|
|
brina_hours_ = set(br.loc[br["brina"], "Timestamp"])
|
|
assert not frz_hours & brina_hours_
|
|
assert frz_hours | brina_hours_ == set(HOURS)
|
|
|
|
|
|
def test_severity_uses_field_stage_when_known():
|
|
bloom, dormant = freezing.detect(_day(3, COLD_DAY), [_field(3, 65, "A"), _field(3, 40, "B")], RUN_DAY)
|
|
assert bloom.details["stage"] == "full_bloom"
|
|
assert 0.10 < bloom.details["kill_fraction"] < 0.50 # -2.5 °C at full bloom
|
|
assert bloom.severity == Severity.MODERATE
|
|
assert bloom.details["in_cover_window"] is True
|
|
assert dormant.status == EVENT and dormant.severity == Severity.NONE
|
|
|
|
|
|
def test_out_of_cover_window_flagged():
|
|
df = _day(3, COLD_DAY, ts=pd.date_range("2025-10-18 07:00", periods=24, freq="h"))
|
|
[a] = freezing.detect(df, [_field(3, 65)], date(2025, 10, 19))
|
|
assert a.status == EVENT and a.details["in_cover_window"] is False
|
|
|
|
|
|
def test_wind_and_wet_bulb_are_reported_not_used_to_detect():
|
|
windy = _day(3, COLD_DAY, wind=[3.0] * 24)
|
|
calm = _day(6, COLD_DAY, wind=[0.5] * 24)
|
|
no_wind = _day(9, COLD_DAY)
|
|
out = freezing.detect(pd.concat([windy, calm, no_wind]), [_field(3), _field(6), _field(9)], RUN_DAY)
|
|
assert [a.status for a in out] == [EVENT] * 3 and len({a.duration_h for a in out}) == 1
|
|
assert [a.details["advective_wind"] for a in out] == [True, False, None]
|
|
assert out[0].details["mean_wind_ms"] == 3.0 and out[2].details["mean_wind_ms"] is None
|
|
rows = freezing.day_hours(windy, RUN_DAY)
|
|
assert rows["wet_bulb_c"].equals(rows[derived.WET_BULB_MIN]) # the shared value
|
|
assert out[0].details["min_wet_bulb_c"] == round(float(rows.loc[rows["freezing"], "wet_bulb_c"].min()), 2)
|
|
|
|
|
|
def test_rows_outside_the_24_hours_are_ignored():
|
|
df = _day(3, [5.0] * 24)
|
|
extra = _day(3, [-10.0, -10.0], ts=pd.to_datetime(["2025-03-11 06:00", "2025-03-12 07:00"]))
|
|
[a] = freezing.detect(pd.concat([df, extra]), [_field(3)], RUN_DAY)
|
|
assert a.status == NO_EVENT and a.details["records"] == DAY
|
|
|
|
|
|
def test_no_records_or_no_daylight_temperature_is_no_data():
|
|
night_only = _day(3, [5.0] * NIGHT, ts=HOURS[DAY:])
|
|
missing, no_day = freezing.detect(night_only, [_field(77), _field(3)], RUN_DAY)
|
|
assert missing.status == NO_DATA
|
|
assert missing.data_notes == ["No records from the station for yesterday's daylight hours."]
|
|
assert no_day.status == NO_DATA
|
|
assert "Air temperature missing for 11 of 11 daylight hours" in " ".join(no_day.data_notes)
|
|
|
|
|
|
def test_missing_sensors_are_noted():
|
|
[a] = freezing.detect(_day(9, COLD_DAY), [_field(9)], RUN_DAY,
|
|
station_sensors={9: {"TemperaturaAria", "UmiditaRelativa"}})
|
|
assert a.status == EVENT
|
|
assert any(n.startswith("No wind sensor at this station") for n in a.data_notes)
|
|
assert any(n.startswith("No solar radiation sensor at this station: daylight = the fixed day "
|
|
"hours (records 07:00-17:00)") for n in a.data_notes)
|
|
|
|
|
|
def test_partly_missing_hours_are_noted():
|
|
solar = [400.0] * DAY + [0.0] * 10 + [None] * 3
|
|
df = _day(566, COLD_DAY, wind=[1.0] * 24, solar=solar).drop(index=[0, 1]) # no 07:00, 08:00 rows
|
|
df.loc[2, freezing.COL_TEMP] = None
|
|
df[freezing.COL_TEMP_FALLBACK] = 0.0
|
|
[a] = freezing.detect(df, [_field(566)], RUN_DAY)
|
|
text = " ".join(a.data_notes)
|
|
assert "Solar radiation missing for 3 of 13 hours of the night window" in text
|
|
assert "Air temperature missing for 2 of 11 daylight hours" in text # the two missing rows
|
|
assert "Hourly minimum temperature missing for 1 of 9 daylight hours: hourly mean used" in text
|
|
|
|
|
|
def test_complete_station_has_no_notes():
|
|
solar = [400.0] * DAY + [0.0] * NIGHT
|
|
[a] = freezing.detect(_day(566, COLD_DAY, wind=[1.0] * 24, solar=solar), [_field(566)], RUN_DAY)
|
|
assert a.status == EVENT and a.data_notes == []
|
|
assert a.details["solar_checked"] is True
|