107 lines
5.4 KiB
Python
107 lines
5.4 KiB
Python
from datetime import date, datetime, timedelta
|
|
|
|
from damage_engine.report import report
|
|
from damage_engine.schemas import EVENT, NO_EVENT, Analysis, Severity
|
|
from damage_engine.severity import grade
|
|
from damage_engine.thresholds import FROST_FALLBACK_BANDS_C, FROST_KILL_BANDS
|
|
|
|
|
|
def _analysis(day, field="F1", status=NO_EVENT, severity=Severity.NONE, block="brina"):
|
|
start = datetime.combine(day, datetime.min.time()) if status == EVENT else None
|
|
return Analysis(block=block, run_day=day, field_id=field, field_name=f"name {field}",
|
|
station_id=3, station_name="Lagnasco", status=status, severity=severity,
|
|
severity_score=0.6 if status == EVENT else 0.0,
|
|
duration_h=4 if status == EVENT else 0, start=start, end=start,
|
|
summary="test", details={"min_bud_temp_c": -3.0})
|
|
|
|
|
|
def test_grade_bands():
|
|
assert grade(0.0, FROST_FALLBACK_BANDS_C)[0] == Severity.NONE # strict: below 0 °C only
|
|
sev, score = grade(-3.1, FROST_FALLBACK_BANDS_C)
|
|
assert sev == Severity.MODERATE and abs(score - 0.6375) < 1e-3
|
|
assert grade(0.95, FROST_KILL_BANDS) == (Severity.EXTREME, 1.0)
|
|
|
|
|
|
def test_analysis_round_trip():
|
|
a = _analysis(date(2025, 3, 12), status=EVENT, severity=Severity.SEVERE)
|
|
assert Analysis.from_dict(a.to_dict()) == a
|
|
|
|
|
|
def test_submit_writes_block_report_and_keeps_last_runs(tmp_path):
|
|
first = date(2025, 3, 1)
|
|
for i in range(report.HISTORY_RUNS + 2):
|
|
day = first + timedelta(days=i)
|
|
text = report.submit("brina", day, [_analysis(day)], root=tmp_path)
|
|
assert "BRINA report" in text
|
|
kept = report.runs("brina", tmp_path)
|
|
assert len(kept) == report.HISTORY_RUNS
|
|
assert kept[0]["run_day"] == first + timedelta(days=report.HISTORY_RUNS + 1)
|
|
assert len(list((tmp_path / "brina").glob("*.txt"))) == report.HISTORY_RUNS
|
|
|
|
|
|
def test_rerun_same_day_replaces_run(tmp_path):
|
|
day = date(2025, 3, 12)
|
|
report.submit("brina", day, [_analysis(day)], root=tmp_path)
|
|
report.submit("brina", day, [_analysis(day, status=EVENT, severity=Severity.LIGHT)], root=tmp_path)
|
|
[run] = report.runs("brina", tmp_path)
|
|
assert run["analyses"][0].status == EVENT
|
|
|
|
|
|
def test_field_report_combines_blocks_and_past_dates(tmp_path):
|
|
d1, d2 = date(2025, 3, 11), date(2025, 3, 12)
|
|
report.submit("brina", d1, [_analysis(d1, status=EVENT, severity=Severity.SEVERE),
|
|
_analysis(d1, field="F2")], root=tmp_path)
|
|
report.submit("brina", d2, [_analysis(d2), _analysis(d2, field="F2")], root=tmp_path)
|
|
report.submit("hail", d2, [_analysis(d2, block="hail")], root=tmp_path)
|
|
|
|
assert report.fields(tmp_path) == {"F1": "name F1", "F2": "name F2"}
|
|
assert report.field_report_dates("F1", tmp_path) == [d2, d1]
|
|
|
|
latest = report.field_report("F1", root=tmp_path)
|
|
assert "[BRINA] run 2025-03-12" in latest and "[HAIL] run 2025-03-12" in latest
|
|
|
|
past = report.field_report("F1", d1, root=tmp_path)
|
|
assert "[BRINA] run 2025-03-11" in past and "SEVERE" in past
|
|
assert "[HAIL] no run yet" in past
|
|
|
|
|
|
def test_reports_show_data_notes(tmp_path):
|
|
day = date(2025, 3, 12)
|
|
a = _analysis(day)
|
|
a.data_notes = ["No wind sensor at this station: calm test not applied."]
|
|
text = report.submit("brina", day, [a], root=tmp_path)
|
|
assert "Missing sensors / data:" in text and "No wind sensor" in text
|
|
assert "Data: No wind sensor" in report.field_report("F1", root=tmp_path)
|
|
|
|
|
|
def test_fields_report_groups_blocks_under_each_field(tmp_path):
|
|
day = date(2025, 6, 12)
|
|
report.submit("flood", day, [_analysis(day, field="F2", block="flood"),
|
|
_analysis(day, field="F1", block="flood")], root=tmp_path)
|
|
report.submit("brina", day, [_analysis(day, field="F1", status=EVENT, severity=Severity.LIGHT),
|
|
_analysis(day, field="F2")], root=tmp_path)
|
|
text = (tmp_path / report.FIELDS_REPORT_FILE).read_text(encoding="utf-8")
|
|
assert text.startswith("REPORT BY FIELD") and "blocks: brina, flood" in text
|
|
f1, f2 = text.index("FIELD name F1 [F1]"), text.index("FIELD name F2 [F2]")
|
|
# field F1 first, with its brina and flood sections, then field F2 with its own
|
|
assert f1 < text.index("[BRINA]", f1) < text.index("[FLOOD]", f1) < f2
|
|
assert f2 < text.index("[BRINA]", f2) < text.index("[FLOOD]", f2)
|
|
assert "LIGHT" in text[f1:f2] and "LIGHT" not in text[f2:]
|
|
|
|
|
|
def test_slots_keep_every_check_of_a_day_and_history_counts_days(tmp_path):
|
|
first = date(2025, 7, 1)
|
|
for i in range(report.HISTORY_RUNS + 1):
|
|
day = first + timedelta(days=i)
|
|
for slot in ("13h", "19h"):
|
|
text = report.submit("hail_daytime", day, [_analysis(day, block="hail_daytime")],
|
|
root=tmp_path, slot=slot)
|
|
assert "HAIL_DAYTIME report - run day 2025-07-11, check 19h" in text
|
|
last = first + timedelta(days=report.HISTORY_RUNS)
|
|
assert [r["slot"] for r in report.day_runs("hail_daytime", last, tmp_path)] == ["13h", "19h"]
|
|
days = {r["run_day"] for r in report.runs("hail_daytime", tmp_path)}
|
|
assert len(days) == report.HISTORY_RUNS and first not in days # 10 days, both slots each
|
|
assert len(list((tmp_path / "hail_daytime").glob("*.json"))) == 2 * report.HISTORY_RUNS
|
|
report.submit("hail_daytime", last, [], root=tmp_path, slot="13h") # same slot: replaced
|
|
assert [len(r["analyses"]) for r in report.day_runs("hail_daytime", last, tmp_path)] == [0, 1]
|