"""SAFETY-118: the peak forecast is read in its own unit.  `python3 -B test_safety118.py`

`peak_forecast_entity` used to be read as a bare number and anything under 50
was multiplied by 1000, so a W sensor reading 17 W was taken as 17 kW. It is
now read through `pv.watts()` by its own `unit_of_measurement`. No unit, or a
unit that is not power: the forecast is not used, as if the option were empty.

1. The reader through the real `read_inputs()`: W, kW, MW, no unit, kWh, bad state.
2. The warning: once, again only when the cause changes, never every cycle.
3. One state read per cycle, as before.
4. The maintenance charge headroom and the charge-only clamp act on the
   converted value (real `cycle()`).
5. The validator, the status row and the strings.

Mutating a module this file grades? Read MUTATION.md first (purge __pycache__).
"""

import asyncio
import logging
from pathlib import Path

from app import main as _main
from app import messages, pv, validate
from app.main import Controller

fails = []


def check(name, cond):
    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
    if not cond:
        fails.append(name)


def st(value, unit=None):
    attrs = {} if unit is None else {"unit_of_measurement": unit}
    return {"state": str(value), "attributes": attrs}


class _Store:
    def __init__(self, data=None):
        self.data = dict(data or {})

    def set(self, k, v):
        self.data[k] = v

    def get_time(self, *a):
        return None

    def set_time(self, *a):
        pass


class _Pub:
    enabled = False

    def publish(self, *a, **k):
        pass

    def close(self):
        pass


class _Hass:
    def __init__(self, fc=None, prices=None):
        self.fc = fc
        self.prices = prices or {}
        self.reads = []
        self.writes = []

    async def limits(self, entity):
        return (-5000.0, 5000.0, 1.0)

    async def state(self, entity):
        self.reads.append(entity)
        if entity == "sensor.fc":
            return self.fc
        v = {"sensor.grid": 0.0, "sensor.soc": 50.0, "sensor.batt": 0.0,
             **self.prices}.get(entity)
        return None if v is None else st(v, "W")

    async def number(self, entity, invert=False):
        obj = await self.state(entity)
        try:
            return float(obj["state"]) if obj else None
        except ValueError:
            return None

    async def set_number(self, entity, value):
        self.writes.append(float(value))
        return True

    async def call(self, domain, service, data):
        return True


OPTS = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
        "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
        "meter_source": "off", "heartbeat_s": 10, "stale_input_s": 15,
        "max_w": 5000.0, "gain": 0.6, "slew_w": 400.0, "deadband_w": 15,
        "step_w": 10, "target_grid_w": -10.0, "stale_input_s": 15,
        "peak_cap_w": 5000, "peak_margin_w": 0,
        "peak_forecast_entity": "sensor.fc"}


class _Rec(logging.Handler):
    def __init__(self):
        super().__init__(logging.DEBUG)
        self.rows = []

    def emit(self, r):
        self.rows.append((r.levelno, r.getMessage()))


def make(fc, store=None, prices=None, **over):
    hass = _Hass(fc, prices)
    c = Controller(dict(OPTS, **over), hass, _Store(store), _Pub(),
                   stop=asyncio.Event())
    c.auto, c.dev_min, c.dev_max = True, -5000.0, 5000.0
    return c, hass


def reads(c):
    asyncio.run(c.read_inputs())
    return c.peak_fc


def peak_of(state):
    c, _h = make(state)
    return reads(c)


print("1. the reader")
CASES = [("W 17", st(17, "W"), 17.0), ("W 820", st(820, "W"), 820.0),
         ("kW 0.82", st(0.82, "kW"), 820.0), ("kW 0.017", st(0.017, "kW"), 17.0),
         ("MW 0.0008", st(0.0008, "MW"), 800.0), ("W 49", st(49, "W"), 49.0),
         ("W 50", st(50, "W"), 50.0), ("kW 30", st(30, "kW"), 30000.0),
         ("W 0", st(0, "W"), 0.0)]
for label, s, want in CASES:
    got = peak_of(s)
    check(f"{label} -> {want} W (got {got})",
          got is not None and abs(got - want) < 1e-6)
for label, s in [("no unit", st(820)), ("empty unit", st(820, "")),
                 ("kWh", st(820, "kWh")), ("percent", st(20, "%")),
                 ("unavailable", st("unavailable", "W")),
                 ("unknown", st("unknown", "W")), ("missing entity", None),
                 ("not a number", st("sunny", "W"))]:
    check(f"{label} -> None (got {peak_of(s)})", peak_of(s) is None)
c, _h = make(st(820, "W"), peak_forecast_entity="")
check("option empty -> None and the entity is never read",
      reads(c) is None and "sensor.fc" not in _h.reads)

print("2. the warning")
rec = _Rec()
lg = logging.getLogger("goodwe")
lg.addHandler(rec)
lg.setLevel(logging.DEBUG)


def warns():
    return [m for lv, m in rec.rows if lv >= logging.WARNING]


