Home Wi-Fi join API (connectWifi SET + scan/status) with docs

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", ...}``.