Add SoftAP battery PoC (apower/BMS/EMS socket read) with docs and tests

Probes 10.100.1.1:9000,18000, sends 0x19/0x1D/0x1E envelopes, extracts
currentBatterySoc/soc/avg(socArray). Includes offline extract_soc tests,
BATTERY_POC docs, README usage, and empirical port-9000 RE notes.

README.md

@@ -11,6 +11,8 @@ Python client library for interacting with FranklinWH AGATE energy storage devic
 - **[Dart AOT Reverse-Engineering Findings](docs/RE_DART_FINDINGS.md)** — detailed protocol
   analysis recovered from the app's `libapp.so` with [blutter](https://github.com/worawit/blutter)
   (transports, wire envelope, command/topic enums, crypto/CRC, BLE framing, hotspot credentials).
+- **[Battery PoC](docs/BATTERY_POC.md)** — read charge % over the SoftAP socket
+  (`python -m franklinwh_agate_client.battery --sn <SN>`).
 
 ## Reverse Engineering Summary
 
@@ -51,6 +53,18 @@ client = AgateClient()
 device_info = client.query_agate_info(device_id)
 ```
 
+## Battery PoC (SoftAP socket)
+
+```bash
+python -m franklinwh_agate_client.battery --sn <20-char-SN> --show-hotspot
+python -m franklinwh_agate_client.battery --sn <20-char-SN>          # e.g. 78%
+python -m franklinwh_agate_client.battery --sn <20-char-SN> --raw
+```
+
+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).
+
 ## Local connect (device hotspot)
 
 The aGate runs its own SoftAP (gateway `10.100.1.1`); joining it is what

docs/BATTERY_POC.md

@@ -0,0 +1,58 @@
+# Battery PoC — direct-connect charge state
+
+Reads the aggregate battery percentage straight from the aGate over its own
+SoftAP. No cloud, no MQTT, no Bluetooth. Recovered from the Dart AOT layer;
+see `RE_DART_FINDINGS.md` §§3–5, 8, 10.
+
+## Requirements
+
+- Host whose Wi-Fi adapter is associated to the device AP
+  (`AP_<sn[11:]>` / `<sn[8:]>`, `wifi.derive_hotspot_credentials`).
+- L2 reachability to `10.100.1.1`. In a container this means a bridged
+  `macvlan`/`ipvlan` address on `10.100.1.0/24` or `network_mode: host`
+  (see repo-root `.devcontainer/docker-compose.yml`). A default Docker
+  bridge (`172.17.0.0/16`) is NATed to a non-`10.100.1.x` source and gets
+  `ECONNREFUSED`.
+- Python `requests` for the HTTP helpers in `device_local.py`.
+
+## What it does
+
+1. Probes TCP ports `9000, 18000` in order (`9000` was open on the tested
+   unit; `0x4650 = 18000` is the app constant but was closed there).
+2. Sends `DEVICE_APOWER_INFO (0x19)`, then `BMS (0x1D)`, `EMS (0x1E)`:
+   compact-JSON `CommunicationObject` → `encrypt_object` (key
+   `hex(sum(sn) & 0xFF)`) → int16LE length framing.
+3. Decrypts the reply and extracts `currentBatterySoc` → `soc` →
+   `avg(socArray)`.
+
+## Usage
+
+```bash
+# show derived hotspot creds only
+python -m franklinwh_agate_client.battery --sn <20-char-SN> --show-hotspot
+
+# print just the percentage
+python -m franklinwh_agate_client.battery --sn <20-char-SN>
+
+# full reply envelope
+python -m franklinwh_agate_client.battery --sn <20-char-SN> --raw
+```
+
+Options: `--host` (default `10.100.1.1`), `--ports 9000,18000`,
+`--key-radix {16,32}` (default 16; 32 is the alternate candidate in the
+disassembly), `--timeout`.
+
+Programmatic:
+
+```python
+from franklinwh_agate_client import get_battery
+out = get_battery("...", host="10.100.1.1")
+print(out["value"], out["source"])  # e.g. 78.0 currentBatterySoc
+```
+
+## Limits
+
+- No live round-trip has succeeded yet from an isolated container netns
+  (all probes `ECONNREFUSED`); first success needs a `10.100.1.x` source.
+- `key_radix`, BLE field order, and `X-Device-*` binding remain open
+  (`RE_DART_FINDINGS.md` Open items).

docs/RE_DART_FINDINGS.md

@@ -193,6 +193,25 @@ authS2C=2    "fwh/s2c/auth/"  mngC2S=5   "fwh/c2s/mng/"     deviceS2C  "fwh/s2c/
   - a `Duration` and retry loop (up to ~2 retries) are involved.
 - `BytesToInt16/Int16ToBytes` LE framing; `Socket Manager send object` / `receive object`.
 
+### Empirical — live aGate on its SoftAP (2026-10-13, from a client on `10.100.1.0/24`)
+
+Full TCP sweep of `10.100.1.1` (ports 1–65535): exactly **two** ports open — **53** and **9000**. No 80/443, no web UI, and **18000 (`0x4650`) is closed** on this unit.
+
+- **Port 53** — authoritative, *closed* DNS: well-formed `NXDOMAIN` for everything
+  (even `example.com`), no PTR for `1.1.0.100.in-addr.arpa`, no forwarding (the SoftAP
+  is an island). Serves some local name pattern we haven't identified yet.
+- **Port 9000** — a **length-prefixed framing service** (int16LE length), consistent
+  with the app's socket layer:
+  - every TLS `ClientHello` gets `UNEXPECTED_EOF` mid-handshake — first two bytes
+    `16 03` = int16LE `790`, so the decoder waits for a 790-byte frame that never
+    completes, then closes on idle;
+  - cleartext HTTP `GET` (`47 45` = `17735`) hangs forever (read timeout).
+  - A *complete* 147-byte frame (int16LE length + encrypted `QUERY_DEVICE_INFO`
+    envelope, `equipNo = X25160132`, key from `sum(utf8(sn))&0xFF`) is accepted
+    but met with **zero bytes** — every command tried (`0x00`/`0x01`/`0x02`/`0x06`)
+    and both key radices (16/32). Content-level rejection; the leading hypothesis is
+    `equipNo` validation against the real serial. **Pending: retry with the true SN.**
+
 ## 9. Bluetooth (BLE) transport — `communication_center/bluetooth`
 
 - `BluetoothManager` (singleton), `BluetoothStatus`: `closed / connected / connecting / interrupted`.
@@ -323,9 +342,15 @@ Still open:
 
 1. Exact BLE frame field order & the 2-byte magic semantics (confirm with `btsnoop`).
 2. Exact `encryptObject` radix + trailing-header handling.
-3. Local HTTP port + TLS behavior for `https://IP:PORT` (device self-signed cert?).
-4. Local socket port confirmation (`18000` candidate) and whether local RPC uses
-   socket vs `/terminal` HTTP vs local MQTT.
+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~~ — **resolved on one unit (2026-10-13): `9000`**
+   (§8 empirical). The `0x4650` constant is not listening; the port may be
+   firmware-/device-dependent. Remaining: first confirmed RPC round-trip
+   (blocked on the real SN — see §8 empirical).
 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/__init__.py

