From 3d59a75f143725f83fe3a4dd75bb7dd6f71c30ee Mon Sep 17 00:00:00 2001 From: kfj001 Date: Fri, 16 Oct 2026 04:08:54 +0000 Subject: [PATCH] Per-pack BMS queries (id arg): batSoc source + host_probe7 --- docs/RE_DART_FINDINGS.md | 11 ++ franklinwh_agate_client/battery.py | 31 +++-- franklinwh_agate_client/device_local.py | 10 +- scripts/host_probe7.py | 162 ++++++++++++++++++++++++ tests/test_battery.py | 22 ++++ 5 files changed, 223 insertions(+), 13 deletions(-) create mode 100644 scripts/host_probe7.py diff --git a/docs/RE_DART_FINDINGS.md b/docs/RE_DART_FINDINGS.md index bed80ec..453ab66 100644 --- a/docs/RE_DART_FINDINGS.md +++ b/docs/RE_DART_FINDINGS.md @@ -306,6 +306,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: diff --git a/franklinwh_agate_client/battery.py b/franklinwh_agate_client/battery.py index 68d7d98..5958ed0 100644 --- a/franklinwh_agate_client/battery.py +++ b/franklinwh_agate_client/battery.py @@ -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]]: @@ -131,12 +135,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: @@ -150,11 +156,13 @@ def get_battery( 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 ( + attempts: Tuple[Tuple[CommunicationCmd, Any], ...] = ( (CommunicationCmd.DEVICE_APOWER_INFO, {"update": 2}), - (CommunicationCmd.BMS, None), + (CommunicationCmd.BMS, {"id": 1}), + (CommunicationCmd.BMS, {"id": 2}), (CommunicationCmd.EMS, None), - ): + ) + for cmd, data in attempts: try: reply = client.send_rpc(cmd, data=data) except (ConnectionError, socket.error, TimeoutError) as exc: @@ -181,7 +189,8 @@ 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), ) diff --git a/franklinwh_agate_client/device_local.py b/franklinwh_agate_client/device_local.py index d1fb23a..512fac6 100644 --- a/franklinwh_agate_client/device_local.py +++ b/franklinwh_agate_client/device_local.py @@ -285,8 +285,14 @@ 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_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) diff --git a/scripts/host_probe7.py b/scripts/host_probe7.py new file mode 100644 index 0000000..44d53f4 --- /dev/null +++ b/scripts/host_probe7.py @@ -0,0 +1,162 @@ +#!/usr/bin/env python3 +"""Host probe v7 — per-pack BMS reads (stdlib only). + +The app's ``queryBmsInfoBySocket`` passes ``arguments={"id": pack}``; the +id-less query answers the all-zero ``id: 0`` slot. Pack ids 1..2 select +the two batteries (``BMS_SN …B1/…B2``); ``batSoc`` is the pack charge. + +Run on the SoftAP-joined host: + python3 host_probe7.py --sn A03X2516013210060005 --ports 9000 +""" + +import argparse +import json +import socket +import time +import zlib + +APOWER, QUERY_INFO, BMS, EMS = 0x51B, 0x44D, 0x6A9, 0x6AB +SNNO = [0] +SOC_KEYS = ("currentBatterySoc", "soc", "batSoc", "batterySoc", "vppSoc") + + +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): + for key in SOC_KEYS: + for hit in walk(reply, key, []): + if isinstance(hit, bool): + continue + try: + return float(hit), key + except (TypeError, ValueError): + continue + return None, None + + +def run(label, host, port, tx, timeout): + rx, err, reply = b"", None, None + try: + with socket.create_connection((host, port), + timeout=timeout) as sock: + 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(" rx (%d B)" % len(rx)) + if reply is not None: + da = reply.get("dataArea", reply) if isinstance(reply, dict) else {} + print(" reply: " + json.dumps(reply, default=str)[:1500]) + if isinstance(reply, dict): + val, src = extract_soc(reply) + print(" ==> soc: %s (source=%s)" % (val, src)) + if err: + print(" error: %s" % err) + return reply + + +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) + args = ap.parse_args() + key = secret_key(args.sn) + print("secret key for %s: %s (radix 16)" % (args.sn, key)) + variants = [ + ("bms/id1", BMS, {"id": 1}, None), + ("bms/id2", BMS, {"id": 2}, None), + ("apower/update2", APOWER, {"update": 2}, None), + ("ems", 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) + run(label, args.host, port, tx, args.timeout) + + +if __name__ == "__main__": + main() diff --git a/tests/test_battery.py b/tests/test_battery.py index 5002861..1206eed 100644 --- a/tests/test_battery.py +++ b/tests/test_battery.py @@ -37,3 +37,25 @@ 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)