franklinwh-agate-client/scripts/agate_battery.sh

#!/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