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/BATTERY_POC.md

@@ -59,6 +59,21 @@ out = get_battery("...", host="10.100.1.1")
 print(out["value"], out["source"])  # e.g. 78.0 currentBatterySoc
 ```
 
+Business operations (one call per question, on a bound client):
+
+```python
+from franklinwh_agate_client.device_local import LocalDeviceClient
+client = LocalDeviceClient("A03X2516013210060005", host="10.100.1.1",
+                           socket_port=9000)
+client.get_battery_soc()        # charge % (apower, then BMS packs 1-2)
+client.get_household_wattage()  # watts (detail, then load, then EMS math)
+client.get_device_inventory()   # IBG/FHP/BMS serials + versions
+```
+
+`get_*` returns `{"value", "source", "detail", "reply", ...}` with
+`value: None` when nothing answered; transport errors propagate so
+multi-port callers (`get_battery`, `get_power`) can move on.
+
 ## Host probe recipe (run on the machine joined to the device AP)
 
 The agent container cannot reach `10.100.1.x`; run this on the host:
@@ -69,10 +84,12 @@ python3 -m franklinwh_agate_client.battery \
   --sn A03X2516013210060005 --host 10.100.1.1 --ports 9000,18000 --probe
 ```
 
-> Prefer the standalone scripts in `scripts/` — they need no install and
-> track the latest corrections: `host_probe4.py` is current (socket-space
-> ids + `opt`-seeded payloads). Copy the file content and run
-> `python3 host_probe4.py --sn A03X2516013210060005 --ports 9000`.
+> Prefer the standalone scripts in `scripts/` — thin wrappers over the
+> library business API (run from the repo root, e.g.
+> `python3 scripts/host_probe7.py --sn …`). Wire bytes always come from
+> `EnvelopeCodec` (`protocol/messages.py`); scripts never reimplement the
+> cipher. Current: `host_probe6.py` (variants), `host_probe7.py` (BMS
+> packs), `host_probe8.py` (wattage).
 
 Expected: `secret key for ...: 1c (radix 32)`, then per-variant `tx`
 (encrypted envelope, starts with the ASCII header
@@ -86,9 +103,21 @@ instead: install the Frida hook (`blutter_out/blutter_frida.js` pattern —
 log `Socket Manager send object: ` / `receive object: `), tap *Refresh* in
 the app while joined to the AP, and compare its TX bytes against ours.
 
-## Limits
+## Status (2026-10-16) — live round-trip works
 
-- No live round-trip has succeeded yet from an isolated container netns
-  (all probes `ECONNREFUSED`); first success needs a `10.100.1.x` source.
+Two-way comms verified against `10.100.1.1:9000` from a host on
+`10.100.1.0/24`: discovery → `1102` device-info reply (versions, `FHP_SN`,
+`BMS_SN`), per-pack BMS reads (`{"id": 1}` / `{"id": 2}` → live `batSoc`),
+EMS telemetry. `get_battery()` runs that exact sequence (discovery,
+apower, BMS packs 1–2, EMS) and returns the first SoC found; an all-zero
+BMS slot is skipped, not reported as 0 %.
+
+`DEVICE_APOWER_INFO` answers `{"result": 1, "reason": -1}` on the tested
+unit — SoC comes from the BMS packs instead.
+
+Remaining limits:
+
+- Needs L2 reachability to `10.100.1.1` (host on the device AP or bridged
+  container); the app's own socket port `18000` is closed on this unit.
 - BLE field order and `X-Device-*` binding remain open
   (`RE_DART_FINDINGS.md` Open items).

docs/RE_DART_FINDINGS.md

@@ -188,8 +188,14 @@ marker model is kept for the unverified Bluetooth path.
    failures keep accumulating ("receive incomplete data").
 
 So requests are `S1 || transform(S2)` (split at the computable header
-length) while responses are `s || 0x58 || transform(tail)` (split at the
-second `0x58` marker). Both halves share the one additive stream.
+length); live socket replies use the identical framing and split
+(`decrypt_wire` locates the header at the 2nd comma — no ids needed up
+front). Both halves share the one additive stream.
+
+Library shape: `protocol/messages.py:EnvelopeCodec` owns the bytes
+(`encode`/`decode` under one device key); `device_local`,
+`battery` and `power` implement the business calls on top; `scripts/`
+are thin CLI wrappers over those calls.
 
 Live corroboration (2026-10-15, PCAPdroid capture of the working app): a
 request frame opens with the ASCII header
@@ -306,6 +312,17 @@ every silent run is fully explained by bytes, with no need for exotic
 causes. Current probes use socket-space ids + radix 16 + `opt` payloads
 via `scripts/host_probe6.py`.
 
+### Live results (2026-10-16, host probes with corrected bytes)
+
+- Discovery + directed `queryDeviceInfo` + `BMS`/`EMS` all answer on
+  `10.100.1.1:9000` (replies: `1102` device info, `1706` BMS, `1708` EMS).
+- `DEVICE_APOWER_INFO` (`1307`) answers `{"result": 1, "reason": -1}` —
+  rejected (args or unsupported on this firmware).
+- `BMS` **requires the pack id** (`queryBmsInfoBySocket` passes
+  `arguments={"id"}`): id-less queries return the all-zero `id: 0` slot;
+  `id: 1`/`2` select the `…B1`/`…B2` packs whose `batSoc`/`batSoh` are the
+  charge/health readings. `scripts/host_probe7.py` reads both packs.
+
 ## 14. Direct-connect flow (manual AP join → return to app)
 
 `sections/menus/direct_connect/` owns the SoftAP onboarding path:
@@ -355,6 +372,83 @@ consistent with all-silent probes.
   socket/HTTP hotspot APIs (`/terminal/*`, `wifi_iot` join) take it from
   there. Same envelope + cipher on both transports (§5, §9).
 
