diff --git a/README.md b/README.md index 9d1efa8..0fa730d 100644 --- a/README.md +++ b/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 --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 diff --git a/scripts/agate_battery.sh b/scripts/agate_battery.sh new file mode 100755 index 0000000..cdb846e --- /dev/null +++ b/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 --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 < device serial (default: ${DEVICE_SN}) + --ip aGate address (default: ${DEVICE_IP}) + --port local socket port (default: ${DEVICE_PORT}) + --timeout 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