adds home battery query

AgateClient takes address= default; attach_local accepts
address/host/gateway_ip per call with join control. An explicit
address skips the SoftAP radio join by default (join=True forces
it). derive_hotspot_credentials and join_device_hotspot accept a
gateway_ip override. battery/power host defaults now use
DEVICE_GATEWAY_IP. Verified live at 192.168.1.142:9000.

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

docs/RE_DART_FINDINGS.md

@@ -396,6 +396,59 @@ Library: `power.get_power()` tries detail → load → EMS in order
 (`franklinwh_agate_client/power.py`); `LocalDeviceClient` gained
 `query_detail_info()` / `query_load_status()`.
 
+## 17. Cloud account binding (installer flow)
+
+Binding a unit to an account is an *installer commissioning* action, not
+a local-socket one (`sections/installer/bind_device/`):
+
+- `BindDeviceProvider.bind()` POSTs a data map to
+  **`RemoteInterface.deviceBind` → `POST {cloud}/manage/deviceBindV2`**,
+  with `protocolVer: "V1.10.01"`, the installer `jobCode`, the
+  `commissionId` (preset-site path via `InstallContext`), and the
+  `BindDeviceEntity` payload. Result parses as
+  `HttpResultEntity<BindDeviceEntity>` (bound `account`, distributor /
+  financier ids/names).
+- Mirror calls: `queryDeviceBindData` →
+  `GET {cloud}​/manage/getGatewayBindData` (pre-check: is this gateway
+  already bound, to whom) and `deviceUnbind` →
+  `POST {cloud}​/manage/unBindDevice` ("Are you sure you want to
+  unbind?", `unbindSuccess` strings).
+- Request auth on cloud calls is a signature, not a session cookie:
+  `RemoteService` computes **`MD5(jsonEncode(body) + "B6F3Zu&aW8XlH9CA")`**
+  and sends it as the **`token`** header (dio `Options`). User login
+  itself is Okta (`sunrun.okta.com`, MFA endpoints); the known cloud
+  bases are `https://e-api.franklinwh.com/api/appapi` and
+  `https://energy.franklinwh.com/*` (exact `/manage` base mapping lives
+  in the `RemoteService` dio init).
+- Locally, bound devices land in the trusted-devices table
+  (`DatabaseManager.getTrustedDevicesTableList`, consulted by
+  `_checkConnectedNotTrustedAp` on direct-connect).
+
+## 18. Joining the aGate to home Wi-Fi (STA onboarding)
+
+Three transports can deliver the credentials; all verified in
+disassembly, socket path live-tested for framing (not the join itself):
+
+- **Socket** (while on the SoftAP): `queryConnectWifi` →
+  `CommunicationService.connectWifi(entity, …)` → `_request` with cmd
+  `connectWifi` (socket `1111`), operate **SET** (`type: 1`), arguments =
+  `ConnectWifiEntity.toJson()`: `ssid`, `password`, `wifi_SSID`,
+  `wifi_Pw` (plus an unencrypted-network flag). Library:
+  `LocalDeviceClient.join_home_wifi(ssid, password)` (+ `opt: 1`
+  seeding via the shared factory path), `scan_home_wifi()`
+  (`scanWifiList`, socket `1109`, no args) to list visible networks,
+  `query_connection_status()` (`connectionStatus`, socket `1113`) to
+  confirm association.
+- **BLE**: ESP protocomm (`prov-session`, `prov-scan`, `prov-config`,
+  `custom-data`, sec0/sec1) — the standard Espressif flow; `prov-config`
+  carries the SSID/password. Used when the phone is not on the SoftAP.
+- **Phone-side**: `wifi_iot` plugin join/leave is the *phone* joining the
+  *device* AP (`wifi.join_device_hotspot`) — the opposite direction.
+
+.. warning:: `join_home_wifi` is state-changing: on success the unit
+   leaves its SoftAP (or roams), dropping the current session. Only run
+   it with physical/rejoin access to the unit.
+
 ## 9. Bluetooth (BLE) transport — `communication_center/bluetooth`
 
 - `BluetoothManager` (singleton), `BluetoothStatus`: `closed / connected / connecting / interrupted`.

franklinwh_agate_client/battery.py

@@ -20,7 +20,7 @@ from typing import Any, Dict, List, Optional, Tuple
 
 from .device_local import LocalDeviceClient
 from .protocol import CommunicationCmd
-from .wifi import derive_hotspot_credentials
+from .wifi import DEVICE_GATEWAY_IP, derive_hotspot_credentials
 
 log = logging.getLogger(__name__)
 
@@ -144,7 +144,7 @@ def _is_empty_pack(reply: Dict[str, Any], result: Dict[str, Any]) -> bool:
 
 def get_battery(
     sn: str,
-    host: str = "10.100.1.1",
+    host: str = DEVICE_GATEWAY_IP,
     ports: Tuple[int, ...] = SOCKET_PORT_CANDIDATES,
     key_radix: int = 16,
     timeout: float = 5.0,
@@ -197,7 +197,7 @@ PROBE_VARIANTS: Tuple[Tuple[str, CommunicationCmd, Any], ...] = (
 
 def probe_device(
     sn: str,
-    host: str = "10.100.1.1",
+    host: str = DEVICE_GATEWAY_IP,
     ports: Tuple[int, ...] = SOCKET_PORT_CANDIDATES,
     key_radix: int = 16,
     timeout: float = 5.0,
@@ -262,7 +262,7 @@ _MQTT_CONNECT = bytes.fromhex("101200044d5154540402003c000670726f62653031")
 
 def probe_transport(
     sn: str,
-    host: str = "10.100.1.1",
+    host: str = DEVICE_GATEWAY_IP,
     ports: Tuple[int, ...] = (9000,),
     key_radix: int = 16,
     timeout: float = 5.0,
@@ -327,7 +327,7 @@ def main(argv: Optional[List[str]] = None) -> int:
         description="Read aGate battery percentage over the device SoftAP socket."
     )
     parser.add_argument("--sn", required=True, help="20-char device serial")
-    parser.add_argument("--host", default="10.100.1.1", help="aGate gateway IP")
+    parser.add_argument("--host", default=DEVICE_GATEWAY_IP, help="aGate gateway IP")
     parser.add_argument(
         "--ports", default=",".join(map(str, SOCKET_PORT_CANDIDATES)),
         help="comma-separated socket ports to probe",

franklinwh_agate_client/client.py

@@ -7,22 +7,39 @@ from __future__ import annotations
 from typing import Any, Optional
 
 from .bluetooth import BluetoothManager
-from .wifi import HotspotProfile, WifiBackend, WifiManager
+from .wifi import DEVICE_GATEWAY_IP, HotspotProfile, WifiBackend, WifiManager
 
 
 class AgateClient:
-    """Main client for FranklinWH AGATE device operations."""
+    """Main client for FranklinWH AGATE device operations.
+
+    :param device_id: optional default device identifier.
+    :param wifi_backend: platform backend used for hotspot joins.
+    :param address: optional default aGate address (IP or hostname).
+        Overrides the ``10.100.1.1`` SoftAP default for direct-connect
+        calls — e.g. the home-LAN address (``192.168.1.x``) once the unit
+        is on the router.  Per-call ``address``/``host``/``gateway_ip``
+        arguments take precedence over this default.
+    """
 
     def __init__(self, device_id: Optional[str] = None,
-                 wifi_backend: Optional[WifiBackend] = None) -> None:
+                 wifi_backend: Optional[WifiBackend] = None,
+                 address: Optional[str] = None) -> None:
         self.device_id = device_id
         self.bluetooth_manager: Optional[BluetoothManager] = None
         self.wifi_manager = WifiManager(backend=wifi_backend)
+        self.address = address
 
     # -- device hotspot (local connect) -------------------------------------
-    def join_hotspot(self, sn: str, **kwargs: Any) -> HotspotProfile:
+    def join_hotspot(self, sn: str, gateway_ip: Optional[str] = None,
-        """Join the device SoftAP (derives creds, joins, verifies)."""
+                     **kwargs: Any) -> HotspotProfile:
-        return self.wifi_manager.join_device_hotspot(sn, **kwargs)
+        """Join the device SoftAP (derives creds, joins, verifies).
+
+        ``gateway_ip`` overrides the address recorded on the profile
+        without changing the radio join.
+        """
+        return self.wifi_manager.join_device_hotspot(
+            sn, gateway_ip=gateway_ip or self.address, **kwargs)
 
     def leave_hotspot(self) -> None:
         """Disconnect from the device hotspot."""
@@ -30,9 +47,23 @@ class AgateClient:
 
     def attach_local(self, sn: str, *, verify_join: bool = True,
                      settle_timeout: float = 30.0, poll_interval: float = 1.0,
+                     address: Optional[str] = None,
+                     host: Optional[str] = None,
+                     gateway_ip: Optional[str] = None,
+                     join: Optional[bool] = None,
                      **local_kwargs: Any):
-        """Join the device hotspot, then return a ``LocalDeviceClient``
+        """Join the device hotspot, then return a ``LocalDeviceClient``.
-        aimed at the device gateway (``https://10.100.1.1:18000`` by default).
+
+        Without an explicit address this keeps the existing behaviour:
+        join the SoftAP and aim at the profile gateway
+        (``https://10.100.1.1:18000`` by default).
+
+        Pass ``address`` (``host``/``gateway_ip`` are aliases) to reach the
+        aGate elsewhere — e.g. its home-LAN IP once it is on the router.
+        An explicit address defaults to *skipping* the hotspot radio join
+        (pass ``join=True`` to force it); ``join=False`` never touches the
+        radio.  The client-level ``address`` default applies when no
+        per-call address is given.
 
         ``verify_join`` gates the association check (TLS verification of the
         device API is a separate concern — pass ``verify=`` in
@@ -45,13 +76,21 @@ class AgateClient:
         (``port``, ``scheme``, ``verify``, ``socket_port``, ``key_radix``,
         ``timeout``, ``extra_headers``, ``session``).
         """
-        profile = self.wifi_manager.join_device_hotspot(
+        explicit = address if address is not None else (
-            sn, verify=verify_join, settle_timeout=settle_timeout,
+            host if host is not None else gateway_ip)
-            poll_interval=poll_interval,
+        target = explicit if explicit is not None else self.address
-        )
+        do_join = (target is None) if join is None else join
+        if do_join:
+            profile = self.wifi_manager.join_device_hotspot(
+                sn, verify=verify_join, settle_timeout=settle_timeout,
+                poll_interval=poll_interval,
+                gateway_ip=target,
+            )
+            target = profile.gateway_ip
         from .device_local import LocalDeviceClient
 
-        return LocalDeviceClient(sn=sn, host=profile.gateway_ip, **local_kwargs)
+        return LocalDeviceClient(sn=sn, host=target or DEVICE_GATEWAY_IP,
+                                 **local_kwargs)
 
     # -- cloud / stub operations (pending RE) -------------------------------
     def query_agate_info(self, device_id: Optional[str] = None):

franklinwh_agate_client/device_local.py

@@ -190,7 +190,7 @@ class LocalDeviceClient:
         if envelope is None:
             envelope = direct_connect_request(
                 cmd, self.sn if equip_no is None else equip_no, data=data,
-                snno=self._sn_seq.next(),
+                snno=self._sn_seq.next(), operate=operate,
             )
         return self.send_envelope(envelope)
 
@@ -212,6 +212,7 @@ class LocalDeviceClient:
         data: Optional[Dict[str, Any]] = None,
         timeout: Optional[float] = None,
         equip_no: Optional[str] = None,
+        operate: CommunicationOperate = CommunicationOperate.QUERY,
     ) -> Dict[str, Any]:
         """Send one envelope, return exact TX/RX bytes plus any parsed reply.
 
@@ -223,7 +224,7 @@ class LocalDeviceClient:
         """
         envelope = direct_connect_request(
             cmd, self.sn if equip_no is None else equip_no, data=data,
-            snno=self._sn_seq.next(),
+            snno=self._sn_seq.next(), operate=operate,
         )
         wire = self.codec.encode(envelope)
         reply_raw = self._socket_roundtrip(
@@ -354,6 +355,34 @@ class LocalDeviceClient:
     def query_modes_v2(self) -> Dict[str, Any]:
         return self.send_rpc(CommunicationCmd.MODE_V2)
 
+    def join_home_wifi(self, ssid: str, password: str = "") -> Dict[str, Any]:
+        """``connectWifi`` — join the aGate to a home Wi-Fi network.
+
+        Mirrors ``WifiOnboardingConnectProvider``: sends a
+        ``ConnectWifiEntity`` shaped payload (``ssid``/``password`` plus
+        the ``wifi_SSID``/``wifi_Pw`` mirrors) with operate SET over the
+        current transport.  Follow with :meth:`query_connection_status`
+        to confirm association.
+
+        .. warning:: State-changing: on success the unit leaves its
+           SoftAP (or roams), so the current socket/HTTP session drops.
+           Only run when you can rejoin the device network afterwards.
+        """
+        return self.send_rpc(
+            CommunicationCmd.CONNECT_WIFI,
+            operate=CommunicationOperate.SET,
+            data={"ssid": ssid, "password": password,
+                  "wifi_SSID": ssid, "wifi_Pw": password},
+        )
+
+    def scan_home_wifi(self) -> Dict[str, Any]:
+        """``scanWifiList`` — networks visible to the aGate (socket 1109)."""
+        return self.send_rpc(CommunicationCmd.SCAN_WIFI_LIST)
+
+    def query_connection_status(self) -> Dict[str, Any]:
+        """``connectionStatus`` — current STA association state."""
+        return self.send_rpc(CommunicationCmd.CONNECTION_STATUS)
+
     # --------------------------------------------------- business operations
     # Intention-revealing reads: each runs the app's query sequence for one
     # question and returns ``{"value", "source", "detail", "reply", ...}``.

franklinwh_agate_client/power.py

@@ -17,10 +17,11 @@ from typing import Any, Dict, List, Optional, Tuple
 
 from .battery import _walk
 from .device_local import LocalDeviceClient
+from .wifi import DEVICE_GATEWAY_IP
 
 log = logging.getLogger(__name__)
 
-HOST = "10.100.1.1"
+HOST = DEVICE_GATEWAY_IP
 PORTS: Tuple[int, ...] = (9000, 18000)
 
 #: Reply keys holding watt readings, in priority order.

franklinwh_agate_client/protocol/envelope.py

@@ -175,6 +175,7 @@ def direct_connect_request(
     equip_no: str,
     data: Optional[Dict[str, Any]] = None,
     snno: int = 0,
+    operate: CommunicationOperate = CommunicationOperate.QUERY,
 ) -> CommunicationObject:
     """Build a *direct-connect* query envelope.
 
@@ -190,13 +191,13 @@ def direct_connect_request(
     ``sendObject`` (see :class:`SequenceNumber`); pass the issued value
     when re-serialising.
     """
-    payload: Dict[str, Any] = {"opt": int(CommunicationOperate.QUERY)}
+    payload: Dict[str, Any] = {"opt": int(operate)}
     if data:
         payload.update(data)
     return CommunicationObject(
         cmd_type=socket_cmd_type(cmd),
         equip_no=equip_no,
-        op_type=CommunicationOperate.QUERY,
+        op_type=int(operate),
         snno=snno,
         data=payload,
     )

franklinwh_agate_client/wifi.py

@@ -65,7 +65,8 @@ class HotspotJoinError(RuntimeError):
     """A join was issued but the association never verified in time."""
 
 
-def derive_hotspot_credentials(sn: str) -> "HotspotProfile":
+def derive_hotspot_credentials(sn: str,
+                               gateway_ip: Optional[str] = None) -> "HotspotProfile":
     """``getDeviceApSsid`` / ``getDeviceApDefaultPwd``.
 
     Offline fallback: for a *20-character* SN the SSID is ``"AP_" + sn[11:]``
@@ -74,6 +75,9 @@ def derive_hotspot_credentials(sn: str) -> "HotspotProfile":
 
     The authoritative values come from the live device
     (``/terminal/getHotSpotInfo`` + ``/terminal/getHosSpotPassword``).
+
+    ``gateway_ip`` overrides the default SoftAP gateway (``10.100.1.1``) —
+    e.g. when the aGate is reached on its home-LAN address instead.
     """
     if len(sn) == VALID_SN_LENGTH:
         ssid = "AP_" + sn[11:]
@@ -83,7 +87,8 @@ def derive_hotspot_credentials(sn: str) -> "HotspotProfile":
                     len(sn), VALID_SN_LENGTH)
         ssid = "AP_" + sn
         password = FALLBACK_PASSWORD
-    return HotspotProfile(ssid=ssid, password=password)
+    return HotspotProfile(ssid=ssid, password=password,
+                          gateway_ip=gateway_ip or DEVICE_GATEWAY_IP)
 
 
 class WifiBackend(ABC):
@@ -308,9 +313,14 @@ class WifiManager:
     # -- device hotspot ------------------------------------------------------
     def join_device_hotspot(self, sn: str, *, verify: bool = True,
                             settle_timeout: float = 30.0,
-                            poll_interval: float = 1.0) -> HotspotProfile:
+                            poll_interval: float = 1.0,
-        """Derive the profile from the SN, join, and verify the association."""
+                            gateway_ip: Optional[str] = None) -> HotspotProfile:
-        profile = derive_hotspot_credentials(sn)
+        """Derive the profile from the SN, join, and verify the association.
+
+        ``gateway_ip`` overrides the device address recorded on the
+        profile (default ``10.100.1.1``) without changing the radio join.
+        """
+        profile = derive_hotspot_credentials(sn, gateway_ip=gateway_ip)
         return self._join(profile, verify=verify,
                           settle_timeout=settle_timeout,
                           poll_interval=poll_interval)

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

tests/test_operations.py

@@ -104,3 +104,30 @@ class TestWattageOp(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+class TestWifiOps(unittest.TestCase):
+    def test_join_home_wifi_shape(self):
+        from franklinwh_agate_client.protocol import direct_connect_request
+        from franklinwh_agate_client.protocol import CommunicationCmd
+        from franklinwh_agate_client.protocol import CommunicationOperate
+        env = direct_connect_request(
+            CommunicationCmd.CONNECT_WIFI, "SN",
+            data={"ssid": "Home", "password": "pw",
+                  "wifi_SSID": "Home", "wifi_Pw": "pw"},
+            operate=CommunicationOperate.SET)
+        d = env.to_dict()
+        self.assertEqual(d["cmdType"], 1111)  # socket id of connectWifi
+        self.assertEqual(d["type"], 1)  # SET
+        self.assertEqual(d["dataArea"]["opt"], 1)
+        self.assertEqual(d["dataArea"]["ssid"], "Home")
+
+    def test_scan_and_status_shape(self):
+        from franklinwh_agate_client.protocol import direct_connect_request
+        from franklinwh_agate_client.protocol import CommunicationCmd
+        scan = direct_connect_request(CommunicationCmd.SCAN_WIFI_LIST, "SN")
+        self.assertEqual(scan.cmd_type, 1109)
+        self.assertEqual(scan.to_dict()["dataArea"], {"opt": 0})
+        status = direct_connect_request(
+            CommunicationCmd.CONNECTION_STATUS, "SN")
+        self.assertEqual(status.cmd_type, 1113)