Compare commits

...
17 Commits
Author SHA1 Message Date
Your Name bff812605f adds home battery query 2026-10-21 20:53:10 +00:00
kfj001 b60b32830f Optional aGate address on the API (home-LAN direct connect)
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.
2026-10-16 08:18:33 +00:00
kfj001 eda78dd7d1 Home Wi-Fi join API (connectWifi SET + scan/status) with docs 2026-10-16 06:08:09 +00:00
kfj001 3c7e74fd0e Document cloud account binding (§17): deviceBindV2 + token signature 2026-10-16 05:57:31 +00:00
kfj001 b79adc5d54 Business operations API: get_battery_soc/get_household_wattage/get_device_inventory 2026-10-16 04:58:33 +00:00
kfj001 363c7cc6a2 EnvelopeCodec message layer; scripts are thin business-API wrappers 2026-10-16 04:44:45 +00:00
kfj001 f15ae42db7 Live power APIs: detail/load/EMS queries + power module + host_probe8 2026-10-16 04:31:51 +00:00
kfj001 c11188dd0b Sync library with live findings: raw socket docs, wire reply parsing, pack-id BMS, empty-slot guard 2026-10-16 04:18:22 +00:00
kfj001 3d59a75f14 Per-pack BMS queries (id arg): batSoc source + host_probe7 2026-10-16 04:08:54 +00:00
kfj001 62e0fd649b Radix 16 (live-verified), universal reply splitter, golden capture vectors, host_probe6 2026-10-16 02:50:23 +00:00
kfj001 0cd6b34a2e Document transport selection (§15): single-mode queries, BLE provisions Wi-Fi 2026-10-16 01:55:30 +00:00
kfj001 58227608a7 Document direct-connect return flow (§14): manual join to socketQueryDeviceInfo 2026-10-16 01:36:27 +00:00
kfj001 186f58da7d Add host_probe5 (session sequencing + close-vs-timeout) 2026-10-16 01:02:25 +00:00
kfj001 bdc98484a9 Seed socket payloads with opt (communicate factory) + host_probe4 2026-10-16 00:55:50 +00:00
kfj001 213a0518d6 Socket-transport cmd space (off_14 ids) from live capture: 1101/1307/1705/1707 + host_probe3 2026-10-16 00:14:42 +00:00
kfj001 2cdc274fe3 Add standalone host_probe2.py (stdlib-only transport probes) 2026-10-15 23:53:41 +00:00
kfj001 ba4b47185d Add --probe2 transport variants (greeting, length prefixes, MQTT, split send) 2026-10-15 23:52:55 +00:00
28 changed files with 2614 additions and 233 deletions

No files matched your search