c, h = make(st(820, "kWh"))
for _ in range(5):
    reads(c)
check(f"kWh: five cycles, one WARNING ({warns()})", len(warns()) == 1)
check("...short, plain, and says it is not used",
      warns()[0] == "The 15-minute demand forecast is not in a power unit, so "
      "it is not used." and len(warns()[0].split()) <= 25)
h.fc = st(820, "MWh")
reads(c)
check("the same cause with another unit warns again (the cause changed)",
      len(warns()) == 2)
h.fc = st(820)
reads(c)
check("no unit is a different cause and warns", len(warns()) == 3
      and "has no unit" in warns()[2])
reads(c)
check("...and only once", len(warns()) == 3)
USED = (logging.INFO, "The 15-minute demand forecast is used again.")


def used():
    return [r for r in rec.rows if r == USED]


h.fc = st("unavailable")
reads(c)
check("unavailable adds no warning and no 'used again'",
      len(warns()) == 3 and not used())
h.fc = st(820, "W")
reads(c)
check("a good unit is read again: one INFO, no new warning",
      c.peak_fc == 820.0 and len(warns()) == 3 and len(used()) == 1
      and rec.rows[-1] == USED)
n = len(warns())
for s in (st(820, "W"), st(0.8, "kW")):
    h.fc = s
    reads(c)
check("W and kW never warn", len(warns()) == n)
rows = [r for r in c.checks() if r.get("id") == "peak_forecast"]
check("a good forecast has no peak_forecast row", not rows)
h.fc = st(820, "kWh")
reads(c)
rows = [r for r in c.checks() if r.get("id") == "peak_forecast"]
check("the status page carries a caution row for the unit fault",
      len(rows) == 1 and rows[0]["warn"] and not rows[0]["ok"]
      and rows[0]["text"] == messages.row_peak_forecast_unit("kWh"))
lg.removeHandler(rec)

print("3. one state read per cycle")
c, h = make(st(0.82, "kW"))
reads(c)
check(f"one read of the forecast entity per read_inputs() ({h.reads})",
      h.reads.count("sensor.fc") == 1)

print("4. maintenance headroom and the charge-only clamp use watts")
seen = []
_real = _main.maintenance_charge_floor


def spy(charge_w, fc, cap):
    out = _real(charge_w, fc, cap)
    seen.append((charge_w, fc, cap, out))
    return out


_main.maintenance_charge_floor = spy
try:
    def cycle(fc, **kw):
        c, h = make(fc, store={"phase": "charge"},
                    prices={"sensor.pnow": 1.0, "sensor.pavg": 2.0},
                    price_now_entity="sensor.pnow",
                    price_avg_entity="sensor.pavg", **kw)
        seen.clear()
        asyncio.run(c.read_inputs())
        asyncio.run(c.cycle())
        return c, h

    # cap 2500 W, charge rate 2000 W: 820 W of forecast leaves 1680 W, 17 W
    # leaves the whole 2000 W (the size rule read 17 as 17 kW and left 0).
    c, h = cycle(st(0.82, "kW"), peak_cap_w=2500)
    check(f"cheap-window headroom is cap - 820 W = 1680 W ({seen})",
          len(seen) == 1 and seen[0][1] == 820.0 and seen[0][0] == 2000.0
          and abs(seen[0][3] - 1680.0) < 1e-6)
    c, h = cycle(st(17, "W"), peak_cap_w=2500)
    check(f"a 17 W forecast leaves the full 2000 W charge rate ({seen})",
          len(seen) == 1 and seen[0][1] == 17.0 and abs(seen[0][3] - 2000.0) < 1e-6)
    # charge-only clamp: not cheap, shaping suspended only if the forecast is
    # over the aim.  6 kW is over 5000 W; 17 W is not (the old rule read it
    # as 17 kW and suspended).
    def clamp(fc):
        c, h = make(fc, store={"phase": "charge"})
        asyncio.run(c.read_inputs())
        asyncio.run(c.cycle())
        return c._maint_shape_suspended
    check("kW 6.0 (6000 W, over the 5000 W aim) suspends the charge-only clamp",
          clamp(st(6.0, "kW")) is True)
    check("W 17 leaves the charge-only clamp in place",
          clamp(st(17, "W")) is False)
    check("kW 0.017 leaves the charge-only clamp in place",
          clamp(st(0.017, "kW")) is False)
    check("no unit: forecast unused, clamp in place, as with the option empty",
          clamp(st(6000)) is False)
finally:
    _main.maintenance_charge_floor = _real

print("5. validator and strings")
def notes(unit, key="peak_forecast_entity"):
    ent = {"sensor.fc": {"entity_id": "sensor.fc", "unit": unit,
                         "state": "1"}}
    return [n for n in validate.config_notes(
        {key: "sensor.fc", "meter_source": "off"}, entities=ent)
        if key in n.keys]


