From 363c7cc6a277a2fac3b9d534ef1f464999d8adff Mon Sep 17 00:00:00 2001 From: kfj001 Date: Fri, 16 Oct 2026 04:44:45 +0000 Subject: [PATCH] EnvelopeCodec message layer; scripts are thin business-API wrappers --- docs/BATTERY_POC.md | 10 +- docs/RE_DART_FINDINGS.md | 10 +- franklinwh_agate_client/battery.py | 25 +-- franklinwh_agate_client/device_local.py | 56 ++--- franklinwh_agate_client/protocol/__init__.py | 2 + franklinwh_agate_client/protocol/messages.py | 64 ++++++ scripts/host_probe2.py | 187 ++++++++-------- scripts/host_probe3.py | 171 +++++---------- scripts/host_probe4.py | 172 +++++---------- scripts/host_probe5.py | 199 ++++++++--------- scripts/host_probe6.py | 218 ++++--------------- scripts/host_probe7.py | 165 +++----------- scripts/host_probe8.py | 140 ++++-------- tests/test_protocol.py | 32 +++ 14 files changed, 551 insertions(+), 900 deletions(-) create mode 100644 franklinwh_agate_client/protocol/messages.py diff --git a/docs/BATTERY_POC.md b/docs/BATTERY_POC.md index abb4b5d..6be3809 100644 --- a/docs/BATTERY_POC.md +++ b/docs/BATTERY_POC.md @@ -69,10 +69,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 diff --git a/docs/RE_DART_FINDINGS.md b/docs/RE_DART_FINDINGS.md index ee68dfd..fd63aff 100644 --- a/docs/RE_DART_FINDINGS.md +++ b/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 diff --git a/franklinwh_agate_client/battery.py b/franklinwh_agate_client/battery.py index 0150fc5..3febd62 100644 --- a/franklinwh_agate_client/battery.py +++ b/franklinwh_agate_client/battery.py @@ -258,12 +258,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: @@ -272,9 +269,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 @@ -297,7 +293,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_wire + from .protocol.messages import EnvelopeCodec results: List[Dict[str, Any]] = [] for port in ports: @@ -330,15 +326,8 @@ def probe_transport( except (ConnectionError, socket.error, TimeoutError) as exc: entry["error"] = f"{type(exc).__name__}: {exc}" if entry["rx"]: - try: - decoded = decrypt_wire(entry["rx"], radix=key_radix) - except ValueError: - decoded = "" - 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) diff --git a/franklinwh_agate_client/device_local.py b/franklinwh_agate_client/device_local.py index 38e74a1..f3d2657 100644 --- a/franklinwh_agate_client/device_local.py +++ b/franklinwh_agate_client/device_local.py @@ -19,7 +19,6 @@ the direct-connect path of the app (see ``docs/RE_DART_FINDINGS.md`` §8, from __future__ import annotations -import json import logging import socket from typing import Any, Dict, Optional @@ -32,12 +31,11 @@ from .protocol import ( CommunicationOperate, SequenceNumber, SocketFrameDecoder, - 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__) @@ -96,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( @@ -196,57 +195,60 @@ class LocalDeviceClient: 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, ) -> 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(), ) - 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: @@ -278,7 +280,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}" diff --git a/franklinwh_agate_client/protocol/__init__.py b/franklinwh_agate_client/protocol/__init__.py index 48bddf5..a1f7fd9 100644 --- a/franklinwh_agate_client/protocol/__init__.py +++ b/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", ] diff --git a/franklinwh_agate_client/protocol/messages.py b/franklinwh_agate_client/protocol/messages.py new file mode 100644 index 0000000..e2a065c --- /dev/null +++ b/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 diff --git a/scripts/host_probe2.py b/scripts/host_probe2.py index 92c48aa..8a81840 100644 --- a/scripts/host_probe2.py +++ b/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__": diff --git a/scripts/host_probe3.py b/scripts/host_probe3.py index 2ffc791..7415fcf 100644 --- a/scripts/host_probe3.py +++ b/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__": diff --git a/scripts/host_probe4.py b/scripts/host_probe4.py index 072a6ad..bcbff7c 100644 --- a/scripts/host_probe4.py +++ b/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__": diff --git a/scripts/host_probe5.py b/scripts/host_probe5.py index 22c4c06..fa81cba 100644 --- a/scripts/host_probe5.py +++ b/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__": diff --git a/scripts/host_probe6.py b/scripts/host_probe6.py index 2f5ff26..f8c8fd7 100644 --- a/scripts/host_probe6.py +++ b/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()) diff --git a/scripts/host_probe7.py b/scripts/host_probe7.py index 44d53f4..4cd97e2 100644 --- a/scripts/host_probe7.py +++ b/scripts/host_probe7.py @@ -1,136 +1,43 @@ #!/usr/bin/env python3 -"""Host probe v7 — per-pack BMS reads (stdlib only). +"""Host probe v7 — per-pack BMS reads via the library business API. 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 +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 time -import zlib +import sys -APOWER, QUERY_INFO, BMS, EMS = 0x51B, 0x44D, 0x6A9, 0x6AB -SNNO = [0] -SOC_KEYS = ("currentBatterySoc", "soc", "batSoc", "batterySoc", "vppSoc") +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): - 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:] +def run(label, client, call): 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()} + reply = call() 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) + print("--- %s: error: %s: %s" % (label, type(exc).__name__, exc)) + return None + print("--- %s (%d B reply)" % (label, len(json.dumps(reply)))) + print(" reply: " + json.dumps(reply, default=str)[:1500]) + result = extract_soc(reply) + print(" ==> soc: %s (source=%s)" % (result["value"], result["source"])) return reply @@ -141,21 +48,17 @@ 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 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), - ] + 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) - run(label, args.host, port, tx, args.timeout) + print("=== port %d" % port) + client = LocalDeviceClient(args.sn, host=args.host, + socket_port=port, timeout=args.timeout) + run("bms/id1", client, lambda: client.query_bms_info(1)) + run("bms/id2", client, lambda: client.query_bms_info(2)) + run("apower/update2", client, lambda: client.send_rpc( + CommunicationCmd.DEVICE_APOWER_INFO, data={"update": 2})) + run("ems", client, client.query_ems_info) if __name__ == "__main__": diff --git a/scripts/host_probe8.py b/scripts/host_probe8.py index 5b4f86b..5b76c57 100644 --- a/scripts/host_probe8.py +++ b/scripts/host_probe8.py @@ -1,109 +1,49 @@ #!/usr/bin/env python3 -"""Host probe v8 — live household wattage (stdlib only). +"""Host probe v8 — live household wattage via the library business API. -Three socket APIs (all verified in disassembly; detail/load pending live -confirmation on this unit, EMS already answered live): -* DEVICE_DETAIL_INFO (1301, no args) -> totalPower [home-page poll] -* LOAD_STATUS (1411, no args) -> SW1ExpPower [per-switch watts] -* EMS (1707, no args) -> grid/load V+A [watts via math] +Three socket APIs: realtime detail (home-page poll) -> ``totalPower``, +per-switch load status -> ``SWxExpPower``, EMS telemetry -> watts math. -Run on the SoftAP-joined host: - python3 host_probe8.py --sn A03X2516013210060005 --ports 9000 +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 time -import zlib +import sys -DETAIL, LOAD, EMS = 0x515, 0x583, 0x6AB -SNNO = [0] +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.power import ( # noqa: E402 + ems_load_watts, + extract_power, +) +from franklinwh_agate_client.protocol import ( # noqa: E402 + 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): - 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:] +def run(label, client, call): 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 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()} + reply = call() 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: - print(" reply: " + json.dumps(reply, default=str)[:2000]) - if err: - print(" error: %s" % err) + print("--- %s: error: %s: %s" % (label, type(exc).__name__, exc)) + return None + print("--- %s" % label) + print(" reply: " + json.dumps(reply, default=str)[:2000]) + result = extract_power(reply) + value, source = result["value"], result["source"] + if value is None: + watts = ems_load_watts(reply) + if watts is not None: + value, source = watts, "ems(loadCurr*V)" + print(" ==> watts: %s (source=%s)" % (value, source)) return reply @@ -114,13 +54,15 @@ 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 16)" % (args.sn, key)) + 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 in (("detail/1301", DETAIL), ("load/1411", LOAD), - ("ems/1707", EMS)): - tx = encrypt(build_envelope(args.sn, cmd), key) - run(label, args.host, port, tx, args.timeout) + print("=== port %d" % port) + client = LocalDeviceClient(args.sn, host=args.host, + socket_port=port, timeout=args.timeout) + run("detail/1301", client, client.query_detail_info) + run("load/1411", client, client.query_load_status) + run("ems/1707", client, client.query_ems_info) if __name__ == "__main__": diff --git a/tests/test_protocol.py b/tests/test_protocol.py index 7549a0b..b408fe7 100644 --- a/tests/test_protocol.py +++ b/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")