import json from datetime import date, datetime from damage_engine.report import report, station_report from damage_engine.schemas import EVENT, NO_DATA, NO_EVENT, Analysis, Severity DAY = date(2026, 4, 22) def _a(block, station, status=NO_EVENT, details=None, field=None, run_day=DAY, **kw): return Analysis(block=block, run_day=run_day, field_id=field or f"ST{station}", field_name=f"field {station}", station_id=station, station_name=f"S{station}", status=status, details=details or {}, **kw) BRINA_EVENT = dict(status=EVENT, severity=Severity.SEVERE, severity_score=0.79, duration_h=5, start=datetime(2026, 4, 22, 1), end=datetime(2026, 4, 22, 5), details={"wind_checked": True, "records": 13, "night_min_temp_c": -1.2, "min_temp_c": -1.2, "min_bud_temp_c": -3.2, "min_wet_bulb_c": -1.58, "severity_method": "kill_fraction", "bbch": 65, "stage": "full_bloom", "kill_fraction": 0.57, "in_cover_window": True, "max_wind_ms": 3.4, "windy_hours": 2}) FLOOD_DRY = {"records": 24, "rain_day_mm": 0.0, "max_1h_mm": 0.0, "max_72h_mm": None, "rain_72h_to_run_mm": 8.6, "rain_10d_mm": 8.8, "reference_10d_mm": None} def _store(root): report.submit("brina", DAY, [_a("brina", 566, **BRINA_EVENT), _a("brina", 3, details={"night_min_temp_c": 9.0})], root=root) report.submit("flood", DAY, [_a("flood", 566, details=FLOOD_DRY), _a("flood", 3, details=FLOOD_DRY)], root=root) def _check(block, name): return next(c for c in block["checks"] if c["check"] == name) def test_report_holds_only_its_station_and_every_block(tmp_path): _store(tmp_path) c = station_report.build(566, DAY, tmp_path) assert [b["block"] for b in c["blocks"]] == list(station_report.EVENTS) # every event, diagram order assert {b["block"] for b in c["blocks"] if b["status"] != "not_run"} == {"brina", "flood"} assert {f["field_id"] for b in c["blocks"] for f in b["fields"]} == {"ST566"} assert c["overall"]["events"] == 1 and c["overall"]["worst_severity"] == "severe" def test_checks_show_value_threshold_and_margin(tmp_path): _store(tmp_path) c = station_report.build(566, DAY, tmp_path) brina, flood = _block(c, "brina"), _block(c, "flood") air = _check(brina, "Lowest air temperature") assert (air["measured"], air["threshold"], air["status"], air["margin"]) == (-1.2, 2.0, "crossed", 3.2) wind = _check(brina, "Highest wind in brina hours") # not a condition: info only assert (wind["measured"], wind["status"]) == (3.4, "info") rain = _check(flood, "Rain in 1 hour (wettest hour)") assert (rain["status"], rain["margin"], rain["threshold_text"]) == ("not_crossed", 30.0, "≥ 30 mm") assert _check(flood, "Rain in the last 10 days")["status"] == "info" # no 5-year average def _block(content, name): return next(b for b in content["blocks"] if b["block"] == name) def test_no_event_night_marks_unstored_values(tmp_path): _store(tmp_path) brina = _block(station_report.build(3, DAY, tmp_path), "brina") assert _check(brina, "Lowest wet-bulb temperature")["status"] == "not_recorded" assert _check(brina, "Lowest air temperature")["status"] == "not_crossed" assert _check(brina, "Highest wind in brina hours")["status"] == "not_applied" # no wind_checked def test_no_data_has_no_checks(tmp_path): report.submit("brina", DAY, [_a("brina", 600, status=NO_DATA, data_notes=["No sensor."])], root=tmp_path) c = station_report.build(600, DAY, tmp_path) assert _block(c, "brina")["checks"] == [] and _block(c, "brina")["data_notes"] == ["No sensor."] assert c["overall"]["headline"] == "No usable data for this station." def test_write_makes_json_and_pdf(tmp_path): _store(tmp_path / "reports") json_path, pdf_path = station_report.write(566, DAY, tmp_path / "reports", tmp_path / "out") assert json_path.name == "2026-04-22.json" and json_path.parent.name == "566" assert json.loads(json_path.read_text(encoding="utf-8"))["station"]["id"] == 566 assert pdf_path.read_bytes()[:5] == b"%PDF-" def test_missing_sensors_compared_with_other_stations(tmp_path): c = station_report.build(153, DAY, tmp_path) # Caraglio: no wind, no solar brina = [m["sensor"] for m in _block(c, "brina")["missing_sensors"]] assert brina == ["VelocitaVento2m", "RadiazioneSolare"] wind = _block(c, "strong_wind")["missing_sensors"][0] assert wind == {"sensor": "VelocitaVento2m", "name": wind["name"]} # no counts of other stations assert _block(c, "flood")["missing_sensors"] == [] # every station has a rain gauge drought = [m["sensor"] for m in _block(c, "drought")["missing_sensors"]] # rain + soil moisture assert drought == ["Suolo1", "Suolo2"] def test_five_year_note_hidden_and_new_block_gets_generic_table(tmp_path): report.submit("flood", DAY, [_a("flood", 3, details=FLOOD_DRY, data_notes=["No 5-year reference average for the 10-day rain: x.", "Rain missing for 2 of 24 hours."])], root=tmp_path) report.submit("new_block", DAY, [_a("new_block", 3, details={"max_size_mm": 21.5, "wind_checked": True})], root=tmp_path) c = station_report.build(3, DAY, tmp_path) assert _block(c, "flood")["data_notes"] == ["Rain missing for 2 of 24 hours."] [size] = _block(c, "new_block")["checks"] assert (size["check"], size["measured"], size["status"]) == ("Max size mm", 21.5, "info") WIND_EVENT = dict(status=EVENT, severity=Severity.MODERATE, severity_score=0.6, duration_h=3, start=datetime(2026, 4, 21, 18), end=datetime(2026, 4, 21, 20), details={"records": 24, "trigger": "hourly_max", "max_wind_ms": 18.5, "max_wind_kmh": 66.6, "max_wind_at": "2026-04-21T19:00", "max_mean_wind_ms": 9.3, "fallback_hours": 0, "strong_wind_hours": 3, "in_cover_window": None}) def test_strong_wind_checks_period_and_severity_text(tmp_path): report.submit("strong_wind", DAY, [ _a("strong_wind", 566, **WIND_EVENT), _a("strong_wind", 3, details={"records": 24, "trigger": "hourly_max", "max_wind_ms": 8.0, "max_wind_kmh": 28.8, "max_wind_at": "2026-04-22T02:00", "max_mean_wind_ms": None, "fallback_hours": 0})], root=tmp_path) wind = _block(station_report.build(566, DAY, tmp_path), "strong_wind") assert wind["window"] == {"start": "2026-04-21T07:00", "end": "2026-04-22T07:00"} top = _check(wind, "Highest wind in the 24 h") assert (top["status"], top["margin"], top["threshold_text"]) == ("crossed", 4.6, "≥ 13.9 m/s") assert top["note"].startswith("At 21 Apr 19:00, 66.6 km/h. ≥ 50 km/h, Beaufort 7") time_ = _check(wind, "Time with wind ≥ 13.9 m/s") assert (time_["measured"], time_["status"], time_["threshold_text"]) == (3, "crossed", "≥ 10 min") assert _check(wind, "Highest hourly mean wind")["status"] == "info" field = wind["fields"][0] assert field["severity_basis"].startswith("Graded on the highest wind") assert field["severity_scale"] == ("Wind: light ≥ 13.9 m/s, moderate ≥ 17.2 m/s, severe ≥ 20.8 m/s, " "extreme ≥ 24.5 m/s (Beaufort 7, 8, 9, 10)") calm = _block(station_report.build(3, DAY, tmp_path), "strong_wind") assert _check(calm, "Highest wind in the 24 h")["status"] == "not_crossed" assert _check(calm, "Time with wind ≥ 13.9 m/s")["status"] == "not_crossed" assert _check(calm, "Highest hourly mean wind")["status"] == "not_recorded" FREEZING_EVENT = dict(status=EVENT, severity=Severity.MODERATE, severity_score=0.56, duration_h=4, start=datetime(2026, 4, 21, 10), end=datetime(2026, 4, 21, 13), details={"records": 13, "daylight_hours": 13, "sunlit_hours_in_night_window": 2, "solar_checked": True, "day_min_temp_c": -2.5, "mean_wind_ms": 3.0, "advective_wind": True, "min_wet_bulb_c": -3.47, "severity_method": "fallback_min_bud_temp", "in_cover_window": None}) def test_freezing_checks_period_and_severity_text(tmp_path): report.submit("freezing", DAY, [_a("freezing", 566, **FREEZING_EVENT), _a("freezing", 153, details={"records": 11, "daylight_hours": 11, "sunlit_hours_in_night_window": 0, "solar_checked": False, "day_min_temp_c": 8.0})], root=tmp_path) frz = _block(station_report.build(566, DAY, tmp_path), "freezing") assert frz["window"] == {"start": "2026-04-21T06:00", "end": "2026-04-22T06:00"} air = _check(frz, "Lowest daylight air temperature") assert (air["status"], air["margin"], air["threshold_text"]) == ("crossed", 2.5, "< 0 °C") assert _check(frz, "Daylight hours below 0 °C")["status"] == "crossed" assert _check(frz, "Mean wind while below 0 °C")["status"] == "crossed" # ≥ 2 m/s assert _check(frz, "Sunlit hours of the night (solar radiation)")["measured"] == 2 assert frz["fields"][0]["severity_basis"] == ("Crop stage not set: graded on the lowest " "daylight air temperature.") quiet = _block(station_report.build(153, DAY, tmp_path), "freezing") # no pyranometer, warm day assert _check(quiet, "Lowest daylight air temperature")["status"] == "not_crossed" assert _check(quiet, "Sunlit hours of the night (solar radiation)")["status"] == "not_applied" assert _check(quiet, "Mean wind while below 0 °C")["status"] == "not_recorded" assert _check(quiet, "Lowest wet-bulb while below 0 °C")["status"] == "not_recorded" HEAVY = {"mm": 90.0, "first_rain": "2026-04-18T12:00", "last_rain": "2026-04-19T18:00", "reached_at": "2026-04-19T18:00", "partial": False} def test_rain_row_3_is_the_last_72h_over_the_threshold(tmp_path): report.submit("flood", DAY, [ _a("flood", 566, details={**FLOOD_DRY, "last_heavy_72h": HEAVY, "wettest_72h": HEAVY}), _a("flood", 3, details={**FLOOD_DRY, "last_heavy_72h": None, "wettest_72h": {**HEAVY, "mm": 41.5, "partial": True}}), _a("flood", 5, details=FLOOD_DRY), # run stored before this row existed ], root=tmp_path) row = _block(station_report.build(566, DAY, tmp_path), "flood")["checks"][2] assert row["check"] == "72-Hour Heavy Rain Watch (18 Apr 07:00 → 22 Apr 07:00)" # the 4 days checked assert (row["measured"], row["status"], row["margin"]) == (90.0, "crossed", 10.0) assert row["note"] == "Rain 18 Apr 12:00 → 19 Apr 18:00; 80 mm reached on 19 Apr 18:00." c = station_report.build(566, DAY, tmp_path) # no new event: past one in the overview assert _block(c, "flood")["notice"]["short"] == "Past: 90 mm in 72 h (18 Apr → 19 Apr)" assert c["overall"]["notices"][0]["text"].startswith("No new event, but 90 mm of rain fell") assert c["overall"]["headline"].startswith("No new damaging event") assert _block(station_report.build(3, DAY, tmp_path), "flood")["notice"] is None none = _block(station_report.build(3, DAY, tmp_path), "flood")["checks"][2] assert (none["measured"], none["status"], none["margin"]) == (41.5, "not_crossed", 38.5) assert none["check"] == "72-Hour Heavy Rain Watch (18 Apr 07:00 → 22 Apr 07:00)" assert none["note"] == ("Rain 18 Apr 12:00 → 19 Apr 18:00. " # the wettest window "Rain also fell just before these days (not counted).") old = _block(station_report.build(5, DAY, tmp_path), "flood")["checks"][2] assert old["status"] == "not_recorded" and old["check"] == "72-Hour Heavy Rain Watch (18 Apr 07:00 → 22 Apr 07:00)" def test_rain_row_3_name_is_the_watch_and_its_4_days(tmp_path): report.submit("flood", DAY, [_a("flood", 3, details={**FLOOD_DRY, "last_heavy_72h": None, "wettest_72h": {**HEAVY, "mm": 63.2}})], root=tmp_path) row = _block(station_report.build(3, DAY, tmp_path), "flood")["checks"][2] assert (row["check"], row["note"]) == ("72-Hour Heavy Rain Watch (18 Apr 07:00 → 22 Apr 07:00)", "Rain 18 Apr 12:00 → 19 Apr 18:00.") later = date(2026, 4, 26) report.submit("flood", later, [_a("flood", 3, run_day=later, details={ **FLOOD_DRY, "last_heavy_72h": None, "wettest_72h": None})], root=tmp_path) row = _block(station_report.build(3, later, tmp_path), "flood")["checks"][2] assert row["check"] == "72-Hour Heavy Rain Watch (22 Apr 07:00 → 26 Apr 07:00)" # follows the run day assert row["note"] == "No rain in the last 4 days." one = {"mm": 0.2, "first_rain": "2026-04-24T06:00", "last_rain": "2026-04-24T06:00", "partial": False} report.submit("flood", later, [_a("flood", 3, run_day=later, details={ **FLOOD_DRY, "last_heavy_72h": None, "wettest_72h": one})], root=tmp_path) row = _block(station_report.build(3, later, tmp_path), "flood")["checks"][2] assert row["note"] == "Rain at 24 Apr 06:00." # one rainy hour def test_rain_row_2_is_the_last_72_hours(tmp_path): intense = dict(status=EVENT, severity=Severity.LIGHT, severity_score=0.1, duration_h=1, details={**FLOOD_DRY, "rain_day_mm": 1.0, "max_72h_mm": 85.0, "rain_72h_to_run_mm": 70.0, "criteria_met": ["intense_72h"], "worst_criterion": "intense_72h", "rain_trigger_ratio": 1.06}) report.submit("flood", DAY, [ _a("flood", 566, details=FLOOD_DRY), _a("flood", 3, details={**FLOOD_DRY, "rain_72h_to_run_mm": 95.0}), # dry day after a storm _a("flood", 5, **intense), # earlier window crossed ], root=tmp_path) rows = {s: _block(station_report.build(s, DAY, tmp_path), "flood")["checks"] for s in (566, 3, 5)} assert [r["check"] for r in rows[566]][:2] == ["Rain in 1 hour (wettest hour)", "Rain in the last 72 hours"] quiet, after_storm, earlier = (rows[s][1] for s in (566, 3, 5)) assert (quiet["measured"], quiet["status"], quiet["margin"], quiet["note"]) == (8.6, "not_crossed", 71.4, "") assert quiet["threshold_text"] == "≥ 80 mm" assert (after_storm["status"], after_storm["margin"]) == ("crossed", 15.0) assert after_storm["note"].startswith("No rain in the last 24 h: not a new event") assert (earlier["measured"], earlier["status"]) == (70.0, "not_crossed") assert earlier["note"].startswith("The event comes from an earlier 72 h window") def test_overview_period_of_rain_includes_the_last_4_days(tmp_path): _store(tmp_path) c = station_report.build(566, DAY, tmp_path) flood = _block(c, "flood") assert flood["window"] == {"start": "2026-04-21T07:00", "end": "2026-04-22T07:00"} assert flood["watch_window"] == {"label": "Last 4 days", "start": "2026-04-18T07:00", "end": "2026-04-22T07:00"} assert _block(c, "brina")["watch_window"] is None def test_new_run_replaces_only_that_day_and_keeps_10_days(tmp_path): _store(tmp_path / "reports") folder = tmp_path / "out" / "566" days = [date(2026, 4, d) for d in range(11, 23)] # 12 days: 11 → 22 April for d in days: station_report.write(566, d, tmp_path / "reports", tmp_path / "out") kept = sorted(p.stem for p in folder.glob("*.json")) assert kept == [d.isoformat() for d in days[-10:]] # the last 10 days stay assert sorted(p.stem for p in folder.glob("*.pdf")) == kept report.submit("flood", DAY, [_a("flood", 566, details={**FLOOD_DRY, "rain_day_mm": 7.5})], root=tmp_path / "reports") # new run of the same day station_report.write(566, DAY, tmp_path / "reports", tmp_path / "out") assert sorted(p.stem for p in folder.glob("*.json")) == kept # no other day removed new = json.loads((folder / "2026-04-22.json").read_text(encoding="utf-8")) assert _block(new, "flood")["details"]["rain_day_mm"] == 7.5 # that day's report replaced HOT_WIND_EVENT = dict(status=EVENT, severity=Severity.MODERATE, severity_score=0.55, duration_h=2, start=datetime(2026, 4, 21, 14), end=datetime(2026, 4, 21, 15), details={"records": 24, "trigger": "hourly_max", "max_temp_c": 42.3, "max_temp_at": "2026-04-21T15:00", "max_mean_temp_c": 40.1, "fallback_hours": 0, "hot_hours": 3, "still_hot_hours": 1, "wind_checked": True, "wind_at_max_temp_ms": 2.4, "humidity_at_max_temp_pct": 18, "direction_checked": False, "hot_wind_hours": 2, "max_hot_wind_temp_c": 42.3, "wind_assumed_hours": 0, "min_humidity_pct": 18, "harvest_checked": False, "in_cover_window": None}) def test_hot_wind_checks_period_and_severity_text(tmp_path): report.submit("hot_wind", DAY, [ _a("hot_wind", 566, **HOT_WIND_EVENT), _a("hot_wind", 3, details={"records": 24, "trigger": "hourly_max", "max_temp_c": 31.0, "max_temp_at": "2026-04-21T16:00", "max_mean_temp_c": 29.5, "fallback_hours": 0, "hot_hours": 0, "still_hot_hours": 0, "wind_checked": False, "wind_at_max_temp_ms": None, "humidity_at_max_temp_pct": None, "direction_checked": False})], root=tmp_path) hot = _block(station_report.build(566, DAY, tmp_path), "hot_wind") assert hot["window"] == {"start": "2026-04-21T07:00", "end": "2026-04-22T07:00"} top = _check(hot, "Highest air temperature in the 24 h") assert (top["status"], top["margin"], top["threshold_text"]) == ("crossed", 2.3, "≥ 40 °C") assert top["note"].startswith("At 21 Apr 15:00. ≥ 40 °C at 2 m (contract)") hours = _check(hot, "Hours at ≥ 40 °C with moving air") assert (hours["measured"], hours["status"], hours["threshold_text"]) == (2, "crossed", "≥ 1 h") assert "1 hour(s) at ≥ 40 °C had still air." in hours["note"] wind = _check(hot, "Wind in the hottest hour") assert (wind["status"], wind["threshold_text"]) == ("crossed", "≥ 1 m/s") direction = _check(hot, "Wind from south-east to south-west") assert (direction["status"], direction["threshold_text"]) == ("not_applied", "135°–225°") assert _check(hot, "Humidity in the hottest hour")["status"] == "info" field = hot["fields"][0] assert field["severity_basis"].startswith("Graded on the highest air temperature") assert field["severity_scale"] == ("Air temperature: light ≥ 40 °C, moderate ≥ 42 °C, " "severe ≥ 44 °C, extreme ≥ 46 °C") mild = _block(station_report.build(3, DAY, tmp_path), "hot_wind") assert _check(mild, "Highest air temperature in the 24 h")["status"] == "not_crossed" assert _check(mild, "Hours at ≥ 40 °C with moving air")["status"] == "not_crossed" assert _check(mild, "Wind in the hottest hour")["status"] == "not_applied" assert _check(mild, "Humidity in the hottest hour")["status"] == "not_recorded" HIGH_TEMPERATURE_EVENT = dict( status=EVENT, severity=Severity.MODERATE, severity_score=0.6, duration_h=2, start=datetime(2026, 4, 21, 14), end=datetime(2026, 4, 21, 15), details={"records": 24, "trigger": "hourly_max", "max_temp_c": 41.2, "max_temp_at": "2026-04-21T15:00", "max_mean_temp_c": 39.5, "fallback_hours": 0, "humidity_at_max_temp_pct": 22, "wet_bulb_at_max_temp_c": 24.1, "solar_checked": True, "sunlit_hours": 10, "max_solar_wm2": 950, "max_fst_c": 51.1, "max_fst_at": "2026-04-21T14:00", "solar_at_max_fst_wm2": 950, "wind_at_max_fst_ms": 1.2, "sunburn_risk_hours": 4, "max_risk_fst_c": 51.1, "hot_hours": 2, "severity_method": "fst", "severity_value_c": 51.1, "in_cover_window": None}) def test_high_temperature_checks_period_and_severity_text(tmp_path): report.submit("high_temperature", DAY, [ _a("high_temperature", 566, **HIGH_TEMPERATURE_EVENT), _a("high_temperature", 3, status=EVENT, severity=Severity.LIGHT, duration_h=1, start=datetime(2026, 4, 21, 15), end=datetime(2026, 4, 21, 15), details={"records": 24, "trigger": "hourly_max", "max_temp_c": 40.4, "max_temp_at": "2026-04-21T15:00", "fallback_hours": 0, "humidity_at_max_temp_pct": None, "wet_bulb_at_max_temp_c": None, "solar_checked": False, "max_fst_c": None, "sunburn_risk_hours": None, "hot_hours": 1, "severity_method": "air", "severity_value_c": 40.4})], root=tmp_path) hot = _block(station_report.build(566, DAY, tmp_path), "high_temperature") assert hot["window"] == {"start": "2026-04-21T07:00", "end": "2026-04-22T07:00"} top = _check(hot, "Highest air temperature in the 24 h") assert (top["status"], top["margin"], top["threshold_text"]) == ("crossed", 1.2, "≥ 40 °C") assert top["note"].startswith("At 21 Apr 15:00. ≥ 40 °C at 2 m (contract)") hours = _check(hot, "Hours at ≥ 40 °C") assert (hours["measured"], hours["status"]) == (2, "crossed") fst = _check(hot, "Highest fruit surface temperature") assert (fst["status"], fst["margin"], fst["threshold_text"]) == ("crossed", 5.1, "≥ 46 °C") assert fst["note"].startswith("At 21 Apr 14:00: sun 950 W/m², wind 1.2 m/s.") risk = _check(hot, "Sunburn risk hours (air ≥ 32 °C, fruit ≥ 46 °C)") assert (risk["measured"], risk["status"]) == (4, "crossed") assert _check(hot, "Wet-bulb in the hottest hour")["status"] == "info" field = hot["fields"][0] assert field["severity_basis"].startswith("Graded on the highest estimated fruit surface temperature (51.1 °C)") assert field["severity_scale"] == ("Fruit surface: light ≥ 46 °C, moderate ≥ 49 °C, " "severe ≥ 52 °C, extreme ≥ 55 °C") no_sun = _block(station_report.build(3, DAY, tmp_path), "high_temperature") assert _check(no_sun, "Highest fruit surface temperature")["status"] == "not_applied" assert _check(no_sun, "Sunburn risk hours (air ≥ 32 °C, fruit ≥ 46 °C)")["status"] == "not_applied" assert _check(no_sun, "Humidity in the hottest hour")["status"] == "not_recorded" assert no_sun["fields"][0]["severity_basis"].startswith("No solar radiation data: graded on the highest air") assert no_sun["fields"][0]["severity_scale"].startswith("Air temperature: light ≥ 40 °C") SNOW_SUSPECTED = dict(status=EVENT, duration_h=3, start=datetime(2026, 4, 21, 23), end=datetime(2026, 4, 22, 1), details={"records": 24, "cold_basis": "wet_bulb", "air_fallback_hours": 0, "min_cold_c": -1.6, "min_cold_at": "2026-04-22T00:00", "cold_hours": 3, "rain_checked": True, "humidity_checked": True, "solar_checked": False, "snow_hours": 3, "snow_water_mm": 4.0, "melt_hours": 0, "melt_water_mm": 0.0, "saturated_hours": 0, "midday_solar_max_wm2": None, "est_new_snow_cm": 4.0, "est_new_snow_max_cm": 4.0, "signals": ["A"], "suspicion": "likely", "in_cover_window": True, "camera_request": {"phenomenon": "snow", "station_id": 3, "photo_at": "2026-04-22T09:00", "level": "likely", "signals": ["A"], "reason": "Snow likely: precipitation in a cold hour."}}) def test_snow_is_shown_as_suspected_with_its_signs_and_photo(tmp_path): report.submit("snow", DAY, [_a("snow", 3, **SNOW_SUSPECTED)], root=tmp_path) c = station_report.build(3, DAY, tmp_path) b = _block(c, "snow") assert b["window"] == {"start": "2026-04-21T07:00", "end": "2026-04-22T07:00"} [f] = b["fields"] assert f["suspected"] and f["camera_request"]["photo_at"] == "2026-04-22T09:00" assert f["severity_basis"].startswith("Not graded") assert c["overall"]["events"] == 0 # not a detected event: a photo decides assert c["overall"]["headline"] == "Snow suspected · photo requested to confirm" assert "22 Apr 09:00" in c["overall"]["suspected"][0]["text"] assert _check(b, "A · Precipitation in cold hours")["status"] == "crossed" assert _check(b, "B · Rain within 12 h after a cold hour")["status"] == "not_crossed" assert _check(b, "D · Highest midday sun after the first suspect hour")["status"] == "not_applied" assert _check(b, "Lowest wet-bulb in the 24 h")["margin"] == 2.6 json_path, pdf_path = station_report.write(3, DAY, tmp_path, out_dir=tmp_path / "out") assert pdf_path.read_bytes()[:5] == b"%PDF-" def test_hail_is_shown_as_suspected_with_its_signs_and_photo(tmp_path): import pandas as pd from damage_engine.detectors import hail from damage_engine.sources.fields import Field hours = pd.date_range("2026-04-21 07:00", "2026-04-22 07:00", freq="h") t = [20.0] * len(hours) t[10:12] = [14.0, 14.0] # 17:00: 6.5 °C drop rain = [0.0] * len(hours) rain[10:12] = [22.0, 2.0] # burst rain at 17:00 df = pd.DataFrame({"CodiceStazione": 3, "Timestamp": hours, hail.COL_T_MEAN: t, hail.COL_T_MIN: [x - 0.5 for x in t], hail.COL_RAIN: rain, hail.COL_WIND: [3.0] * len(hours), hail.COL_WIND_MEAN: [1.0] * len(hours)}) [a] = hail.detect(df, [Field(id="ST3", name="field 3", station_id=3)], DAY) assert a.status == EVENT and a.details["suspicion"] == "likely" report.submit("hail", DAY, [a], root=tmp_path) c = station_report.build(3, DAY, tmp_path) b = _block(c, "hail") assert b["window"] == {"start": "2026-04-21T07:00", "end": "2026-04-22T07:00"} [f] = b["fields"] assert f["suspected"] and f["camera_request"]["photo_at"] == "2026-04-22T09:00" assert f["severity_basis"].startswith("Not graded") assert c["overall"]["events"] == 0 assert c["overall"]["headline"] == "Hail suspected · photo requested to confirm" burst = _check(b, "A · Rain in the wettest hour") assert (burst["status"], burst["margin"]) == ("crossed", 2.0) assert _check(b, "A · Highest rain rate")["status"] == "not_applied" assert _check(b, "B · Largest temperature drop")["measured"] == 6.5 assert _check(b, "C · Highest wind (gust)")["status"] == "not_crossed" assert _check(b, "Signs in one storm")["measured"] == 2 json_path, pdf_path = station_report.write(3, DAY, tmp_path, out_dir=tmp_path / "out") assert pdf_path.read_bytes()[:5] == b"%PDF-" DROUGHT_MONTHS = [ {"month": "2026-01", "label": "Jan 2026", "days": 31, "missing_days": 0, "rain_mm": 10.0, "normal_mm": 52.7, "rain_pct_of_normal": 19.0, "et0_mm": 20.0, "water_balance_mm": -10.0}, {"month": "2026-02", "label": "Feb 2026", "days": 28, "missing_days": 2, "rain_mm": 20.0, "normal_mm": 61.5, "rain_pct_of_normal": 32.5, "et0_mm": 30.0, "water_balance_mm": -10.0}, {"month": "2026-03", "label": "Mar 2026", "days": 31, "missing_days": 0, "rain_mm": 30.0, "normal_mm": 82.2, "rain_pct_of_normal": 36.5, "et0_mm": 50.0, "water_balance_mm": -20.0}, ] DROUGHT_EVENT = dict(status=EVENT, severity=Severity.MODERATE, severity_score=0.6, duration_h=90 * 24, start=datetime(2026, 1, 1), end=datetime(2026, 4, 1), details={"months": DROUGHT_MONTHS, "window_label": "Jan–Mar 2026", "window_months": 3, "days": 90, "missing_days": 2, "rain_mm": 60.0, "normal_mm": 196.4, "rain_pct_of_normal": 30.5, "deficit_mm": 136.4, "last_month_label": "Mar 2026", "et0_mm": 100.0, "water_balance_mm": -40.0, "water_balance_months": 3, "soil_moisture": None, "window_start": "2026-01-01T00:00", "window_end": "2026-04-01T00:00"}) def test_drought_shows_rain_against_the_average_month_by_month(tmp_path): report.submit("drought", date(2026, 4, 1), [_a("drought", 566, **DROUGHT_EVENT)], root=tmp_path) _store(tmp_path) # daily blocks of 22 Apr c = station_report.build(566, DAY, tmp_path) b = _block(c, "drought") assert b["window"]["label"] == "Jan–Mar 2026" and b["monthly"] and b["stale"] pct = _check(b, "Rain as % of the average") assert (pct["measured"], pct["status"], pct["margin"]) == (30.5, "crossed", 29.5) assert pct["threshold_text"] == "≤ 60 % of the average (= 117.8 mm)" total = _check(b, "Rain in the last 3 months (Jan–Mar 2026)") assert total["threshold_text"] == "average 196.4 mm" and "136.4 mm below" in total["note"] feb = _check(b, "Rain in Feb 2026") assert (feb["measured"], feb["threshold_text"], feb["status"]) == (20.0, "average 61.5 mm", "info") assert _check(b, "Water balance: rain − ET0 (Jan–Mar 2026)")["measured"] == -40.0 assert _check(b, "Soil moisture, mean of Mar 2026")["status"] == "not_applied" [f] = b["fields"] assert f["period_text"] == "Jan–Mar 2026 (90 days)" and f["duration_text"] == "90 d" assert f["severity_scale"].startswith("Rain / average: light ≤ 60 %") json_path, pdf_path = station_report.write(566, DAY, tmp_path, out_dir=tmp_path / "out") assert pdf_path.read_bytes()[:5] == b"%PDF-" def test_hail_daytime_suspicion_is_merged_into_the_hail_section(tmp_path): report.submit("hail", DAY, [_a("hail", 3, summary="No hail suspected: no rain in the 24 hours. " "Hail-risk day: the station is checked again " "today at 13:00 and 19:00.", details={"hail_risk": True, "storm_signs": 0})], root=tmp_path) quiet = _a("hail_daytime", 3, summary="No hail suspected: no rain in the 6 hours.", details={"check": "daytime", "window_start": "2026-04-22T07:00", "window_end": "2026-04-22T13:00"}) request = {"phenomenon": "hail", "station_id": 3, "photo_at": "2026-04-22T19:00", "level": "likely", "signals": ["A", "B"], "reason": "Hail likely: burst rain; x."} storm = _a("hail_daytime", 3, status=EVENT, duration_h=2, start=datetime(2026, 4, 22, 16), end=datetime(2026, 4, 22, 17), summary="Hail likely: burst rain of 24 mm.", details={"check": "daytime", "window_start": "2026-04-22T13:00", "window_end": "2026-04-22T19:00", "camera_request": request, "in_cover_window": None}, data_notes=["No wind sensor (VelocitaVento2m) at this station: x."]) report.submit("hail_daytime", DAY, [quiet], root=tmp_path, slot="13h") report.submit("hail_daytime", DAY, [storm], root=tmp_path, slot="19h") c = station_report.build(3, DAY, tmp_path) assert not any(b["block"] == "hail_daytime" for b in c["blocks"]) b = _block(c, "hail") assert b["status"] == EVENT assert b["watch_window"] == {"label": "Daytime checks", "start": "2026-04-22T07:00", "end": "2026-04-22T19:00"} [f] = b["fields"] assert f["suspected"] and f["camera_request"]["photo_at"] == "2026-04-22T19:00" assert (f["start"], f["end"], f["duration_h"]) == ("2026-04-22T16:00", "2026-04-22T17:00", 2) assert "Daytime check 07:00 → 13:00: No hail suspected" in f["summary"] assert "Daytime check 13:00 → 19:00: Hail likely" in f["summary"] assert [d["slot"] for d in f["daytime_checks"]] == ["13h", "19h"] assert b["data_notes"] == ["Daytime check 13:00 → 19:00: No wind sensor (VelocitaVento2m) at " "this station: x."] assert c["overall"]["headline"] == "Hail suspected · photo requested to confirm" json_path, pdf_path = station_report.write(3, DAY, tmp_path, out_dir=tmp_path / "out") assert pdf_path.read_bytes()[:5] == b"%PDF-"