from datetime import date, datetime

import numpy as np
import pandas as pd
import pytest

from damage_engine import thresholds
from damage_engine.detectors import strong_wind
from damage_engine.schemas import EVENT, NO_DATA, NO_EVENT, Severity
from damage_engine.sources.fields import Field

RUN_DAY = date(2026, 9, 15)
# the 24 h checked for RUN_DAY: records 08:00 of 14 Sep ... 07:00 of 15 Sep
HOURS = pd.date_range("2026-09-14 08:00", "2026-09-15 07:00", freq="h")
CALM = [3.0] * 24


def _day(station, gust, mean=None, ts=None):
    ts = HOURS[:len(gust)] if ts is None else ts
    mean = [g / 2 if g is not None and not np.isnan(g) else np.nan for g in gust] if mean is None else mean
    return pd.DataFrame({"CodiceStazione": station, "Timestamp": ts,
                         strong_wind.COL_MAX: gust, strong_wind.COL_MEAN: mean})


def _field(station, bbch=None):
    return Field(id=f"ST{station}", name=f"field {station}", station_id=station, bbch=bbch)


def _with_storm(values, at=10):
    gust = list(CALM)
    gust[at:at + len(values)] = values
    return gust


def test_window_is_the_24h_ending_at_07():
    assert strong_wind.RUN_HOUR == 7
    assert strong_wind.day_window(RUN_DAY) == (datetime(2026, 9, 14, 7), datetime(2026, 9, 15, 7))


def test_trigger_is_the_hourly_maximum():
    assert thresholds.STRONG_WIND["use_mean_or_gust"] == "hourly_max"
    assert strong_wind.trigger_columns() == (strong_wind.COL_MAX, strong_wind.COL_MEAN)


def test_storm_is_an_event_graded_on_beaufort():
    # 18:00-20:00: 15.0, 18.5, 14.0 m/s hourly maximum, means far below the trigger
    [a] = strong_wind.detect(_day(3, _with_storm([15.0, 18.5, 14.0])), [_field(3)], RUN_DAY, {3: "Lagnasco"})
    assert a.status == EVENT and a.block == "strong_wind" and a.station_name == "Lagnasco"
    assert a.duration_h == 3
    assert a.start == datetime(2026, 9, 14, 18) and a.end == datetime(2026, 9, 14, 20)
    assert a.details["max_wind_ms"] == 18.5 and a.details["max_wind_kmh"] == 66.6
    assert a.details["max_wind_at"] == "2026-09-14T19:00"
    assert a.details["max_mean_wind_ms"] == 9.2           # rounded to 0.1 m/s
    assert a.details["strong_wind_hours"] == 3
    assert a.severity == Severity.MODERATE                     # Beaufort 8: 17.2 <= 18.5 < 20.8
    assert a.details["in_cover_window"] is None                # stage unknown
    assert "18.5 m/s (66.6 km/h)" in a.summary


