"""SAFETY-137: the 0 W of shutdown and of the Stop button is the last write every
unit hears.  `python3 test_safety137_last_write.py`

A cycle write already on its way (0.3-0.6 s, or hung) must not land after the
0 W, for a GoodWe unit and for a fake unit that does not hand back, at N=1 and
N=2, at shutdown and at Stop. The control cycle (and the heartbeat) issue no
new write once shutdown has begun. The whole wind-down stays inside
HAND_BACK_WORST_S (6 s) when a write hangs.

Mutating a module this file grades? Read MUTATION.md first (purge __pycache__).
The mutants this file was run against are listed at the end.
"""

import asyncio
import sys
import time
from pathlib import Path

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))

from app import main as M                                    # noqa: E402
from app.drivers import Driver, Probe, REGISTRY             # noqa: E402
from app.main import Controller                             # noqa: E402
from app.store import DEFAULTS                              # noqa: E402

fails = []


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


class PlainFake(Driver):
    """A unit that does not hand back (so, like GoodWe, no release()). A write
    of anything but 0 W takes `delay` s to land; every write is logged when it
    LANDS (`landed`) and when it STARTS (`started`)."""
    name = "plainfake"
    writer_key = "host"
    brand = "Fake"
    hands_back = False
    delay = 0.0

    def __init__(self, host="", port=502, unit_id=1, rating=2000.0):
        self.host, self.port, self.unit_id = host, port, unit_id
        self.rating, self.power = rating, 0.0
        self.landed, self.started = [], []

    @classmethod
    def from_entry(cls, entry):
        return cls(entry.get("host", ""), entry.get("port", 502),
                   entry.get("unit_id", 1))

    @classmethod
    def actuator_id_of(cls, entry):
        host = str(entry.get("host") or "").strip()
        return f"{host}:{entry.get('port', 502)}:{entry.get('unit_id', 1)}" \
            if host else ""

    def configured(self):
        return bool(self.host)

    def actuator_id(self):
        return f"{self.host}:{self.port}:{self.unit_id}"

    def soc_source(self):
        return "fake-soc"

    def batt_source(self):
        return "fake-power"

    async def limits(self):
        return -self.rating, self.rating

    async def read(self):
        return 60.0, self.power, {}

    async def write(self, w):
        self.started.append((time.monotonic(), float(w)))
        if w != 0 and self.delay:
            await asyncio.sleep(self.delay)
        self.landed.append((time.monotonic(), float(w)))
        self.power = w
        return True

    async def probe(self):
        return Probe(-self.rating, self.rating, 60.0, self.power, {})


REGISTRY["plainfake"] = PlainFake


class MemStore:
    def __init__(self):
        self.data = dict(DEFAULTS)

    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, values):
        pass

    def close(self):
        pass


class SlowHass:
    """Home Assistant for the GoodWe units. A set_number of anything but 0 W
    takes `delay` s to land; the log gets the write when it lands."""

    def __init__(self):
        self.delay = 0.0
        self.landed, self.started = [], []        # (t, entity, value)
        self.numbers = {"sensor.grid": 3000.0, "sensor.soc_a": 50.0,
                        "sensor.batt_a": 0.0, "sensor.soc_b": 50.0,
                        "sensor.batt_b": 0.0}

    async def number(self, entity, invert=False):
        return self.numbers.get(entity)

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

    async def set_number(self, entity, value):
        self.started.append((time.monotonic(), entity, value))
        if value != 0 and self.delay:
            await asyncio.sleep(self.delay)
        self.landed.append((time.monotonic(), entity, value))
        return True

    async def state(self, entity):
        return None

    async def call(self, *a):
        return True


def gw(name, suffix):
    return dict(name=name, setpoint_entity=f"number.sp_{suffix}",
                soc_entity=f"sensor.soc_{suffix}",
                batt_entity=f"sensor.batt_{suffix}", max_w=3000,
                bms_soc_floor=10)


def fk(name):
    return dict(name=name, driver="plainfake", host="10.9.9.9", port=502,
                unit_id=3, max_w=2000, bms_soc_floor=10)


ROSTERS = {
    "N=1 GoodWe": [gw("inverter-a", "a")],
    "N=1 fake": [fk("inverter-a")],
    "N=2 GoodWe+GoodWe": [gw("inverter-a", "a"), gw("inverter-b", "b")],
    "N=2 GoodWe+fake": [gw("inverter-a", "a"), fk("inverter-b")],
}


def site(inverters, delay):
    opts = dict(meter_source="off", meter_entity="sensor.grid", max_w=5000,
                gain=0.6, slew_w=0, deadband_w=15, target_grid_w=0, step_w=10,
                heartbeat_s=10, stale_input_s=15, shadow_mode="off",
                inverters=inverters)
    hass = SlowHass()
    hass.delay = delay
    c = Controller(opts, hass, MemStore(), Pub(), stop=asyncio.Event())
    for inv in c.inverters:
        inv.dev_min, inv.dev_max = -3000.0, 3000.0
        if isinstance(inv.driver, PlainFake):
            inv.driver.delay = delay
    c.auto = True
    return c, hass


def per_unit(c, hass, kind="landed"):
    """name -> [(t, value)] in the order the writes LANDED (or started)."""
    out = {}
    for inv in c.inverters:
        d = inv.driver
        if isinstance(d, PlainFake):
            out[inv.name] = list(getattr(d, kind))
        else:
            ent = d.setpoint_entity
            out[inv.name] = [(t, v) for t, e, v in getattr(hass, kind)
                             if e == ent]
    return out