+16
View File
@@ -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
+37 -3
View File
@@ -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,6 +84,13 @@ python3 -m franklinwh_agate_client.battery \
--sn A03X2516013210060005 --host 10.100.1.1 --ports 9000,18000 --probe
```
> 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
`{"cmdType":25,"equipNo":"A03X2516013210060005",`) and `rx` bytes.
@@ -81,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).
+226 -29
View File
@@ -106,10 +106,17 @@ Null/empty payload ⇒ `len == 0`, `crc == ""`, no `dataArea` key.
- The `communicate` factory (`0xc9264c`) builds the `data` map from the
caller's `arguments` (e.g. `queryAPowerData` passes `{"update": 2}` with
cmd `deviceApowerInfo`/`0x19`, operate `query`) and calls the ctor with
a named `data:` argument.
a named `data:` argument. On socket mode it **always seeds the map with
`"opt"`** (the operation type — `0` for query) before merging arguments,
so the app's apower query sends
`dataArea = {"opt": 0, "update": 2}` and its device-info query sends
`{"opt": 0, "minProtocolVer": "V1.00.00"}` (empty/non-query args still
yield `{"opt": 0}`). This matches the live capture sizes (§8).
## 4. `CommunicationCmd` (raw → name)
`off_8` values (the enum's first id field):
```
0x00 queryDeviceInfo 0x13 pairPower 0x26 smartLoad2
0x01 networkInfo 0x14 processPairPower 0x27 rebootDevice
@@ -132,6 +139,17 @@ Null/empty payload ⇒ `len == 0`, `crc == ""`, no `dataArea` key.
```
(missing: 0x09, 0x1b, 0x2c, 0x31 — not surfaced as instances.)
**Two id spaces.** `CommunicationCmd::cmd(cmd, mode)` (0xc92894) returns the
*second* id field (`off_14` in the pool) for `socket`/`bluetooth`/`local`
modes and throws `"do not support http communication"` for `http`. The
socket-transport ids live in `SOCKET_CMD_TYPE`
(`protocol/commands.py`) — e.g. `queryDeviceInfo → 1101 (0x44D)`,
`deviceApowerInfo → 1307 (0x51B)`, `bms → 1705`, `ems → 1707`. The `off_8`
values above are **not** used on the socket/BLE wire. Verified live: a
PCAPdroid capture of the working app shows
`{"cmdType":1101,"equipNo":"00000000",…}` with the S1-plaintext-first
framing (§5).
## 5. Encryption — `communication_center/core/encryption.dart`
**`generateSecretKey(String equipNo)`** = hex of `(sum(utf8(equipNo)) & 0xFF)` (via
@@ -143,14 +161,20 @@ unpadded. E.g. SN `A03X2516013210060005` → byte sum `1052 & 0xFF = 28` → `"1
2. `S1 = parts[0] + "," + parts[1] + ","` (header, *with* trailing comma);
`a = UTF-8(S1)`.
3. `S2 = join(",", parts[2..])` (the rest); `b = UTF-8(S2)`.
4. `keyInt = int.tryParse(key, radix: 32)` (literal `0x20` at the call site;
null → 0).
4. `keyInt = int.tryParse(key)` (default radix **16** — two live frames
decrypt cleanly only with radix 16 and then verify byte-exact on
`len`/CRC; an earlier radix-32 reading of a nearby `0x20` literal is
superseded).
5. `out[i] = (b[i] + i + len(a) + keyInt) & 0xFF` for all i (so each tail
byte is shifted by its *absolute* JSON index plus the key).
6. Result = `a ++ out` — the plaintext header goes on the wire **first**,
transformed tail after (`addAll` onto the header copy).
**`decryptData(bytes)`** (single arg, the accumulated receive buffer):
as recovered from the disassembly (two-`0x58`-marker scan, `s + "}"`
equipNo probe, `s + "," + T` return). NOTE (2026-10-16): live socket
replies decrypt with the *request-side* split instead (see below); the
marker model is kept for the unverified Bluetooth path.
1. `i = indexWhere(byte == 0x58)`, `j = indexWhere(byte == 0x58, start: i+1)`;
either missing → return `""`.
2. `s = utf8(wire[0..j])`; `probe = jsonDecode(s + "}")`, null → `""`.
@@ -164,8 +188,23 @@ unpadded. E.g. SN `A03X2516013210060005` → byte sum `1052 & 0xFF = 28` → `"1
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
`{"cmdType":1101,"equipNo":"00000000",` — plaintext `S1` first, exactly as
derived, with the socket-space id (§4). A second frame shows
`{"cmdType":1102,"equipNo":"1006005A03X25160132",` (larger payload,
direction under analysis — see §8). Note the captures were exported as
text, so non-UTF-8 bytes were replaced with U+FFFD and only the ASCII
prefixes survive; a binary `.pcap` export is still wanted.
## 6. Checksums & byte order
@@ -237,22 +276,178 @@ authS2C=2 "fwh/s2c/auth/" mngC2S=5 "fwh/c2s/mng/" deviceS2C "fwh/s2c/
### 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.
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. No `9000` reference exists anywhere in the app (Dart or Java) — but see the 2026-10-15 update below.
- **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).
- Earlier probes are superseded: they used an int16LE length prefix (the
socket layer sends **raw**), the old `transform(b) || a` order (correct
is `a || transform(b)` — §5), a truncated `equipNo`, and default radix
16. All are corrected in the current PoC; see `BATTERY_POC.md` for the
host probe recipe with the true SN.
- **Port 9000** — accepts TCP, reads one frame, answers zero bytes, closes —
for *every* probe (TLS ClientHello, HTTP GET, MQTT CONNECT, exact-spec
envelopes raw/split/length-prefixed, greeting-first reads).
### Update (2026-10-16) — binary capture fully decoded
Raw `.bin` payloads (no text mangling) decrypt completely offline with
radix 16 (`tests/fixtures/*.bin` are golden vectors):
- Discovery request (153 B): `{"cmdType":1101,"equipNo":"00000000",…,
"dataArea":{"opt":0,"minProtocolVer":"V1.00.00"}}`, key `"80"`/n=128.
- Device-info reply (1154 B): `cmdType` **1102**, `equipNo`
`10060005A03X25160132` (the unit's IBG SN — label SN
`A03X2516013210060005` shares its trailing 12 chars), `len` 1024 exact,
CRC exact, `dataArea.result == 0` with the full version inventory
(`FHP_NUM: 2`, two `FHP_SN`, two `BMS_SN`, `protocolVer V1.12.36`).
Encrypted with the **gateway** key (`"1c"`/n=28) while addressed to the
IBG SN — the gateway is the crypto endpoint; sub-unit addressing rides
inside.
Reply rules learned: same `S1 || transform(S2)` framing both directions;
reply id observed as request + 1 (`1101 → 1102`); universal split at the
2nd comma needs no ids up front (`decrypt_wire`). The remaining unknown
is the live endpoint (dst IP:port + direction per capture).
Superseded notes: pre-2026-10-16 probes additionally used the wrong key
radix (32 instead of 16) and mostly omitted the `opt`-seeded payload, so
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:
- The *manual connect* button (`hotspot_guide_widget.dart`) calls
`goToSystemWifiSettings()` → `OpenSystemWifiSetting` method channel →
OS Wi-Fi picker. The app goes to background.
- On return, the page (`viewDidLoad`, plus the provider-state listener in
`direct_connect.dart`) re-evaluates: with a device SN selected it calls
`DirectConnectProvider.socketQueryDeviceInfo()` (also fired from
`onSelectHotspot` once wifi state settles).
- `socketQueryDeviceInfo` first awaits `CommunicationService.directConnect()`,
which is true when `connectedFhpAp()` (current SSID is the FHP AP) **or** a
BLE device is connected — otherwise it aborts silently.
- It then issues `queryDeviceInfo` in the *current* transport mode
(`_queryDeviceInfoFuture`, args `{"minProtocolVer": "V1.00.00"}`, or
`{"installInfo": 2}` for mqtt mode which is HTTP-forwarded instead).
In socket mode `_connectBeforeRequest` dials
`getWifiGatewayIP():18000` (`"10.100.1.1"` fallback) with ~2 retries.
- On the reply it updates the device model
(`updateFromDeviceInfoInitDataEntity`) and continues with
`_queryTrustedDeviceModel` → `_checkConnectedNotTrustedAp`.
- The `communicate` factory always seeds socket payloads with `"opt"`
(operation type, `0` for query), so the on-air discovery frame is shaped
`{"opt": 0, "minProtocolVer": "V1.00.00"}` (§3).
Implication: on the SoftAP the app speaks **only** to the gateway IP,
port **18000**. Port `9000` appears nowhere in the app; `18000` is closed
on the tested unit, so even the app's own socket path cannot work there —
consistent with all-silent probes.
## 15. Transport selection — one mode at a time, BLE provisions Wi-Fi
- `DirectConnectProvider.initSelectedMode`: on the FHP AP → `socket`;
otherwise, if a BLE device is connected (or a previous selection exists)
→ `bluetooth` (+ `initScan`); else `socket`. Every request carries its
transport in `selectedMode` (`field_5b`); `sendObjectOrganized` routes to
exactly one manager. Queries never run on both transports at once.
- Switching AP → socket explicitly tears down BLE first
(`onSelectHotspot`: `disconnectBluetooth` behind the
`switchBluetoothToHotspotTip` dialog).
- `directConnect()` (provider and service variants) is true when on the FHP
AP **or** BLE-connected — the two are alternative routes to one device.
- BLE and Wi-Fi *cooperate* at the commissioning layer, not the query
layer: ESP BLE provisioning (`prov-session/scan/config`, sec0/sec1)
delivers the home Wi-Fi SSID/password to the device **over BLE**; the
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`
@@ -382,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
+163 -28
View File
@@ -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,20 +125,41 @@ 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 = 32,
key_radix: int = 16,
timeout: float = 5.0,
) -> Dict[str, Any]:
"""Query the aggregate charge percentage over the local TCP socket.
Mirrors the app's ``queryAPowerData`` first (``DEVICE_APOWER_INFO`` with
``data={"update": 2}``), then ``BMS`` and ``EMS`` 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.
``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:
@@ -142,26 +167,17 @@ def get_battery(
sn=sn, host=host, socket_port=port,
key_radix=key_radix, timeout=timeout,
)
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
@@ -173,16 +189,17 @@ 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 = 32,
key_radix: int = 16,
timeout: float = 5.0,
) -> List[Dict[str, Any]]:
"""Send every :data:`PROBE_VARIANTS` envelope; return exact TX/RX bytes."""
@@ -206,23 +223,125 @@ def probe_device(
return results
def _recv_all(sock: socket.socket, idle: float = 4.0) -> bytes:
"""Drain a socket until idle timeout or close."""
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 _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
from .protocol.envelope import SequenceNumber
from .protocol.messages import EnvelopeCodec
_apower_tx_seq = getattr(_apower_tx, "_seq", None)
if _apower_tx_seq is None:
_apower_tx_seq = _apower_tx._seq = SequenceNumber() # type: ignore[attr-defined]
env = direct_connect_request(
CommunicationCmd.DEVICE_APOWER_INFO, sn,
data={"update": 2}, snno=_apower_tx_seq.next(),
)
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
#: a local broker on the probed port (CONNACK starts ``20 02``).
_MQTT_CONNECT = bytes.fromhex("101200044d5154540402003c000670726f62653031")
def probe_transport(
sn: str,
host: str = DEVICE_GATEWAY_IP,
ports: Tuple[int, ...] = (9000,),
key_radix: int = 16,
timeout: float = 5.0,
) -> List[Dict[str, Any]]:
"""Transport-level probes against one service port.
The envelope bytes are already verified byte-for-byte against the
disassembly; these variants isolate *framing/handshake* behaviour:
server-first greeting, immediate raw send with a long read, length
prefixes (LE16/BE16/LE32), an MQTT CONNECT, and a double-send over one
persistent connection. Each entry carries exact ``tx``/``rx`` bytes.
"""
from .protocol.messages import EnvelopeCodec
results: List[Dict[str, Any]] = []
for port in ports:
tx, envelope = _apower_tx(sn, key_radix)
def run(label: str, tx_bytes: bytes, pre_read: float = 0.0,
post_idle: float = 4.0, split_send: bool = False) -> None:
entry: Dict[str, Any] = {
"variant": label, "port": port, "tx": tx_bytes,
"rx_pre": b"", "rx": b"", "reply": None, "error": None,
}
try:
with socket.create_connection(
(host, port), timeout=timeout) as sock:
if pre_read:
entry["rx_pre"] = _recv_all(sock, idle=pre_read)
if tx_bytes:
if split_send:
half = len(tx_bytes) // 2
sock.sendall(tx_bytes[:half])
sock.settimeout(0.5)
try:
entry["rx_pre"] += sock.recv(65536)
except (socket.timeout, TimeoutError):
pass
sock.sendall(tx_bytes[half:])
else:
sock.sendall(tx_bytes)
entry["rx"] = _recv_all(sock, idle=post_idle)
except (ConnectionError, socket.error, TimeoutError) as exc:
entry["error"] = f"{type(exc).__name__}: {exc}"
if entry["rx"]:
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)
run("raw/long-read", tx, post_idle=8.0)
run("raw/split-send", tx, split_send=True)
run("le16-prefix", len(tx).to_bytes(2, "little") + tx)
run("be16-prefix", len(tx).to_bytes(2, "big") + tx)
run("le32-prefix", len(tx).to_bytes(4, "little") + tx)
run("mqtt-connect", _MQTT_CONNECT)
return results
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("--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",
)
parser.add_argument("--key-radix", type=int, default=32, choices=(16, 32))
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")
parser.add_argument("--probe", action="store_true",
help="send probe variants and print exact TX/RX bytes")
parser.add_argument("--probe2", action="store_true",
help="transport-level probes (greeting, prefixes, "
"MQTT, split send) with exact TX/RX bytes")
args = parser.parse_args(argv)
if args.show_hotspot:
@@ -254,6 +373,22 @@ def main(argv: Optional[List[str]] = None) -> int:
if out.get("error"):
print(f" error: {out['error']}")
return 0
if args.probe2:
for out in probe_transport(args.sn, host=args.host, ports=ports,
key_radix=args.key_radix,
timeout=args.timeout):
print(f"--- {out['variant']} -> {args.host}:{out['port']}")
print(f" tx ({len(out['tx'])} B): {out['tx'].hex() or '(none)'}")
pre = out.get("rx_pre") or b""
if pre:
print(f" rx_pre ({len(pre)} B): {pre.hex()}")
rx = out.get("rx") or b""
print(f" rx ({len(rx)} B): {rx.hex() or '(empty)'}")
if out.get("reply") is not None:
print(" reply: " + json.dumps(out["reply"], default=str))
if out.get("error"):
print(f" error: {out['error']}")
return 0
try:
result = get_battery(args.sn, host=args.host, ports=ports,
key_radix=args.key_radix, timeout=args.timeout)
+48 -9
View File
@@ -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):
+227 -42
View File
@@ -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,13 +31,11 @@ from .protocol import (
CommunicationOperate,
SequenceNumber,
SocketFrameDecoder,
decrypt_data,
decrypt_request,
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__)
@@ -64,8 +66,8 @@ class LocalDeviceClient:
:param socket_port: local TCP socket port (candidate 18000).
:param verify: TLS verification for the device API (device likely serves a
self-signed cert, so ``False`` is the practical default).
:param key_radix: key parse radix for the envelope cipher (32, per the
``int.tryParse(key, radix: 32)`` call in both directions).
:param key_radix: key parse radix for the envelope cipher (16 —
default ``int.tryParse`` radix, verified against live captures).
:param timeout: per-request timeout in seconds.
"""
@@ -77,7 +79,7 @@ class LocalDeviceClient:
scheme: str = "https",
verify: Optional[bool] = False,
socket_port: Optional[int] = None,
key_radix: int = 32,
key_radix: int = 16,
timeout: float = 10.0,
extra_headers: Optional[Dict[str, str]] = None,
session: Optional[requests.Session] = None,
@@ -92,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(
@@ -181,63 +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, equip_no or 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,
)
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))
reply = decrypt_data(raw, radix=self.key_radix)
if not reply:
raw = self._socket_roundtrip(
frame_socket_payload(self.codec.encode(envelope)))
try:
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:
decoded = decrypt_data(reply_raw, radix=self.key_radix)
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:
@@ -268,18 +280,43 @@ class LocalDeviceClient:
break
if collect_only:
return bytes(received)
if decrypt_data(bytes(received), radix=self.key_radix):
return bytes(received)
try:
self.codec.decode(bytes(received))
except ValueError:
raise ConnectionError(
f"no reply frame from {self.socket_host}:{self.socket_port}"
)
return bytes(received)
# ------------------------------------------------------- convenience RPCs
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)
@@ -294,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)
@@ -311,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
+166
View File
@@ -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())
+19 -1
View File
@@ -4,7 +4,13 @@ Everything here is hardware-independent and covered by unit tests. See
``docs/RE_DART_FINDINGS.md`` for the reverse-engineering evidence.
"""
from .commands import CommunicationCmd, CommunicationMode, CommunicationOperate
from .commands import (
CommunicationCmd,
CommunicationMode,
CommunicationOperate,
SOCKET_CMD_TYPE,
socket_cmd_type,
)
from .crc import (
CRC16_INIT,
CRC16_POLY,
@@ -20,10 +26,12 @@ from .crypto import (
decrypt_data,
decrypt_object,
decrypt_request,
decrypt_wire,
encrypt_object,
generate_secret_key,
header_len_for,
split_envelope,
split_wire,
)
from .envelope import (
KEY_ORDER,
@@ -35,8 +43,12 @@ from .envelope import (
direct_connect_request,
)
from .framing import (
BLE_HEADER_LEN,
MAGIC_ACK,
MAGIC_DATA,
MAGIC_MTU,
ble_frame_sn,
ble_frame_total_len,
build_ble_ack_frame,
build_ble_data_frame,
chunk_for_mtu,
@@ -49,11 +61,14 @@ from .framing import (
verify_frame_crc,
)
from .framing import SocketFrameDecoder
from .messages import EnvelopeCodec
__all__ = [
"CommunicationCmd",
"CommunicationMode",
"CommunicationOperate",
"SOCKET_CMD_TYPE",
"socket_cmd_type",
"CRC16_INIT",
"CRC16_POLY",
"crc16_modbus",
@@ -69,6 +84,8 @@ __all__ = [
"decrypt_object",
"decrypt_request",
"decrypt_data",
"decrypt_wire",
"split_wire",
"header_len_for",
"KEY_ORDER",
"SNNO_MIN",
@@ -90,4 +107,5 @@ __all__ = [
"chunk_for_mtu",
"frame_socket_payload",
"SocketFrameDecoder",
"EnvelopeCodec",
]
@@ -84,3 +84,73 @@ class CommunicationCmd(IntEnum):
BREAKER_COM_STATE = 0x32
BREAKER_LOAD_STATUS = 0x33
BREAKER_IDENTIFY_ME = 0x34
#: ``cmdType`` values on the socket/Bluetooth/local transports.
#:
#: ``CommunicationCmd::cmd(cmd, mode)`` returns the *second* id field
#: (``off_14`` in the pool) for ``socket``/``bluetooth``/``local`` modes and
#: throws for ``http`` — verified live: a PCAPdroid capture of the app shows
#: ``{"cmdType":1101, ...}`` (``QUERY_DEVICE_INFO`` here, ``off_8`` ``0x00``)
#: with the S1-plaintext-first framing (see ``docs/RE_DART_FINDINGS.md``
#: §5). The ``off_8`` values above are *not* used on the wire for these
#: transports. Entries mapped to ``None`` have no pool id (``0xFFFF…``).
SOCKET_CMD_TYPE = {
CommunicationCmd.QUERY_DEVICE_INFO: 0x44D,
CommunicationCmd.NETWORK_INFO: 0x45D,
CommunicationCmd.RESOURCE_ADDRESS: 0x4B5,
CommunicationCmd.SUNSPEC_MODBUS: 0x4B5,
CommunicationCmd.NETWORK_SWITCH: 0x45F,
CommunicationCmd.FEEDING_GRID_POWER: 0x6A5,
CommunicationCmd.MODE: 0x57B,
CommunicationCmd.MODE_V2: 0x6BD,
CommunicationCmd.MODE_SETTING_V2: 0x6BF,
CommunicationCmd.SOLAR_INFO: 0x76F,
CommunicationCmd.ENERGY_MANAGE_INFO: 0x4B3,
CommunicationCmd.GENERATOR_INFO: 0x76D,
CommunicationCmd.BLUETOOTH_HEARTBEAT: 0x7D7,
CommunicationCmd.SCAN_WIFI_LIST: 0x455,
CommunicationCmd.CONNECT_WIFI: 0x457,
CommunicationCmd.CONNECTION_STATUS: 0x459,
CommunicationCmd.NETWORK_CHANNEL_STATUS: 0x46D,
CommunicationCmd.NETWORK_CHANNEL: 0x46B,
CommunicationCmd.PAIR_POWER: 0x44F,
CommunicationCmd.PROCESS_PAIR_POWER: 0x451,
CommunicationCmd.CHECK_BLUETOOTH_PASSWORD: 0x7D5,
CommunicationCmd.GO_OFF_GRID_STATUS: 0x6BB,
CommunicationCmd.A_POWER_LED_SWITCH: None,
CommunicationCmd.DEVICE_DETAIL_INFO: 0x515,
CommunicationCmd.DEVICE_APOWER_INFO: 0x51B,
CommunicationCmd.DEVICE_ALARMS: 0x725,
CommunicationCmd.PE: 0x6A7,
CommunicationCmd.BMS: 0x6A9,
CommunicationCmd.EMS: 0x6AB,
CommunicationCmd.MPPT: 0x6B5,
CommunicationCmd.A_HUB: 0x771,
CommunicationCmd.CT_TEST_REVERSE: None,
CommunicationCmd.DEVICE_DATA: None,
CommunicationCmd.LOAD_STATUS: 0x583,
CommunicationCmd.LOAD_STATUS_2: 0x587,
CommunicationCmd.SMART_LOAD: 0x581,
CommunicationCmd.SMART_LOAD_2: 0x585,
CommunicationCmd.REBOOT_DEVICE: 0x6B9,
CommunicationCmd.DEBUG_MODE: 0x71D,
CommunicationCmd.POWER_ON: 0x71F,
CommunicationCmd.CHARGE_DEVICE: 0x721,
CommunicationCmd.CLEAR_FAULT: 0x6AD,
CommunicationCmd.DEVICE_ADDRESS_INFO: 0x4B1,
CommunicationCmd.OVERLOAD_CUT_OFF: 0x6A5,
CommunicationCmd.WIFI_FREQUECY: 0x46F,
CommunicationCmd.BREAKER_COMMISSION_NOTIFY: None,
CommunicationCmd.BREAKER_COM_STATE: None,
CommunicationCmd.BREAKER_LOAD_STATUS: 0x471,
CommunicationCmd.BREAKER_IDENTIFY_ME: 0x473,
}
def socket_cmd_type(cmd: CommunicationCmd) -> int:
"""Wire ``cmdType`` for *cmd* on the socket transport (``cmd()``)."""
value = SOCKET_CMD_TYPE[cmd]
if value is None:
raise KeyError(f"{cmd} has no socket-transport command id")
return int(value)
+72 -23
View File
@@ -5,42 +5,42 @@ Reconstructed from the blutter disassembly; see
Request wire form of an envelope::
s = compact JSON, e.g. {"cmdType":25,"equipNo":"SN",...}
s = compact JSON, e.g. {"cmdType":1101,"equipNo":"SN",...}
S1 = parts[0] + "," + parts[1] + "," # quartet header, trailing comma
S2 = ",".join(parts[2:]) # the rest of the JSON
a = UTF-8(S1) # sent PLAINTEXT first
n = int.tryParse(key, radix: 32) or 0
n = int.tryParse(key) or 0 # default radix 16
wire = a || bytes((S2[i] + i + len(a) + n) & 0xFF)
i.e. the two leading JSON parts go on the wire **unchanged first**, and the
tail is transformed byte-wise with a position-dependent additive stream
(index ``i`` is relative to the tail; ``i + len(a)`` is the absolute JSON
index). The receiver that knows the serial can compute ``len(a)`` up front
(``header_len_for``) and split the wire deterministically.
index). Both directions use this same framing (verified live: a 1154-byte
device reply decrypts with the identical split — see ``decrypt_wire``).
Responses use a different framing (``decrypt_data``): the wire holds the
plaintext JSON prefix ``s`` (which itself contains the serial's ``X``),
a literal ``0x58`` separator, then the encoded tail::
.. note:: An earlier revision claimed radix 32 from a ``0x20`` literal
next to the ``tryParse`` call sites. Two independent live frames (an
app discovery request and a device info reply) decrypt cleanly only
with radix 16 (``int("80", 16) == 128``, ``int("1c", 16) == 28``) and
then verify byte-exact (``len``/uppercase CRC-32). Ground truth wins:
the ``0x20`` belongs to a neighbouring call.
wire = UTF-8(s) || 0x58 || encode(tail)
``decrypt_data`` locates the *second* ``0x58`` byte ``j``, takes
``s = UTF-8(wire[:j])``, probes ``s + "}"`` as JSON to recover ``equipNo``
(and hence the key), decodes the tail with the same stream in reverse, and
returns ``s + "," + tail_plain``. Anything unexpected yields ``""`` (the
app's ``decryptData`` returns ``""`` on marker/parse failure and the socket
layer keeps accumulating bytes).
Confirmed from the disassembly:
Confirmed from the disassembly *and live traffic*:
* ``generateSecretKey`` = ``sum(utf8(sn)) & 0xFF`` rendered as lowercase hex
with **no** zero padding (Dart ``_toPow2String(16)``).
* The key is parsed with ``int.tryParse(key, radix: 32)`` — the literal
``0x20`` (32) is materialised right before the call in both the encrypt
and decrypt paths.
with **no** zero padding (Dart ``_toPow2String(16)``). E.g. ``"1c"`` for
the gateway SN, ``"80"`` for the empty/placeholder SN.
* The key is parsed with ``int.tryParse(key)`` — default radix **16**,
verified against two live frames (``tests/fixtures/*.bin``).
* ``encryptObject(object, key)`` takes the object in the first slot and the
key in the second; its result is ``a ++ transform(b)`` (plaintext header
first — ``addAll(transformed)`` onto the header copy).
* Replies use the same ``S1 || transform(S2)`` framing (the device info
reply decrypts identically). The reply ``cmdType`` observed live is the
request id + 1 (``1101`` → ``1102``); replies are therefore parsed with
:func:`decrypt_wire`, which locates the header without knowing the ids
up front. ``decrypt_data`` (``0x58``-marker model) is kept for the
Bluetooth path, which is unverified live.
"""
from __future__ import annotations
@@ -48,8 +48,10 @@ from __future__ import annotations
import json
from typing import Optional, Tuple
#: Key radix used by both ``encryptObject`` and ``decryptData`` (literal 0x20).
DEFAULT_KEY_RADIX = 32
#: Key radix used by the envelope cipher (default ``tryParse`` radix).
#: Verified live: ``int("80", 16) == 128`` and ``int("1c", 16) == 28``
#: decode two independent captured frames byte-exact.
DEFAULT_KEY_RADIX = 16
#: Response separator byte (``0x58``, ASCII ``X``).
MARKER = 0x58
@@ -143,6 +145,53 @@ def decrypt_request(
decrypt_object = decrypt_request
def split_wire(wire: bytes) -> Tuple[bytes, bytes]:
"""Split any envelope wire into ``(a, b)`` at the 2nd comma.
The header ``S1 = parts[0] + "," + parts[1] + ","`` is always ASCII
with no commas inside, so the first two commas on the wire are its
structural boundaries — no ids or serials needed up front. Works for
requests and replies alike.
"""
raw = bytes(wire)
first = raw.find(b",")
if first < 0:
raise ValueError("no header comma in wire buffer")
second = raw.find(b",", first + 1)
if second < 0:
raise ValueError("truncated header in wire buffer")
return raw[: second + 1], raw[second + 1:]
def decrypt_wire(wire: bytes, radix: int = DEFAULT_KEY_RADIX) -> str:
"""Decode one complete envelope wire (request or reply) to JSON.
Splits via :func:`split_wire`, recovers ``equipNo`` from the
``a + "}"`` probe (exactly how the receiver learns which key to use),
and reverses the additive stream. Raises ``ValueError`` on any
failure (use :func:`decrypt_data`-style ``""`` handling when
accumulating).
"""
a, b = split_wire(wire)
try:
# ``a`` ends with the header's trailing comma — drop it so the
# probe is valid JSON (mirrors the ``s + "}"`` equipNo probe).
probe = json.loads(a[:-1].decode("utf-8") + "}")
except (UnicodeDecodeError, ValueError) as exc:
raise ValueError(f"header probe is not JSON: {exc}") from exc
if not isinstance(probe, dict):
raise ValueError("header probe is not an object")
key = generate_secret_key(str(probe.get("equipNo", "")))
key_int = _parse_key(key, radix)
plain_b = bytes(
(byte - i - len(a) - key_int) & 0xFF for i, byte in enumerate(b)
)
try:
return (a + plain_b).decode("utf-8")
except UnicodeDecodeError as exc:
raise ValueError(f"tail is not UTF-8: {exc}") from exc
def decrypt_data(wire: bytes, radix: int = DEFAULT_KEY_RADIX) -> str:
"""``decryptData`` — decode one accumulated *response* buffer.
+17 -7
View File
@@ -25,7 +25,7 @@ import json
import time
from typing import Any, Dict, Optional
from .commands import CommunicationCmd, CommunicationOperate
from .commands import CommunicationCmd, CommunicationOperate, socket_cmd_type
from .crc import crc32
#: Exact key order used by the app's ``toJson`` (``dataArea`` conditional).
@@ -175,19 +175,29 @@ 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.
Mirrors the Dart ``CommunicationObject.communicate`` factory +
``CommunicationService._query*ByDirectConnect`` family
(``docs/RE_DART_FINDINGS.md`` §3). Note ``snno`` is assigned by
``sendObject`` (see :class:`SequenceNumber`); pass the issued value when
re-serialising.
(``docs/RE_DART_FINDINGS.md`` §3). The wire ``cmdType`` is the
*socket-transport* id (``CommunicationCmd::cmd`` for socket mode —
e.g. ``DEVICE_APOWER_INFO`` → ``1307``, verified live). On socket
mode the factory always seeds the payload with ``"opt"`` (the
operation type, ``0`` for query) and merges the caller's arguments
after it — e.g. the app's ``queryAPowerData`` sends
``dataArea = {"opt": 0, "update": 2}``. Note ``snno`` is assigned by
``sendObject`` (see :class:`SequenceNumber`); pass the issued value
when re-serialising.
"""
payload: Dict[str, Any] = {"opt": int(operate)}
if data:
payload.update(data)
return CommunicationObject(
cmd_type=int(cmd),
cmd_type=socket_cmd_type(cmd),
equip_no=equip_no,
op_type=CommunicationOperate.QUERY,
op_type=int(operate),
snno=snno,
data=data,
data=payload,
)
+114 -47
View File
@@ -1,12 +1,23 @@
"""Frame builders for the two byte transports.
* **BLE** — ``communication_center/bluetooth/bluetooth_frame.dart`` (see
``docs/RE_DART_FINDINGS.md`` §9). Frame layout and magics are confirmed;
the *exact* on-air order of the magic relative to the frame and the MTU
continuation scheme still need a live capture.
* **Local TCP socket** — ``communication_center/socket`` uses ``int16ToBytes``
/ ``bytesToInt16`` (little-endian) framing (see §8). A length-prefixed
frame is the working candidate; adjust once a capture is available.
``docs/RE_DART_FINDINGS.md`` §9). On-air layout, settled from
``createBluetoothFrame``::
magic[2] ++ crc[2] ++ int16LE(totalLen) ++ int16LE(sn) ++ content
* ``totalLen = 6 + len(content)`` (2 crc-slot + 2 len + 2 sn bytes).
* CRC-16/MODBUS over the buffer with ``00 00`` in the crc slots, rendered
as the app's swapped-lowercase 4-hex string; each hex *pair* is then run
through ``int.parse(pair, radix: 32)`` (literal ``0x20`` — quirky but
explicit) and written into slots 2..3.
* Magics: ``5A A5`` data (default), ``4A A4`` ack, ``2A A2`` MTU announce.
* The encrypted envelope (same bytes as the socket path) is split into
MTU-sized slices when longer than the negotiated MTU; every slice goes
out as its own frame with its own ``_increaseBluetoothSN()`` value.
The receiver dedups by sn, accumulates ``content`` parts and runs the
same ``decryptData`` over the concatenation.
* **Local TCP socket** — raw encrypted bytes, no framing (see §8).
"""
from __future__ import annotations
@@ -20,12 +31,79 @@ from .crc import crc16_hex_string, crc16_modbus, int16_to_le_bytes, le_bytes_to_
MAGIC_ACK = b"\x4a\xa4"
#: First two bytes of a **data** frame (receiving one triggers the app's ack).
MAGIC_DATA = b"\x5a\xa5"
#: Magic of the MTU-announce frame (``_sendMtu`` sends ``mtuToData()``).
MAGIC_MTU = b"\x2a\xa2"
MAGIC_NAMES = {
MAGIC_ACK[0:2]: "ack",
MAGIC_DATA[0:2]: "data",
MAGIC_MTU[0:2]: "mtu",
}
#: On-air header length: magic(2) + crc(2) + totalLen(2) + sn(2).
BLE_HEADER_LEN = 8
def _ble_crc_bytes(body_with_zero_slots: bytes) -> bytes:
"""CRC slot bytes the way ``createBluetoothFrame`` derives them.
CRC-16/MODBUS over the frame with ``00 00`` slots, swapped-lowercase
4-hex string, then ``int.parse(pair, radix: 32)`` per pair (the
disassembly materialises ``0x20`` for the radix). Values can exceed
one byte in theory; the GATT write path must narrow them, so mask
here (assumption — confirm against a capture).
"""
hexstr = crc16_hex_string(body_with_zero_slots)
return bytes(int(hexstr[i:i + 2], 32) & 0xFF for i in (0, 2))
def create_ble_frame(
content: bytes,
sn: int,
magic: bytes = MAGIC_DATA,
) -> bytes:
"""``createBluetoothFrame(content, magic, sn)`` — full on-air frame.
Layout::
magic ++ crc[2] ++ int16LE(6 + len(content)) ++ int16LE(sn) ++ content
"""
total = 6 + len(content)
buf = bytearray(
bytes(magic[:2])
+ b"\x00\x00"
+ int16_to_le_bytes(total)
+ int16_to_le_bytes(sn)
+ bytes(content)
)
buf[2:4] = _ble_crc_bytes(bytes(buf))
return bytes(buf)
def build_ble_data_frame(content: bytes, sn: int) -> bytes:
"""Full on-air data frame (magic ``0x5A 0xA5``)."""
return create_ble_frame(content, sn, MAGIC_DATA)
def build_ble_ack_frame(content: bytes, sn: int) -> bytes:
"""Full on-air ACK frame (magic ``0x4A 0xA4``)."""
return create_ble_frame(content, sn, MAGIC_ACK)
def magic_in_data(data: bytes) -> Optional[str]:
"""Return ``"ack"`` / ``"data"`` / ``"mtu"`` if known; else ``None``."""
return MAGIC_NAMES.get(bytes(data[:2]))
def ble_frame_sn(data: bytes) -> int:
"""``sn`` field of a frame (int16LE at offset 6)."""
return le_bytes_to_int16(data[6:8])
def ble_frame_total_len(data: bytes) -> int:
"""``totalLen`` field of a frame (int16LE at offset 4)."""
return le_bytes_to_int16(data[4:6])
# In createBluetoothFrame the CRC placeholder sits at byte offsets 2..5 of the
# frame body (content ++ [00 00 00 00] ++ int16LE(totalLen) ++ int16LE(value) ++ sn).
_CRC_SLOTS = slice(2, 4)
@@ -77,51 +155,40 @@ def magic_in_data(data: bytes) -> Optional[str]:
return MAGIC_NAMES.get(bytes(data[:2]))
def sn_in_data(data: bytes, sn: str) -> bool:
"""True if the frame (after any magic) ends with the device SN."""
sn_bytes = sn.encode("utf-8")
return data.endswith(sn_bytes)
def sn_in_data(data: bytes, sn: int) -> bool:
"""True if the frame's ``sn`` field equals *sn*."""
if len(data) < BLE_HEADER_LEN:
return False
return ble_frame_sn(data) == int(sn)
def content_in_data(data: bytes, sn: str) -> Optional[bytes]:
"""Extract the ``content`` bytes of a data/ack frame.
Layout after the 2-byte magic is fixed-size except for ``content``, so::
content = frame[2 : len(frame) - len(sn) - FRAME_TAIL_SIZE]
"""
body = data[2:] if len(data) >= 2 else data
sn_bytes = sn.encode("utf-8")
if not body.endswith(sn_bytes):
def content_in_data(data: bytes) -> Optional[bytes]:
"""Frame ``content`` (everything after the 8-byte header)."""
if len(data) < BLE_HEADER_LEN:
return None
end = len(body) - len(sn_bytes) - _FRAME_TAIL_SIZE
if end < 2: # need at least the 4-byte crc slots
return None
return body[2:end]
return bytes(data[BLE_HEADER_LEN:])
def verify_frame_crc(data: bytes) -> bool:
"""Recompute the CRC-16 over the frame body and compare slots 2..3."""
body = data[2:] if len(data) >= 2 else data
if len(body) < 4:
"""Recompute the CRC-16 over the frame (zeroed slots) and compare."""
if len(data) < BLE_HEADER_LEN:
return False
candidate = body[:]
candidate[2:4] = b"\x00\x00\x00\x00"
return bytes.fromhex(crc16_hex_string(bytes(candidate))) == body[2:4]
candidate = bytearray(data)
candidate[2:4] = b"\x00\x00"
return _ble_crc_bytes(bytes(candidate)) == bytes(data[2:4])
def chunk_for_mtu(frame: bytes, mtu: int = 20) -> List[bytes]:
"""Split a frame for GATT writes.
def chunk_for_mtu(encrypted: bytes, mtu: int = 20) -> List[bytes]:
"""Split the encrypted envelope into MTU-sized slices.
First chunk carries the 2-byte magic; the rest are raw continuation
chunks. The app's ``mtuToData`` does something similar — the exact
continuation framing (length headers? re-magic?) needs confirmation.
Mirrors the ``sendObject`` over-length branch (``ceil(len/mtu)`` slices
via ``sublist``): every slice is sent as its own BLE frame with its own
``_increaseBluetoothSN()`` value (see :func:`create_ble_frame`).
"""
if mtu <= 2:
raise ValueError("mtu must exceed the 2-byte magic")
if len(frame) <= mtu:
return [frame]
return [frame[i:i + mtu] for i in range(0, len(frame), mtu)]
if mtu <= 0:
raise ValueError("mtu must be positive")
return [bytes(encrypted[i:i + mtu])
for i in range(0, len(encrypted), mtu)] or [b""]
def magicEquals(a: bytes, b: bytes) -> bool:
@@ -145,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.
"""
@@ -157,11 +223,12 @@ class SocketFrameDecoder:
def feed(self, data: bytes) -> List[str]:
"""Append *data*; return a list of complete decoded JSON strings."""
from .crypto import decrypt_data # local import: cheap, no cycles
from .crypto import decrypt_wire # local import: cheap, no cycles
self._buf.extend(data)
decoded = decrypt_data(bytes(self._buf))
if not decoded:
try:
decoded = decrypt_wire(bytes(self._buf))
except ValueError:
return []
try:
json.loads(decoded)
@@ -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
+15 -5
View File
@@ -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)
+411
View File
@@ -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
+126
View File
@@ -0,0 +1,126 @@
#!/usr/bin/env python3
"""Host probe v2 — transport framing variants via the library.
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``.
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
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 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 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):
return bytes(out), "idle-timeout"
if not chunk:
return bytes(out), "closed-by-peer"
out.extend(chunk)
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()
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()]:
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__":
main()
+77
View File
@@ -0,0 +1,77 @@
#!/usr/bin/env python3
"""Host probe v3 — socket-id variants via the library business API.
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 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 sys
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,
)
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():
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()
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)
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__":
main()
+75
View File
@@ -0,0 +1,75 @@
#!/usr/bin/env python3
"""Host probe v4 — ``opt``-seeded payloads via the library business API.
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 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 sys
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,
)
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():
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()
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)
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__":
main()
+129
View File
@@ -0,0 +1,129 @@
#!/usr/bin/env python3
"""Host probe v5 — session sequencing via the library.
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
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 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 recv_all_verbose(sock: socket.socket, idle: float = 8.0):
out = bytearray()
sock.settimeout(idle)
while True:
try:
chunk = sock.recv(65536)
except (socket.timeout, TimeoutError):
return bytes(out), "idle-timeout(%.1fs)" % idle
if not chunk:
return bytes(out), "closed-by-peer"
out.extend(chunk)
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()
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:
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__":
main()
+66
View File
@@ -0,0 +1,66 @@
#!/usr/bin/env python3
"""Host probe v6 — envelope variants via the library business API.
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 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 sys
sys.path.insert(0, os.path.dirname(os.path.dirname(
os.path.abspath(__file__))))
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,18000")
ap.add_argument("--timeout", type=float, default=5.0)
args = ap.parse_args()
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__":
raise SystemExit(main())
+60
View File
@@ -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())
+62
View File
@@ -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())
Binary file not shown.
+2
View File
@@ -0,0 +1,2 @@
{"cmdType":1101,"equipNo":"00000000",Ç ÌåÜÙÐ#(#'Úóëòõïïð÷úò÷ðç9568ìýùð;5?ô␍ ù;K=ýÿ%+
MK_M.`TQ,ndfk3*'jgmPsqwshus^n|-F/d@>ABADE8”•
+36
View File
@@ -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"}))
+133
View File
@@ -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)
+44
View File
@@ -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()
+139 -39
View File
@@ -6,6 +6,7 @@ No radio/hardware needed. Vectors mirror the blutter disassembly of
"""
import json
import os
import unittest
from franklinwh_agate_client.protocol import (
@@ -13,15 +14,18 @@ from franklinwh_agate_client.protocol import (
CommunicationCmd,
CommunicationObject,
SequenceNumber,
socket_cmd_type,
SocketFrameDecoder,
decrypt_data,
decrypt_request,
decrypt_wire,
direct_connect_request,
encrypt_object,
frame_socket_payload,
generate_secret_key,
header_len_for,
split_envelope,
split_wire,
)
from franklinwh_agate_client.protocol.crypto import MARKER
@@ -38,13 +42,32 @@ def _apower_json(snno=1, time_stamp=1792105196):
return env.to_json_string()
class TestSocketCmdSpace(unittest.TestCase):
def test_socket_ids(self):
self.assertEqual(socket_cmd_type(CommunicationCmd.QUERY_DEVICE_INFO), 1101)
self.assertEqual(socket_cmd_type(CommunicationCmd.DEVICE_APOWER_INFO), 1307)
self.assertEqual(socket_cmd_type(CommunicationCmd.BMS), 1705)
self.assertEqual(socket_cmd_type(CommunicationCmd.EMS), 1707)
def test_direct_request_uses_socket_id(self):
env = direct_connect_request(CommunicationCmd.DEVICE_APOWER_INFO, SN)
self.assertEqual(env.cmd_type, 1307)
self.assertTrue(env.to_json_string().startswith('{"cmdType":1307,'))
def test_discovery_shape_like_capture(self):
env = direct_connect_request(CommunicationCmd.QUERY_DEVICE_INFO, "")
body = env.to_json_string()
self.assertTrue(body.startswith(
'{"cmdType":1101,"equipNo":"00000000",'))
class TestSecretKey(unittest.TestCase):
def test_real_sn_key(self):
self.assertEqual(generate_secret_key(SN), "1c")
def test_key_radix_is_32(self):
self.assertEqual(DEFAULT_KEY_RADIX, 32)
self.assertEqual(int(generate_secret_key(SN), 32), 44)
def test_key_radix_is_16(self):
self.assertEqual(DEFAULT_KEY_RADIX, 16)
self.assertEqual(int(generate_secret_key(SN), 16), 28)
def test_empty_sn_key(self):
self.assertEqual(generate_secret_key(""), "0")
@@ -61,30 +84,28 @@ class TestEnvelope(unittest.TestCase):
"crc", "dataArea"],
)
self.assertNotIn("data", d)
# socket payload is always seeded with the operation type first
self.assertEqual(d["dataArea"], {"opt": 0, "update": 2})
def test_no_data_means_seven_keys(self):
def test_no_data_still_carries_opt(self):
env = direct_connect_request(CommunicationCmd.BMS, SN, data=None)
d = env.to_dict()
self.assertEqual(
list(d),
["cmdType", "equipNo", "type", "timeStamp", "snno", "len", "crc"],
)
self.assertEqual(env.length, 0)
self.assertEqual(env.crc, "")
self.assertIn("dataArea", d)
self.assertEqual(d["dataArea"], {"opt": 0})
self.assertEqual(env.length, len(b'{"opt":0}'))
self.assertTrue(env.crc)
def test_empty_map_means_no_data_area(self):
def test_empty_map_still_carries_opt(self):
env = direct_connect_request(CommunicationCmd.BMS, SN, data={})
self.assertNotIn("dataArea", env.to_dict())
self.assertEqual(env.length, 0)
self.assertEqual(env.crc, "")
self.assertEqual(env.to_dict()["dataArea"], {"opt": 0})
def test_len_crc_auto_uppercase(self):
env = direct_connect_request(
CommunicationCmd.DEVICE_APOWER_INFO, SN, data={"update": 2})
self.assertEqual(env.length, len(b'{"update":2}'))
self.assertEqual(env.length, len(b'{"opt":0,"update":2}'))
import zlib
self.assertEqual(
env.crc, format(zlib.crc32(b'{"update":2}'), "X"))
env.crc, format(zlib.crc32(b'{"opt":0,"update":2}'), "X"))
self.assertEqual(env.crc, env.crc.upper())
def test_int_fields(self):
@@ -126,11 +147,11 @@ class TestCipher(unittest.TestCase):
def test_header_len_matches(self):
a, _ = split_envelope(_apower_json())
self.assertEqual(header_len_for(0x19, SN), len(a))
self.assertEqual(header_len_for(1307, SN), len(a))
def test_stream_is_absolute_index_plus_n(self):
a, b = split_envelope(_apower_json())
n = int(generate_secret_key(SN), 32)
n = int(generate_secret_key(SN), 16)
wire = encrypt_object(_apower_json(), generate_secret_key(SN))
tail = wire[len(a):]
for i, byte in enumerate(b):
@@ -141,20 +162,14 @@ class TestCipher(unittest.TestCase):
wire = encrypt_object(body, generate_secret_key(SN))
self.assertEqual(
decrypt_request(wire, generate_secret_key(SN),
cmd_type=0x19, equip_no=SN),
cmd_type=1307, equip_no=SN),
body,
)
def test_decrypt_response_roundtrip(self):
def test_wire_roundtrip(self):
body = _apower_json()
cut = body.index('"type"')
s, tail = body[:cut - 1], body[cut:]
n = int(generate_secret_key(SN), 32)
prefix_len = len((s + ",").encode("utf-8"))
enc = bytes((byte + k + prefix_len + n) & 0xFF
for k, byte in enumerate(tail.encode("utf-8")))
resp = s.encode("utf-8") + bytes((MARKER,)) + enc
self.assertEqual(decrypt_data(resp), body)
wire = encrypt_object(body, generate_secret_key(SN))
self.assertEqual(decrypt_wire(wire), body)
def test_decrypt_data_failures_are_empty(self):
self.assertEqual(decrypt_data(b""), "")
@@ -171,22 +186,107 @@ class TestSocketFraming(unittest.TestCase):
self.assertEqual(frame_socket_payload(payload), payload)
def test_decoder_accumulates_until_json(self):
body = _apower_json()
cut = body.index('"type"')
s, tail = body[:cut - 1], body[cut:]
n = int(generate_secret_key(SN), 32)
prefix_len = len((s + ",").encode("utf-8"))
enc = bytes((byte + k + prefix_len + n) & 0xFF
for k, byte in enumerate(tail.encode("utf-8")))
resp = s.encode("utf-8") + bytes((MARKER,)) + enc
wire = open(os.path.join(FIXTURES, "discovery_req.bin"),
"rb").read()
dec = SocketFrameDecoder()
# first half: incomplete -> nothing yet
self.assertEqual(dec.feed(resp[:len(resp) // 2]), [])
self.assertEqual(dec.feed(wire[:len(wire) // 2]), [])
# rest: one complete JSON string
out = dec.feed(resp[len(resp) // 2:])
out = dec.feed(wire[len(wire) // 2:])
self.assertEqual(len(out), 1)
self.assertEqual(json.loads(out[0])["equipNo"], SN)
self.assertEqual(json.loads(out[0])["equipNo"], "00000000")
if __name__ == "__main__":
unittest.main()
FIXTURES = os.path.join(os.path.dirname(__file__), "fixtures")
class TestLiveCaptures(unittest.TestCase):
"""Golden vectors: two real frames (app request + device reply)."""
def test_key_radix_is_16(self):
self.assertEqual(DEFAULT_KEY_RADIX, 16)
self.assertEqual(generate_secret_key("00000000"), "80")
self.assertEqual(int("80", 16), 128)
self.assertEqual(int(generate_secret_key(SN), 16), 28)
def test_discovery_request_vector(self):
wire = open(os.path.join(FIXTURES, "discovery_req.bin"), "rb").read()
self.assertEqual(len(wire), 153)
a, b = split_wire(wire)
self.assertEqual(a, b'{"cmdType":1101,"equipNo":"00000000",')
self.assertEqual(
decrypt_wire(wire),
'{"cmdType":1101,"equipNo":"00000000","type":0,'
'"timeStamp":1792117904,"snno":1,"len":37,"crc":"A474C16F",'
'"dataArea":{"opt":0,"minProtocolVer":"V1.00.00"}}',
)
def test_device_info_reply_vector(self):
wire = open(os.path.join(FIXTURES, "device_info_reply.bin"),
"rb").read()
self.assertEqual(len(wire), 1154)
a, b = split_wire(wire)
self.assertEqual(
a, b'{"cmdType":1102,"equipNo":"10060005A03X25160132",')
reply = json.loads(decrypt_wire(wire))
self.assertEqual(reply["cmdType"], 1102)
self.assertEqual(reply["equipNo"], "10060005A03X25160132")
self.assertEqual(reply["len"], 1024)
data = reply["dataArea"]
self.assertEqual(data["opt"], 0)
self.assertEqual(data["result"], 0)
self.assertEqual(data["IBG_SN"], "10060005A03X25160132")
self.assertEqual(data["FHP_NUM"], 2)
self.assertEqual(len(data["FHP_SN"]), 2)
self.assertEqual(len(data["BMS_SN"]), 2)
# len/crc integrity, as the device checks them
import zlib
compact = json.dumps(data, separators=(",", ":"),
ensure_ascii=False)
self.assertEqual(reply["len"], len(compact.encode("utf-8")))
self.assertEqual(
reply["crc"], format(zlib.crc32(compact.encode()), "X"))
def test_split_wire_errors(self):
with self.assertRaises(ValueError):
split_wire(b"no commas here")
with self.assertRaises(ValueError):
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")