check("validator: W no note", not notes("W"))
check("validator: kW no note", not notes("kW"))
check("validator: MW no note", not notes("MW"))
nu, kw = notes(""), notes("kWh")
check(f"validator: no unit warns ({nu})", len(nu) == 1 and nu[0].tier == "warn"
      and "has no unit" in nu[0].text)
check(f"validator: kWh warns ({kw})", len(kw) == 1 and kw[0].tier == "warn"
      and "not in a power unit" in kw[0].text)
check("validator: the false 'taken as watts' claim is gone for this entity",
      not any("taken as watts" in n.text for n in nu + kw + notes("kW")))
check("validator: another watt key keeps its claim",
      len(notes("kW", "batt_entity")) == 1
      and "taken as watts" in notes("kW", "batt_entity")[0].text)
check("WATT_ENTITY_KEYS no longer holds the peak forecast",
      "peak_forecast_entity" not in validate.WATT_ENTITY_KEYS
      and "batt_entity" in validate.WATT_ENTITY_KEYS)
print("   EN no unit :", messages.peak_forecast_unit(""))
print("   EN bad unit:", messages.peak_forecast_unit("kWh"))
print("   NL no unit :", messages.peak_forecast_unit("", "nl"))
print("   NL bad unit:", messages.peak_forecast_unit("kWh", "nl"))
check("the strings differ by cause and by language",
      len({messages.peak_forecast_unit(u, l) for u in ("", "kWh")
           for l in ("en", "nl")}) == 4)
print("6. a missing reading leaves the unit fault as it was (F1)")


lg.addHandler(rec)


def prow(c):
    return [r for r in c.checks() if r.get("id") == "peak_forecast"]


def flap(first, gap, last):
    """first, gap, first again (or last): the log rows and the status row
    after each cycle."""
    rec.rows.clear()
    c, h = make(first)
    out = []
    for s in (first, gap, last):
        h.fc = s
        reads(c)
        out.append((len(warns()), len(used()), len(prow(c)), c.peak_fc))
    return out


for label, first, gap in [
        ("kWh -> unavailable -> kWh", st(820, "kWh"), st("unavailable", "kWh")),
        ("kWh -> unknown -> kWh", st(820, "kWh"), st("unknown", "kWh")),
        ("kWh -> failed read -> kWh", st(820, "kWh"), None),
        ("kWh -> non-numeric -> kWh", st(820, "kWh"), st("sunny", "kWh")),
        ("no unit -> failed read -> no unit", st(820), None),
        ("no unit -> unavailable -> no unit", st(820), st("unavailable"))]:
    r = flap(first, gap, first)
    check(f"{label}: one WARNING, no 'used again', row on all 3 ({r})",
          [x[0] for x in r] == [1, 1, 1] and [x[1] for x in r] == [0, 0, 0]
          and [x[2] for x in r] == [1, 1, 1]
          and [x[3] for x in r] == [None, None, None])
r = flap(st(820, "kWh"), st("unavailable", "kWh"), st(820, "W"))
check(f"kWh -> unavailable -> W: the fault clears only on the W reading ({r})",
      [x[1] for x in r] == [0, 0, 1] and [x[2] for x in r] == [1, 1, 0]
      and [x[0] for x in r] == [1, 1, 1] and r[2][3] == 820.0)
rec.rows.clear()
c, h = make(st(820, "W"))
for s in (st(820, "W"), st("unavailable", "W"), None, st("sunny", "W")):
    h.fc = s
    reads(c)
    check(f"good unit, reading {s and s['state']!r}: no row, no log",
          not prow(c) and not warns() and not used())
rec.rows.clear()
c, h = make(st(820, "kWh"))
reads(c)
for s in (st("unavailable", "kWh"), None):
    h.fc = s
    reads(c)
    check("fault row stays while the reading is missing",
          len(prow(c)) == 1 and prow(c)[0]["text"]
          == messages.row_peak_forecast_unit("kWh"))
c, h = make(st(820, "W"))
h.fc = st("unavailable", "W")
reads(c)
check("good-unit sensor only unavailable: the fault row stays away",
      not prow(c))
lg.removeHandler(rec)
check("has_reading: numbers only",
      pv.has_reading(st(0, "kWh")) and pv.has_reading(st("1.5"))
      and not pv.has_reading(None) and not pv.has_reading(st("unavailable"))
      and not pv.has_reading(st("unknown")) and not pv.has_reading(st("x")))
check("pv.unit_fault: None for a good unit and for a missing reading (has_reading tells them apart)",
      pv.unit_fault(st(1, "kW")) is None and pv.unit_fault(None) is None
      and pv.unit_fault(st("unavailable")) is None
      and pv.unit_fault(st(1)) == "" and pv.unit_fault(st(1, "kWh")) == "kWh")
src = (Path(__file__).parent / "app" / "main.py").read_text(encoding="utf-8")
check("the size rule is gone from main.py", "abs(raw) < 50" not in src)

print()
print("FAILED: " + "; ".join(fails) if fails else "all checks passed")
raise SystemExit(1 if fails else 0)