dashboard.py

#!/usr/bin/env python3
"""
Local web dashboard for the Axpert inverter. Nothing leaves this machine.

  ./venv/bin/python dashboard.py          then open http://localhost:8770

One process at a time can hold the USB device, so stop this before running
axpert.py by hand (or just use the dashboard's own raw view).
"""
import csv, datetime, json, os, threading, time
from collections import deque
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from axpert import (MODES, QBMS, QPIGS, QPIRI, STATUS2_BITS,
                    STATUS_BITS, parse, parse_flags, parse_status_bits,
                    NetInverter, open_inverter, parse_warnings)

HERE = os.path.dirname(os.path.abspath(__file__))
CSV_PATH = os.path.join(HERE, "history.csv")
EVENTS_PATH = os.path.join(HERE, "events.csv")
BRIDGE_STATE_PATH = os.path.join(HERE, "bridge_state.json")
CONFIG_PATH = os.path.join(HERE, "config.json")

FAST_INTERVAL = 5      # live stats
SLOW_INTERVAL = 120    # settings, BMS, energy counters
DAILY_INTERVAL = 1800  # the inverter's own per-day history
HISTORY_DAYS = 14
HISTORY_POINTS = 1440  # 2 hours at 5s
CSV_MAX_BYTES = 40 * 1024 * 1024
CSV_KEEP_ROWS = 300_000

CSV_COLUMNS = ["timestamp", "mode", "grid_v", "load_w", "load_pct", "batt_v",
               "batt_soc", "batt_charge_a", "batt_discharge_a", "pv_v", "pv_a",
               "pv_w", "temp_c", "grid_hz", "out_v", "load_va", "bms_charge_limit_a"]
EVENT_MODES = {"L": "grid", "B": "battery"}
BMS_INTERVAL = 30      # the BMS's moving charge limit; shows when it's holding solar back
GRID_BUCKET_S = 300    # grid-quality chart resolution

DEFAULT_CONFIG = {
    "battery_kwh": None,          # e.g. 5.12 — enables runtime estimates
    "battery_reserve_percent": 20,  # inverter program 12: grid takes over here
    "battery_return_percent": 60,   # inverter program 13: back to battery here
    "battery_cutoff_percent": 10,   # inverter program 29: floor during a blackout
    "port": 8770,
    "bridge_url": None,             # e.g. "http://inverter.local" to read via the ESP32
    "bridge_token": None,           # must match esp32/src/bridge_secrets.h
}


def load_config():
    cfg = dict(DEFAULT_CONFIG)
    if os.path.exists(CONFIG_PATH):
        try:
            with open(CONFIG_PATH) as fh:
                cfg.update(json.load(fh))
        except (OSError, ValueError) as e:
            print(f"config.json ignored: {e}")
    return cfg


def seconds_between(a, b):
    fmt = "%Y-%m-%dT%H:%M:%S"
    return time.mktime(time.strptime(b, fmt)) - time.mktime(time.strptime(a, fmt))


def num(raw, default=0.0):
    try:
        return float(raw)
    except (TypeError, ValueError):
        return default


