from datetime import date, datetime import numpy as np import pandas as pd import pytest from damage_engine.detectors import hot_wind from damage_engine.schemas import EVENT, NO_DATA, NO_EVENT, Severity from damage_engine.sources.fields import Field RUN_DAY = date(2026, 7, 20) # the 24 h checked for RUN_DAY: records 08:00 of 19 Jul ... 07:00 of 20 Jul HOURS = pd.date_range("2026-07-19 08:00", "2026-07-20 07:00", freq="h") MILD = [30.0] * 24 BREEZE = [2.5] * 24 ALL_SENSORS = {"TemperaturaAria", "UmiditaRelativa", "VelocitaVento2m"} def _day(station, t_max, wind=None, t_mean=None, wind_max=None, rh=None, ts=None): ts = HOURS[:len(t_max)] if ts is None else ts n = len(ts) return pd.DataFrame({ "CodiceStazione": station, "Timestamp": ts, hot_wind.COL_T_MAX: t_max, hot_wind.COL_T_MEAN: [t - 2 if t is not None and not np.isnan(t) else np.nan for t in t_max] if t_mean is None else t_mean, hot_wind.COL_WIND: BREEZE[:n] if wind is None else wind, hot_wind.COL_WIND_MAX: [np.nan] * n if wind_max is None else wind_max, hot_wind.COL_RH: [25.0] * n if rh is None else rh, }) def _field(station, bbch=None): return Field(id=f"ST{station}", name=f"field {station}", station_id=station, bbch=bbch) def _with_heat(values, at=6, base=MILD): t = list(base) t[at:at + len(values)] = values return t def test_window_is_the_24h_ending_at_07(): assert hot_wind.RUN_HOUR == 7 assert hot_wind.day_window(RUN_DAY) == (datetime(2026, 7, 19, 7), datetime(2026, 7, 20, 7)) def test_hot_moving_air_is_an_event_graded_on_temperature(): # 14:00-16:00: 40.5, 42.3, 41.0 °C hourly maximum, breeze 2.5 m/s df = _day(3, _with_heat([40.5, 42.3, 41.0])) [a] = hot_wind.detect(df, [_field(3)], RUN_DAY, {3: "Lagnasco"}, {3: ALL_SENSORS}) assert a.status == EVENT and a.block == "hot_wind" and a.station_name == "Lagnasco" assert a.duration_h == 3 assert a.start == datetime(2026, 7, 19, 14) and a.end == datetime(2026, 7, 19, 16) d = a.details assert d["max_temp_c"] == 42.3 and d["max_temp_at"] == "2026-07-19T15:00" assert d["max_hot_wind_temp_c"] == 42.3 and d["hot_wind_hours"] == 3 and d["hot_hours"] == 3 assert d["max_mean_temp_c"] == 40.3 assert d["wind_checked"] is True and d["wind_assumed_hours"] == 0 assert d["wind_at_max_temp_ms"] == 2.5 and d["min_humidity_pct"] == 25 assert d["direction_checked"] is False and d["harvest_checked"] is False assert a.severity == Severity.MODERATE # 42 <= 42.3 < 44 assert d["in_cover_window"] is None # stage and harvest unknown assert "42.3 °C" in a.summary assert hot_wind.DIRECTION_NOTE in a.data_notes assert any("30 days before harvest" in n for n in a.data_notes) def test_trigger_is_at_least_40(): [hit] = hot_wind.detect(_day(3, _with_heat([40.0])), [_field(3)], RUN_DAY) [miss] = hot_wind.detect(_day(3, _with_heat([39.9])), [_field(3)], RUN_DAY) assert hit.status == EVENT and hit.severity == Severity.LIGHT and hit.duration_h == 1 assert miss.status == NO_EVENT and miss.severity == Severity.NONE assert miss.details["max_temp_c"] == 39.9 and miss.details["hot_hours"] == 0 assert miss.summary == "No hot wind: highest air temperature 39.9 °C." @pytest.mark.parametrize("temp, level", [(42.0, Severity.MODERATE), (44.0, Severity.SEVERE), (46.0, Severity.EXTREME), (48.0, Severity.EXTREME)]) def test_temperature_bands(temp, level): [a] = hot_wind.detect(_day(3, _with_heat([temp])), [_field(3)], RUN_DAY) assert a.severity == level def test_still_air_is_not_hot_wind(): wind = list(BREEZE) wind[6] = 0.6 # 14:00: 41 °C, still air [a] = hot_wind.detect(_day(3, _with_heat([41.0]), wind=wind), [_field(3)], RUN_DAY, station_sensors={3: ALL_SENSORS}) assert a.status == NO_EVENT assert a.details["hot_hours"] == 1 and a.details["still_hot_hours"] == 1 assert a.summary.endswith("1 hour(s) at 40 °C or more had still air.") def test_wind_exactly_at_the_minimum_counts_as_moving(): wind = list(BREEZE) wind[6] = 1.0 [a] = hot_wind.detect(_day(3, _with_heat([41.0]), wind=wind), [_field(3)], RUN_DAY) assert a.status == EVENT def test_severity_uses_the_hot_wind_hours_only(): wind = list(BREEZE) wind[7] = 0.2 # the 45 °C hour had still air [a] = hot_wind.detect(_day(3, _with_heat([41.0, 45.0]), wind=wind), [_field(3)], RUN_DAY) assert a.status == EVENT and a.duration_h == 1 assert a.details["max_temp_c"] == 45.0 and a.details["max_hot_wind_temp_c"] == 41.0 assert a.severity == Severity.LIGHT def test_station_without_anemometer_assumes_moving_air(): df = _day(9, _with_heat([41.0]), wind=[np.nan] * 24) [a] = hot_wind.detect(df, [_field(9)], RUN_DAY, station_sensors={9: {"TemperaturaAria", "UmiditaRelativa"}}) assert a.status == EVENT assert a.details["wind_checked"] is False and a.details["wind_assumed_hours"] == 1 assert a.details["wind_at_max_temp_ms"] is None assert ("No wind sensor (VelocitaVento2m) at this station: air movement not checked, " "taken as moving (worst case).") in a.data_notes def test_missing_wind_hours_are_taken_as_moving(): wind = list(BREEZE) wind[6] = np.nan [a] = hot_wind.detect(_day(3, _with_heat([41.0]), wind=wind), [_field(3)], RUN_DAY, station_sensors={3: ALL_SENSORS}) assert a.status == EVENT and a.details["wind_assumed_hours"] == 1 assert "Wind missing for 1 of 24 hours: air taken as moving in those hours (worst case)." in a.data_notes def test_missing_mean_wind_uses_the_maximum(): wind, wind_max = list(BREEZE), [np.nan] * 24 wind[6], wind_max[6] = np.nan, 0.8 # even the maximum is still air [a] = hot_wind.detect(_day(3, _with_heat([41.0]), wind=wind, wind_max=wind_max), [_field(3)], RUN_DAY) assert a.status == NO_EVENT and a.details["still_hot_hours"] == 1 assert "Hourly mean wind missing for 1 of 24 hours: hourly maximum used." in a.data_notes def test_missing_maximum_temperature_uses_the_mean(): t_max = _with_heat([np.nan]) t_mean = [28.0] * 24 t_mean[6] = 40.2 # only the mean of 14:00 is known [a] = hot_wind.detect(_day(3, t_max, t_mean=t_mean), [_field(3)], RUN_DAY) assert a.status == EVENT and a.details["fallback_hours"] == 1 assert any("Hourly maximum temperature missing for 1 of 24 hours: hourly mean used" in n for n in a.data_notes) def test_hours_outside_the_24h_are_ignored(): ts = pd.date_range("2026-07-19 07:00", "2026-07-20 08:00", freq="h") # 26 records t = [45.0] + MILD + [45.0] [a] = hot_wind.detect(_day(3, t, ts=ts, wind=[2.5] * 26, rh=[25.0] * 26), [_field(3)], RUN_DAY) assert a.status == NO_EVENT and a.details["records"] == 24 def test_cover_window_fruit_set_to_31_october(): df = _day(3, _with_heat([41.0])) [after, before] = hot_wind.detect(df, [_field(3, bbch=75), _field(3, bbch=65)], RUN_DAY) assert after.details["in_cover_window"] is None # fruit set, harvest date unknown assert before.details["in_cover_window"] is False # still in bloom assert after.severity == before.severity # severity does not depend on the stage assert hot_wind.in_cover_window(date(2026, 11, 2), 80) is False def test_missing_humidity_is_noted_only(): df = _day(3, _with_heat([41.0]), rh=[np.nan] * 24) [a] = hot_wind.detect(df, [_field(3)], RUN_DAY, station_sensors={3: ALL_SENSORS}) assert a.status == EVENT and a.details["min_humidity_pct"] is None assert "No humidity data in the 24 hours: humidity not reported (not used to detect)." in a.data_notes def test_missing_hours_are_noted_and_not_checked(): df = _day(3, _with_heat([41.0])).drop(index=[0, 1, 2]) # 3 hours with no record [a] = hot_wind.detect(df, [_field(3)], RUN_DAY) assert a.status == EVENT and a.details["records"] == 21 assert "Air temperature missing for 3 of 24 hours: those hours not checked." in a.data_notes def test_no_temperature_values_is_no_data(): df = _day(3, [np.nan] * 24, t_mean=[np.nan] * 24) [a] = hot_wind.detect(df, [_field(3)], RUN_DAY) assert a.status == NO_DATA assert a.data_notes == ["No air temperature data in the 24 hours checked: hot wind cannot be judged."] def test_missing_columns_do_not_break_the_block(): df = pd.DataFrame({"CodiceStazione": 3, "Timestamp": HOURS}) # no columns at all [a] = hot_wind.detect(df, [_field(3)], RUN_DAY) assert a.status == NO_DATA def test_station_without_records_is_no_data(): [a] = hot_wind.detect(pd.DataFrame(), [_field(3)], RUN_DAY) assert a.status == NO_DATA and a.data_notes == ["No records from the station for the last 24 hours."]