@@ -33,7 +33,20 @@ __all__ = [
     "HotspotProfile",
     "HotspotJoinError",
     "derive_hotspot_credentials",
+    "extract_soc",
+    "get_battery",
+    "SOCKET_PORT_CANDIDATES",
     "DEVICE_GATEWAY_IP",
     "FALLBACK_PASSWORD",
     "VALID_SN_LENGTH",
 ]
+
+
+def __getattr__(name: str):
+    # Lazy so `python -m franklinwh_agate_client.battery` does not import
+    # the submodule twice (avoids a runpy RuntimeWarning).
+    if name in {"extract_soc", "get_battery", "SOCKET_PORT_CANDIDATES"}:
+        from . import battery as _battery
+
+        return getattr(_battery, name)
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

franklinwh_agate_client/battery.py

@@ -0,0 +1,207 @@
+"""Battery charge-state proof of concept.
+
+Direct-connect (SoftAP) read path recovered from the app's Dart AOT layer
+(see ``docs/RE_DART_FINDINGS.md``): join the device hotspot, open the local
+TCP socket, send the encrypted ``CommunicationObject`` envelope, decrypt the
+reply, and pull the aggregate state-of-charge out of it.
+
+No cloud, no MQTT, no Bluetooth needed. Requires L2 reachability to the
+aGate gateway (``10.100.1.1``) — i.e. run on the host whose Wi-Fi adapter is
+associated to the device AP, or in a container bridged onto ``10.100.1.0/24``.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import socket
+from typing import Any, Dict, List, Optional, Tuple
+
+from .device_local import LocalDeviceClient
+from .protocol import CommunicationCmd
+from .wifi import derive_hotspot_credentials
+
+log = logging.getLogger(__name__)
+
+#: Ports probed in order. ``9000`` was observed open on one live unit
+#: (``docs/RE_DART_FINDINGS.md`` §8); ``18000`` (``0x4650``) is the constant
+#: in the app's connect path but was closed on that unit.
+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")
+
+
+def _as_dict(data: Any) -> Optional[Dict[str, Any]]:
+    if isinstance(data, dict):
+        return data
+    if isinstance(data, str):
+        try:
+            parsed = json.loads(data)
+        except (ValueError, TypeError):
+            return None
+        return parsed if isinstance(parsed, dict) else None
+    return None
+
+
+def _walk(obj: Any, key: str) -> List[Any]:
+    """Collect every value stored under ``key`` anywhere in nested dicts/lists."""
+    found: List[Any] = []
+    if isinstance(obj, dict):
+        for k, v in obj.items():
+            if k == key:
+                found.append(v)
+            found.extend(_walk(v, key))
+    elif isinstance(obj, list):
+        for item in obj:
+            found.extend(_walk(item, key))
+    return found
+
+
+def _to_float(value: Any) -> Optional[float]:
+    try:
+        if isinstance(value, bool):
+            return None
+        num = float(value)
+    except (TypeError, ValueError):
+        return None
+    return num
+
+
+def extract_soc(reply: Dict[str, Any]) -> Dict[str, Any]:
+    """Pull the aggregate charge percentage out of a parsed reply envelope.
+
+    Returns ``{"value": float|None, "source": str|None, "detail": {...}}``.
+    ``source`` names the key the value came from; ``detail`` carries the
+    supporting fields (``socArray``, ``batteryCount``, charge/discharge
+    power) when present.
+    """
+    data = _as_dict(reply.get("data", reply)) or {}
+
+    for key in SOC_KEYS:
+        for container in (data, reply):
+            for hit in _walk(container, key):
+                num = _to_float(hit)
+                if num is not None:
+                    return {
+                        "value": num,
+                        "source": key,
+                        "detail": _soc_detail(data),
+                    }
+    soc_arrays = _walk(data, "socArray")
+    for arr in soc_arrays:
+        if isinstance(arr, list) and arr:
+            nums = [_to_float(v) for v in arr]
+            nums = [n for n in nums if n is not None]
+            if nums:
+                return {
+                    "value": sum(nums) / len(nums),
+                    "source": "socArray(avg)",
+                    "detail": _soc_detail(data),
+                }
+    return {"value": None, "source": None, "detail": _soc_detail(data)}
+
+
+def _soc_detail(data: Dict[str, Any]) -> Dict[str, Any]:
+    detail: Dict[str, Any] = {}
+    for key in (
+        "socArray",
+        "batteryCount",
+        "batteryList",
+        "chargePower",
+        "dischargePower",
+        "currentBatterySoc",
+        "soc",
+    ):
+        hits = _walk(data, key)
+        if hits:
+            detail[key] = hits[0] if len(hits) == 1 else hits
+    # Always record where we looked so failures are debuggable.
+    return detail
+
+
+def get_battery(
+    sn: str,
+    host: str = "10.100.1.1",
+    ports: Tuple[int, ...] = SOCKET_PORT_CANDIDATES,
+    key_radix: int = 16,
+    timeout: float = 5.0,
+) -> Dict[str, Any]:
+    """Query the aggregate charge percentage over the local TCP socket.
+
+    Tries ``DEVICE_APOWER_INFO`` (0x19) first, then ``BMS`` (0x1D) and
+    ``EMS`` (0x1E) as fallbacks, probing each candidate port. 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:
+        client = LocalDeviceClient(
+            sn=sn, host=host, socket_port=port,
+            key_radix=key_radix, timeout=timeout,
+        )
+        for cmd in (
+            CommunicationCmd.DEVICE_APOWER_INFO,
+            CommunicationCmd.BMS,
+            CommunicationCmd.EMS,
+        ):
+            try:
+                reply = client.send_rpc(cmd)
+            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})
+            if result["value"] is not None:
+                return result
+            last_err = ValueError(f"reply on port {port} cmd {cmd} has no SoC field")
+    raise ConnectionError(f"no battery reply via {host}:{list(ports)}") from last_err
+
+
+def main(argv: Optional[List[str]] = None) -> int:
+    parser = argparse.ArgumentParser(
+        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(
+        "--ports", default=",".join(map(str, SOCKET_PORT_CANDIDATES)),
+        help="comma-separated socket ports to probe",
+    )
+    parser.add_argument("--key-radix", type=int, default=16, choices=(16, 32))
+    parser.add_argument("--timeout", type=float, default=5.0)
+    parser.add_argument("--show-hotspot", action="store_true",
+                        help="print derived AP SSID/password and exit")
+    parser.add_argument("--raw", action="store_true", help="print full reply JSON")
+    args = parser.parse_args(argv)
+
+    if args.show_hotspot:
+        profile = derive_hotspot_credentials(args.sn)
+        print(json.dumps(
+            {"ssid": profile.ssid, "password": profile.password,
+             "gateway_ip": profile.gateway_ip}, indent=2))
+        return 0
+
+    ports = tuple(int(p) for p in args.ports.split(",") if p.strip())
+    try:
+        result = get_battery(args.sn, host=args.host, ports=ports,
+                             key_radix=args.key_radix, 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}%" if value is not None else "unknown")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_battery.py

@@ -0,0 +1,39 @@
+"""Offline tests for battery SoC extraction (no radio/hardware needed)."""
+
+import unittest
+
+from franklinwh_agate_client.battery import extract_soc
+
+
+class TestExtractSoc(unittest.TestCase):
+    def test_current_battery_soc_wins(self):
+        reply = {"cmdType": 25, "data": {"currentBatterySoc": 78, "soc": 10}}
+        out = extract_soc(reply)
+        self.assertEqual(out["value"], 78)
+        self.assertEqual(out["source"], "currentBatterySoc")
+
+    def test_soc_fallback(self):
+        out = extract_soc({"data": {"soc": "55"}})
+        self.assertEqual(out["value"], 55)
+        self.assertEqual(out["source"], "soc")
+
+    def test_soc_array_averaged(self):
+        out = extract_soc({"data": {"socArray": [80, 90]}})
+        self.assertEqual(out["value"], 85)
+        self.assertEqual(out["source"], "socArray(avg)")
+
+    def test_nested_data_string(self):
+        import json
+
+        reply = {"data": json.dumps({"aPower": {"soc": 62}})}
+        out = extract_soc(reply)
+        self.assertEqual(out["value"], 62)
+
+    def test_missing_soc(self):
+        out = extract_soc({"data": {"foo": 1}})
+        self.assertIsNone(out["value"])
+        self.assertIsNone(out["source"])
+
+
+if __name__ == "__main__":
+    unittest.main()