async def cycles(c, n=2):
    for _ in range(n):
        await c.read_inputs()
        await c.cycle()


async def in_flight(c):
    """A cycle write on its way to every unit (what rig 5b had): the real
    cycle, whose writes are authorised and take `delay` s to land."""
    tasks = [asyncio.ensure_future(cycles(c, 1))]
    await asyncio.sleep(0.1)
    assert all(not t.done() for t in tasks)
    return tasks


async def finish(tasks):
    for t in tasks:
        t.cancel()
    for t in tasks:
        try:
            await t
        except BaseException:           # noqa: BLE001 - cancelled or timed out
            pass


def last_is_zero(c, hass):
    got = per_unit(c, hass)
    return all(w and w[-1][1] == 0.0 for w in got.values()), got


def scenario(label, inverters, how, delay):
    """how: 'shutdown' or 'stop'. Returns (seconds, ok, detail)."""
    c, hass = site(inverters, delay)

    async def go():
        tasks = await in_flight(c)
        t0 = time.monotonic()
        if how == "shutdown":
            await c.shutdown()
        else:
            await c.handle_action("auto_toggle")        # Stop
        dt = time.monotonic() - t0
        # let a write that is still on its way land, to see if it overtakes
        await asyncio.sleep(0.8 if delay < 5 else 0.0)
        await finish(tasks)
        return dt

    dt = asyncio.run(go())
    ok, got = last_is_zero(c, hass)
    n_sent = {n: len(w) for n, w in got.items()}
    check(f"{label}: the cycle write was on its way at {how} (test sees it)",
          all(len(w) >= 2 for w in per_unit(c, hass, "started").values()))
    print(f"    {label}: {how}, delay {delay} s: {dt:.2f} s, "
          f"landed per unit {n_sent}, last "
          f"{ {n: (w[-1][1] if w else None) for n, w in got.items()} }")
    return dt, ok, got


print("SAFETY-137 - the 0 W is the last write")

print("1. a slow write on its way (0.3 s, 0.6 s) when shutdown / Stop starts")
for label, roster in ROSTERS.items():
    for how in ("shutdown", "stop"):
        for delay in (0.3, 0.6):
            dt, ok, got = scenario(label, roster, how, delay)
            check(f"{label}, {how}, {delay} s in flight: the last write per "
                  f"unit is 0 W", ok)
            check(f"{label}, {how}, {delay} s in flight: the slow write "
                  f"landed first, then the 0 W (not dropped, not overtaken)",
                  all(len(w) >= 2 and w[-2][1] != 0.0
                      for w in got.values()))
            check(f"{label}, {how}: took about the write, not the budget "
                  f"({dt:.2f} s)", dt < delay + 1.0)

print("2. a hung write on its way: the 0 W still reaches every unit, in time")
for label, roster in ROSTERS.items():
    for how in ("shutdown", "stop"):
        dt, ok, got = scenario(label, roster, how, 3600.0)
        check(f"{label}, {how}, hung write: 0 W landed on every unit and is "
              f"its last write", ok)
        check(f"{label}, {how}, hung write: inside HAND_BACK_WORST_S "
              f"({dt:.2f} s <= {M.HAND_BACK_WORST_S:.0f} s)",
              dt <= M.HAND_BACK_WORST_S)
        check(f"{label}, {how}, hung write: the 0 W goes after the settle "
              f"wait ({M.SETTLE_BUDGET_S:.0f} s), not after the write's own "
              f"budget ({dt:.2f} s)", dt <= M.SETTLE_BUDGET_S + 0.5)

print("3. no new write once shutdown has begun (cycle, heartbeat)")
for label, roster in ROSTERS.items():
    c, hass = site(roster, 0.4)

    async def go3():
        tasks = await in_flight(c)
        s = asyncio.ensure_future(c.shutdown())
        await asyncio.sleep(0)                      # shutdown has begun
        begun = time.monotonic()
        await cycles(c, 3)                          # would write: grid 3000 W
        await c.send_heartbeats(10 ** 9, 10)        # every unit is due
        await asyncio.wait_for(s, 10)
        await asyncio.sleep(0.8)
        await finish(tasks)
        return begun

    begun = asyncio.run(go3())
    starts = per_unit(c, hass, "started")
    late = {n: [v for t, v in w if t >= begun and v != 0.0]
            for n, w in starts.items()}
    check(f"{label}: no non-zero write started after shutdown began "
          f"({late})", all(not v for v in late.values()))
    ok, got = last_is_zero(c, hass)
    check(f"{label}: and the last write per unit is 0 W", ok)
    # sanity: the same cycles DO write before shutdown
    c2, hass2 = site(roster, 0.0)
    asyncio.run(cycles(c2, 3))
    before = per_unit(c2, hass2, "started")
    check(f"{label}: control, same cycles before shutdown, write non-zero "
          f"(the test can see a write)",
          all(any(v != 0.0 for _t, v in w) for w in before.values()))

print()
if fails:
    print(f"FAILED {len(fails)}:")
    for f in fails:
        print("  -", f)
    sys.exit(1)
print("all ok")

# MUTANTS this file was run against (each must turn it red):
#  (a) _await_inflight does nothing (no wait for in-flight writes), GoodWe too
#  (b) the `elif self.stopping: return` gate removed (cycle not stopped)
#  (c) `last=True` not passed by command_site_release (Stop/shutdown 0 W is a
#      plain write)
#  (d) the wait unbounded (no timeout): red on the settle-bound check