adds home battery query

README.md

@@ -65,6 +65,22 @@ Probes `10.100.1.1:9000,18000`, sends `0x19/0x1D/0x1E`, prints
 `currentBatterySoc → soc → avg(socArray)`. Must run where the Wi-Fi adapter
 is on the device AP — see [Battery PoC](docs/BATTERY_POC.md).
 
+## Battery report (pretty terminal output)
+
+`scripts/agate_battery.sh` bakes in the device (serial + LAN address),
+queries the aGate's local socket and renders the state of charge, health,
+cell voltages/temperatures and the live power flow as a colourised report:
+
+```bash
+./scripts/agate_battery.sh                 # baked-in device
+./scripts/agate_battery.sh --sn <SN> --ip <IP>
+./scripts/agate_battery.sh --json          # raw device replies
+./scripts/agate_battery.sh --no-color      # plain text
+```
+
+Env overrides: `AGATE_SN`, `AGATE_IP`, `AGATE_PORT`, `AGATE_TIMEOUT`,
+`NO_COLOR`. Colour is auto-disabled when stdout is not a TTY.
+
 ## Local connect (device hotspot)
 
 The aGate runs its own SoftAP (gateway `10.100.1.1`); joining it is what

scripts/agate_battery.sh

@@ -0,0 +1,411 @@
+#!/usr/bin/env bash
+#
+# agate_battery.sh -- FranklinWH AGATE battery state, pretty-printed.
+#
+# Bakes in the device under test (serial number + LAN address) and runs the
+# library's direct-connect battery read over the aGate's local TCP socket
+# (the encrypted CommunicationObject envelope -- no cloud, no BLE), then
+# renders the answer as a colourised terminal report.
+#
+#   ./scripts/agate_battery.sh                    # baked-in device
+#   ./scripts/agate_battery.sh --sn <SN> --ip <IP>
+#   ./scripts/agate_battery.sh --json             # raw replies as JSON
+#   ./scripts/agate_battery.sh --no-color         # plain text
+#
+# Env overrides: AGATE_SN, AGATE_IP, AGATE_PORT, AGATE_TIMEOUT, NO_COLOR.
+#
+set -euo pipefail
+
+# ---------------------------------------------------------------- config ---
+DEVICE_SN="${AGATE_SN:-A03X2516013210060005}"
+DEVICE_IP="${AGATE_IP:-192.168.1.142}"
+DEVICE_PORT="${AGATE_PORT:-9000}"
+TIMEOUT="${AGATE_TIMEOUT:-8}"
+JSON_OUT=0
+
+SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd -- "${SCRIPT_DIR}/.." && pwd)"
+export PYTHONPATH="${REPO_ROOT}${PYTHONPATH:+:${PYTHONPATH}}"
+
+usage() {
+    cat <<EOF
+usage: ${0##*/} [options]
+
+  --sn <serial>   device serial (default: ${DEVICE_SN})
+  --ip <addr>     aGate address (default: ${DEVICE_IP})
+  --port <n>      local socket port (default: ${DEVICE_PORT})
+  --timeout <s>   per-request timeout (default: ${TIMEOUT})
+  --json          dump the raw device replies as JSON
+  --no-color      disable ANSI colour
+  -h, --help      show this help
+EOF
+}
+
+while [[ $# -gt 0 ]]; do
+    case "$1" in
+        --sn)      DEVICE_SN="$2"; shift 2 ;;
+        --ip)      DEVICE_IP="$2"; shift 2 ;;
+        --port)    DEVICE_PORT="$2"; shift 2 ;;
+        --timeout) TIMEOUT="$2"; shift 2 ;;
+        --json)    JSON_OUT=1; shift ;;
+        --no-color) NO_COLOR=1; shift ;;
+        -h|--help) usage; exit 0 ;;
+        *) echo "unknown option: $1" >&2; usage >&2; exit 2 ;;
+    esac
+done
+
+# --------------------------------------------------------------- colour ---
+if [[ -t 1 && -z "${NO_COLOR:-}" && "${TERM:-dumb}" != "dumb" ]]; then
+    AGATE_COLOR=1
+else
+    AGATE_COLOR=0
+fi
+export AGATE_COLOR
+
+# ------------------------------------------------------------- preflight ---
+if ! command -v python3 >/dev/null 2>&1; then
+    printf '\033[31m✗ python3 not found\033[0m\n' >&2
+    exit 1
+fi
+
+if ! python3 -c 'import requests' >/dev/null 2>&1; then
+    printf '\033[31m✗ the requests package is missing (pip install requests)\033[0m\n' >&2
+    exit 1
+fi
+
+if command -v timeout >/dev/null 2>&1; then
+    if ! timeout "${TIMEOUT}" bash -c "exec 3<>/dev/tcp/${DEVICE_IP}/${DEVICE_PORT}" 2>/dev/null; then
+        printf '\033[31m✗ %s:%s is unreachable\033[0m\n' "${DEVICE_IP}" "${DEVICE_PORT}" >&2
+        printf '  the aGate local socket only answers on its own network\n' >&2
+        exit 1
+    fi
+fi
+
+# ------------------------------------------------------------------ query ---
+export AGATE_SN="$DEVICE_SN" DEVICE_IP DEVICE_PORT TIMEOUT JSON_OUT
+
+python3 - <<'PY'
+"""Query the aGate and render a colourised terminal report."""
+import json
+import os
+import re
+import socket
+import sys
+import time
+
+from franklinwh_agate_client.device_local import LocalDeviceClient
+from franklinwh_agate_client.protocol import CommunicationCmd
+
+SN = os.environ["AGATE_SN"]
+HOST = os.environ["DEVICE_IP"]
+PORT = int(os.environ["DEVICE_PORT"])
+TIMEOUT = float(os.environ["TIMEOUT"])
+JSON_OUT = os.environ.get("JSON_OUT") == "1"
+COLOR = os.environ.get("AGATE_COLOR") == "1"
+
+WIDTH = 74
+#: label column width inside the boxes
+LABELS = 13
+#: usable width for a value that follows a label (inner width minus the
+#: leading space of the box line and the label column)
+VALUE_WIDTH = WIDTH - 2 - 1 - LABELS
+#: usable width for a full-width box line
+LINE_WIDTH = WIDTH - 2 - 1
+
+# ------------------------------------------------------------------ colour ---
+_C = {
+    "off": "\033[0m", "bold": "\033[1m", "dim": "\033[2m",
+    "red": "\033[31m", "green": "\033[32m", "yellow": "\033[33m",
+    "blue": "\033[34m", "magenta": "\033[35m", "cyan": "\033[36m",
+    "grey": "\033[90m", "bgreen": "\033[92m", "bcyan": "\033[96m",
+    "byellow": "\033[93m", "bred": "\033[91m",
+}
+
+
+def col(name, text):
+    if not COLOR:
+        return text
+    return f"{_C[name]}{text}{_C['off']}"
+
+
+_ANSI = re.compile(r"\x1b\[[0-9;]*m")
+
+
+def pad(text, width, align="left"):
+    """Pad to a visible width, ignoring ANSI escapes."""
+    visible = len(_ANSI.sub("", text))
+    gap = width - visible
+    if gap <= 0:
+        return text
+    return (" " * gap + text) if align == "right" else (text + " " * gap)
+
+
+def kv(label, value, width=13):
+    return col("grey", label.ljust(width)) + value
+
+
+def box_top(title=""):
+    inner = WIDTH - 2
+    if title:
+        head = f" {title} "
+        line = "─" + head + "─" * max(0, inner - 1 - len(head))
+    else:
+        line = "─" * inner
+    return col("cyan", "╭") + col("dim", line) + col("cyan", "╮")
+
+
+def box_bottom():
+    return col("cyan", "╰") + col("dim", "─" * (WIDTH - 2)) + col("cyan", "╯")
+
+
+def box_line(content=""):
+    return (col("cyan", "│") + pad(" " + content, WIDTH - 2)
+            + col("cyan", "│"))
+
+
+def blank():
+    return box_line()
+
+
+SPARK = "▁▂▃▄▅▆▇█"
+
+
+def sparkline(values):
+    """Unicode bar chart of a numeric series."""
+    vals = [float(v) for v in values]
+    if not vals:
+        return ""
+    lo, hi = min(vals), max(vals)
+    if hi - lo < 1e-9:
+        return SPARK[3] * len(vals)
+    return "".join(
+        SPARK[int(round((v - lo) / (hi - lo) * (len(SPARK) - 1)))] for v in vals)
+
+
+def soc_bar(pct, width=26):
+    """Charge bar, coloured by level (green >= 50, yellow >= 20, red below)."""
+    pct = max(0.0, min(100.0, pct))
+    filled = int(round(width * pct / 100.0))
+    tone = "bgreen" if pct >= 50 else ("byellow" if pct >= 20 else "bred")
+    return col(tone, "█" * filled) + col("grey", "░" * (width - filled))
+
+
+def num(value, unit="", digits=1):
+    if value is None:
+        return col("grey", "n/a")
+    return f"{value:,.{digits}f} {unit}" if unit else f"{value:,.{digits}f}"
+
+
+def signed(value, unit="", digits=1):
+    if value is None:
+        return col("grey", "n/a")
+    text = f"{value:+,.{digits}f}"
+    return f"{text} {unit}" if unit else text
+
+
+def pack(items, budget, sep=" | "):
+    """Greedily pack short strings into lines no wider than *budget*."""
+    lines, current = [], ""
+    for item in items:
+        candidate = item if not current else f"{current}{sep}{item}"
+        if len(_ANSI.sub("", candidate)) <= budget or not current:
+            current = candidate
+        else:
+            lines.append(current)
+            current = item
+    if current:
+        lines.append(current)
+    return lines
+
+
+# ------------------------------------------------------------------ query ---
+client = LocalDeviceClient(sn=SN, host=HOST, socket_port=PORT, timeout=TIMEOUT)
+
+raw = {}
+
+
+def rpc(name, cmd, data=None, equip_no=None):
+    """Send one envelope; never raise -- a failed read is just empty."""
+    try:
+        reply = client.send_rpc(cmd, data=data, equip_no=equip_no)
+    except Exception as exc:  # noqa: BLE001 -- report, do not explode
+        raw[name] = {"error": f"{type(exc).__name__}: {exc}"}
+        return {}
+    raw[name] = reply
+    return reply.get("dataArea", {}) if isinstance(reply, dict) else {}
+
+
+# best-effort discovery probe (mirrors the app's direct-connect open)
+try:
+    client.send_rpc(CommunicationCmd.QUERY_DEVICE_INFO, equip_no="",
+                    data={"minProtocolVer": "V1.00.00"})
+except Exception:
+    pass
+
+dev = rpc("device_info", CommunicationCmd.QUERY_DEVICE_INFO,
+          {"minProtocolVer": "V1.00.00"})
+detail = rpc("detail", CommunicationCmd.DEVICE_DETAIL_INFO)
+apower = rpc("apower", CommunicationCmd.DEVICE_APOWER_INFO, {"update": 2})
+
+# pack ids come from the BMS_SN suffixes (…B1 / …B2)
+bms_sns = [str(s) for s in (dev.get("BMS_SN") or [])]
+pack_ids = []
+for sn in bms_sns:
+    match = re.search(r"([0-9]+)\s*$", sn)
+    pack_ids.append(int(match.group(1)) if match else len(pack_ids) + 1)
+if not pack_ids:
+    pack_ids = [1, 2]
+
+packs = []
+for pid in pack_ids:
+    data = rpc(f"bms{pid}", CommunicationCmd.BMS, {"id": pid})
+    if not data or data.get("result") != 0:
+        continue
+    packs.append({
+        "id": pid,
+        "sn": bms_sns[pack_ids.index(pid)] if pid in bms_sns else "",
+        "soc": data.get("batSoc"),
+        "soh": data.get("batSoh"),
+        "volt": data.get("batTotalVolt"),
+        "curr": data.get("currGrp"),
+        "alarm": data.get("alarmLevel"),
+        "temps": data.get("batTemp") or [],
+        "volts": data.get("batVolt") or [],
+    })
+
+if JSON_OUT:
+    print(json.dumps(raw, indent=2, default=str))
+    sys.exit(0)
+
+if not packs and not detail:
+    print(col("bred", "✗ the device did not answer any battery query"))
+    sys.exit(2)
+
+# ------------------------------------------------------------------ render ---
+print()
+print(box_top("FRANKLINWH AGATE"))
+print(box_line())
+print(box_line(kv("serial", col("bold", SN))))
+print(box_line(kv("address", f"{HOST}:{PORT}  "
+                   + col("grey", "(local aGate socket)"))))
+if dev.get("protocolVer"):
+    print(box_line(kv("protocol", dev["protocolVer"])))
+fhp = [str(s) for s in (dev.get("FHP_SN") or [])]
+if fhp:
+    for i, s in enumerate(fhp):
+        print(box_line(kv("FHP unit %d" % (i + 1), s)))
+mode = detail.get("name") or detail.get("mode")
+if mode:
+    print(box_line(kv("mode", f"{mode}"
+                      + (col("grey", f"  (status {detail['run_status']})")
+                         if detail.get("run_status") is not None else ""))))
+print(box_line())
+print(box_bottom())
+print()
+
+# -- battery --------------------------------------------------------------
+print(box_top("BATTERY"))
+print(box_line())
+
+pack_socs = [p["soc"] for p in packs if p.get("soc") is not None]
+overall = sum(pack_socs) / len(pack_socs) if pack_socs else None
+gw_soc = detail.get("soc")
+
+if overall is not None:
+    bar = soc_bar(overall)
+    print(box_line(kv("overall", bar + "  " + col("bold", f"{overall:.1f} %"))))
+    if gw_soc is not None and abs(gw_soc - overall) >= 0.5:
+        print(box_line(kv("", col("grey",
+                                 f"gateway aggregate reads {gw_soc:.1f} %"))))
+    print(box_line())
+
+for p in packs:
+    soc = p["soc"]
+    label = f"Pack {p['id']}"
+    if p["sn"]:
+        label += col("grey", f"  {p['sn'][-3:]}")
+    print(box_line(kv(label, (soc_bar(soc) + "  " + col("bold", f"{soc:.1f} %")
+                              if soc is not None
+                              else col("grey", "no data")))))
+    meta = []
+    if p["soh"] is not None:
+        meta.append(col("grey", "SoH ") + col("bcyan", f"{p['soh']:.1f} %"))
+    if p["volt"] is not None:
+        meta.append(col("grey", "voltage ")
+                    + col("bcyan", f"{p['volt']:.2f} V"))
+    if p["curr"] is not None:
+        flow = "discharging" if p["curr"] < 0 else "charging"
+        meta.append(col("grey", "current ")
+                    + col("byellow" if p["curr"] < 0 else "bgreen",
+                          f"{signed(p['curr'], 'A')} {flow}"))
+    if p["temps"]:
+        meta.append(col("grey", "temp ")
+                    + col("bcyan", f"{min(p['temps']):.1f}–"
+                                   f"{max(p['temps']):.1f} °C"))
+    alarm = p["alarm"]
+    meta.append(col("grey", "alarms ")
+                + (col("bgreen", "none") if not alarm
+                   else col("bred", f"level {alarm}")))
+    for line in pack(meta, VALUE_WIDTH):
+        print(box_line(kv("", line)))
+    if p["volts"]:
+        print(box_line(kv("", col("grey", "cells")
+                          + "  " + col("bcyan", sparkline(p["volts"]))
+                          + col("grey", f"  {min(p['volts']):.0f}–"
+                                f"{max(p['volts']):.0f} mV"
+                                f"  ({len(p['volts'])})"))))
+    if p["temps"]:
+        print(box_line(kv("", col("grey", "temps ")
+                          + "  " + col("bcyan", sparkline(p["temps"]))
+                          + col("grey", "  °C"))))
+    print(box_line())
+
+if not packs:
+    print(box_line(col("grey", "no BMS pack answered")))
+    print(box_line())
+
+print(box_bottom())
+print()
+
+# -- power ----------------------------------------------------------------
+if detail:
+    print(box_top("POWER"))
+    print(box_line())
+    flows = (
+        ("load", detail.get("p_load"), "W", "byellow"),
+        ("FHP", detail.get("p_fhp"), "W", "bcyan"),
+        ("solar", detail.get("p_sun"), "W", "byellow"),
+        ("generator", detail.get("p_gen"), "W", "bgreen"),
+    )
+    row = "   ".join(
+        col("grey", f"{name} ") + col(tone, num(val, unit))
+        for name, val, unit, tone in flows if val is not None)
+    for line in pack([row], LINE_WIDTH, sep="   "):
+        print(box_line(line))
+    energy = (
+        ("load", detail.get("kwh_load")),
+        ("solar", detail.get("kwh_sun")),
+        ("FHP charge", detail.get("kwh_fhp_chg")),
+        ("FHP discharge", detail.get("kwh_fhp_di")),
+    )
+    items = [col("grey", f"{name} ") + col("bcyan", f"{val:.2f} kWh")
+             for name, val in energy if val is not None]
+    if items:
+        print(box_line())
+        for line in pack(items, LINE_WIDTH, sep="   "):
+            print(box_line(line))
+    print(box_line())
+    print(box_bottom())
+    print()
+
+stamp = (raw.get("detail") or {}).get("timeStamp")
+stamp_txt = ""
+if stamp:
+    try:
+        stamp_txt = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(stamp))
+    except (ValueError, OSError, OverflowError):
+        stamp_txt = str(stamp)
+print(col("grey", "  device clock " + (stamp_txt or "unknown")
+          + "   local " + time.strftime("%Y-%m-%d %H:%M:%S")))
+print()
+PY