class Poller(threading.Thread):
    """Owns the USB device and keeps the newest reading in memory."""

    daemon = True

    def __init__(self, config):
        super().__init__()
        self.config = config
        self.lock = threading.Lock()          # guards self.state / self.history
        self.lock_device = threading.Lock()   # guards the USB handle
        self.inv = None
        self.state = {"connected": False, "error": "starting up"}
        self.history = deque(maxlen=HISTORY_POINTS)
        self.slow = {}
        self._writes = 0
        self._last_sample = None
        self._prev_event_state = None
        self._held_load = None
        self._acc_load = None
        self._typical_load = None
        self._gtb = deque(maxlen=12)
        self._bms_limit = ""
        self._last_bms = 0
        self._warn_names = None
        self._grid_cache = {}
        self._pending_since = {}
        self.daily = {}
        self.events = deque(maxlen=400)
        self.energy = {"date": time.strftime("%Y-%m-%d"), "solar_wh": 0.0,
                       "load_wh": 0.0, "grid_wh": 0.0, "grid_on_h": 0.0}
        self._load_history()
        self._load_events()
        self._seed_energy()

    # -- history ----------------------------------------------------------
    def _migrate_csv(self):
        """Rewrite the log under the current header when columns were added;
        old rows get blanks for the new fields."""
        try:
            with open(CSV_PATH, newline="") as fh:
                header = next(csv.reader(fh), None)
                if header == CSV_COLUMNS:
                    return
                fh.seek(0)
                rows = list(csv.DictReader(fh))
        except (OSError, StopIteration):
            return
        tmp = CSV_PATH + ".tmp"
        with open(tmp, "w", newline="") as fh:
            w = csv.DictWriter(fh, fieldnames=CSV_COLUMNS, extrasaction="ignore")
            w.writeheader()
            w.writerows(rows)
        os.replace(tmp, CSV_PATH)

    def _load_history(self):
        if not os.path.exists(CSV_PATH):
            return
        self._migrate_csv()
        try:
            with open(CSV_PATH, newline="") as fh:
                rows = list(csv.DictReader(fh))[-HISTORY_POINTS:]
        except OSError:
            return
        for r in rows:
            self.history.append({
                "t": r.get("timestamp", ""),
                "solar_w": num(r.get("pv_w")),
                "load_w": num(r.get("load_w")),
                "soc": num(r.get("batt_soc")),
                "batt_v": num(r.get("batt_v")),
                "grid_v": num(r.get("grid_v")),
                "grid_hz": num(r.get("grid_hz")),
            })

    # -- events: the moments things change ---------------------------------
    def _load_events(self):
        if not os.path.exists(EVENTS_PATH):
            return
        try:
            with open(EVENTS_PATH, newline="") as fh:
                for r in csv.DictReader(fh):
                    self.events.append(r)
        except OSError:
            pass

    def _note_event(self, kind, text, ts=None):
        row = {"timestamp": ts or time.strftime("%Y-%m-%dT%H:%M:%S"), "kind": kind, "text": text}
        self.events.append(row)
        new = not os.path.exists(EVENTS_PATH)
        try:
            with open(EVENTS_PATH, "a", newline="") as fh:
                w = csv.DictWriter(fh, fieldnames=["timestamp", "kind", "text"])
                if new:
                    w.writeheader()
                w.writerow(row)
        except OSError:
            pass

    def _detect_events(self, grid_on, mode, charging, discharging, ts=None):
        ts = ts or time.strftime("%Y-%m-%dT%H:%M:%S")
        was = self._prev_event_state
        now = {"grid": grid_on, "mode": mode, "charging": charging, "discharging": discharging}
        if was is None:                      # first reading after a restart
            self._prev_event_state = now
            self._note_event("monitor", "Monitoring started", ts)
            return
        # the inverter only sees the grid after your switch, so "gone" can mean either
        if now["grid"] != was["grid"]:
            self._note_event("grid", "Grid power reached the inverter" if now["grid"]
                             else "Grid power stopped reaching the inverter (blackout or switch)", ts)
        if now["mode"] != was["mode"]:
            self._note_event("mode", f"Switched to {EVENT_MODES.get(mode, MODES.get(mode, mode))}", ts)
        # when solar and the house load are close, the battery flickers between a
        # trickle in and a trickle out; only log a change that holds for a minute
        for key, kind, start, stop in (
                ("charging", "charge", "Started charging", "Stopped charging"),
                ("discharging", "discharge", "Started drawing from the battery",
                 "Stopped drawing from the battery")):
            if now[key] == was[key]:
                self._pending_since.pop(key, None)
                continue
            since = self._pending_since.setdefault(key, ts)
            if seconds_between(since, ts) >= 60:
                self._note_event(kind, start if now[key] else stop, since)
                self._pending_since.pop(key, None)
            else:
                now[key] = was[key]
        self._prev_event_state = now

    def _detect_warnings(self, warn_raw):
        # "Line fail" is already logged as a grid event
        names = {w["name"] for w in parse_warnings(warn_raw)} - {"Line fail"}
        if self._warn_names is not None:
            for n in sorted(names - self._warn_names):
                self._note_event("warning", f"Inverter warning: {n}")
            for n in sorted(self._warn_names - names):
                self._note_event("warning", f"Warning cleared: {n}")
        self._warn_names = names

    # -- grid quality: voltage, frequency and on/off over a window ---------------
    def grid_report(self, hours):
        hours = max(1, min(hours, 24 * 7))
        hit = self._grid_cache.get(hours)
        if hit and time.time() - hit[0] < 60:
            return hit[1]
        start = time.time() - hours * 3600
        since = time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(start))
        rows = []
        try:
            with open(CSV_PATH, "rb") as fh:
                fh.seek(0, os.SEEK_END)
                # ~100 bytes a row, one row per 5 s, plus slack
                fh.seek(max(0, fh.tell() - int(hours * 3600 / 5 * 130)))
                lines = fh.read().decode("utf-8", "replace").splitlines()
            for r in csv.DictReader(lines[1:], fieldnames=CSV_COLUMNS):
                if (r["timestamp"] or "") >= since and r["timestamp"][:2] == "20":
                    rows.append(r)
        except OSError:
            pass

        n = int(hours * 3600 // GRID_BUCKET_S)
        buckets = [{"on": 0, "seen": 0, "v": [], "hz": [], "solar": [], "load": [], "soc": None}
                   for _ in range(n)]
        held = None
        cuts, longest, off_since, prev_on, on_s = 0, 0, None, None, 0.0
        prev_t = None
        for r in rows:
            t = time.mktime(time.strptime(r["timestamp"], "%Y-%m-%dT%H:%M:%S"))
            i = int((t - start) // GRID_BUCKET_S)
            if not 0 <= i < n:
                continue
            v, hz = num(r["grid_v"]), num(r.get("grid_hz"))
            if hz < 40:                        # 0 while the grid is down or not logged
                hz = 0
            on = v > 50
            b = buckets[i]
            b["seen"] += 1
            # line mode reports 0 W most of the time: carry the last real reading
            # through the stretch, and skip the sample if there isn't one yet
            load = num(r["load_w"])
            if r["mode"] != "L":
                held = None
            if r["mode"] == "L" and load == 0:
                load = held
            elif load > 0 and r["mode"] == "L":
                held = load
            b["solar"].append(num(r["pv_w"]))
            if load is not None:
                b["load"].append(load)
            b["soc"] = num(r["batt_soc"])
            if on:
                b["on"] += 1
                b["v"].append(v)
                if hz:
                    b["hz"].append(hz)
            if prev_t is not None and t - prev_t < 120 and prev_on:
                on_s += t - prev_t
            if prev_on and not on:
                cuts += 1
                off_since = t
            elif on and prev_on is False and off_since:
                longest = max(longest, t - off_since)
                off_since = None
            prev_on, prev_t = on, t
        if off_since and prev_t:
            longest = max(longest, prev_t - off_since)

        out = []
        for i, b in enumerate(buckets):
            item = {"t": time.strftime("%H:%M", time.localtime(start + i * GRID_BUCKET_S)),
                    "on": round(b["on"] / b["seen"], 2) if b["seen"] else None}
            if b["v"]:
                item.update(v_min=min(b["v"]), v_max=max(b["v"]),
                            v=round(sum(b["v"]) / len(b["v"]), 1))
            if b["hz"]:
                item.update(hz_min=min(b["hz"]), hz_max=max(b["hz"]),
                            hz=round(sum(b["hz"]) / len(b["hz"]), 2))
            if b["seen"]:
                item.update(solar=round(sum(b["solar"]) / len(b["solar"])),
                            load=round(sum(b["load"]) / len(b["load"])) if b["load"] else None, soc=b["soc"])
            out.append(item)
        vs = [v for b in buckets for v in b["v"]]
        loads = [x for b in buckets for x in b["load"]]
        hzs = [h for b in buckets for h in b["hz"]]
        report = {
            "hours": hours, "bucket_s": GRID_BUCKET_S, "buckets": out,
            "covered_h": round(sum(1 for b in buckets if b["seen"]) * GRID_BUCKET_S / 3600, 1),
            "grid_on_h": round(on_s / 3600, 1), "cuts": cuts, "longest_cut_h": round(longest / 3600, 1),
            "v_min": min(vs) if vs else None, "v_max": max(vs) if vs else None,
            "v_avg": round(sum(vs) / len(vs), 1) if vs else None,
            "low_v_pct": round(100 * sum(1 for v in vs if v < 207) / len(vs)) if vs else None,
            "hz_min": min(hzs) if hzs else None, "hz_max": max(hzs) if hzs else None,
            "load_avg_w": sum(loads) / len(loads) if loads else None,
        }
        self._grid_cache[hours] = (time.time(), report)
        return report

    # -- the inverter's own day-by-day counters -----------------------------
    def _poll_daily(self, inv):
        """QED/QLD only answer for dates on the inverter's own clock, which may
        be years off from ours — so ask it what day it thinks it is first."""
        qt = inv.query("QT")
        if not qt or len(qt) != 14:
            return
        try:
            d0 = datetime.datetime.strptime(qt, "%Y%m%d%H%M%S")
        except ValueError:
            return
        days = []
        for i in range(HISTORY_DAYS):
            d = d0 - datetime.timedelta(days=i)
            pv = inv.query("QED" + d.strftime("%Y%m%d"))
            load = inv.query("QLD" + d.strftime("%Y%m%d"))
            if pv is None:
                continue
            days.append({"label": d.strftime("%-d"), "date": d.strftime("%a %-d %b"),
                         "offset": i, "solar_wh": num(pv), "load_wh": num(load)})
        daily = {
            "days": list(reversed(days)),
            "month_solar_wh": num(inv.query("QEM" + d0.strftime("%Y%m"))),
            "month_load_wh": num(inv.query("QLM" + d0.strftime("%Y%m"))),
            "year_solar_wh": num(inv.query("QEY" + d0.strftime("%Y"))),
            "year_load_wh": num(inv.query("QLY" + d0.strftime("%Y"))),
            "month_label": d0.strftime("%B"),
        }
        if days:
            daily["today_solar_wh"] = days[0]["solar_wh"]
            daily["today_load_wh"] = days[0]["load_wh"]
        with self.lock:
            self.daily = daily

    def _seed_energy(self):
        """Rebuild today's running totals from the log, so a restart doesn't zero them."""
        if not os.path.exists(CSV_PATH):
            return
        today = time.strftime("%Y-%m-%d")
        prev = None
        try:
            with open(CSV_PATH, newline="") as fh:
                for r in csv.DictReader(fh):
                    stamp = r.get("timestamp", "")
                    if not stamp.startswith(today):
                        continue
                    if prev is not None:
                        self._accumulate(prev, stamp)
                    prev = r
        except OSError:
            return
        self._last_sample = prev

    def _accumulate(self, row, next_stamp):
        """Add one sample's worth of energy, using the real gap between readings."""
        try:
            a = time.mktime(time.strptime(row["timestamp"], "%Y-%m-%dT%H:%M:%S"))
            b = time.mktime(time.strptime(next_stamp, "%Y-%m-%dT%H:%M:%S"))
        except (ValueError, KeyError):
            return
        dt = min(max(b - a, 0), 60) / 3600.0   # cap gaps so downtime isn't counted
        if not dt:
            return
        solar, load = num(row.get("pv_w")), num(row.get("load_w"))
        load = self._hold_load(row, load, a)
        self.energy["solar_wh"] += solar * dt
        self.energy["load_wh"] += load * dt
        if num(row.get("grid_v")) > 50:
            self.energy["grid_on_h"] = self.energy.get("grid_on_h", 0.0) + dt
        # the grid only supplies anything in line mode: the house, plus whatever
        # the battery takes beyond what solar gives
        if row.get("mode") == "L":
            batt_w = num(row.get("batt_v")) * (num(row.get("batt_charge_a"))
                                               - num(row.get("batt_discharge_a")))
            self.energy["grid_wh"] += (load + max(0.0, batt_w - solar)) * dt

    def _hold_load(self, row, load, t):
        """Line mode reports 0 W most of the time at light load; carry the last
        real reading through the line-mode stretch."""
        if row.get("mode") != "L":
            self._acc_load = None
            return load
        if load == 0:
            return self._acc_load or 0.0
        self._acc_load = load
        return load

    def _roll_energy(self, row):
        today = row["timestamp"][:10]
        if self.energy["date"] != today:
            self.energy = {"date": today, "solar_wh": 0.0, "load_wh": 0.0, "grid_wh": 0.0, "grid_on_h": 0.0}
            self._last_sample = None
        if self._last_sample is not None:
            self._accumulate(self._last_sample, row["timestamp"])
        self._last_sample = row

    def _append_csv(self, row):
        new = not os.path.exists(CSV_PATH)
        with open(CSV_PATH, "a", newline="") as fh:
            w = csv.DictWriter(fh, fieldnames=CSV_COLUMNS)
            if new:
                w.writeheader()
            w.writerow(row)
        self._writes += 1
        if self._writes % 2000 == 0:
            self._trim_csv()

    def _trim_csv(self):
        """Keep the log from growing forever — roughly the last 3 weeks."""
        try:
            if os.path.getsize(CSV_PATH) < CSV_MAX_BYTES:
                return
            with open(CSV_PATH, newline="") as fh:
                lines = fh.readlines()
            keep = [lines[0]] + lines[-CSV_KEEP_ROWS:]
            tmp = CSV_PATH + ".tmp"
            with open(tmp, "w", newline="") as fh:
                fh.writelines(keep)
            os.replace(tmp, CSV_PATH)
        except OSError as e:
            print(f"could not trim history.csv: {e}")

    # -- polling ----------------------------------------------------------
    def ask(self, cmd):
        """One-off query for the /api/raw endpoint. Caller holds lock_device."""
        return self.inv.query(cmd) if self.inv else None

    def run(self):
        last_slow = last_daily = 0.0
        while True:
            try:
                with self.lock_device:
                    if self.inv is None:
                        self.inv = open_inverter(self.config.get("bridge_url"),
                                                 self.config.get("bridge_token"))
                        if isinstance(self.inv, NetInverter):
                            try:
                                self._bridge_sync(self.inv)
                            except Exception as e:   # never let backfill block live readings
                                print(f"backfill from ESP32 skipped: {e}", flush=True)
                    self._poll_fast(self.inv)
                    if time.time() - last_slow > SLOW_INTERVAL:
                        self._poll_slow(self.inv)
                        last_slow = time.time()
                    if time.time() - last_daily > DAILY_INTERVAL:
                        self._poll_daily(self.inv)
                        last_daily = time.time()
            except Exception as e:                      # USB unplugged, busy, etc.
                with self.lock:
                    self.state = {"connected": False, "error": str(e)}
                if self.inv is not None:
                    try:
                        self.inv.close()
                    except Exception:
                        pass
                    self.inv = None
                time.sleep(3)
                continue
            time.sleep(FAST_INTERVAL)

    def _poll_slow(self, inv):
        slow = {}
        for cmd, key in [("QID", "serial"), ("QVFW", "firmware"), ("QPI", "protocol"), ("QMN", "model")]:
            slow[key] = inv.query(cmd)
        r = inv.query("QPIRI")
        if r:
            slow["rating"] = parse(QPIRI, r)
        r = inv.query("QBMS")
        if r:
            slow["bms"] = parse(QBMS, r)
        r = inv.query("QFLAG")
        if r:
            slow["flags"] = parse_flags(r)
        for cmd, key in [("QET", "pv_total_wh"), ("QLT", "load_total_wh")]:
            v = inv.query(cmd)
            if v:
                slow[key] = num(v)
        r = inv.query("QT")
        if r and len(r) == 14:
            clock = f"{r[0:4]}-{r[4:6]}-{r[6:8]} {r[8:10]}:{r[10:12]}:{r[12:14]}"
            slow["clock"] = clock
            try:
                t = time.mktime(time.strptime(clock, "%Y-%m-%d %H:%M:%S"))
                slow["clock_drift_seconds"] = round(t - time.time())
            except ValueError:
                pass
        if isinstance(inv, NetInverter):
            try:
                slow["bridge"] = inv.status()
            except Exception:
                pass
        with self.lock:
            self.slow = slow

    def _poll_fast(self, inv):
        raw = inv.query("QPIGS")
        if not raw:
            raise RuntimeError("no answer to QPIGS")
        s = {k: v["value"] for k, v in parse(QPIGS, raw).items()}
        mode = (inv.query("QMOD") or "?").strip()
        warn_raw = inv.query("QPIWS") or ""
        if time.time() - self._last_bms > BMS_INTERVAL:
            self._last_bms = time.time()
            b = inv.query("QBMS")
            if b:
                self._bms_limit = str(int(num(parse(QBMS, b)["bms_max_charge_current"]["value"])))
        self._detect_warnings(warn_raw)

        stamp = time.strftime("%Y-%m-%dT%H:%M:%S")
        row = {
            "timestamp": stamp, "mode": mode,
            "grid_v": s.get("grid_voltage"), "load_w": s.get("ac_out_w"),
            "load_pct": s.get("load_percent"), "batt_v": s.get("battery_voltage"),
            "batt_soc": s.get("battery_capacity"),
            "batt_charge_a": s.get("battery_charge_current"),
            "batt_discharge_a": s.get("battery_discharge_current"),
            "pv_v": s.get("pv_voltage"), "pv_a": s.get("pv_current"),
            "pv_w": s.get("pv_charging_power"), "temp_c": s.get("heatsink_temp"),
            "grid_hz": s.get("grid_frequency"), "out_v": s.get("ac_out_voltage"),
            "load_va": s.get("ac_out_va"), "bms_charge_limit_a": self._bms_limit,
        }
        self._ingest(row)
        self._typical_load = self.grid_report(24).get("load_avg_w")
        with self.lock:
            self.state = self._derive(s, mode, warn_raw, list(self.history))

    def _ingest(self, row):
        self._roll_energy(row)
        self._append_csv(row)
        self._detect_events(num(row["grid_v"]) > 50, row["mode"],
                            num(row["batt_charge_a"]) > 0,
                            num(row["batt_discharge_a"]) > 0, row["timestamp"])
        with self.lock:
            self.history.append({"t": row["timestamp"], "solar_w": num(row["pv_w"]),
                                 "load_w": num(row["load_w"]), "soc": num(row["batt_soc"]),
                                 "batt_v": num(row["batt_v"]), "grid_v": num(row["grid_v"]),
                                 "grid_hz": num(row.get("grid_hz"))})

    # -- backfill from the ESP32's own log ------------------------------------
    def _bridge_sync(self, inv):
        """The ESP32 records every 10 s whether or not this dashboard runs.
        Pull whatever it saw while we were away, so the history has no gaps."""
        st = inv.status()
        if not st.get("time_synced") and inv.token:
            inv.set_time(time.time())     # no internet NTP: lend it our clock
        if not st.get("history_ready"):
            return
        try:
            with open(BRIDGE_STATE_PATH) as fh:
                since = json.load(fh).get("last_seq", 0)
        except (OSError, ValueError):
            since = 0
        if since >= st.get("history_next_seq", 0):
            since = 0                     # its log was wiped; start over

        rows = []
        while True:
            page = inv.history(since, 2000)
            rows += page
            if page:
                since = int(page[-1]["seq"])
            if len(page) < 2000:
                break

        # records made before the ESP32 knew the time borrow it from a later
        # record in the same boot
        offset = {}
        for r in rows:
            if int(r["unix"]):
                offset.setdefault(r["boot"], int(r["unix"]) - int(r["uptime"]))
        rated_va = num((self.slow.get("rating") or {}).get("ac_out_rating_va", {}).get("value"), 6000)
        last_ts, added = self._last_csv_timestamp(), 0
        for r in rows:
            unix = int(r["unix"]) or (offset[r["boot"]] + int(r["uptime"]) if r["boot"] in offset else 0)
            if not unix:
                continue
            ts = time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(unix))
            if ts <= last_ts:
                continue                  # this dashboard already recorded that moment live
            row = {
                "timestamp": ts, "mode": r["mode"], "grid_v": f"{int(r['grid_v10']) / 10:.1f}",
                "load_w": r["load_w"], "load_pct": str(int(r["load_va"]) * 100 // int(rated_va)),
                "batt_v": f"{int(r['batt_v100']) / 100:.2f}", "batt_soc": r["soc"],
                "batt_charge_a": r["charge_a"], "batt_discharge_a": r["discharge_a"],
                "pv_v": f"{int(r['pv_v10']) / 10:.1f}", "pv_a": f"{int(r['pv_a10']) / 10:.1f}",
                "pv_w": r["pv_w"], "temp_c": r["temp_c"], "load_va": r["load_va"],
            }
            if self._prev_event_state is None:   # don't log "monitoring started" mid-gap
                self._prev_event_state = {"grid": num(row["grid_v"]) > 50, "mode": row["mode"],
                                          "charging": num(row["batt_charge_a"]) > 0,
                                          "discharging": num(row["batt_discharge_a"]) > 0}
            self._ingest(row)
            last_ts, added = ts, added + 1

        with open(BRIDGE_STATE_PATH, "w") as fh:
            json.dump({"last_seq": since}, fh)
        if added >= 30:   # 5+ minutes; shorter blips are filled silently
            self._note_event("monitor", f"Filled a gap with {added} readings the ESP32 kept on its own")

    def _last_csv_timestamp(self):
        try:
            with open(CSV_PATH, "rb") as fh:
                fh.seek(0, os.SEEK_END)
                fh.seek(max(0, fh.tell() - 4096))
                lines = fh.read().decode("utf-8", "replace").strip().splitlines()
        except OSError:
            return ""
        last = lines[-1].split(",")[0] if lines else ""
        return "" if last == "timestamp" else last

    # -- turn raw numbers into something a human can read ------------------
    def _derive(self, s, mode, warn_raw, history):
        cfg = self.config
        solar_w = num(s.get("pv_charging_power"))
        load_w = num(s.get("ac_out_w"))
        load_va = num(s.get("ac_out_va"))
        # in line mode the inverter's output meter reads 0/0 most of the time at
        # light loads; carry the last real value through the line-mode stretch
        if mode != "L":
            self._held_load = None
        elif load_va == 0 and self._held_load:
            load_w, load_va = self._held_load
        elif load_va > 0:
            self._held_load = (load_w, load_va)
        grid_v = num(s.get("grid_voltage"))
        batt_v = num(s.get("battery_voltage"))
        soc = num(s.get("battery_capacity"))
        charge_a = num(s.get("battery_charge_current"))
        discharge_a = num(s.get("battery_discharge_current"))
        temp = num(s.get("heatsink_temp"))

        battery_w = round(batt_v * (charge_a - discharge_a))
        grid_on = grid_v > 50
        # what the grid contributes = whatever solar and battery aren't covering
        # the grid only feeds the house in line mode; otherwise this is rounding noise
        grid_w = max(0.0, load_w - solar_w + battery_w) if grid_on and mode == "L" else 0.0

        # in line mode the house runs straight off the grid and solar only charges the
        # battery, so anything the battery takes beyond solar came from the grid
        # battery current is whole amps (~53 W steps), so average over ~1 min
        if mode == "L":
            self._gtb.append(battery_w - solar_w)
        else:
            self._gtb.clear()
        grid_to_batt_w = max(0.0, sum(self._gtb) / len(self._gtb)) if self._gtb else 0.0
        if mode == "L":
            grid_w = load_w + grid_to_batt_w

        if charge_a > 0 and discharge_a == 0:
            batt_dir = "charging"
        elif discharge_a > 0:
            batt_dir = "discharging"
        else:
            batt_dir = "idle"

        # in line mode SBU hands back to the battery at program 13, so that's where grid charging ends
        charge_target = cfg.get("battery_return_percent", 100) if mode == "L" else 100
        runtime_hours = blackout_hours = None
        if cfg.get("battery_kwh") and self._typical_load:
            # how long the battery would carry the house if the grid cut right now
            usable_cut = cfg["battery_kwh"] * max(0.0, soc - cfg.get("battery_cutoff_percent", 10)) / 100
            blackout_hours = usable_cut * 1000 / self._typical_load
        if cfg.get("battery_kwh"):
            # with grid available it hands over at program 12; in a blackout it runs to program 29
            floor = cfg["battery_reserve_percent"] if grid_on else cfg.get("battery_cutoff_percent", 10)
            usable = cfg["battery_kwh"] * max(0.0, soc - floor) / 100
            if batt_dir == "discharging" and abs(battery_w) > 10:
                runtime_hours = usable * 1000 / abs(battery_w)
            elif batt_dir == "charging" and battery_w > 10:
                to_go = cfg["battery_kwh"] * max(0.0, charge_target - soc) / 100
                runtime_hours = -to_go * 1000 / battery_w   # negative = time to charge_target

        rated_va = num((self.slow.get("rating") or {}).get("ac_out_rating_va", {}).get("value"), 6000)
        # everything the inverter draws that never reaches the output: its own
        # electronics plus conversion losses
        overhead = round(solar_w - battery_w + grid_w - load_w) if not grid_on else None

        return {
            "connected": True,
            "updated": time.strftime("%H:%M:%S"),
            "rated_va": rated_va,
            "source": "esp32" if isinstance(self.inv, NetInverter) else "usb",
            "power_factor": round(load_w / load_va, 2) if load_va > 0 else None,
            "overhead_w": overhead,
            "status_flags": (parse_status_bits(s.get("status_bits"), STATUS_BITS)
                             + parse_status_bits(s.get("status_bits2"), STATUS2_BITS)),
            "energy_today": dict(self.energy),
            "daily": dict(self.daily),
            "events": list(self.events)[-40:][::-1],
            "mode": mode,
            "mode_text": "On grid" if mode == "L" else MODES.get(mode, mode),
            "solar_w": solar_w,
            "load_w": load_w,
            "load_va": load_va,
            "load_percent": num(s.get("load_percent")) or int(load_va / rated_va * 100),
            "grid_v": grid_v,
            "grid_on": grid_on,
            "grid_w": round(grid_w),
            "grid_to_batt_w": round(grid_to_batt_w),
            "battery_v": batt_v,
            "battery_soc": soc,
            "battery_w": battery_w,
            "battery_direction": batt_dir,
            "battery_charge_a": charge_a,
            "battery_discharge_a": discharge_a,
            "pv_v": num(s.get("pv_voltage")),
            "pv_a": num(s.get("pv_current")),
            "temp_c": temp,
            "runtime_hours": runtime_hours,
            "charge_target": charge_target,
            "blackout_hours": blackout_hours,
            "typical_load_w": round(self._typical_load) if self._typical_load else None,
            "warnings": parse_warnings(warn_raw),
            "raw": s,
            "history": history[-360:],
            "slow": self.slow,
            "config": {k: v for k, v in cfg.items() if k != "bridge_token"},
        }

    def snapshot(self):
        with self.lock:
            return dict(self.state)


class Handler(BaseHTTPRequestHandler):
    poller = None

    def log_message(self, *a):
        pass

    def _send(self, code, body, ctype):
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path.startswith("/api/raw"):
            from urllib.parse import parse_qs, urlparse
            cmd = (parse_qs(urlparse(self.path).query).get("cmd") or [""])[0].upper()
            # queries only — a P-command could change an inverter setting
            if not cmd.isalnum() or not cmd.startswith("Q"):
                self._send(400, b'{"error":"only Q... query commands allowed"}',
                           "application/json")
                return
            with self.poller.lock_device:
                reply = self.poller.ask(cmd)
            self._send(200, json.dumps({"cmd": cmd, "reply": reply}).encode(),
                       "application/json")
        elif self.path.startswith("/api/grid"):
            from urllib.parse import parse_qs, urlparse
            try:
                hours = int((parse_qs(urlparse(self.path).query).get("hours") or ["24"])[0])
            except ValueError:
                hours = 24
            self._send(200, json.dumps(self.poller.grid_report(hours)).encode(), "application/json")
        elif self.path.startswith("/api/state"):
            body = json.dumps(self.poller.snapshot()).encode()
            self._send(200, body, "application/json")
        elif self.path in ("/", "/index.html"):
            try:
                with open(os.path.join(HERE, "dashboard.html"), "rb") as fh:
                    self._send(200, fh.read(), "text/html; charset=utf-8")
            except OSError:
                self._send(500, b"dashboard.html missing", "text/plain")
        else:
            self._send(404, b"not found", "text/plain")


def main():
    cfg = load_config()
    poller = Poller(cfg)
    poller.start()
    Handler.poller = poller
    port = cfg.get("port", 8765)
    print(f"Inverter dashboard: http://localhost:{port}")
    ThreadingHTTPServer(("127.0.0.1", port), Handler).serve_forever()


if __name__ == "__main__":
    main()