addition of the iot3 protocol and lock bolt v2 functionality

custom_components/wyzeapi/__init__.py

@@ -29,6 +29,8 @@
     API_KEY,
 )
 from .coordinator import WyzeLockBoltCoordinator
+from .iot3_coordinator import WyzeLockBoltV2Coordinator
+from .iot3_service import Iot3Service
 from .token_manager import TokenManager
 
 PLATFORMS = [
@@ -195,19 +197,46 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 async def setup_coordinators(
     hass: HomeAssistant, config_entry: ConfigEntry, client: Wyzeapy
 ):
-    """Set up coordinators for Wyze devices that require Bluetooth."""
-    # Check if Bluetooth is active and functioning
+    """Set up coordinators for Wyze Lock Bolt devices (BLE and IoT3)."""
+    from .const import IOT3_MODELS
+
+    lock_service = await client.lock_service
+    all_locks = await lock_service.get_locks()
+    coordinators = hass.data[DOMAIN][config_entry.entry_id].setdefault(
+        "coordinators", {}
+    )
+
+    # IoT3 devices have product_type "Common" so get_locks() won't find them.
+    # Search the full device list by product_model instead.
+    all_devices = await lock_service.get_object_list()
+    iot3_devices = [d for d in all_devices if d.product_model in IOT3_MODELS]
+    # Store them so lock.py can find them later
+    hass.data[DOMAIN][config_entry.entry_id]["iot3_devices"] = iot3_devices
+
+    # IoT3 coordinators for DX-family locks (no Bluetooth needed)
+    iot3_locks = iot3_devices
+    if iot3_locks:
+        iot3_service = Iot3Service(hass, config_entry)
+        hass.data[DOMAIN][config_entry.entry_id]["iot3_service"] = iot3_service
+        for lock in iot3_locks:
+            _LOGGER.info(
+                "Setting up IoT3 coordinator for %s (%s)",
+                lock.nickname,
+                lock.product_model,
+            )
+            coordinators[lock.mac] = WyzeLockBoltV2Coordinator(
+                hass, iot3_service, lock
+            )
+
+    # BLE coordinators for YD_BT1 Lock Bolt v1
     if bluetooth.async_scanner_count(hass, connectable=True) == 0:
-        _LOGGER.info(
-            "Bluetooth is not active or no scanners available. Skipping WyzeLockBoltCoordinator setup."
-        )
+        if any(l.product_model == "YD_BT1" for l in all_locks):
+            _LOGGER.info(
+                "Bluetooth is not active. Skipping WyzeLockBoltCoordinator setup."
+            )
         return
 
-    lock_service = await client.lock_service
-    for lock in await lock_service.get_locks():
+    for lock in all_locks:
         if lock.product_model == "YD_BT1":
-            coordinators = hass.data[DOMAIN][config_entry.entry_id].setdefault(
-                "coordinators", {}
-            )
             coordinators[lock.mac] = WyzeLockBoltCoordinator(hass, lock_service, lock)
             await coordinators[lock.mac].update_lock_info()

custom_components/wyzeapi/const.py

@@ -26,3 +26,13 @@
 YDBLE_LOCK_STATE_UUID = "00002220-0000-6b63-6f6c-2e6b636f6f6c"
 YDBLE_UART_RX_UUID = "6e400003-b5a3-f393-e0a9-e50e24dcca9e"
 YDBLE_UART_TX_UUID = "6e400002-b5a3-f393-e0a9-e50e24dcca9e"
+
+# IoT3 API for DX-family devices (Lock Bolt v2, Palm Lock, etc.)
+IOT3_BASE_URL = "https://app.wyzecam.com/app/v4/iot3"
+IOT3_GET_PROPERTY_PATH = "/app/v4/iot3/get-property"
+IOT3_RUN_ACTION_PATH = "/app/v4/iot3/run-action"
+IOT3_APP_HOST = "https://app.wyzecam.com"
+OLIVE_SIGNING_SECRET = "wyze_app_secret_key_132"
+OLIVE_APP_ID = "9319141212m2ik"
+OLIVE_APP_INFO = "wyze_android_3.11.0.758"
+IOT3_MODELS = {"DX_LB2", "DX_PVLOC"}

custom_components/wyzeapi/iot3_coordinator.py

@@ -0,0 +1,82 @@
+"""DataUpdateCoordinator for Wyze DX-family devices via IoT3 API."""
+
+from __future__ import annotations
+
+from datetime import timedelta
+import logging
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .iot3_service import Iot3Service
+
+_LOGGER = logging.getLogger(__name__)
+
+UPDATE_INTERVAL = timedelta(seconds=30)
+
+
+class WyzeLockBoltV2Coordinator(DataUpdateCoordinator):
+    """Coordinator for Wyze Lock Bolt v2 (DX_LB2) via IoT3 cloud API."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        iot3_service: Iot3Service,
+        lock,
+    ):
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=f"Wyze Lock Bolt V2 {lock.nickname}",
+            update_interval=UPDATE_INTERVAL,
+        )
+        self._iot3_service = iot3_service
+        self._lock = lock
+        self._current_command: str | None = None
+
+    async def _async_update_data(self) -> dict:
+        """Poll IoT3 get-property for current lock state."""
+        if self._current_command is not None:
+            return self.data or {}
+
+        try:
+            result = await self._iot3_service.get_properties(self._lock.mac)
+        except Exception as exc:
+            raise UpdateFailed(f"Error fetching lock state: {exc}") from exc
+
+        if result.get("code") != "1":
+            raise UpdateFailed(
+                f"IoT3 API returned error: {result.get('msg', 'unknown')}"
+            )
+
+        props = result.get("data", {}).get("props", {})
+        return {
+            "locked": props.get("lock::lock-status", None),
+            "door_open": not props.get("lock::door-status", True),
+            "online": props.get("iot-device::iot-state", False),
+            "battery_level": props.get("battery::battery-level", None),
+            "power_source": props.get("battery::power-source", None),
+            "firmware_ver": props.get("device-info::firmware-ver", None),
+        }
+
+    async def lock_unlock(self, command: str):
+        """Execute lock or unlock command."""
+        self._current_command = command
+        self.async_update_listeners()
+        try:
+            if command == "lock":
+                result = await self._iot3_service.lock(self._lock.mac)
+            else:
+                result = await self._iot3_service.unlock(self._lock.mac)
+
+            if result.get("code") != "1":
+                _LOGGER.error(
+                    "Lock %s command failed: %s",
+                    command,
+                    result.get("msg", "unknown"),
+                )
+        except Exception:
+            _LOGGER.exception("Failed to %s lock %s", command, self._lock.nickname)
+        finally:
+            self._current_command = None
+            await self.async_request_refresh()

custom_components/wyzeapi/iot3_service.py

@@ -0,0 +1,160 @@
+"""IoT3 API service for Wyze DX-family devices (Lock Bolt v2, Palm Lock, etc.)."""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import logging
+import random
+import time
+import uuid
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_USERNAME
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import (
+    ACCESS_TOKEN,
+    IOT3_APP_HOST,
+    IOT3_GET_PROPERTY_PATH,
+    IOT3_RUN_ACTION_PATH,
+    OLIVE_APP_ID,
+    OLIVE_APP_INFO,
+    OLIVE_SIGNING_SECRET,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class Iot3Service:
+    """Client for the Wyze IoT3 API (DX-family devices)."""
+
+    def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry):
+        self._hass = hass
+        self._config_entry = config_entry
+        self._phone_id = str(uuid.uuid4())
+        self._session = async_get_clientsession(hass)
+
+    @property
+    def _access_token(self) -> str:
+        return self._config_entry.data.get(ACCESS_TOKEN, "")
+
+    @property
+    def username(self) -> str:
+        return self._config_entry.data.get(CONF_USERNAME, "")
+
+    def _compute_signature(self, body: str) -> str:
+        access_key = self._access_token + OLIVE_SIGNING_SECRET
+        secret = hashlib.md5(access_key.encode()).hexdigest()
+        return hmac.new(secret.encode(), body.encode(), hashlib.md5).hexdigest()
+
+    def _build_headers(self, body: str) -> dict:
+        return {
+            "access_token": self._access_token,
+            "appid": OLIVE_APP_ID,
+            "appinfo": OLIVE_APP_INFO,
+            "appversion": "3.11.0.758",
+            "env": "Prod",
+            "phoneid": self._phone_id,
+            "requestid": uuid.uuid4().hex,
+            "Signature2": self._compute_signature(body),
+            "Content-Type": "application/json; charset=utf-8",
+        }
+
+    @staticmethod
+    def _extract_model(device_mac: str) -> str:
+        """Extract model from MAC (e.g., DX_LB2_80482C9C659C -> DX_LB2)."""
+        parts = device_mac.split("_")
+        if len(parts) >= 3:
+            return "_".join(parts[:2])
+        return device_mac
+
+    async def _post(self, path: str, payload: dict) -> dict:
+        body = json.dumps(payload)
+        headers = self._build_headers(body)
+        url = f"{IOT3_APP_HOST}{path}"
+        try:
+            async with self._session.post(url, headers=headers, data=body) as resp:
+                result = await resp.json()
+                if result.get("code") != "1":
+                    _LOGGER.warning(
+                        "IoT3 API error on %s: code=%s msg=%s",
+                        path,
+                        result.get("code"),
+                        result.get("msg"),
+                    )
+                return result
+        except Exception as exc:
+            _LOGGER.error("IoT3 API request failed for %s: %s", path, exc)
+            raise
+
+    async def get_properties(
+        self,
+        device_mac: str,
+        props: list[str] | None = None,
+    ) -> dict:
+        """Get device properties via IoT3 get-property endpoint."""
+        if props is None:
+            props = [
+                "lock::lock-status",
+                "lock::door-status",
+                "iot-device::iot-state",
+                "battery::battery-level",
+                "battery::power-source",
+                "device-info::firmware-ver",
+            ]
+
+        ts = int(time.time() * 1000)
+        payload = {
+            "nonce": str(ts),
+            "payload": {
+                "cmd": "get_property",
+                "props": props,
+                "tid": random.randint(1000, 99999),
+                "ts": ts,
+                "ver": 1,
+            },
+            "targetInfo": {
+                "id": device_mac,
+                "model": self._extract_model(device_mac),
+            },
+        }
+        return await self._post(IOT3_GET_PROPERTY_PATH, payload)
+
+    async def run_action(
+        self,
+        device_mac: str,
+        action: str,
+    ) -> dict:
+        """Run an action (e.g., lock::lock, lock::unlock) via IoT3 run-action endpoint."""
+        ts = int(time.time() * 1000)
+        payload = {
+            "nonce": str(ts),
+            "payload": {
+                "action": action,
+                "cmd": "run_action",
+                "params": {
+                    "action_id": random.randint(10000, 99999),
+                    "type": 1,
+                    "username": self.username,
+                },
+                "tid": random.randint(1000, 99999),
+                "ts": ts,
+                "ver": 1,
+            },
+            "targetInfo": {
+                "id": device_mac,
+                "model": self._extract_model(device_mac),
+            },
+        }
+        return await self._post(IOT3_RUN_ACTION_PATH, payload)
+
+    async def lock(self, device_mac: str) -> dict:
+        """Lock the device."""
+        return await self.run_action(device_mac, "lock::lock")
+
+    async def unlock(self, device_mac: str) -> dict:
+        """Unlock the device."""
+        return await self.run_action(device_mac, "lock::unlock")