150 lines
7.0 KiB
Python
150 lines
7.0 KiB
Python
from datetime import date, datetime
|
|
|
|
import pandas as pd
|
|
|
|
from damage_engine.detectors import brina
|
|
from damage_engine.schemas import EVENT, NO_DATA, NO_EVENT, Severity
|
|
from damage_engine.sources.fields import Field
|
|
|
|
RUN_DAY = date(2025, 3, 12)
|
|
|
|
|
|
def _night(station, temps, rh, wind=None, solar=None):
|
|
ts = pd.date_range("2025-03-11 18:00", periods=len(temps), freq="h")
|
|
df = pd.DataFrame({"CodiceStazione": station, "Timestamp": ts,
|
|
brina.COL_TEMP: temps, brina.COL_RH: rh, brina.COL_RH_MAX: rh})
|
|
if wind is not None:
|
|
df[brina.COL_WIND] = wind
|
|
if solar is not None:
|
|
df[brina.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)
|
|
|
|
|
|
COLD_NIGHT = [3, 2, 1, 0.5, -0.5, -1.5, -2.5, -3, -2, -1, 0.5, 2, 3]
|
|
|
|
|
|
def test_night_window():
|
|
assert brina.night_window(RUN_DAY) == (datetime(2025, 3, 11, 18), datetime(2025, 3, 12, 6))
|
|
|
|
|
|
def test_cold_calm_humid_night_is_brina_fallback_severity():
|
|
df = _night(3, COLD_NIGHT, [95] * 13, [0.5] * 13)
|
|
[a] = brina.detect(df, [_field(3)], RUN_DAY, {3: "Lagnasco"})
|
|
assert a.status == EVENT and a.block == "brina"
|
|
assert a.field_id == "ST3" and a.station_name == "Lagnasco"
|
|
assert a.duration_h == 6 # not the 0.5 °C hours: wet-bulb +0.22 °C (3A formula)
|
|
assert a.details["min_bud_temp_c"] == -5 # -3 °C air - 2 °C radiative offset
|
|
assert a.severity == Severity.SEVERE # fallback bands: -4 >= x > -6
|
|
assert a.details["severity_method"] == "fallback_min_bud_temp"
|
|
assert a.details["in_cover_window"] is None # stage unknown
|
|
|
|
|
|
def test_severity_uses_field_stage_when_known():
|
|
df = _night(3, COLD_NIGHT, [95] * 13, [0.5] * 13)
|
|
bloom, dormant = brina.detect(df, [_field(3, 65, "A"), _field(3, 40, "B")], RUN_DAY)
|
|
assert bloom.details["stage"] == "full_bloom"
|
|
assert bloom.details["kill_fraction"] > 0.9
|
|
assert bloom.severity == Severity.EXTREME
|
|
assert bloom.details["in_cover_window"] is True
|
|
assert dormant.status == EVENT and dormant.severity == Severity.NONE
|
|
|
|
|
|
def test_dry_or_daytime_night_is_not_brina():
|
|
temps = [-3] * 13
|
|
dry = _night(5, temps, [40] * 13)
|
|
sunny = _night(293, temps, [95] * 13, [0.5] * 13, [500.0] * 13)
|
|
out = brina.detect(pd.concat([dry, sunny]), [_field(5), _field(293)], RUN_DAY)
|
|
assert [a.status for a in out] == [NO_EVENT] * 2
|
|
assert all(a.severity == Severity.NONE for a in out)
|
|
|
|
|
|
def test_windy_sub_zero_night_hour_is_brina_without_ice_test():
|
|
windy_humid = _night(3, [-3] * 13, [95] * 13, [5.0] * 13) # no ice deposit at bud = air
|
|
windy_dry = _night(6, [-3] * 13, [40] * 13, [5.0] * 13)
|
|
windy_zero = _night(10, [0.0] * 13, [40] * 13, [5.0] * 13) # contract: "below 0 °C"
|
|
windy_mild = _night(30, [1.5] * 13, [40] * 13, [5.0] * 13) # air <= 2 °C, wet-bulb <= 0, not sub-zero
|
|
calm_humid = _night(544, [-3] * 13, [95] * 13, [0.5] * 13) # calm: ice test, buds 2 °C colder
|
|
calm_dry = _night(566, [-3] * 13, [40] * 13, [0.5] * 13) # calm and dry: no ice, no brina
|
|
out = brina.detect(pd.concat([windy_humid, windy_dry, windy_zero, windy_mild, calm_humid, calm_dry]),
|
|
[_field(i) for i in (3, 6, 10, 30, 544, 566)], RUN_DAY)
|
|
assert [a.status for a in out] == [EVENT, EVENT, NO_EVENT, NO_EVENT, EVENT, NO_EVENT]
|
|
hum, dry, calm = out[0], out[1], out[4]
|
|
assert hum.duration_h == 13 and hum.details["min_bud_temp_c"] == -3 # bud = air when windy
|
|
assert hum.details["windy_hours"] == 13 and hum.details["max_wind_ms"] == 5.0
|
|
assert dry.severity == hum.severity == Severity.MODERATE # fallback: -3 °C
|
|
assert calm.details["min_bud_temp_c"] == -5 and calm.details["windy_hours"] == 0
|
|
rows = brina.brina_hours(windy_mild)
|
|
assert (rows["wet_bulb_c"] <= 0).all() # only "below 0 °C" stops it
|
|
|
|
|
|
def test_station_without_wind_sensor_still_evaluated():
|
|
[a] = brina.detect(_night(9, [-5] * 13, [92] * 13), [_field(9)], RUN_DAY, {9: "Centallo"})
|
|
assert a.status == EVENT
|
|
assert a.details["wind_checked"] is False
|
|
assert a.severity == Severity.EXTREME # bud -7 °C < -6 °C
|
|
|
|
|
|
def test_field_without_station_records_is_no_data():
|
|
[a] = brina.detect(_night(3, COLD_NIGHT, [95] * 13), [_field(77)], RUN_DAY)
|
|
assert a.status == NO_DATA
|
|
|
|
|
|
def test_out_of_cover_window_flagged():
|
|
df = _night(3, COLD_NIGHT, [95] * 13, [0.5] * 13)
|
|
[a] = brina.detect(df, [_field(3, 65)], date(2025, 10, 19))
|
|
assert a.details["in_cover_window"] is False
|
|
|
|
|
|
def test_missing_sensors_and_data_are_noted():
|
|
df = _night(9, [-5] * 13, [92] * 13) # no wind, no solar columns
|
|
[a] = brina.detect(df, [_field(9)], RUN_DAY, station_sensors={9: {"TemperaturaAria", "UmiditaRelativa"}})
|
|
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") for n in a.data_notes)
|
|
|
|
|
|
def test_partly_missing_hours_are_noted():
|
|
solar = [0.0] * 10 + [None] * 3
|
|
rh = [95] * 12 + [None]
|
|
df = _night(293, COLD_NIGHT, rh, [0.5] * 13, solar)
|
|
[a] = brina.detect(df, [_field(293)], RUN_DAY)
|
|
assert "Solar radiation missing for 3 of 13 hours" in " ".join(a.data_notes)
|
|
assert "Temperature or humidity missing for 1 of 13 hours" in " ".join(a.data_notes)
|
|
|
|
|
|
def test_complete_station_has_no_notes_and_no_records_is_noted():
|
|
full = _night(3, COLD_NIGHT, [95] * 13, [0.5] * 13, [0.0] * 13)
|
|
ok, missing = brina.detect(full, [_field(3), _field(77)], RUN_DAY)
|
|
assert ok.data_notes == []
|
|
assert missing.data_notes == ["No records from the station last night."]
|
|
|
|
|
|
def test_3a_wet_bulb_is_reported_but_not_used_to_detect():
|
|
df = _night(566, COLD_NIGHT, [95] * 13, [0.5] * 13, [1.0] * 13)
|
|
df[brina.COL_WET_BULB_3A] = [5.0] * 13 # warm on purpose: must not block the event
|
|
[a] = brina.detect(df, [_field(566)], RUN_DAY)
|
|
assert a.status == EVENT and a.duration_h == 6 # same as without TBulbMed
|
|
assert a.details["min_wet_bulb_3a_c"] == 5.0
|
|
[b] = brina.detect(_night(3, COLD_NIGHT, [95] * 13, [0.5] * 13), [_field(3)], RUN_DAY)
|
|
assert "min_wet_bulb_3a_c" not in b.details # station without TBulbMed
|
|
|
|
|
|
def test_solar_column_is_hourly_irradiance():
|
|
assert brina.COL_SOLAR == "RadiazioneSolareMax" # W/m²; RadiazioneSolareInt is MJ/m², daily only
|
|
|
|
|
|
def test_wet_bulb_is_the_shared_lowest_of_the_hour():
|
|
df = _night(3, COLD_NIGHT, [95] * 13, [0.5] * 13)
|
|
rows = brina.brina_hours(df)
|
|
assert rows["wet_bulb_c"].equals(rows["WetBulbMin"]) # one value for the whole pipeline
|
|
|
|
|
|
def test_missing_max_humidity_uses_mean_and_is_noted():
|
|
df = _night(3, COLD_NIGHT, [95] * 13, [0.5] * 13).drop(columns=[brina.COL_RH_MAX])
|
|
[a] = brina.detect(df, [_field(3)], RUN_DAY)
|
|
assert a.duration_h == 6 # RH max = RH mean here: same result
|
|
assert "Hourly maximum humidity missing for 13 of 13 hours" in " ".join(a.data_notes)
|