+## 16. Live power readings
+
+- `queryDeviceRealTimeData` — what the home page polls
+  (`DeviceRealTimeDataMixin`, `startPollingDeviceRealTimeData`): cmd
+  `deviceDetailInfo` (socket `1301`), no arguments on socket
+  (`{"refreshData": 2}` is added only for mqtt mode). Reply `dataArea`:
+  `apowerCount`, **`totalPower`** (the household-draw headline).
+- `queryLoadStatusInfo` — cmd `loadStatus` (socket `1411`), no arguments.
+  Reply per smart switch: `Sw1Volt`/`Sw2Volt`, `SW1Curr`, `SW1ExpPower`
+  (`SW2…`), `SW1ExpEnergy`, `index`. (`loadStatus2`, socket `1415`, is the
+  newer variant.)
+- `queryEmsInfoBySocket` — cmd `ems` (socket `1707`), no arguments. Reply
+  carries `gridVol1/2`, `gridCurr1/2`, `loadCurr1/2`, `gridFreq`, relay
+  states. Household draw ≈
+  `(loadCurr1 + loadCurr2) × (gridVol1 + gridVol2) / 2`
+  (library: `power.ems_load_watts`).
+- `InstallSunRunSummaryEntity` (`homeloadPower`, `solarPower`,
+  `gridPower`, `apowersPower`) is parsed in the bean factory layer, not
+  off the local socket — cloud/HTTP-side entity.
+
+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`.
@@ -483,23 +577,25 @@ keepAlive 10 s, LWT, QoS 0, event names), `MethodHandlerManager` handler list,
 
 Still open:
 
-1. Exact BLE frame field order & the 2-byte magic semantics (confirm with `btsnoop`).
-2. ~~Exact `encryptObject` radix + trailing-header handling~~ — **resolved
-   from disassembly (see §5): radix 32, plaintext header first
-   (`S1 || transform(S2)`), `decryptData` marker model. Remaining: first
-   confirmed live round-trip.**
+1. Live verification of the BLE frame layout (settled from disassembly in
+   §9 — `magic ++ crc ++ totalLen ++ sn ++ content`, radix-32 CRC pairs —
+   but no live BLE exchange yet) and the GATT service/notify
+   characteristics (write char matches `0000f002-…`).
+2. ~~Exact `encryptObject` radix + framing~~ — **resolved live (2026-10-16,
+   see §5 and `tests/fixtures/*.bin`): radix 16, plaintext header first
+   (`S1 || transform(S2)`) both directions.** (An earlier radix-32 reading
+   of a nearby `0x20` literal is superseded.) The `decryptData`
+   `0x58`-marker model is kept for the unverified Bluetooth path only.
 3. ~~Local HTTP port + TLS behavior for `https://IP:PORT`~~ — **revised (2026-10-13,
    §8 empirical):** no HTTP/TLS port exists on the device at all (only 53 + 9000).
    Either local mode is socket-only on this firmware, or the HTTP service is
    state-dependent and wasn't running. Revisit if a `/terminal` response is ever
    captured from the SoftAP.
-4. ~~Local socket port confirmation~~ — **partially resolved: `9000` open on
-   one unit** (§8 empirical); the `0x4650` constant is not listening there,
-   so the port may be firmware-/device-dependent. Remaining: first confirmed
-   RPC round-trip — the corrected request bytes (raw `S1 || transform(S2)`,
-   radix 32, int `snno`, seconds `timeStamp`, true SN
-   `A03X2516013210060005`) are ready via
-   `battery --probe`; needs a host on `10.100.1.0/24`.
+4. ~~Local socket round-trip~~ — **done (2026-10-16): full two-way RPC on
+   `10.100.1.1:9000`** (discovery → `1102` info, per-pack BMS `batSoc`,
+   EMS telemetry). Remaining: the live endpoint's exact role for port 9000
+   vs the app's `18000` constant (closed here), and the `1102 = req + 1`
+   reply-numbering rule across more commands.
 5. Which class in `classes5.dex` sets the `X-Device-*` headers and their exact values
    (string-pool location confirmed; class binding in progress — §13).
 6. (Tooling, optional) Enable REA's DEX tools: obtain a **jadx-headless-mcp 0.7.1** JAR

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__)
 
@@ -30,7 +30,11 @@ log = logging.getLogger(__name__)
 SOCKET_PORT_CANDIDATES: Tuple[int, ...] = (9000, 18000)
 
 #: Reply keys searched for the aggregate percentage, in priority order.
-SOC_KEYS: Tuple[str, ...] = ("currentBatterySoc", "soc", "batterySoc", "vppSoc")
+#: ``batSoc`` is the per-pack charge from the BMS socket query
+#: (``queryBmsInfoBySocket`` passes ``arguments={"id": pack}`` — without
+#: the pack id the device answers the all-zero ``id: 0`` slot).
+SOC_KEYS: Tuple[str, ...] = ("currentBatterySoc", "soc", "batSoc",
+                             "batterySoc", "vppSoc")
 
 
 def _as_dict(data: Any) -> Optional[Dict[str, Any]]:
@@ -121,9 +125,26 @@ def _soc_detail(data: Dict[str, Any]) -> Dict[str, Any]:
     return detail
 
 
+def _is_empty_pack(reply: Dict[str, Any], result: Dict[str, Any]) -> bool:
+    """True when a BMS reply is the all-zero placeholder slot.
+
+    A genuine 0 % pack still reports health/voltage (``batSoh``,
+    ``batTotalVolt``); the unaddressed ``id: 0`` slot reports literal
+    zeros with ``result == 0``.  Returning ``0.0`` for that slot would
+    mask the real packs, so the caller keeps probing instead.
+    """
+    if result.get("source") != "batSoc" or result.get("value") != 0:
+        return False
+    data = reply.get("dataArea", reply)
+    if not isinstance(data, dict):
+        return False
+    return (data.get("result") == 0 and data.get("batSoh") == 0
+            and data.get("batTotalVolt") == 0)
+
+
 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,
@@ -131,12 +152,14 @@ def get_battery(
     """Query the aggregate charge percentage over the local TCP socket.
 
     Mirrors the app's ``queryAPowerData`` first (``DEVICE_APOWER_INFO`` with
-    ``data={"opt": 0, "update": 2}``), then ``BMS`` and ``EMS`` as fallbacks,
-    probing each candidate port. Like the app's direct-connect flow, a
-    discovery query (``QUERY_DEVICE_INFO``, empty serial) opens each port's
-    sequence — its reply also confirms two-way crypto before the battery
-    read. Returns the ``extract_soc`` dict plus ``cmd``/``port``/``reply``
-    context. Raises the last socket error if no port answers.
+    ``data={"opt": 0, "update": 2}``), then per-pack ``BMS`` socket queries
+    (``queryBmsInfoBySocket`` passes the pack ``id`` — ``batSoc`` is the
+    pack charge), then ``EMS`` as a fallback, probing each candidate port.
+    Like the app's direct-connect flow, a discovery query
+    (``QUERY_DEVICE_INFO``, empty serial) opens each port's sequence — its
+    reply also confirms two-way crypto before the battery read. Returns
+    the ``extract_soc`` dict plus ``cmd``/``port``/``reply`` context.
+    Raises the last socket error if no port answers.
     """
     last_err: Optional[Exception] = None
     for port in ports:
@@ -145,31 +168,16 @@ def get_battery(
             key_radix=key_radix, timeout=timeout,
         )
         try:
-            client.send_rpc(CommunicationCmd.QUERY_DEVICE_INFO,
-                            equip_no="",
-                            data={"minProtocolVer": "V1.00.00"})
-        except Exception as exc:  # discovery is best-effort; keep going
-            log.debug("port %d discovery failed: %s", port, exc)
-        for cmd, data in (
-            (CommunicationCmd.DEVICE_APOWER_INFO, {"update": 2}),
-            (CommunicationCmd.BMS, None),
-            (CommunicationCmd.EMS, None),
-        ):
-            try:
-                reply = client.send_rpc(cmd, data=data)
+            result = client.get_battery_soc()
         except (ConnectionError, socket.error, TimeoutError) as exc:
             last_err = exc
-                log.debug("port %d cmd %s failed: %s", port, cmd, exc)
-                break  # port unreachable — try next port
-            except Exception as exc:  # envelope-level failure; try next cmd
-                last_err = exc
-                log.debug("port %d cmd %s error: %s", port, cmd, exc)
-                continue
-            result = extract_soc(reply)
-            result.update({"cmd": str(cmd), "port": port, "reply": reply})
+            log.debug("port %d failed: %s", port, exc)
+            continue  # port unreachable — try next port
         if result["value"] is not None:
+            result.update({"port": port})
             return result
-            last_err = ValueError(f"reply on port {port} cmd {cmd} has no SoC field")
+        last_err = ValueError(
+            f"no SoC field in replies via {host}:{port}")
     raise ConnectionError(f"no battery reply via {host}:{list(ports)}") from last_err
 
 
@@ -181,14 +189,15 @@ PROBE_VARIANTS: Tuple[Tuple[str, CommunicationCmd, Any], ...] = (
     ("apower/none", CommunicationCmd.DEVICE_APOWER_INFO, None),
     ("apower/empty", CommunicationCmd.DEVICE_APOWER_INFO, {}),
     ("queryDeviceInfo/none", CommunicationCmd.QUERY_DEVICE_INFO, None),
-    ("bms/none", CommunicationCmd.BMS, None),
+    ("bms/id1", CommunicationCmd.BMS, {"id": 1}),
+    ("bms/id2", CommunicationCmd.BMS, {"id": 2}),
     ("ems/none", CommunicationCmd.EMS, None),
 )
 
 
 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,
@@ -231,12 +240,9 @@ def _recv_all(sock: socket.socket, idle: float = 4.0) -> bytes:
 
 def _apower_tx(sn: str, key_radix: int) -> Tuple[bytes, Dict[str, Any]]:
     """Canonical ``queryAPowerData`` wire bytes (fresh snno/timeStamp)."""
-    from .protocol import (
-        direct_connect_request,
-        encrypt_object,
-        generate_secret_key,
-    )
+    from .protocol import direct_connect_request
     from .protocol.envelope import SequenceNumber
+    from .protocol.messages import EnvelopeCodec
 
     _apower_tx_seq = getattr(_apower_tx, "_seq", None)
     if _apower_tx_seq is None:
@@ -245,9 +251,8 @@ def _apower_tx(sn: str, key_radix: int) -> Tuple[bytes, Dict[str, Any]]:
         CommunicationCmd.DEVICE_APOWER_INFO, sn,
         data={"update": 2}, snno=_apower_tx_seq.next(),
     )
-    wire = encrypt_object(
-        env.to_json_string(), generate_secret_key(sn), radix=key_radix)
-    return wire, env.to_dict()
+    codec = EnvelopeCodec.for_serial(sn, radix=key_radix)
+    return codec.encode(env), env.to_dict()
 
 
 #: Minimal MQTT v3.1.1 CONNECT (client id ``probe01``) — rules out / confirms
@@ -257,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,
@@ -270,7 +275,7 @@ def probe_transport(
     prefixes (LE16/BE16/LE32), an MQTT CONNECT, and a double-send over one
     persistent connection.  Each entry carries exact ``tx``/``rx`` bytes.
     """
-    from .protocol import decrypt_data
+    from .protocol.messages import EnvelopeCodec
 
     results: List[Dict[str, Any]] = []
     for port in ports:
@@ -303,12 +308,8 @@ def probe_transport(
             except (ConnectionError, socket.error, TimeoutError) as exc:
                 entry["error"] = f"{type(exc).__name__}: {exc}"
             if entry["rx"]:
-                decoded = decrypt_data(entry["rx"], radix=key_radix)
-                if decoded:
-                    try:
-                        entry["reply"] = json.loads(decoded)
-                    except ValueError:
-                        entry["reply"] = {"_ undecodable": decoded}
+                entry["reply"] = EnvelopeCodec.for_serial(
+                    sn, radix=key_radix).try_decode(entry["rx"])
             results.append(entry)
 
         run("greet-first/read-then-raw", tx, pre_read=4.0)
@@ -326,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:
-        """Join the device SoftAP (derives creds, joins, verifies)."""
-        return self.wifi_manager.join_device_hotspot(sn, **kwargs)
+    def join_hotspot(self, sn: str, gateway_ip: Optional[str] = None,
+                     **kwargs: Any) -> HotspotProfile:
+        """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``
-        aimed at the device gateway (``https://10.100.1.1:18000`` by default).
+        """Join the device hotspot, then return a ``LocalDeviceClient``.
+
+        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``).
         """
+        explicit = address if address is not None else (
+            host if host is not None else gateway_ip)
+        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

@@ -1,20 +1,24 @@
 """Local (direct-connect) device client.
 
 Talks to the aGate **on its own network** after joining the device hotspot —
-the ``mode == local`` path of the app (see ``docs/RE_DART_FINDINGS.md`` §8,
-§11).  Two channels:
+the direct-connect path of the app (see ``docs/RE_DART_FINDINGS.md`` §8,
+§11, §14).  Two channels:
 
 * **HTTP** to ``https://IP:PORT/terminal/...`` — the ``/terminal/*`` local
   endpoint family (hotspot info, aGate info, overall info, MQTT relay, ...).
-  Scheme/port are *not* confirmed (Open item #3); defaults are candidates.
-* **TCP socket** to ``10.100.1.1:18000`` (candidate port — Open item #4)
-  carrying the encrypted :class:`~franklinwh_agate_client.protocol.CommunicationObject`
-  envelope with little-endian int16 length framing.
+  Scheme/port are *not* confirmed live (Open item #3); defaults follow the
+  app's ``RemoteService`` shape.
+* **TCP socket** to ``10.100.1.1:9000`` — verified live 2026-10-16.  The
+  app dials the gateway IP port ``18000`` (``0x4650``), which is closed on
+  the tested unit; ``9000`` answers the same encrypted
+  :class:`~franklinwh_agate_client.protocol.CommunicationObject` envelope
+  sent **raw** (``dart:io _Socket.add``, no length prefix), with socket-id
+  ``cmdType`` values (``SOCKET_CMD_TYPE``) and ``opt``-seeded payloads.
+  Replies use the identical framing with id ``request + 1``.
 """
 
 from __future__ import annotations
 
-import json
 import logging
 import socket
 from typing import Any, Dict, Optional
@@ -27,14 +31,11 @@ from .protocol import (
     CommunicationOperate,
     SequenceNumber,
     SocketFrameDecoder,
-    decrypt_data,
-    decrypt_request,
-    decrypt_wire,
     direct_connect_request,
-    encrypt_object,
     frame_socket_payload,
     generate_secret_key,
 )
+from .protocol.messages import EnvelopeCodec
 from .wifi import DEVICE_GATEWAY_IP
 
 log = logging.getLogger(__name__)
@@ -93,6 +94,7 @@ class LocalDeviceClient:
         self.key_radix = int(key_radix)
         self.timeout = timeout
         self.key = generate_secret_key(sn)
+        self.codec = EnvelopeCodec(self.key, radix=self.key_radix)
         self._sn_seq = SequenceNumber()
         self._session = session or requests.Session()
         self._session.headers.update(
@@ -182,67 +184,72 @@ class LocalDeviceClient:
         ``increaseSN`` counter, serialised to compact JSON,
         ``encrypt_object``-ed, and sent **raw** (``_Socket.add``, no length
         prefix).  The reply is accumulated chunk by chunk and run through
-        ``decrypt_data`` until it yields a JSON object.
+        ``decrypt_wire`` (2nd-comma split, equipNo-derived key) until it
+        yields a JSON object.
         """
         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)
 
     def send_envelope(self, envelope: CommunicationObject) -> Dict[str, Any]:
-        wire = encrypt_object(envelope.to_json_string(), self.key,
-                              radix=self.key_radix)
-        raw = self._socket_roundtrip(frame_socket_payload(wire))
+        raw = self._socket_roundtrip(
+            frame_socket_payload(self.codec.encode(envelope)))
         try:
-            reply = decrypt_wire(raw, radix=self.key_radix)
+            reply = self.codec.decode(raw).to_dict()
         except ValueError:
             raise ConnectionError(
                 f"undecodable reply ({len(raw)} bytes) from "
                 f"{self.socket_host}:{self.socket_port}"
             )
-        return json.loads(reply)
+        return reply
 
     def send_probe(
         self,
         cmd: CommunicationCmd,
         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.
 
-        Never raises on a device-level failure: ``reply`` is the parsed JSON
-        (or ``None``), ``reply_raw`` the exact bytes received (``b""`` when
-        the device accepted the frame but answered with zero bytes).
+        Never raises on a device-level failure: ``reply`` is the parsed
+        envelope dict (or ``None``), ``reply_raw`` the exact bytes received
+        (``b""`` when the device accepted the frame but answered with zero
+        bytes).  ``equip_no`` overrides the serial (``""`` reproduces the
+        app's empty-SN discovery frame).
         """
         envelope = direct_connect_request(
-            cmd, self.sn, data=data, snno=self._sn_seq.next(),
+            cmd, self.sn if equip_no is None else equip_no, data=data,
+            snno=self._sn_seq.next(), operate=operate,
         )
-        wire = encrypt_object(envelope.to_json_string(), self.key,
-                              radix=self.key_radix)
+        wire = self.codec.encode(envelope)
         reply_raw = self._socket_roundtrip(
             frame_socket_payload(wire), timeout=timeout, collect_only=True,
         )
-        reply: Optional[Dict[str, Any]] = None
-        if reply_raw:
-            try:
-                decoded = decrypt_wire(reply_raw, radix=self.key_radix)
-            except ValueError:
-                decoded = ""
-            if decoded:
-                try:
-                    reply = json.loads(decoded)
-                except ValueError:
-                    reply = None
         return {
             "cmd": str(cmd),
             "envelope": envelope.to_dict(),
             "tx": wire,
             "reply_raw": reply_raw,
-            "reply": reply,
+            "reply": self.codec.try_decode(reply_raw),
         }
 
+    def send_raw(self, payload: bytes,
+                 timeout: Optional[float] = None) -> bytes:
+        """Low-level escape hatch: one connection, send *payload*, drain.
+
+        Used by transport-level probes (framing/handshake variants) that
+        are not valid envelopes.  Returns the exact bytes received (``b""``
+        when the peer answers with nothing).  Envelope traffic must use
+        :meth:`send_envelope` / :meth:`send_probe` instead.
+        """
+        return self._socket_roundtrip(bytes(payload), timeout=timeout,
+                                      collect_only=True)
+
     def _socket_roundtrip(self, framed: bytes,
                           timeout: Optional[float] = None,
                           collect_only: bool = False) -> bytes:
@@ -274,7 +281,7 @@ class LocalDeviceClient:
         if collect_only:
             return bytes(received)
         try:
-            decrypt_wire(bytes(received), radix=self.key_radix)
+            self.codec.decode(bytes(received))
         except ValueError:
             raise ConnectionError(
                 f"no reply frame from {self.socket_host}:{self.socket_port}"
@@ -285,8 +292,31 @@ class LocalDeviceClient:
     def query_device_info(self) -> Dict[str, Any]:
         return self.send_rpc(CommunicationCmd.QUERY_DEVICE_INFO)
 
-    def query_bms_info(self) -> Dict[str, Any]:
-        return self.send_rpc(CommunicationCmd.BMS)
+    def query_detail_info(self) -> Dict[str, Any]:
+        """``queryDeviceRealTimeData`` — realtime headline (socket id 1301).
+
+        No arguments.  The reply ``dataArea`` carries ``totalPower``
+        (the household draw the home page polls) and ``apowerCount``.
+        """
+        return self.send_rpc(CommunicationCmd.DEVICE_DETAIL_INFO)
+
+    def query_load_status(self) -> Dict[str, Any]:
+        """``queryLoadStatusInfo`` — per smart-switch watts (socket id 1411).
+
+        No arguments.  The reply ``dataArea`` carries per-channel
+        ``Sw1Volt``/``SW1Curr``/``SW1ExpPower`` (and ``SW2…``) plus
+        ``index``.
+        """
+        return self.send_rpc(CommunicationCmd.LOAD_STATUS)
+
+    def query_bms_info(self, pack_id: int = 1) -> Dict[str, Any]:
+        """``queryBmsInfoBySocket`` — BMS pack query (``arguments={"id"}``).
+
+        The pack index selects the battery (``BMS_SN`` ``…B1``/``…B2`` in
+        the device-info reply); omitting it yields the all-zero ``id: 0``
+        slot.  ``batSoc``/``batSoh`` carry the pack charge/health.
+        """
+        return self.send_rpc(CommunicationCmd.BMS, data={"id": int(pack_id)})
 
     def query_mppt_info(self) -> Dict[str, Any]:
         return self.send_rpc(CommunicationCmd.MPPT)
@@ -301,7 +331,13 @@ class LocalDeviceClient:
         return self.send_rpc(CommunicationCmd.EMS)
 
     def query_a_power(self) -> Dict[str, Any]:
-        return self.send_rpc(CommunicationCmd.DEVICE_APOWER_INFO)
+        """``queryAPowerData`` — apower query (``arguments={"update": 2}``).
+
+        Note: the tested unit answers ``{"result": 1, "reason": -1}`` here;
+        prefer :meth:`query_bms_info` / :meth:`get_battery_soc` for charge.
+        """
+        return self.send_rpc(CommunicationCmd.DEVICE_APOWER_INFO,
+                             data={"update": 2})
 
     def query_warning_list(self) -> Dict[str, Any]:
         return self.send_rpc(CommunicationCmd.DEVICE_ALARMS)
@@ -318,3 +354,145 @@ 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", ...}``.
+    # Transport errors propagate (caller picks the next port/device);
+    # answered-but-empty replies yield ``value: None``.
+
+    def get_device_inventory(self) -> Dict[str, Any]:
+        """Discovery + device info: identities, versions, FHP/BMS serials.
+
+        Mirrors the direct-connect open: a best-effort discovery query
+        (empty serial), then ``queryDeviceInfo``.  Returns the parsed
+        device-info reply plus an ``identity`` summary (``ibg_sn``,
+        ``fhp_sn``/``bms_sn`` lists, ``protocol_ver``).
+        """
+        try:
+            self.send_rpc(CommunicationCmd.QUERY_DEVICE_INFO,
+                          equip_no="",
+                          data={"minProtocolVer": "V1.00.00"})
+        except Exception:
+            log.debug("discovery probe failed (best-effort)", exc_info=True)
+        reply = self.send_rpc(
+            CommunicationCmd.QUERY_DEVICE_INFO,
+            data={"minProtocolVer": "V1.00.00"})
+        data = reply.get("dataArea", {})
+        if not isinstance(data, dict):
+            data = {}
+        return {
+            "reply": reply,
+            "ibg_sn": reply.get("equipNo"),
+            "fhp_sn": list(data.get("FHP_SN", []) or []),
+            "bms_sn": list(data.get("BMS_SN", []) or []),
+            "protocol_ver": data.get("protocolVer"),
+            "result": data.get("result"),
+        }
+
+    def get_battery_soc(self) -> Dict[str, Any]:
+        """Aggregate battery charge percentage for this device.
+
+        Tries apower, then per-pack BMS reads (``id`` 1..2 — the id-less
+        slot answers all zeros), skipping empty placeholder slots.
+        Returns ``extract_soc`` fields plus ``reply``/``cmd`` context;
+        ``value`` is ``None`` when nothing answered with charge data.
+        """
+        from .battery import _is_empty_pack, extract_soc
+
+        try:  # best-effort open, mirrors the direct-connect flow
+            self.send_rpc(CommunicationCmd.QUERY_DEVICE_INFO,
+                          equip_no="",
+                          data={"minProtocolVer": "V1.00.00"})
+        except Exception:
+            log.debug("discovery probe failed (best-effort)", exc_info=True)
+        last: Dict[str, Any] = {"value": None, "source": None,
+                                "detail": {}, "reply": None, "cmd": None}
+        for cmd, data in (
+            (CommunicationCmd.DEVICE_APOWER_INFO, {"update": 2}),
+            (CommunicationCmd.BMS, {"id": 1}),
+            (CommunicationCmd.BMS, {"id": 2}),
+        ):
+            try:
+                reply = self.send_rpc(cmd, data=data)
+            except (ConnectionError, socket.error, TimeoutError):
+                raise
+            except Exception as exc:  # envelope-level failure; next attempt
+                log.debug("get_battery_soc %s error: %s", cmd, exc)
+                continue
+            result = extract_soc(reply)
+            result.update({"reply": reply, "cmd": str(cmd)})
+            if (result["value"] is not None
+                    and not _is_empty_pack(reply, result)):
+                return result
+            last = result
+        if last["value"] is not None and _is_empty_pack(
+                last.get("reply") or {}, last):
+            # every attempt hit placeholder slots: unknown, not 0 %.
+            last = dict(last, value=None, source=None)
+        return last
+
+    def get_household_wattage(self) -> Dict[str, Any]:
+        """Live household wattage draw for this device.
+
+        Tries realtime detail (``totalPower``), then per-switch load
+        status, then EMS current×voltage math.  Returns ``extract_power``
+        fields plus ``reply``/``cmd`` context; ``value`` is ``None`` when
+        nothing answered with power data.
+        """
+        from .power import _power_detail, ems_load_watts, extract_power
+
+        last: Dict[str, Any] = {"value": None, "source": None,
+                                "detail": {}, "reply": None, "cmd": None}
+        for name, call in (
+            ("detail", self.query_detail_info),
+            ("load", self.query_load_status),
+            ("ems", self.query_ems_info),
+        ):
+            try:
+                reply = call()
+            except (ConnectionError, socket.error, TimeoutError):
+                raise
+            except Exception as exc:  # envelope-level failure; next attempt
+                log.debug("get_household_wattage %s error: %s", name, exc)
+                continue
+            result = extract_power(reply)
+            if result["value"] is None and name == "ems":
+                watts = ems_load_watts(reply)
+                if watts is not None:
+                    result = {"value": watts, "source": "ems(loadCurr*V)",
+                              "detail": _power_detail(
+                                  reply.get("dataArea", reply))}
+            result.update({"reply": reply, "cmd": name})
+            if result["value"] is not None:
+                return result
+            last = result
+        return last

franklinwh_agate_client/power.py

@@ -0,0 +1,166 @@
+"""Live household power over the local socket.
+
+The app's home page polls ``queryDeviceRealTimeData`` (socket
+``DEVICE_DETAIL_INFO``) for its energy-flow screen; per-switch watts come
+from ``queryLoadStatusInfo`` (socket ``LOAD_STATUS``); grid/load
+current+voltage telemetry comes from ``queryEmsInfoBySocket``.  See
+``docs/RE_DART_FINDINGS.md`` §4 (ids) and §16.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import socket
+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 = DEVICE_GATEWAY_IP
+PORTS: Tuple[int, ...] = (9000, 18000)
+
+#: Reply keys holding watt readings, in priority order.
+POWER_KEYS: Tuple[str, ...] = (
+    "totalPower",
+    "homeloadPower",
+    "gridPower",
+    "solarPower",
+    "SW1ExpPower",
+    "SW2ExpPower",
+)
+
+
+def _to_float(value: Any) -> Optional[float]:
+    try:
+        if isinstance(value, bool):
+            return None
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def extract_power(reply: Dict[str, Any]) -> Dict[str, Any]:
+    """Pull watt readings out of a parsed reply envelope.
+
+    Returns ``{"value": float|None, "source": str|None, "detail": {...}}``.
+    ``value`` is the first ``POWER_KEYS`` hit (headline: ``totalPower``);
+    ``detail`` carries the supporting fields (per-switch watts/currents,
+    grid volts/amps) when present.
+    """
+    data = reply.get("dataArea", reply)
+    if isinstance(data, str):
+        try:
+            data = json.loads(data)
+        except (ValueError, TypeError):
+            data = {}
+    for key in POWER_KEYS:
+        for hit in _walk(reply, key):
+            num = _to_float(hit)
+            if num is not None:
+                return {"value": num, "source": key,
+                        "detail": _power_detail(data)}
+    return {"value": None, "source": None, "detail": _power_detail(data)}
+
+
+def _power_detail(data: Any) -> Dict[str, Any]:
+    detail: Dict[str, Any] = {}
+    if not isinstance(data, dict):
+        return detail
+    for key in (
+        "totalPower", "apowerCount", "homeloadPower", "solarPower",
+        "gridPower", "SW1ExpPower", "SW2ExpPower", "Sw1Volt", "Sw2Volt",
+        "SW1Curr", "gridVol1", "gridVol2", "gridCurr1", "gridCurr2",
+        "loadCurr1", "loadCurr2", "gridLineVol", "gridFreq",
+    ):
+        hits = _walk(data, key)
+        if hits:
+            detail[key] = hits[0] if len(hits) == 1 else hits
+    return detail
+
+
+def ems_load_watts(reply: Dict[str, Any]) -> Optional[float]:
+    """Estimate household draw from an EMS reply, when present.
+
+    ``(loadCurr1 + loadCurr2) * (gridVol1 + gridVol2) / 2`` — per-leg
+    current times mean leg voltage.  Returns ``None`` if the fields are
+    absent (the EMS reply on the tested unit carries live grid Vol/Curr).
+    """
+    data = reply.get("dataArea", reply)
+    if not isinstance(data, dict):
+        return None
+    try:
+        amps = float(data["loadCurr1"]) + float(data["loadCurr2"])
+        volts = (float(data["gridVol1"]) + float(data["gridVol2"])) / 2
+    except (KeyError, TypeError, ValueError):
+        return None
+    return amps * volts
+
+
+def get_power(
+    sn: str,
+    host: str = HOST,
+    ports: Tuple[int, ...] = PORTS,
+    key_radix: int = 16,
+    timeout: float = 5.0,
+) -> Dict[str, Any]:
+    """Read live household wattage over the local TCP socket.
+
+    Tries realtime detail (``DEVICE_DETAIL_INFO`` → ``totalPower``),
+    then per-switch load status, then EMS telemetry, probing each port.
+    Returns the ``extract_power`` dict plus ``cmd``/``port``/``reply``.
+    """
+    last_err: Optional[Exception] = None
+    for port in ports:
+        client = LocalDeviceClient(
+            sn=sn, host=host, socket_port=port,
+            key_radix=key_radix, timeout=timeout,
+        )
+        try:
+            result = client.get_household_wattage()
+        except (ConnectionError, socket.error, TimeoutError) as exc:
+            last_err = exc
+            log.debug("port %d failed: %s", port, exc)
+            continue  # port unreachable — try next port
+        if result["value"] is not None:
+            result.update({"port": port})
+            return result
+        last_err = ValueError(
+            f"no power field in replies via {host}:{port}")
+    raise ConnectionError(f"no power reply via {host}:{list(ports)}") from last_err
+
+
+def main(argv: Optional[List[str]] = None) -> int:
+    parser = argparse.ArgumentParser(
+        description="Read live household wattage over the device SoftAP socket.")
+    parser.add_argument("--sn", required=True, help="20-char device serial")
+    parser.add_argument("--host", default=HOST, help="aGate gateway IP")
+    parser.add_argument("--ports", default=",".join(map(str, PORTS)),
+                        help="comma-separated socket ports to probe")
+    parser.add_argument("--timeout", type=float, default=5.0)
+    parser.add_argument("--raw", action="store_true",
+                        help="print full reply JSON")
+    args = parser.parse_args(argv)
+
+    ports = tuple(int(p) for p in args.ports.split(",") if p.strip())
+    try:
+        result = get_power(args.sn, host=args.host, ports=ports,
+                           timeout=args.timeout)
+    except Exception as exc:
+        print(f"ERROR: {exc}")
+        return 1
+    if args.raw:
+        print(json.dumps(result["reply"], indent=2, default=str))
+    else:
+        value = result["value"]
+        print(f"{value:g}W ({result['source']})"
+              if value is not None else "unknown")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

franklinwh_agate_client/protocol/__init__.py

@@ -61,6 +61,7 @@ from .framing import (
     verify_frame_crc,
 )
 from .framing import SocketFrameDecoder
+from .messages import EnvelopeCodec
 
 __all__ = [
     "CommunicationCmd",
@@ -106,4 +107,5 @@ __all__ = [
     "chunk_for_mtu",
     "frame_socket_payload",
     "SocketFrameDecoder",
+    "EnvelopeCodec",
 ]

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/protocol/framing.py

@@ -212,10 +212,9 @@ def frame_socket_payload(payload: bytes) -> bytes:
 class SocketFrameDecoder:
     """Accumulating socket receiver (``SocketManager._onSocketReceiveData``).
 
-    Each incoming chunk is appended to a growable buffer and ``decrypt_data``
-    is attempted over the whole buffer (inside try/catch in the app); a
-    non-empty decoded string that parses as JSON is one complete reply, at
-    which point the buffer is cleared.  Use :meth:`feed` for raw chunks and
+    Each incoming chunk is appended to a growable buffer and ``decrypt_wire``
+    is attempted over the whole buffer; a decoded string that parses as
+    JSON is one complete reply, at which point the buffer is cleared.  Use :meth:`feed` for raw chunks and
     :meth:`feed_json` when you want parsed envelopes back.
     """
 

franklinwh_agate_client/protocol/messages.py

@@ -0,0 +1,64 @@
+"""Low-level wire messages: envelopes to bytes and back.
+
+:class:`EnvelopeCodec` owns everything byte-level about one device key:
+serialising a :class:`CommunicationObject` to encrypted TX bytes and
+parsing received RX bytes back into an envelope.  Business logic (which
+command/arguments to send, retries, SoC extraction) lives one layer up in
+``device_local`` / ``battery`` / ``power``; standalone scripts must go
+through those layers, never reimplement this.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any, Dict
+
+from .crypto import (
+    DEFAULT_KEY_RADIX,
+    decrypt_wire,
+    encrypt_object,
+    generate_secret_key,
+)
+from .envelope import CommunicationObject
+
+
+class EnvelopeCodec:
+    """Encrypt/decrypt envelopes under one device key.
+
+    :param key: ``generate_secret_key`` output for the device serial.
+    :param radix: key-parse radix (16 — verified against live captures).
+    """
+
+    def __init__(self, key: str, radix: int = DEFAULT_KEY_RADIX) -> None:
+        self.key = str(key)
+        self.radix = int(radix)
+
+    @classmethod
+    def for_serial(cls, equip_no: str,
+                   radix: int = DEFAULT_KEY_RADIX) -> "EnvelopeCodec":
+        """Codec keyed by a device serial (derives ``generate_secret_key``)."""
+        return cls(generate_secret_key(equip_no), radix=radix)
+
+    # ---------------------------------------------------------------- encode
+    def encode(self, envelope: CommunicationObject) -> bytes:
+        """``toJsonString`` → ``S1 || transform(S2)`` TX bytes."""
+        return encrypt_object(envelope.to_json_string(), self.key,
+                              radix=self.radix)
+
+    # ---------------------------------------------------------------- decode
+    def decode(self, wire: bytes) -> CommunicationObject:
+        """RX bytes → envelope (2nd-comma split, key from ``equipNo``).
+
+        Raises :class:`ValueError` on truncated/non-JSON/undecodable
+        input — callers accumulating a stream treat that as "keep
+        reading".
+        """
+        return CommunicationObject.from_dict(
+            json.loads(decrypt_wire(bytes(wire), radix=self.radix)))
+
+    def try_decode(self, wire: bytes) -> Dict[str, Any] | None:
+        """Like :meth:`decode` but returns the dict, or ``None``."""
+        try:
+            return self.decode(wire).to_dict()
+        except (ValueError, TypeError):
+            return None

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:
-        """Derive the profile from the SN, join, and verify the association."""
-        profile = derive_hotspot_credentials(sn)
+                            poll_interval: float = 1.0,
+                            gateway_ip: Optional[str] = None) -> HotspotProfile:
+        """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

scripts/host_probe2.py

@@ -1,105 +1,64 @@
 #!/usr/bin/env python3
-"""Standalone host probe — no package install, no git needed (stdlib only).
+"""Host probe v2 — transport framing variants via the library.
 
-Drop this file on the machine joined to the aGate SoftAP and run:
-    python3 host_probe2.py --sn A03X2516013210060005
+Envelope bytes come from ``EnvelopeCodec``; only the on-the-wire framing
+varies (raw, length prefixes, split send, MQTT poke, greeting-first read).
+Raw payloads go through ``LocalDeviceClient.send_raw``.
 
-Replicates ``battery --probe2``: the canonical queryAPowerData envelope
-(verified byte-for-byte against the Dart AOT disassembly) sent with
-transport-level variants. Prints exact TX/RX hex for each.
+Run from the repo root on the SoftAP-joined host:
+    python3 scripts/host_probe2.py --sn A03X2516013210060005 --ports 9000
 """
 
 import argparse
 import json
+import os
 import socket
+import sys
 import time
-import zlib
 
-SNNO = [0]  # 1-based increaseSN counter (wraps at 9999)
+sys.path.insert(0, os.path.dirname(os.path.dirname(
+    os.path.abspath(__file__))))
+
+from franklinwh_agate_client.device_local import (  # noqa: E402
+    LocalDeviceClient,
+)
+from franklinwh_agate_client.protocol import (  # noqa: E402
+    CommunicationCmd,
+    direct_connect_request,
+    generate_secret_key,
+)
+from franklinwh_agate_client.protocol.envelope import (  # noqa: E402
+    SequenceNumber,
+)
+from franklinwh_agate_client.protocol.messages import (  # noqa: E402
+    EnvelopeCodec,
+)
+
+#: Minimal MQTT v3.1.1 CONNECT (client id ``probe01``).
+_MQTT_CONNECT = bytes.fromhex("101200044d5154540402003c000670726f62653031")
+
+_seq = SequenceNumber()
 
 
-def secret_key(sn):
-    return format(sum(sn.encode()) & 0xFF, "x")
+def canonical_tx(sn: str, codec: EnvelopeCodec) -> bytes:
+    """App-shaped apower query bytes (fresh snno/timeStamp)."""
+    env = direct_connect_request(
+        CommunicationCmd.DEVICE_APOWER_INFO, sn,
+        data={"update": 2}, snno=_seq.next())
+    return codec.encode(env)
 
 
-def build_envelope(sn, cmd=25, data=None, snno=None, ts=None):
-    SNNO[0] = (SNNO[0] % 9999) + 1
-    env = {
-        "cmdType": cmd,
-        "equipNo": sn,
-        "type": 0,
-        "timeStamp": ts if ts is not None else int(time.time()),
-        "snno": snno if snno is not None else SNNO[0],
-        "len": 0,
-        "crc": "",
-    }
-    if data:
-        compact = json.dumps(data, separators=(",", ":"))
-        env["len"] = len(compact.encode())
-        env["crc"] = format(zlib.crc32(compact.encode()), "X")
-        env["dataArea"] = data
-    return json.dumps(env, separators=(",", ":"))
-
-
-def encrypt(json_str, key):
-    parts = json_str.split(",")
-    a = (parts[0] + "," + parts[1] + ",").encode()
-    b = ",".join(parts[2:]).encode()
-    n = int(key, 32)
-    return a + bytes((b[i] + i + len(a) + n) & 0xFF
-                     for i in range(len(b)))
-
-
-def recv_all(sock, idle=4.0):
+def recv_all_verbose(sock: socket.socket, idle: float = 4.0):
     out = bytearray()
     sock.settimeout(idle)
     while True:
         try:
             chunk = sock.recv(65536)
         except (socket.timeout, TimeoutError):
-            break
+            return bytes(out), "idle-timeout"
         if not chunk:
-            break
+            return bytes(out), "closed-by-peer"
         out.extend(chunk)
-    return bytes(out)
-
-
-MQTT_CONNECT = bytes.fromhex("101200044d5154540402003c000670726f62653031")
-
-
-def run(label, host, port, tx, timeout, pre_read=0.0, post_idle=4.0,
-        split_send=False):
-    entry = (label, tx, b"", b"", None)
-    rx_pre, rx = b"", b""
-    err = None
-    try:
-        with socket.create_connection((host, port),
-                                      timeout=timeout) as sock:
-            if pre_read:
-                rx_pre = recv_all(sock, idle=pre_read)
-            if tx:
-                if split_send:
-                    half = len(tx) // 2
-                    sock.sendall(tx[:half])
-                    try:
-                        sock.settimeout(0.5)
-                        rx_pre += sock.recv(65536)
-                    except (socket.timeout, TimeoutError):
-                        pass
-                    sock.sendall(tx[half:])
-                else:
-                    sock.sendall(tx)
-                rx = recv_all(sock, idle=post_idle)
-    except (ConnectionError, socket.error, TimeoutError) as exc:
-        err = "%s: %s" % (type(exc).__name__, exc)
-    print("--- %s -> %s:%d" % (label, host, port))
-    print("  tx (%d B): %s" % (len(tx), tx.hex() or "(none)"))
-    if rx_pre:
-        print("  rx_pre (%d B): %s" % (len(rx_pre), rx_pre.hex()))
-    print("  rx (%d B): %s" % (len(rx), rx.hex() or "(empty)"))
-    if err:
-        print("  error: %s" % err)
-    return rx
 
 
 def main():
@@ -109,24 +68,58 @@ def main():
     ap.add_argument("--ports", default="9000")
     ap.add_argument("--timeout", type=float, default=5.0)
     args = ap.parse_args()
-
-    key = secret_key(args.sn)
-    print("secret key for %s: %s (radix 32)" % (args.sn, key))
-    tx = encrypt(build_envelope(args.sn, data={"update": 2}), key)
+    codec = EnvelopeCodec.for_serial(args.sn)
+    print("secret key for %s: %s (radix 16)"
+          % (args.sn, generate_secret_key(args.sn)))
     for port in [int(p) for p in args.ports.split(",") if p.strip()]:
-        run("greet-first/read-then-raw", args.host, port, tx,
-            args.timeout, pre_read=4.0)
-        run("raw/long-read", args.host, port, tx, args.timeout,
-            post_idle=8.0)
-        run("raw/split-send", args.host, port, tx, args.timeout,
-            split_send=True)
-        run("le16-prefix", args.host, port,
-            len(tx).to_bytes(2, "little") + tx, args.timeout)
-        run("be16-prefix", args.host, port,
-            len(tx).to_bytes(2, "big") + tx, args.timeout)
-        run("le32-prefix", args.host, port,
-            len(tx).to_bytes(4, "little") + tx, args.timeout)
-        run("mqtt-connect", args.host, port, MQTT_CONNECT, args.timeout)
+        client = LocalDeviceClient(args.sn, host=args.host,
+                                   socket_port=port, timeout=args.timeout)
+        tx = canonical_tx(args.sn, codec)
+
+        def show(label, tx_bytes, rx, how=""):
+            print("--- %s tx=%dB rx=%dB %s"
+                  % (label, len(tx_bytes), len(rx), how))
+            if rx:
+                print("  rx: %s" % rx.hex()[:400])
+
+        # greeting-first: connect, read before sending, then send
+        try:
+            with socket.create_connection(
+                    (args.host, port), timeout=args.timeout) as sock:
+                pre, _ = recv_all_verbose(sock, idle=4.0)
+                sock.sendall(tx)
+                rx, how = recv_all_verbose(sock, idle=8.0)
+                show("greet-first/read-then-raw", tx, rx,
+                     "pre=%dB %s" % (len(pre), how))
+        except (ConnectionError, socket.error, TimeoutError) as exc:
+            print("--- greet-first: error: %s" % exc)
+
+        for label, payload in (
+            ("raw/long-read", tx),
+            ("le16-prefix", len(tx).to_bytes(2, "little") + tx),
+            ("be16-prefix", len(tx).to_bytes(2, "big") + tx),
+            ("le32-prefix", len(tx).to_bytes(4, "little") + tx),
+            ("mqtt-connect", _MQTT_CONNECT),
+        ):
+            try:
+                t0 = time.time()
+                rx = client.send_raw(payload, timeout=args.timeout)
+                show(label, payload, rx, "%.1fs" % (time.time() - t0))
+            except (ConnectionError, socket.error, TimeoutError) as exc:
+                print("--- %s: error: %s" % (label, exc))
+
+        # split send over one connection
+        try:
+            with socket.create_connection(
+                    (args.host, port), timeout=args.timeout) as sock:
+                half = len(tx) // 2
+                sock.sendall(tx[:half])
+                time.sleep(0.5)
+                sock.sendall(tx[half:])
+                rx, how = recv_all_verbose(sock)
+                show("raw/split-send", tx, rx, how)
+        except (ConnectionError, socket.error, TimeoutError) as exc:
+            print("--- split-send: error: %s" % exc)
 
 
 if __name__ == "__main__":

scripts/host_probe3.py

@@ -1,114 +1,45 @@
 #!/usr/bin/env python3
-"""Host probe v3 — socket-transport command ids (stdlib only).
+"""Host probe v3 — socket-id variants via the library business API.
 
-A PCAPdroid capture of the working app showed ``{"cmdType":1101, ...}`` on
-the wire: ``CommunicationCmd::cmd`` returns the *second* id field for
-socket mode (QUERY_DEVICE_INFO 1101, APOWER 1307, BMS 1705, EMS 1707).
-Earlier probes used the small ids (25/0/29/30) and were dropped silently.
+Historical sweep that established the socket-transport command space
+(``1101/1307/1705/1707`` instead of the ``off_8`` ids). Superseded by
+``host_probe6.py`` for routine use; kept as a variant-list driver over
+``LocalDeviceClient.send_probe``.
 
-Run on the SoftAP-joined host:
-    python3 host_probe3.py --sn A03X2516013210060005 --ports 9000
+Run from the repo root on the SoftAP-joined host:
+    python3 scripts/host_probe3.py --sn A03X2516013210060005 --ports 9000
 """
 
 import argparse
 import json
+import os
 import socket
-import time
-import zlib
+import sys
 
-# Wire cmdType per CommunicationCmd::cmd (socket mode).
-APOWER, QUERY_INFO, BMS, EMS = 0x51B, 0x44D, 0x6A9, 0x6AB
-SNNO = [0]
+sys.path.insert(0, os.path.dirname(os.path.dirname(
+    os.path.abspath(__file__))))
 
+from franklinwh_agate_client.battery import extract_soc  # noqa: E402
+from franklinwh_agate_client.device_local import (  # noqa: E402
+    LocalDeviceClient,
+)
+from franklinwh_agate_client.protocol import (  # noqa: E402
+    CommunicationCmd,
+    generate_secret_key,
+)
 
-def secret_key(sn):
-    return format(sum(sn.encode()) & 0xFF, "x")
-
-
-def build_envelope(sn, cmd, data=None, snno=None, ts=None, equip=None):
-    SNNO[0] = (SNNO[0] % 9999) + 1
-    env = {"cmdType": cmd, "equipNo": equip if equip is not None else sn,
-           "type": 0,
-           "timeStamp": ts if ts is not None else int(time.time()),
-           "snno": snno if snno is not None else SNNO[0], "len": 0, "crc": ""}
-    if data:
-        compact = json.dumps(data, separators=(",", ":"))
-        env["len"] = len(compact.encode())
-        env["crc"] = format(zlib.crc32(compact.encode()), "X")
-        env["dataArea"] = data
-    return json.dumps(env, separators=(",", ":"))
-
-
-def encrypt(json_str, key):
-    parts = json_str.split(",")
-    a = (parts[0] + "," + parts[1] + ",").encode()
-    b = ",".join(parts[2:]).encode()
-    n = int(key, 32)
-    return a + bytes((b[i] + i + len(a) + n) & 0xFF
-                     for i in range(len(b)))
-
-
-def decrypt_data(wire, radix=32):
-    try:
-        i = wire.find(b"\x58")
-        if i < 0:
-            return ""
-        j = wire.find(b"\x58", i + 1)
-        if j < 0:
-            return ""
-        s = wire[:j].decode()
-        probe = json.loads(s + "}")
-        if not isinstance(probe, dict):
-            return ""
-        n = int(secret_key(str(probe.get("equipNo", ""))), radix)
-        prefix = s + ","
-        tail = wire[j + 1:]
-        pl = len(prefix.encode())
-        plain = bytes((t - k - pl - n) & 0xFF for k, t in enumerate(tail))
-        return prefix + plain.decode()
-    except Exception:
-        return ""
-
-
-def recv_all(sock, idle=6.0):
-    out = bytearray()
-    sock.settimeout(idle)
-    while True:
-        try:
-            chunk = sock.recv(65536)
-        except (socket.timeout, TimeoutError):
-            break
-        if not chunk:
-            break
-        out.extend(chunk)
-    return bytes(out)
-
-
-def run(label, host, port, tx, timeout, pre_read=0.0):
-    rx_pre, rx, err, reply = b"", b"", None, None
-    try:
-        with socket.create_connection((host, port),
-                                      timeout=timeout) as sock:
-            if pre_read:
-                rx_pre = recv_all(sock, idle=pre_read)
-            if tx:
-                sock.sendall(tx)
-                rx = recv_all(sock)
-            if rx:
-                dec = decrypt_data(rx)
-                reply = json.loads(dec) if dec else {"_raw": rx.hex()}
-    except (ConnectionError, socket.error, TimeoutError) as exc:
-        err = "%s: %s" % (type(exc).__name__, exc)
-    print("--- %s -> %s:%d" % (label, host, port))
-    print("  tx (%d B): %s" % (len(tx), tx.hex() or "(none)"))
-    if rx_pre:
-        print("  rx_pre (%d B): %s" % (len(rx_pre), rx_pre.hex()))
-    print("  rx (%d B): %s" % (len(rx), rx.hex() or "(empty)"))
-    if reply is not None:
-        print("  reply: " + json.dumps(reply, default=str)[:2000])
-    if err:
-        print("  error: %s" % err)
-    return rx
+VARIANTS = (
+    # (label, cmd, data, equip_no override or None for the client SN)
+    ("queryInfo/1101+SN", CommunicationCmd.QUERY_DEVICE_INFO,
+     {"minProtocolVer": "V1.00.00"}, None),
+    ("queryInfo/1101+empty", CommunicationCmd.QUERY_DEVICE_INFO,
+     {"minProtocolVer": "V1.00.00"}, ""),
+    ("apower/1307+update2", CommunicationCmd.DEVICE_APOWER_INFO,
+     {"update": 2}, None),
+    ("apower/1307+none", CommunicationCmd.DEVICE_APOWER_INFO, None, None),
+    ("bms/1705", CommunicationCmd.BMS, None, None),
+    ("ems/1707", CommunicationCmd.EMS, None, None),
+)
 
 
 def main():
@@ -118,26 +49,28 @@ def main():
     ap.add_argument("--ports", default="9000")
     ap.add_argument("--timeout", type=float, default=5.0)
     args = ap.parse_args()
-    key = secret_key(args.sn)
-    print("secret key for %s: %s (radix 32)" % (args.sn, key))
-    variants = [
-        ("queryInfo/1101+SN", QUERY_INFO, None, None),
-        ("queryInfo/1101+empty-SN", QUERY_INFO, None, ""),
-        ("apower/1307+update2", APOWER, {"update": 2}, None),
-        ("apower/1307+none", APOWER, None, None),
-        ("bms/1705", BMS, None, None),
-        ("ems/1707", EMS, None, None),
-    ]
+    print("secret key for %s: %s (radix 16)"
+          % (args.sn, generate_secret_key(args.sn)))
     for port in [int(p) for p in args.ports.split(",") if p.strip()]:
-        for label, cmd, data, equip in variants:
-            tx = encrypt(build_envelope(
-                args.sn, cmd, data,
-                equip=args.sn if equip is None else equip or "00000000",
-            ), key)
-            if run(label, args.host, port, tx, args.timeout,
-                   pre_read=4.0 if label.startswith("queryInfo/1101+SN")
-                   else 0.0):
-                return
+        client = LocalDeviceClient(args.sn, host=args.host,
+                                   socket_port=port, timeout=args.timeout)
+        for label, cmd, data, equip in VARIANTS:
+            try:
+                out = client.send_probe(cmd, data=data, timeout=args.timeout,
+                                        equip_no=equip)
+            except (ConnectionError, socket.error, TimeoutError) as exc:
+                print("--- %s: error: %s: %s"
+                      % (label, type(exc).__name__, exc))
+                continue
+            rx = out.get("reply_raw") or b""
+            print("--- %s tx=%dB rx=%dB"
+                  % (label, len(out["tx"]), len(rx)))
+            if out.get("reply") is not None:
+                print("  reply: "
+                      + json.dumps(out["reply"], default=str)[:1500])
+                soc = extract_soc(out["reply"])
+                print("  soc: %s (source=%s)" % (soc["value"],
+                                                 soc["source"]))
 
 
 if __name__ == "__main__":

scripts/host_probe4.py

@@ -1,117 +1,43 @@
 #!/usr/bin/env python3
-"""Host probe v4 — socket ids + ``opt``-seeded payloads (stdlib only).
+"""Host probe v4 — ``opt``-seeded payloads via the library business API.
 
-Two corrections vs v3, both from disassembly + live capture:
-* wire ``cmdType`` is the socket id (1101/1307/1705/1707);
-* the ``communicate`` factory seeds every socket payload with
-  ``"opt": 0`` first (operation type), then merges arguments, e.g.
-  apower → ``{"opt": 0, "update": 2}``; queryDeviceInfo →
-  ``{"opt": 0, "minProtocolVer": "V1.00.00"}``.
+Historical sweep from the ``communicate``-factory finding (socket payloads
+always start with ``"opt": 0``). Now built into
+``direct_connect_request``; kept as a variant-list driver over
+``LocalDeviceClient.send_probe``.
 
-Run on the SoftAP-joined host:
-    python3 host_probe4.py --sn A03X2516013210060005 --ports 9000
+Run from the repo root on the SoftAP-joined host:
+    python3 scripts/host_probe4.py --sn A03X2516013210060005 --ports 9000
 """
 
 import argparse
 import json
+import os
 import socket
-import time
-import zlib
+import sys
 
-APOWER, QUERY_INFO, BMS, EMS = 0x51B, 0x44D, 0x6A9, 0x6AB
-SNNO = [0]
+sys.path.insert(0, os.path.dirname(os.path.dirname(
+    os.path.abspath(__file__))))
 
+from franklinwh_agate_client.battery import extract_soc  # noqa: E402
+from franklinwh_agate_client.device_local import (  # noqa: E402
+    LocalDeviceClient,
+)
+from franklinwh_agate_client.protocol import (  # noqa: E402
+    CommunicationCmd,
+    generate_secret_key,
+)
 
-def secret_key(sn):
-    return format(sum(sn.encode()) & 0xFF, "x")
-
-
-def build_envelope(sn, cmd, args=None, snno=None, ts=None, equip=None):
-    SNNO[0] = (SNNO[0] % 9999) + 1
-    data = {"opt": 0}
-    if args:
-        data.update(args)
-    compact = json.dumps(data, separators=(",", ":"))
-    env = {"cmdType": cmd, "equipNo": equip if equip is not None else sn,
-           "type": 0,
-           "timeStamp": ts if ts is not None else int(time.time()),
-           "snno": snno if snno is not None else SNNO[0],
-           "len": len(compact.encode()),
-           "crc": format(zlib.crc32(compact.encode()), "X"),
-           "dataArea": data}
-    return json.dumps(env, separators=(",", ":"))
-
-
-def encrypt(json_str, key):
-    parts = json_str.split(",")
-    a = (parts[0] + "," + parts[1] + ",").encode()
-    b = ",".join(parts[2:]).encode()
-    n = int(key, 32)
-    return a + bytes((b[i] + i + len(a) + n) & 0xFF
-                     for i in range(len(b)))
-
-
-def decrypt_data(wire, radix=32):
-    try:
-        i = wire.find(b"\x58")
-        if i < 0:
-            return ""
-        j = wire.find(b"\x58", i + 1)
-        if j < 0:
-            return ""
-        s = wire[:j].decode()
-        probe = json.loads(s + "}")
-        if not isinstance(probe, dict):
-            return ""
-        n = int(secret_key(str(probe.get("equipNo", ""))), radix)
-        prefix = s + ","
-        tail = wire[j + 1:]
-        pl = len(prefix.encode())
-        return prefix + bytes((t - k - pl - n) & 0xFF
-                              for k, t in enumerate(tail)).decode()
-    except Exception:
-        return ""
-
-
-def recv_all(sock, idle=8.0):
-    out = bytearray()
-    sock.settimeout(idle)
-    while True:
-        try:
-            chunk = sock.recv(65536)
-        except (socket.timeout, TimeoutError):
-            break
-        if not chunk:
-            break
-        out.extend(chunk)
-    return bytes(out)
-
-
-def run(label, host, port, tx, timeout, pre_read=0.0):
-    rx_pre, rx, err, reply = b"", b"", None, None
-    try:
-        with socket.create_connection((host, port),
-                                      timeout=timeout) as sock:
-            if pre_read:
-                rx_pre = recv_all(sock, idle=pre_read)
-            if tx:
-                sock.sendall(tx)
-                rx = recv_all(sock)
-            if rx:
-                dec = decrypt_data(rx)
-                reply = json.loads(dec) if dec else {"_raw": rx.hex()}
-    except (ConnectionError, socket.error, TimeoutError) as exc:
-        err = "%s: %s" % (type(exc).__name__, exc)
-    print("--- %s -> %s:%d" % (label, host, port))
-    print("  tx (%d B): %s" % (len(tx), tx.hex() or "(none)"))
-    if rx_pre:
-        print("  rx_pre (%d B): %s" % (len(rx_pre), rx_pre.hex()))
-    print("  rx (%d B): %s" % (len(rx), rx.hex() or "(empty)"))
-    if reply is not None:
-        print("  reply: " + json.dumps(reply, default=str)[:3000])
-    if err:
-        print("  error: %s" % err)
-    return rx
+VARIANTS = (
+    ("queryInfo+minProto+SN", CommunicationCmd.QUERY_DEVICE_INFO,
+     {"minProtocolVer": "V1.00.00"}, None),
+    ("queryInfo+minProto+empty", CommunicationCmd.QUERY_DEVICE_INFO,
+     {"minProtocolVer": "V1.00.00"}, ""),
+    ("apower+opt+update2", CommunicationCmd.DEVICE_APOWER_INFO,
+     {"update": 2}, None),
+    ("bms+opt", CommunicationCmd.BMS, None, None),
+    ("ems+opt", CommunicationCmd.EMS, None, None),
+)
 
 
 def main():
@@ -121,26 +47,28 @@ def main():
     ap.add_argument("--ports", default="9000")
     ap.add_argument("--timeout", type=float, default=5.0)
     args = ap.parse_args()
-    key = secret_key(args.sn)
-    print("secret key for %s: %s (radix 32)" % (args.sn, key))
-    variants = [
-        ("queryInfo+minProto+SN", QUERY_INFO, {"minProtocolVer": "V1.00.00"},
-         None),
-        ("queryInfo+minProto+empty", QUERY_INFO, {"minProtocolVer": "V1.00.00"},
-         ""),
-        ("apower+opt+update2", APOWER, {"update": 2}, None),
-        ("bms+opt", BMS, None, None),
-        ("ems+opt", EMS, None, None),
-    ]
+    print("secret key for %s: %s (radix 16)"
+          % (args.sn, generate_secret_key(args.sn)))
     for port in [int(p) for p in args.ports.split(",") if p.strip()]:
-        for label, cmd, data, equip in variants:
-            tx = encrypt(build_envelope(
-                args.sn, cmd, data,
-                equip=args.sn if equip is None else equip or "00000000",
-            ), key)
-            if run(label, args.host, port, tx, args.timeout,
-                   pre_read=4.0 if "empty" not in label else 0.0):
-                return
+        client = LocalDeviceClient(args.sn, host=args.host,
+                                   socket_port=port, timeout=args.timeout)
+        for label, cmd, data, equip in VARIANTS:
+            try:
+                out = client.send_probe(cmd, data=data, timeout=args.timeout,
+                                        equip_no=equip)
+            except (ConnectionError, socket.error, TimeoutError) as exc:
+                print("--- %s: error: %s: %s"
+                      % (label, type(exc).__name__, exc))
+                continue
+            rx = out.get("reply_raw") or b""
+            print("--- %s tx=%dB rx=%dB"
+                  % (label, len(out["tx"]), len(rx)))
+            if out.get("reply") is not None:
+                print("  reply: "
+                      + json.dumps(out["reply"], default=str)[:1500])
+                soc = extract_soc(out["reply"])
+                print("  soc: %s (source=%s)" % (soc["value"],
+                                                 soc["source"]))
 
 
 if __name__ == "__main__":

scripts/host_probe5.py

@@ -1,59 +1,52 @@
 #!/usr/bin/env python3
-"""Host probe v5 — session behaviour on 9000 (stdlib only).
+"""Host probe v5 — session sequencing via the library.
 
-v4 (correct ids + opt payloads) is still met with silence, so this probe
-tests *session* hypotheses instead of bytes:
-* persistent connection: discovery → directed queries on ONE socket, with
-  app-like sequential snno (the app reuses its SocketManager connection;
-  the field captures show a discovery frame minutes before directed ones);
-* close-vs-timeout distinction for every read (immediate close = active
-  reject; 8 s idle + open conn = device thinking/forwarding upstream);
-* high-snno opens (in case the device tracks sequence per sender).
+One persistent connection carrying discovery → directed queries with
+app-like sequential ``snno``, plus close-vs-timeout reporting and
+high-snno opens. Payload bytes come from ``direct_connect_request`` +
+``EnvelopeCodec``; only the connection orchestration lives here.
+
+Run from the repo root on the SoftAP-joined host:
+    python3 scripts/host_probe5.py --sn A03X2516013210060005 --ports 9000
 """
 
 import argparse
 import json
+import os
 import socket
+import sys
 import time
-import zlib
 
-APOWER, QUERY_INFO, BMS = 0x51B, 0x44D, 0x6A9
-SNNO = [100]
+sys.path.insert(0, os.path.dirname(os.path.dirname(
+    os.path.abspath(__file__))))
+
+from franklinwh_agate_client.device_local import (  # noqa: E402
+    LocalDeviceClient,
+)
+from franklinwh_agate_client.protocol import (  # noqa: E402
+    CommunicationCmd,
+    direct_connect_request,
+    generate_secret_key,
+)
+from franklinwh_agate_client.protocol.envelope import (  # noqa: E402
+    SequenceNumber,
+)
+from franklinwh_agate_client.protocol.messages import (  # noqa: E402
+    EnvelopeCodec,
+)
+
+_seq = SequenceNumber(start=100)
 
 
-def secret_key(sn):
-    return format(sum(sn.encode()) & 0xFF, "x")
+def tx_for(codec: EnvelopeCodec, sn: str, cmd: CommunicationCmd,
+           data, equip=None, snno=None) -> bytes:
+    env = direct_connect_request(
+        cmd, sn if equip is None else equip, data=data,
+        snno=_seq.next() if snno is None else snno)
+    return codec.encode(env)
 
 
-def build_envelope(sn, cmd, args=None, snno=None, ts=None, equip=None):
-    if snno is None:
-        SNNO[0] = (SNNO[0] % 9999) + 1
-        snno = SNNO[0]
-    data = {"opt": 0}
-    if args:
-        data.update(args)
-    compact = json.dumps(data, separators=(",", ":"))
-    env = {"cmdType": cmd, "equipNo": equip if equip is not None else sn,
-           "type": 0,
-           "timeStamp": ts if ts is not None else int(time.time()),
-           "snno": snno,
-           "len": len(compact.encode()),
-           "crc": format(zlib.crc32(compact.encode()), "X"),
-           "dataArea": data}
-    return json.dumps(env, separators=(",", ":"))
-
-
-def encrypt(json_str, key):
-    parts = json_str.split(",")
-    a = (parts[0] + "," + parts[1] + ",").encode()
-    b = ",".join(parts[2:]).encode()
-    n = int(key, 32)
-    return a + bytes((b[i] + i + len(a) + n) & 0xFF
-                     for i in range(len(b)))
-
-
-def recv_all_verbose(sock, idle=8.0):
-    """Return (bytes, how) where how is closed|idle-timeout."""
+def recv_all_verbose(sock: socket.socket, idle: float = 8.0):
     out = bytearray()
     sock.settimeout(idle)
     while True:
@@ -66,47 +59,6 @@ def recv_all_verbose(sock, idle=8.0):
         out.extend(chunk)
 
 
-def one_shot(label, host, port, tx, timeout, snno):
-    try:
-        with socket.create_connection((host, port),
-                                      timeout=timeout) as sock:
-            t0 = time.time()
-            sock.sendall(tx)
-            rx, how = recv_all_verbose(sock)
-            dt = time.time() - t0
-    except (ConnectionError, socket.error, TimeoutError) as exc:
-        print("--- %s snno=%d -> %s:%d" % (label, snno, host, port))
-        print("  tx (%d B): %s" % (len(tx), tx.hex()))
-        print("  error: %s: %s" % (type(exc).__name__, exc))
-        return b""
-    print("--- %s snno=%d -> %s:%d (%.1fs, %s)" % (
-        label, snno, host, port, dt, how))
-    print("  tx (%d B): %s" % (len(tx), tx.hex()))
-    print("  rx (%d B): %s" % (len(rx), rx.hex() or "(empty)"))
-    return rx
-
-
-def session(label, host, port, frames, timeout):
-    print("=== %s -> %s:%d (one connection)" % (label, host, port))
-    try:
-        with socket.create_connection((host, port),
-                                      timeout=timeout) as sock:
-            for name, tx in frames:
-                t0 = time.time()
-                sock.sendall(tx)
-                rx, how = recv_all_verbose(sock)
-                dt = time.time() - t0
-                print("--- %s tx=%dB rx=%dB (%.1fs, %s)" % (
-                    name, len(tx), len(rx), dt, how))
-                if rx:
-                    print("  rx: %s" % rx.hex()[:3000])
-                if how == "closed-by-peer" and not rx:
-                    print("  (peer closed; stopping session)")
-                    break
-    except (ConnectionError, socket.error, TimeoutError) as exc:
-        print("  error: %s: %s" % (type(exc).__name__, exc))
-
-
 def main():
     ap = argparse.ArgumentParser()
     ap.add_argument("--sn", required=True)
@@ -114,32 +66,63 @@ def main():
     ap.add_argument("--ports", default="9000")
     ap.add_argument("--timeout", type=float, default=5.0)
     args = ap.parse_args()
-    key = secret_key(args.sn)
-    print("secret key for %s: %s (radix 32)" % (args.sn, key))
-
-    def tx(cmd, data, equip=None, snno=None):
-        return encrypt(build_envelope(
-            args.sn, cmd, data, snno=snno,
-            equip=args.sn if equip is None else equip or "00000000"), key)
-
+    codec = EnvelopeCodec.for_serial(args.sn)
+    print("secret key for %s: %s (radix 16)"
+          % (args.sn, generate_secret_key(args.sn)))
     ports = [int(p) for p in args.ports.split(",") if p.strip()]
     for port in ports:
-        # A. persistent session: discovery first, then directed (app order)
-        session("discovery-then-directed", args.host, port, [
-            ("disc/1101+empty",
-             tx(QUERY_INFO, {"minProtocolVer": "V1.00.00"}, equip="")),
-            ("info/1101+SN",
-             tx(QUERY_INFO, {"minProtocolVer": "V1.00.00"})),
-            ("apower/1307",
-             tx(APOWER, {"update": 2})),
-            ("bms/1705", tx(BMS, None)),
-        ], args.timeout)
-        # B. high-snno fresh opens
-        one_shot("apower-high-snno", args.host, port,
-                 tx(APOWER, {"update": 2}, snno=500), args.timeout, 500)
-        one_shot("info-high-snno", args.host, port,
-                 tx(QUERY_INFO, {"minProtocolVer": "V1.00.00"}, snno=501),
-                 args.timeout, 501)
+        print("=== discovery-then-directed -> %s:%d (one connection)"
+              % (args.host, port))
+        try:
+            with socket.create_connection(
+                    (args.host, port), timeout=args.timeout) as sock:
+                for name, cmd, data, equip in (
+                    ("disc/1101+empty", CommunicationCmd.QUERY_DEVICE_INFO,
+                     {"minProtocolVer": "V1.00.00"}, ""),
+                    ("info/1101+SN", CommunicationCmd.QUERY_DEVICE_INFO,
+                     {"minProtocolVer": "V1.00.00"}, None),
+                    ("apower/1307", CommunicationCmd.DEVICE_APOWER_INFO,
+                     {"update": 2}, None),
+                    ("bms/1705", CommunicationCmd.BMS, {"id": 1}, None),
+                ):
+                    payload = tx_for(codec, args.sn, cmd, data, equip)
+                    t0 = time.time()
+                    sock.sendall(payload)
+                    rx, how = recv_all_verbose(sock)
+                    print("--- %s tx=%dB rx=%dB (%.1fs, %s)"
+                          % (name, len(payload), len(rx),
+                             time.time() - t0, how))
+                    if rx:
+                        print("  rx: %s" % rx.hex()[:400])
+                        parsed = codec.try_decode(rx)
+                        if parsed is not None:
+                            print("  reply: " + json.dumps(
+                                parsed, default=str)[:800])
+                    if how == "closed-by-peer" and not rx:
+                        print("  (peer closed; stopping session)")
+                        break
+        except (ConnectionError, socket.error, TimeoutError) as exc:
+            print("  error: %s: %s" % (type(exc).__name__, exc))
+
+        client = LocalDeviceClient(args.sn, host=args.host,
+                                   socket_port=port, timeout=args.timeout)
+        for label, cmd, data, snno in (
+            ("apower-high-snno", CommunicationCmd.DEVICE_APOWER_INFO,
+             {"update": 2}, 500),
+            ("info-high-snno", CommunicationCmd.QUERY_DEVICE_INFO,
+             {"minProtocolVer": "V1.00.00"}, 501),
+        ):
+            env = direct_connect_request(cmd, args.sn, data=data, snno=snno)
+            payload = codec.encode(env)
+            try:
+                rx = client.send_raw(payload, timeout=args.timeout)
+            except (ConnectionError, socket.error, TimeoutError) as exc:
+                print("--- %s: error: %s" % (label, exc))
+                continue
+            print("--- %s tx=%dB rx=%dB" % (label, len(payload), len(rx)))
+            parsed = codec.try_decode(rx)
+            if parsed is not None:
+                print("  reply: " + json.dumps(parsed, default=str)[:800])
 
 
 if __name__ == "__main__":

scripts/host_probe6.py

@@ -1,194 +1,66 @@
 #!/usr/bin/env python3
-"""Host probe v6 — verified-against-capture wire format (stdlib only).
+"""Host probe v6 — envelope variants via the library business API.
 
-Locked by two live frames (tests/fixtures/*.bin in the repo):
-* key radix 16 (``int("80",16)=128``, ``int("1c",16)=28``);
-* ``S1 || transform(S2)`` both directions; reply id = request id + 1
-  observed (1101 -> 1102);
-* socket payloads seeded with ``"opt": 0``;
-* replies parsed by splitting at the 2nd comma (no ids needed up front).
+Sends the ``battery.PROBE_VARIANTS`` envelopes through
+``LocalDeviceClient.send_probe`` (sequence numbers, ``opt``-seeded
+payloads, socket ids and cipher all come from the library) and prints
+exact TX/RX bytes plus any parsed reply/SoC.
 
-Run on the SoftAP-joined host:
-    python3 host_probe6.py --sn A03X2516013210060005 --ports 9000
+Run from the repo root on the SoftAP-joined host:
+    python3 scripts/host_probe6.py --sn A03X2516013210060005 --ports 9000
 """
 
 import argparse
 import json
+import os
 import socket
-import time
-import zlib
+import sys
 
-APOWER, QUERY_INFO, BMS, EMS = 0x51B, 0x44D, 0x6A9, 0x6AB
-SNNO = [0]
-SOC_KEYS = ("currentBatterySoc", "soc", "batterySoc", "vppSoc")
+sys.path.insert(0, os.path.dirname(os.path.dirname(
+    os.path.abspath(__file__))))
 
-
-def secret_key(sn):
-    return format(sum(sn.encode()) & 0xFF, "x")
-
-
-def build_envelope(sn, cmd, args=None, snno=None, ts=None, equip=None):
-    if snno is None:
-        SNNO[0] = (SNNO[0] % 9999) + 1
-        snno = SNNO[0]
-    data = {"opt": 0}
-    if args:
-        data.update(args)
-    compact = json.dumps(data, separators=(",", ":"))
-    env = {"cmdType": cmd, "equipNo": equip if equip is not None else sn,
-           "type": 0,
-           "timeStamp": ts if ts is not None else int(time.time()),
-           "snno": snno,
-           "len": len(compact.encode()),
-           "crc": format(zlib.crc32(compact.encode()), "X"),
-           "dataArea": data}
-    return json.dumps(env, separators=(",", ":"))
-
-
-def encrypt(json_str, key):
-    parts = json_str.split(",")
-    a = (parts[0] + "," + parts[1] + ",").encode()
-    b = ",".join(parts[2:]).encode()
-    n = int(key, 16)
-    return a + bytes((b[i] + i + len(a) + n) & 0xFF
-                     for i in range(len(b)))
-
-
-def decrypt_wire(wire):
-    first = wire.find(b",")
-    second = wire.find(b",", first + 1)
-    if first < 0 or second < 0:
-        return ""
-    a, b = wire[:second + 1], wire[second + 1:]
-    try:
-        probe = json.loads(a[:-1].decode() + "}")
-    except Exception:
-        return ""
-    if not isinstance(probe, dict):
-        return ""
-    n = int(secret_key(str(probe.get("equipNo", ""))), 16)
-    try:
-        return (a + bytes((t - i - len(a) - n) & 0xFF
-                          for i, t in enumerate(b))).decode()
-    except Exception:
-        return ""
-
-
-def recv_all(sock, idle=8.0):
-    out = bytearray()
-    sock.settimeout(idle)
-    while True:
-        try:
-            chunk = sock.recv(65536)
-        except (socket.timeout, TimeoutError):
-            break
-        if not chunk:
-            break
-        out.extend(chunk)
-    return bytes(out)
-
-
-def walk(obj, key, found):
-    if isinstance(obj, dict):
-        for k, v in obj.items():
-            if k == key:
-                found.append(v)
-            walk(v, key, found)
-    elif isinstance(obj, list):
-        for item in obj:
-            walk(item, key, found)
-    return found
-
-
-def extract_soc(reply):
-    data = reply.get("dataArea", reply)
-    if isinstance(data, str):
-        try:
-            data = json.loads(data)
-        except ValueError:
-            data = {}
-    for key in SOC_KEYS:
-        hits = walk(reply, key, [])
-        for hit in hits:
-            if isinstance(hit, bool):
-                continue
-            try:
-                return float(hit), key
-            except (TypeError, ValueError):
-                continue
-    arrs = walk(data if isinstance(data, dict) else {}, "socArray", [])
-    for arr in arrs:
-        if isinstance(arr, list) and arr:
-            try:
-                nums = [float(v) for v in arr
-                        if not isinstance(v, bool)]
-                if nums:
-                    return sum(nums) / len(nums), "socArray(avg)"
-            except (TypeError, ValueError):
-                continue
-    return None, None
-
-
-def run(label, host, port, tx, timeout, pre_read=0.0):
-    rx_pre, rx, err, reply = b"", b"", None, None
-    try:
-        with socket.create_connection((host, port),
-                                      timeout=timeout) as sock:
-            if pre_read:
-                rx_pre = recv_all(sock, idle=pre_read)
-            if tx:
-                sock.sendall(tx)
-                rx = recv_all(sock)
-            if rx:
-                dec = decrypt_wire(rx)
-                reply = json.loads(dec) if dec else {"_raw": rx.hex()}
-    except (ConnectionError, socket.error, TimeoutError) as exc:
-        err = "%s: %s" % (type(exc).__name__, exc)
-    print("--- %s -> %s:%d" % (label, host, port))
-    print("  tx (%d B): %s" % (len(tx), tx.hex() or "(none)"))
-    if rx_pre:
-        print("  rx_pre (%d B): %s" % (len(rx_pre), rx_pre.hex()))
-    print("  rx (%d B): %s" % (len(rx), rx.hex()[:400] or "(empty)"))
-    if reply is not None:
-        print("  reply: " + json.dumps(reply, default=str)[:3000])
-        if isinstance(reply, dict):
-            val, src = extract_soc(reply)
-            print("  soc: %s (source=%s)" % (val, src))
-    if err:
-        print("  error: %s" % err)
-    return rx
+from franklinwh_agate_client.battery import (  # noqa: E402
+    extract_soc,
+    probe_device,
+)
+from franklinwh_agate_client.protocol import (  # noqa: E402
+    generate_secret_key,
+)
 
 
 def main():
     ap = argparse.ArgumentParser()
     ap.add_argument("--sn", required=True)
     ap.add_argument("--host", default="10.100.1.1")
-    ap.add_argument("--ports", default="9000")
+    ap.add_argument("--ports", default="9000,18000")
     ap.add_argument("--timeout", type=float, default=5.0)
     args = ap.parse_args()
-    key = secret_key(args.sn)
-    print("secret key for %s: %s (radix 16)" % (args.sn, key))
-    variants = [
-        ("discovery/1101+empty", QUERY_INFO,
-         {"minProtocolVer": "V1.00.00"}, ""),
-        ("queryInfo/1101+SN", QUERY_INFO,
-         {"minProtocolVer": "V1.00.00"}, None),
-        ("apower/1307+SN", APOWER, {"update": 2}, None),
-        ("apower/1307+IBG", APOWER, {"update": 2},
-         "10060005A03X25160132"),
-        ("bms/1705+SN", BMS, None, None),
-        ("ems/1707+SN", EMS, None, None),
-    ]
-    for port in [int(p) for p in args.ports.split(",") if p.strip()]:
-        for label, cmd, data, equip in variants:
-            tx = encrypt(build_envelope(
-                args.sn, cmd, data,
-                equip=args.sn if equip is None else equip or "00000000",
-            ), key)
-            if run(label, args.host, port, tx, args.timeout,
-                   pre_read=4.0 if label.startswith("discovery") else 0.0):
-                pass
+    print("secret key for %s: %s (radix 16)"
+          % (args.sn, generate_secret_key(args.sn)))
+    ports = tuple(int(p) for p in args.ports.split(",") if p.strip())
+    try:
+        results = probe_device(args.sn, host=args.host, ports=ports,
+                               timeout=args.timeout)
+    except (ConnectionError, socket.error, TimeoutError) as exc:
+        print("ERROR: %s: %s" % (type(exc).__name__, exc))
+        return 1
+    for out in results:
+        print("--- %s -> %s:%d cmd=%s"
+              % (out["variant"], args.host, out["port"], out["cmd"]))
+        if out.get("envelope") is not None:
+            print("  envelope: " + json.dumps(out["envelope"], default=str))
+        tx = out.get("tx") or b""
+        print("  tx (%d B): %s" % (len(tx), tx.hex()))
+        rx = out.get("reply_raw") or b""
+        print("  rx (%d B): %s" % (len(rx), rx.hex()[:400] or "(empty)"))
+        if out.get("reply") is not None:
+            print("  reply: " + json.dumps(out["reply"], default=str)[:2000])
+            soc = extract_soc(out["reply"])
+            print("  soc: %s (source=%s)" % (soc["value"], soc["source"]))
+        if out.get("error"):
+            print("  error: %s" % out["error"])
+    return 0
 
 
 if __name__ == "__main__":
-    main()
+    raise SystemExit(main())

scripts/host_probe7.py

@@ -0,0 +1,60 @@
+#!/usr/bin/env python3
+"""Host probe v7 — battery charge via the business operations API.
+
+One call per port: ``LocalDeviceClient.get_battery_soc`` (discovery,
+apower, per-pack BMS reads). Prints the percentage.
+
+Run from the repo root on the SoftAP-joined host:
+    python3 scripts/host_probe7.py --sn A03X2516013210060005 --ports 9000
+"""
+
+import argparse
+import json
+import os
+import socket
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(
+    os.path.abspath(__file__))))
+
+from franklinwh_agate_client.device_local import (  # noqa: E402
+    LocalDeviceClient,
+)
+from franklinwh_agate_client.protocol import (  # noqa: E402
+    generate_secret_key,
+)
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--sn", required=True)
+    ap.add_argument("--host", default="10.100.1.1")
+    ap.add_argument("--ports", default="9000")
+    ap.add_argument("--timeout", type=float, default=5.0)
+    ap.add_argument("--raw", action="store_true",
+                    help="print the full reply envelope")
+    args = ap.parse_args()
+    print("secret key for %s: %s (radix 16)"
+          % (args.sn, generate_secret_key(args.sn)))
+    for port in [int(p) for p in args.ports.split(",") if p.strip()]:
+        client = LocalDeviceClient(args.sn, host=args.host,
+                                   socket_port=port, timeout=args.timeout)
+        try:
+            result = client.get_battery_soc()
+        except (ConnectionError, socket.error, TimeoutError) as exc:
+            print("=== port %d: error: %s: %s"
+                  % (port, type(exc).__name__, exc))
+            continue
+        print("=== port %d" % port)
+        if args.raw:
+            print(json.dumps(result.get("reply"), indent=2, default=str))
+            return 0
+        value = result["value"]
+        print("==> soc: %s%% (source=%s, cmd=%s)"
+              % (value, result["source"], result["cmd"])
+              if value is not None else "==> soc: unknown")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/host_probe8.py

@@ -0,0 +1,62 @@
+#!/usr/bin/env python3
+"""Host probe v8 — household wattage via the business operations API.
+
+One call per port: ``LocalDeviceClient.get_household_wattage`` (realtime
+detail → per-switch load → EMS math). Prints the watts.
+
+Run from the repo root on the SoftAP-joined host:
+    python3 scripts/host_probe8.py --sn A03X2516013210060005 --ports 9000
+"""
+
+import argparse
+import json
+import os
+import socket
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(
+    os.path.abspath(__file__))))
+
+from franklinwh_agate_client.device_local import (  # noqa: E402
+    LocalDeviceClient,
+)
+from franklinwh_agate_client.protocol import (  # noqa: E402
+    generate_secret_key,
+)
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--sn", required=True)
+    ap.add_argument("--host", default="10.100.1.1")
+    ap.add_argument("--ports", default="9000")
+    ap.add_argument("--timeout", type=float, default=5.0)
+    ap.add_argument("--raw", action="store_true",
+                    help="print the full reply envelope")
+    args = ap.parse_args()
+    print("secret key for %s: %s (radix 16)"
+          % (args.sn, generate_secret_key(args.sn)))
+    for port in [int(p) for p in args.ports.split(",") if p.strip()]:
+        client = LocalDeviceClient(args.sn, host=args.host,
+                                   socket_port=port, timeout=args.timeout)
+        try:
+            result = client.get_household_wattage()
+        except (ConnectionError, socket.error, TimeoutError) as exc:
+            print("=== port %d: error: %s: %s"
+                  % (port, type(exc).__name__, exc))
+            continue
+        print("=== port %d" % port)
+        if args.raw:
+            print(json.dumps(result.get("reply"), indent=2, default=str))
+            return 0
+        value = result["value"]
+        if value is not None:
+            print("==> watts: %gW (source=%s, cmd=%s)"
+                  % (value, result["source"], result["cmd"]))
+        else:
+            print("==> watts: unknown")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_battery.py

@@ -37,3 +37,39 @@ class TestExtractSoc(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+class TestBmsSoc(unittest.TestCase):
+    def test_batsoc_key(self):
+        from franklinwh_agate_client.battery import SOC_KEYS
+        self.assertIn("batSoc", SOC_KEYS)
+
+    def test_batsoc_extracted(self):
+        reply = {"cmdType": 1706,
+                 "dataArea": {"opt": 0, "result": 0, "id": 1, "batSoc": 87,
+                              "batSoh": 100}}
+        out = extract_soc(reply)
+        self.assertEqual(out["value"], 87)
+        self.assertEqual(out["source"], "batSoc")
+
+    def test_bms_query_carries_pack_id(self):
+        from franklinwh_agate_client.protocol import (
+            CommunicationCmd, direct_connect_request)
+        env = direct_connect_request(CommunicationCmd.BMS, "SN",
+                                     data={"id": 2})
+        self.assertEqual(env.to_dict()["dataArea"], {"opt": 0, "id": 2})
+        self.assertEqual(env.cmd_type, 1705)
+
+    def test_empty_pack_slot_skipped(self):
+        from franklinwh_agate_client.battery import _is_empty_pack
+        reply = {"dataArea": {"opt": 0, "result": 0, "id": 0, "batSoc": 0,
+                              "batSoh": 0, "batTotalVolt": 0}}
+        self.assertTrue(_is_empty_pack(
+            reply, {"value": 0.0, "source": "batSoc"}))
+
+    def test_genuine_zero_pack_kept(self):
+        from franklinwh_agate_client.battery import _is_empty_pack
+        reply = {"dataArea": {"opt": 0, "result": 0, "id": 1, "batSoc": 0,
+                              "batSoh": 96, "batTotalVolt": 512}}
+        self.assertFalse(_is_empty_pack(
+            reply, {"value": 0.0, "source": "batSoc"}))

tests/test_operations.py

@@ -0,0 +1,133 @@
+"""Offline tests for the business operations (transport stubbed)."""
+
+import unittest
+
+from franklinwh_agate_client.device_local import LocalDeviceClient
+from franklinwh_agate_client.protocol import CommunicationCmd
+
+
+def _client(handler):
+    client = LocalDeviceClient("SN", host="127.0.0.1", timeout=0.1)
+    calls = []
+
+    def fake_send_rpc(cmd, equip_no=None, operate=None, data=None,
+                      envelope=None):
+        calls.append((cmd, equip_no, data))
+        return handler(cmd, equip_no, data)
+
+    client.send_rpc = fake_send_rpc
+    return client, calls
+
+
+class TestBatterySocOp(unittest.TestCase):
+    def test_apower_win(self):
+        client, calls = _client(
+            lambda cmd, eq, data: {"dataArea": {"currentBatterySoc": 78}})
+        out = client.get_battery_soc()
+        self.assertEqual(out["value"], 78)
+        self.assertEqual(out["source"], "currentBatterySoc")
+
+    def test_bms_pack_fallback(self):
+        def handler(cmd, eq, data):
+            if cmd == CommunicationCmd.DEVICE_APOWER_INFO:
+                return {"dataArea": {"result": 1, "reason": -1}}
+            if (data or {}).get("id") == 1:
+                return {"dataArea": {"opt": 0, "result": 0, "id": 1,
+                                     "batSoc": 64, "batSoh": 99,
+                                     "batTotalVolt": 512}}
+            return {"dataArea": {"opt": 0, "result": 0, "id": 0,
+                                 "batSoc": 0, "batSoh": 0,
+                                 "batTotalVolt": 0}}
+
+        client, calls = _client(handler)
+        out = client.get_battery_soc()
+        self.assertEqual(out["value"], 64)
+        self.assertEqual(out["source"], "batSoc")
+        # discovery + apower + bms id=1 (empty id=0 never needed)
+        ids = [d.get("id") if isinstance(d, dict) else None
+               for _, _, d in calls]
+        self.assertIn(1, ids)
+
+    def test_empty_slots_skipped(self):
+        client, _ = _client(
+            lambda cmd, eq, data: {"dataArea": {"opt": 0, "result": 0,
+                                                "id": 0, "batSoc": 0,
+                                                "batSoh": 0,
+                                                "batTotalVolt": 0}})
+        out = client.get_battery_soc()
+        self.assertIsNone(out["value"])
+
+    def test_transport_error_propagates(self):
+        client, _ = _client(None)
+
+        def boom(cmd, equip_no=None, operate=None, data=None,
+                 envelope=None):
+            raise ConnectionError("down")
+
+        client.send_rpc = boom
+        with self.assertRaises(ConnectionError):
+            client.get_battery_soc()
+
+
+class TestWattageOp(unittest.TestCase):
+    def test_detail_win(self):
+        client, _ = _client(
+            lambda cmd, eq, data: {"dataArea": {"apowerCount": 2,
+                                               "totalPower": 1872.5}})
+        out = client.get_household_wattage()
+        self.assertEqual(out["value"], 1872.5)
+        self.assertEqual(out["source"], "totalPower")
+
+    def test_ems_math_fallback(self):
+        def handler(cmd, eq, data):
+            if cmd == CommunicationCmd.EMS:
+                return {"dataArea": {"gridVol1": 120.0, "gridVol2": 120.0,
+                                     "loadCurr1": 5.0, "loadCurr2": 5.0}}
+            return {"dataArea": {"result": 1, "reason": -1}}
+
+        client, _ = _client(handler)
+        out = client.get_household_wattage()
+        self.assertAlmostEqual(out["value"], 1200.0)
+        self.assertEqual(out["source"], "ems(loadCurr*V)")
+
+    def test_inventory(self):
+        client, _ = _client(
+            lambda cmd, eq, data: {
+                "cmdType": 1102, "equipNo": "IBG",
+                "dataArea": {"opt": 0, "result": 0, "FHP_SN": ["A", "B"],
+                             "BMS_SN": ["C"], "protocolVer": "V1"}})
+        out = client.get_device_inventory()
+        self.assertEqual(out["ibg_sn"], "IBG")
+        self.assertEqual(out["fhp_sn"], ["A", "B"])
+        self.assertEqual(out["protocol_ver"], "V1")
+
+
+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)

tests/test_power.py

@@ -0,0 +1,44 @@
+"""Offline tests for live-power extraction (no radio/hardware needed)."""
+
+import unittest
+
+from franklinwh_agate_client.power import (
+    ems_load_watts,
+    extract_power,
+)
+
+
+class TestExtractPower(unittest.TestCase):
+    def test_total_power_wins(self):
+        reply = {"cmdType": 1301,
+                 "dataArea": {"apowerCount": 2, "totalPower": 1872.5}}
+        out = extract_power(reply)
+        self.assertEqual(out["value"], 1872.5)
+        self.assertEqual(out["source"], "totalPower")
+
+    def test_switch_power_fallback(self):
+        reply = {"dataArea": {"index": 1, "Sw1Volt": 120.1,
+                              "SW1Curr": 3.2, "SW1ExpPower": 384.0}}
+        out = extract_power(reply)
+        self.assertEqual(out["value"], 384.0)
+        self.assertEqual(out["source"], "SW1ExpPower")
+
+    def test_missing_power(self):
+        out = extract_power({"dataArea": {"result": 1, "reason": -1}})
+        self.assertIsNone(out["value"])
+        self.assertIsNone(out["source"])
+
+    def test_ems_load_watts_math(self):
+        reply = {"dataArea": {"gridVol1": 120.4, "gridVol2": 120.5,
+                              "loadCurr1": 7.8, "loadCurr2": 7.8,
+                              "gridLineVol": 2410}}
+        watts = ems_load_watts(reply)
+        self.assertAlmostEqual(watts, 15.6 * 120.45, places=1)
+
+    def test_ems_load_watts_missing(self):
+        self.assertIsNone(ems_load_watts({"dataArea": {}}))
+        self.assertIsNone(ems_load_watts({}))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_protocol.py

@@ -258,3 +258,35 @@ class TestLiveCaptures(unittest.TestCase):
             split_wire(b'{"cmdType":1101')
         with self.assertRaises(ValueError):
             decrypt_wire(b"not json at all")
+
+
+class TestEnvelopeCodec(unittest.TestCase):
+    def test_for_serial_roundtrip(self):
+        from franklinwh_agate_client.protocol.messages import EnvelopeCodec
+        codec = EnvelopeCodec.for_serial(SN)
+        self.assertEqual(codec.key, generate_secret_key(SN))
+        env = direct_connect_request(CommunicationCmd.BMS, SN,
+                                     data={"id": 1})
+        wire = codec.encode(env)
+        back = codec.decode(wire)
+        self.assertEqual(back.to_dict(), env.to_dict())
+
+    def test_decode_fixtures(self):
+        from franklinwh_agate_client.protocol.messages import EnvelopeCodec
+        disc = open(os.path.join(FIXTURES, "discovery_req.bin"),
+                    "rb").read()
+        info = open(os.path.join(FIXTURES, "device_info_reply.bin"),
+                    "rb").read()
+        self.assertEqual(
+            EnvelopeCodec.for_serial("").decode(disc).cmd_type, 1101)
+        reply = EnvelopeCodec.for_serial(SN).decode(info)
+        self.assertEqual(reply.cmd_type, 1102)
+        self.assertEqual(reply.to_dict()["dataArea"]["FHP_NUM"], 2)
+
+    def test_try_decode_none(self):
+        from franklinwh_agate_client.protocol.messages import EnvelopeCodec
+        codec = EnvelopeCodec.for_serial(SN)
+        self.assertIsNone(codec.try_decode(b""))
+        self.assertIsNone(codec.try_decode(b"not,a wire"))
+        with self.assertRaises(ValueError):
+            codec.decode(b"not,a wire")