251 lines
12 KiB
Python
251 lines
12 KiB
Python
from datetime import date, datetime
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from damage_engine.detectors import high_temperature as ht
|
|
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(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
|
|
NOON = 6 # index of the 14:00 record
|
|
SUN = [800.0] * 11 + [0.0] * 13 # 08:00 ... 18:00 sunny, then night
|
|
BREEZE = [2.0] * 24
|
|
ALL_SENSORS = {"TemperaturaAria", "UmiditaRelativa", "VelocitaVento2m", "RadiazioneSolare"}
|
|
NO_SOLAR = {"TemperaturaAria", "UmiditaRelativa"}
|
|
|
|
|
|
def _day(station, t_max, solar=None, wind=None, t_mean=None, rh=None, ts=None):
|
|
ts = HOURS[:len(t_max)] if ts is None else ts
|
|
n = len(ts)
|
|
nan = [np.nan] * n
|
|
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)
|
|
rh = [30.0] * n if rh is None else rh
|
|
return pd.DataFrame({
|
|
"CodiceStazione": station, "Timestamp": ts,
|
|
derived.T_MAX: t_max, derived.T_MEAN: t_mean,
|
|
derived.T_MIN: [t - 4 if not np.isnan(t) else np.nan for t in t_mean],
|
|
derived.RH_MIN: rh, derived.RH_MEAN: rh, derived.RH_MAX: rh,
|
|
ht.COL_SOLAR: nan if solar is None else solar,
|
|
ht.COL_WIND: nan if wind is None else wind,
|
|
})
|
|
|
|
|
|
def _field(station, bbch=None):
|
|
return Field(id=f"ST{station}", name=f"field {station}", station_id=station, bbch=bbch)
|
|
|
|
|
|
def _with(values, at=NOON, base=MILD):
|
|
out = list(base)
|
|
out[at:at + len(values)] = values
|
|
return out
|
|
|
|
|
|
def test_window_is_the_24h_ending_at_07():
|
|
assert ht.RUN_HOUR == 7
|
|
assert ht.day_window(RUN_DAY) == (datetime(2026, 7, 19, 7), datetime(2026, 7, 20, 7))
|
|
|
|
|
|
def test_fruit_surface_temperature_formula():
|
|
fst = ht.fruit_surface_c([40.0, 40.0, 40.0, 40.0, 40.0, 40.0],
|
|
[1000.0, 1000.0, 299.0, 300.0, np.nan, 1000.0],
|
|
[0.0, 2.0, 0.0, 0.0, 0.0, np.nan])
|
|
assert fst[0] == pytest.approx(55.0) # calm, full sun: + 15 °C
|
|
assert fst[1] == pytest.approx(40 + 15 / 1.6) # wind cools the fruit
|
|
assert fst[2] == 40.0 # below 300 W/m²: no sun gain
|
|
assert fst[3] == pytest.approx(44.5) # 300 W/m² counts
|
|
assert np.isnan(fst[4]) # no solar value: no FST
|
|
assert fst[5] == pytest.approx(55.0) # no wind value: calm (worst case)
|
|
|
|
|
|
def test_event_without_solar_is_graded_on_the_air():
|
|
# 14:00-16:00: 40.5, 42.3, 41.0 °C hourly maximum, no pyranometer (11 of 14 stations)
|
|
df = _day(3, _with([40.5, 42.3, 41.0]), wind=BREEZE)
|
|
[a] = ht.detect(df, [_field(3)], RUN_DAY, {3: "Lagnasco"}, {3: NO_SOLAR | {"VelocitaVento2m"}})
|
|
assert a.status == EVENT and a.block == "high_temperature" 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["hot_hours"] == 3 and d["max_mean_temp_c"] == 40.3
|
|
assert d["severity_method"] == "air" and d["severity_value_c"] == 42.3
|
|
assert d["solar_checked"] is False and d["max_fst_c"] is None
|
|
assert d["sunburn_risk_hours"] is None
|
|
assert a.severity == Severity.MODERATE # air bands: 42 <= 42.3 < 44
|
|
assert d["in_cover_window"] is None # stage unknown
|
|
assert a.summary == "High temperature for 3 h, highest air temperature 42.3 °C: severity moderate."
|
|
assert any(n.startswith("No solar radiation sensor (RadiazioneSolare) at this station") for n in a.data_notes)
|
|
|
|
|
|
def test_trigger_is_at_least_40():
|
|
[hit] = ht.detect(_day(3, _with([40.0])), [_field(3)], RUN_DAY)
|
|
[miss] = ht.detect(_day(3, _with([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
|
|
assert miss.summary == "No high temperature: highest air temperature 39.9 °C."
|
|
|
|
|
|
@pytest.mark.parametrize("temp, level", [(42.0, Severity.MODERATE), (44.0, Severity.SEVERE),
|
|
(46.0, Severity.EXTREME)])
|
|
def test_air_bands_without_solar(temp, level):
|
|
[a] = ht.detect(_day(3, _with([temp])), [_field(3)], RUN_DAY)
|
|
assert a.severity == level
|
|
|
|
|
|
def test_event_with_solar_is_graded_on_the_fruit_surface():
|
|
# 14:00: 41 °C, 900 W/m², wind 1 m/s -> FST = 41 + 13.5 / 1.3 = 51.4 °C
|
|
solar, wind = _with([900.0], base=SUN), _with([1.0], base=BREEZE)
|
|
df = _day(566, _with([41.0]), solar=solar, wind=wind)
|
|
[a] = ht.detect(df, [_field(566)], RUN_DAY, station_sensors={566: ALL_SENSORS})
|
|
d = a.details
|
|
assert a.status == EVENT and d["severity_method"] == "fst"
|
|
assert d["max_fst_c"] == 51.4 and d["max_fst_at"] == "2026-07-19T14:00"
|
|
assert d["solar_at_max_fst_wm2"] == 900 and d["wind_at_max_fst_ms"] == 1.0
|
|
assert d["sunlit_hours"] == 11 and d["solar_checked"] is True
|
|
assert a.severity == Severity.MODERATE # 49 <= 51.4 < 52
|
|
assert "fruit surface about 51.4 °C" in a.summary
|
|
assert a.data_notes == []
|
|
|
|
|
|
def test_fst_below_46_is_an_event_without_severity():
|
|
# 14:00: 40 °C but a weak sun and wind: FST = 40 + 6 / 1.9 = 43.2 °C
|
|
solar, wind = _with([400.0], base=[0.0] * 24), _with([3.0], base=BREEZE)
|
|
[a] = ht.detect(_day(566, _with([40.0]), solar=solar, wind=wind), [_field(566)], RUN_DAY)
|
|
assert a.status == EVENT and a.severity == Severity.NONE
|
|
assert a.details["severity_method"] == "fst" and a.details["max_fst_c"] == 43.2
|
|
|
|
|
|
def test_severity_uses_the_hottest_fruit_of_the_day():
|
|
# 13:00: 39 °C in full calm sun (FST 54) comes before the 40 °C hour (cloudy, FST 40)
|
|
t = _with([39.0, 40.0], at=NOON - 1)
|
|
solar = _with([1000.0, 100.0], at=NOON - 1, base=SUN)
|
|
wind = _with([0.0, 2.0], at=NOON - 1, base=BREEZE)
|
|
[a] = ht.detect(_day(566, t, solar=solar, wind=wind), [_field(566)], RUN_DAY)
|
|
assert a.status == EVENT and a.duration_h == 1
|
|
assert a.details["max_fst_c"] == 54.0 and a.severity == Severity.SEVERE
|
|
|
|
|
|
def test_sunburn_risk_below_40_is_information_only():
|
|
# 14:00: 35 °C, 1000 W/m², wind 0.5 -> FST 48.0 °C; 15:00: 31 °C (FST 46) is too cool
|
|
t = _with([35.0, 31.0])
|
|
solar = _with([1000.0, 1000.0], base=SUN)
|
|
wind = _with([0.5, 0.0], base=BREEZE)
|
|
[a] = ht.detect(_day(566, t, solar=solar, wind=wind), [_field(566)], RUN_DAY)
|
|
assert a.status == NO_EVENT and a.severity == Severity.NONE
|
|
assert a.details["sunburn_risk_hours"] == 1 and a.details["max_risk_fst_c"] == 48.0
|
|
assert a.summary == ("No high temperature: highest air temperature 35 °C. Sunburn risk: fruit "
|
|
"surface about 48 °C in 1 hour(s) (below the contract's 40 °C air, "
|
|
"not insurable).")
|
|
|
|
|
|
def test_no_sunburn_risk_on_a_mild_sunny_day():
|
|
[a] = ht.detect(_day(566, MILD, solar=SUN, wind=BREEZE), [_field(566)], RUN_DAY)
|
|
assert a.status == NO_EVENT and a.details["sunburn_risk_hours"] == 0
|
|
assert a.details["max_risk_fst_c"] is None
|
|
|
|
|
|
def test_missing_wind_in_sunlit_hours_is_calm():
|
|
wind = _with([np.nan], base=BREEZE)
|
|
[a] = ht.detect(_day(566, _with([40.0]), solar=_with([1000.0], base=SUN), wind=wind),
|
|
[_field(566)], RUN_DAY, station_sensors={566: ALL_SENSORS})
|
|
assert a.details["max_fst_c"] == 55.0 and a.severity == Severity.EXTREME
|
|
assert ("Wind missing for 1 sunlit hour(s): fruit surface temperature computed for calm "
|
|
"air in those hours (worst case).") in a.data_notes
|
|
|
|
|
|
def test_solar_station_without_anemometer_is_calm():
|
|
[a] = ht.detect(_day(9, _with([40.0]), solar=SUN), [_field(9)], RUN_DAY,
|
|
station_sensors={9: NO_SOLAR | {"RadiazioneSolare"}})
|
|
assert a.details["wind_at_max_fst_ms"] is None
|
|
assert ("No wind sensor (VelocitaVento2m) at this station: fruit surface temperature "
|
|
"computed for calm air (worst case).") in a.data_notes
|
|
|
|
|
|
def test_missing_solar_hours_are_noted():
|
|
solar = list(SUN)
|
|
solar[NOON] = np.nan
|
|
[a] = ht.detect(_day(566, _with([41.0]), solar=solar, wind=BREEZE), [_field(566)], RUN_DAY,
|
|
station_sensors={566: ALL_SENSORS})
|
|
assert a.status == EVENT and a.details["severity_method"] == "fst"
|
|
assert any(n.startswith("Solar radiation missing for 1 of 24 hours") for n in a.data_notes)
|
|
|
|
|
|
def test_solar_sensor_without_data_uses_the_air():
|
|
[a] = ht.detect(_day(566, _with([42.0]), wind=BREEZE), [_field(566)], RUN_DAY,
|
|
station_sensors={566: ALL_SENSORS})
|
|
assert a.details["severity_method"] == "air" and a.severity == Severity.MODERATE
|
|
assert any(n.startswith("No solar radiation data in the 24 hours") for n in a.data_notes)
|
|
|
|
|
|
def test_missing_maximum_temperature_uses_the_mean():
|
|
t_max = _with([np.nan])
|
|
t_mean = _with([40.2], base=[28.0] * 24) # only the mean of 14:00 is known
|
|
[a] = ht.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_humidity_and_wet_bulb_of_the_hottest_hour():
|
|
rh = _with([20.0], base=[50.0] * 24)
|
|
[a] = ht.detect(_day(3, _with([41.0]), rh=rh), [_field(3)], RUN_DAY)
|
|
assert a.details["humidity_at_max_temp_pct"] == 20
|
|
assert a.details["wet_bulb_at_max_temp_c"] == round(float(derived.wet_bulb_c(41.0, 20.0)), 1)
|
|
|
|
|
|
def test_missing_humidity_is_noted_only():
|
|
[a] = ht.detect(_day(3, _with([41.0]), rh=[np.nan] * 24), [_field(3)], RUN_DAY,
|
|
station_sensors={3: NO_SOLAR})
|
|
assert a.status == EVENT and a.details["humidity_at_max_temp_pct"] is None
|
|
assert a.details["wet_bulb_at_max_temp_c"] is None
|
|
assert ("No humidity data in the 24 hours: humidity and wet-bulb not reported "
|
|
"(not used to detect).") in a.data_notes
|
|
|
|
|
|
def test_cover_window_fruit_set_to_31_october():
|
|
df = _day(3, _with([41.0]))
|
|
[after, before] = ht.detect(df, [_field(3, bbch=75), _field(3, bbch=65)], RUN_DAY)
|
|
assert after.details["in_cover_window"] is True and after.details["bbch"] == 75
|
|
assert before.details["in_cover_window"] is False # still in bloom
|
|
assert after.severity == before.severity # severity does not depend on the stage
|
|
assert ht.in_cover_window(date(2026, 11, 2), 80) is False
|
|
|
|
|
|
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
|
|
[a] = ht.detect(_day(3, [45.0] + MILD + [45.0], ts=ts), [_field(3)], RUN_DAY)
|
|
assert a.status == NO_EVENT and a.details["records"] == 24
|
|
|
|
|
|
def test_missing_hours_are_noted_and_not_checked():
|
|
df = _day(3, _with([41.0])).drop(index=[0, 1, 2]) # 3 hours with no record
|
|
[a] = ht.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] = ht.detect(df, [_field(3)], RUN_DAY)
|
|
assert a.status == NO_DATA
|
|
assert a.data_notes == ["No air temperature data in the 24 hours checked: high temperature "
|
|
"cannot be judged."]
|
|
|
|
|
|
def test_missing_columns_do_not_break_the_block():
|
|
df = pd.DataFrame({"CodiceStazione": 3, "Timestamp": HOURS}) # no columns at all
|
|
[a] = ht.detect(df, [_field(3)], RUN_DAY)
|
|
assert a.status == NO_DATA
|
|
|
|
|
|
def test_station_without_records_is_no_data():
|
|
[a] = ht.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."]
|