From eda78dd7d179b0202a9999a798217eb9d2b9c970 Mon Sep 17 00:00:00 2001 From: kfj001 Date: Fri, 16 Oct 2026 06:08:09 +0000 Subject: [PATCH] Home Wi-Fi join API (connectWifi SET + scan/status) with docs --- docs/RE_DART_FINDINGS.md | 25 +++++++++++++++ franklinwh_agate_client/device_local.py | 33 ++++++++++++++++++-- franklinwh_agate_client/protocol/envelope.py | 5 +-- tests/test_operations.py | 27 ++++++++++++++++ 4 files changed, 86 insertions(+), 4 deletions(-) diff --git a/docs/RE_DART_FINDINGS.md b/docs/RE_DART_FINDINGS.md index dc9efee..1457afd 100644 --- a/docs/RE_DART_FINDINGS.md +++ b/docs/RE_DART_FINDINGS.md @@ -424,6 +424,31 @@ a local-socket one (`sections/installer/bind_device/`): (`DatabaseManager.getTrustedDevicesTableList`, consulted by `_checkConnectedNotTrustedAp` on direct-connect). +## 18. Joining the aGate to home Wi-Fi (STA onboarding) + +Three transports can deliver the credentials; all verified in +disassembly, socket path live-tested for framing (not the join itself): + +- **Socket** (while on the SoftAP): `queryConnectWifi` → + `CommunicationService.connectWifi(entity, …)` → `_request` with cmd + `connectWifi` (socket `1111`), operate **SET** (`type: 1`), arguments = + `ConnectWifiEntity.toJson()`: `ssid`, `password`, `wifi_SSID`, + `wifi_Pw` (plus an unencrypted-network flag). Library: + `LocalDeviceClient.join_home_wifi(ssid, password)` (+ `opt: 1` + seeding via the shared factory path), `scan_home_wifi()` + (`scanWifiList`, socket `1109`, no args) to list visible networks, + `query_connection_status()` (`connectionStatus`, socket `1113`) to + confirm association. +- **BLE**: ESP protocomm (`prov-session`, `prov-scan`, `prov-config`, + `custom-data`, sec0/sec1) — the standard Espressif flow; `prov-config` + carries the SSID/password. Used when the phone is not on the SoftAP. +- **Phone-side**: `wifi_iot` plugin join/leave is the *phone* joining the + *device* AP (`wifi.join_device_hotspot`) — the opposite direction. + +.. warning:: `join_home_wifi` is state-changing: on success the unit + leaves its SoftAP (or roams), dropping the current session. Only run + it with physical/rejoin access to the unit. + ## 9. Bluetooth (BLE) transport — `communication_center/bluetooth` - `BluetoothManager` (singleton), `BluetoothStatus`: `closed / connected / connecting / interrupted`. diff --git a/franklinwh_agate_client/device_local.py b/franklinwh_agate_client/device_local.py index d9ae7e9..16995a5 100644 --- a/franklinwh_agate_client/device_local.py +++ b/franklinwh_agate_client/device_local.py @@ -190,7 +190,7 @@ class LocalDeviceClient: if envelope is None: envelope = direct_connect_request( cmd, self.sn if equip_no is None else equip_no, data=data, - snno=self._sn_seq.next(), + snno=self._sn_seq.next(), operate=operate, ) return self.send_envelope(envelope) @@ -212,6 +212,7 @@ class LocalDeviceClient: 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. @@ -223,7 +224,7 @@ class LocalDeviceClient: """ envelope = direct_connect_request( cmd, self.sn if equip_no is None else equip_no, data=data, - snno=self._sn_seq.next(), + snno=self._sn_seq.next(), operate=operate, ) wire = self.codec.encode(envelope) reply_raw = self._socket_roundtrip( @@ -354,6 +355,34 @@ 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", ...}``. diff --git a/franklinwh_agate_client/protocol/envelope.py b/franklinwh_agate_client/protocol/envelope.py index fff4c26..65d0557 100644 --- a/franklinwh_agate_client/protocol/envelope.py +++ b/franklinwh_agate_client/protocol/envelope.py @@ -175,6 +175,7 @@ def direct_connect_request( equip_no: str, data: Optional[Dict[str, Any]] = None, snno: int = 0, + operate: CommunicationOperate = CommunicationOperate.QUERY, ) -> CommunicationObject: """Build a *direct-connect* query envelope. @@ -190,13 +191,13 @@ def direct_connect_request( ``sendObject`` (see :class:`SequenceNumber`); pass the issued value when re-serialising. """ - payload: Dict[str, Any] = {"opt": int(CommunicationOperate.QUERY)} + payload: Dict[str, Any] = {"opt": int(operate)} if data: payload.update(data) return CommunicationObject( cmd_type=socket_cmd_type(cmd), equip_no=equip_no, - op_type=CommunicationOperate.QUERY, + op_type=int(operate), snno=snno, data=payload, ) diff --git a/tests/test_operations.py b/tests/test_operations.py index e8bd25c..a69215f 100644 --- a/tests/test_operations.py +++ b/tests/test_operations.py @@ -104,3 +104,30 @@ class TestWattageOp(unittest.TestCase): 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)