def test_trigger_is_at_least_13_9():
    [hit] = strong_wind.detect(_day(3, _with_storm([13.9])), [_field(3)], RUN_DAY)
    [miss] = strong_wind.detect(_day(3, _with_storm([13.8])), [_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_wind_ms"] == 13.8
    assert miss.summary == "No strong wind: highest wind 13.8 m/s (49.7 km/h)."


@pytest.mark.parametrize("wind, level", [(17.2, Severity.MODERATE), (20.8, Severity.SEVERE),
                                         (24.5, Severity.EXTREME), (30.0, Severity.EXTREME)])
def test_beaufort_bands(wind, level):
    [a] = strong_wind.detect(_day(3, _with_storm([wind])), [_field(3)], RUN_DAY)
    assert a.severity == level


def test_hours_outside_the_24h_are_ignored():
    ts = pd.date_range("2026-09-14 07:00", "2026-09-15 08:00", freq="h")     # 26 records
    gust = [25.0] + CALM + [25.0]                               # storms at 07:00 (14th), 08:00 (15th)
    [a] = strong_wind.detect(_day(3, gust, ts=ts), [_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_storm([16.0]))
    [after, before] = strong_wind.detect(df, [_field(3, bbch=72), _field(3, bbch=65)], RUN_DAY)
    assert after.details["in_cover_window"] is True            # fruit set (BBCH > 69)
    assert before.details["in_cover_window"] is False          # still in bloom
    assert after.severity == before.severity                   # severity does not depend on the stage
    assert strong_wind.in_cover_window(date(2026, 11, 2), 75) is False


def test_missing_maximum_uses_the_mean():
    gust = _with_storm([np.nan, 16.0])
    mean = [1.5] * 24
    mean[10] = 15.0                                             # only the mean of 18:00 is known
    [a] = strong_wind.detect(_day(3, gust, mean=mean), [_field(3)], RUN_DAY)
    assert a.status == EVENT and a.duration_h == 2 and a.details["fallback_hours"] == 1
    assert any("Hourly maximum wind missing for 1 of 24 hours: hourly mean used" in n for n in a.data_notes)


def test_missing_hours_are_noted_and_not_checked():
    df = _day(3, _with_storm([20.0])).drop(index=[0, 1, 2])     # 3 hours with no record
    [a] = strong_wind.detect(df, [_field(3)], RUN_DAY)
    assert a.status == EVENT and a.details["records"] == 21
    assert "Wind missing for 3 of 24 hours: those hours not checked." in a.data_notes


def test_no_wind_values_is_no_data():
    df = _day(3, [np.nan] * 24, mean=[np.nan] * 24)
    [a] = strong_wind.detect(df, [_field(3)], RUN_DAY)
    assert a.status == NO_DATA
    assert a.data_notes == ["No wind data in the 24 hours checked: strong wind cannot be judged."]


def test_missing_columns_do_not_break_the_block():
    df = pd.DataFrame({"CodiceStazione": 3, "Timestamp": HOURS})          # no wind columns at all
    [a] = strong_wind.detect(df, [_field(3)], RUN_DAY)
    assert a.status == NO_DATA


def test_station_without_records_is_no_data():
    [a] = strong_wind.detect(_day(3, CALM), [_field(6)], RUN_DAY)
    assert a.status == NO_DATA and a.data_notes == ["No records from the station for the last 24 hours."]


def test_station_without_wind_sensor_is_no_data(monkeypatch):
    station = type("S", (), {"id": 3, "name": "Lagnasco", "sensors": ["VelocitaVento2m"]})()
    monkeypatch.setattr(strong_wind, "stations_with", lambda *s: [station])
    monkeypatch.setattr(strong_wind, "all_fields", lambda: [_field(3), _field(153)])
    monkeypatch.setattr(strong_wind, "fetch_hourly", lambda ids, start, end, cols: _day(3, CALM))
    out = {a.station_id: a for a in strong_wind.run(RUN_DAY)}
    assert out[3].status == NO_EVENT
    assert out[153].status == NO_DATA
    assert out[153].data_notes == ["No wind sensor (VelocitaVento2m) at this station: "
                                   "strong wind cannot be judged."]


def test_run_asks_for_the_24h_records(monkeypatch):
    calls = []
    station = type("S", (), {"id": 3, "name": "Lagnasco", "sensors": ["VelocitaVento2m"]})()
    monkeypatch.setattr(strong_wind, "stations_with", lambda *s: [station])
    monkeypatch.setattr(strong_wind, "all_fields", lambda: [_field(3)])
    monkeypatch.setattr(strong_wind, "fetch_hourly",
                        lambda ids, start, end, cols: calls.append((ids, start, end, cols)) or _day(3, CALM))
    strong_wind.run(RUN_DAY)
    assert calls == [([3], datetime(2026, 9, 14, 8), datetime(2026, 9, 15, 7), strong_wind.COLUMNS)]


def test_unknown_trigger_setting_is_an_error(monkeypatch):
    monkeypatch.setitem(thresholds.STRONG_WIND, "use_mean_or_gust", "gust")
    with pytest.raises(ValueError, match="use_mean_or_gust"):
        strong_wind.trigger_columns()


def test_hourly_mean_setting_uses_the_mean(monkeypatch):
    monkeypatch.setitem(thresholds.STRONG_WIND, "use_mean_or_gust", "hourly_mean")
    gust = _with_storm([20.0])
    mean = [2.0] * 24
    [a] = strong_wind.detect(_day(3, gust, mean=mean), [_field(3)], RUN_DAY)
    assert a.status == NO_EVENT and a.details["max_wind_ms"] == 2.0