Zigbee Gateway Support

.gitignore

@@ -2,4 +2,5 @@
 __pycache__
 .tox
 tuyadebug/
-.pre-commit-config.yaml
+.pre-commit-config.yaml
+.idea

README.md

@@ -16,6 +16,7 @@ The following Tuya device types are currently supported:
 * Fans
 * Climates
 * Vacuums
+* * Zigbee and Bluetooth gateways and their attached devices
 
 Energy monitoring (voltage, current, watts, etc.) is supported for compatible devices.
 

custom_components/localtuya/__init__.py

@@ -19,12 +19,14 @@
     CONF_PLATFORM,
     CONF_REGION,
     CONF_USERNAME,
+    CONF_FRIENDLY_NAME,
+    CONF_MODEL,
     EVENT_HOMEASSISTANT_STOP,
     SERVICE_RELOAD,
 )
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
-from homeassistant.helpers.device_registry import DeviceEntry
+from homeassistant.helpers import device_registry
 from homeassistant.helpers.event import async_track_time_interval
 
 from .cloud_api import TuyaCloudApi
@@ -39,6 +41,9 @@
     DATA_DISCOVERY,
     DOMAIN,
     TUYA_DEVICES,
+    CONF_GATEWAY_DEVICE_ID,
+    CONF_IS_GATEWAY,
+    CONF_PROTOCOL_VERSION,
 )
 from .discovery import TuyaDiscovery
 
@@ -139,10 +144,8 @@ def _device_discovered(device):
             )
             new_data[ATTR_UPDATED_AT] = str(int(time.time() * 1000))
             hass.config_entries.async_update_entry(entry, data=new_data)
-            device = hass.data[DOMAIN][TUYA_DEVICES][device_id]
-            if not device.connected:
-                device.async_connect()
-        elif device_id in hass.data[DOMAIN][TUYA_DEVICES]:
+
+        if device_id in hass.data[DOMAIN][TUYA_DEVICES]:
             # _LOGGER.debug("Device %s found with IP %s", device_id, device_ip)
 
             device = hass.data[DOMAIN][TUYA_DEVICES][device_id]
@@ -261,12 +264,34 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
 
     async def setup_entities(device_ids):
         platforms = set()
+        sub_devices = []
         for dev_id in device_ids:
-            entities = entry.data[CONF_DEVICES][dev_id][CONF_ENTITIES]
+            device_entry = entry.data[CONF_DEVICES][dev_id]
+            entities = device_entry[CONF_ENTITIES]
             platforms = platforms.union(
                 set(entity[CONF_PLATFORM] for entity in entities)
             )
-            hass.data[DOMAIN][TUYA_DEVICES][dev_id] = TuyaDevice(hass, entry, dev_id)
+            device = TuyaDevice(hass, entry, dev_id)
+            hass.data[DOMAIN][TUYA_DEVICES][dev_id] = device
+
+            # Register gateway device manually to HA
+            if device_entry.get(CONF_IS_GATEWAY, False):
+                dr = device_registry.async_get(hass)
+                dr.async_get_or_create(
+                    config_entry_id=entry.entry_id,
+                    identifiers={(DOMAIN, f"local_{dev_id}")},
+                    name=device_entry[CONF_FRIENDLY_NAME],
+                    manufacturer="Tuya",
+                    model=f"{device_entry[CONF_MODEL]} ({dev_id})",
+                    sw_version=device_entry[CONF_PROTOCOL_VERSION],
+                )
+
+            if device.gateway_device_id:
+                sub_devices.append(device)
+        # at this point, the gateway should have been created,
+        # and we can register the sub-device
+        for sub_device in sub_devices:
+            sub_device.register_in_gateway()
 
         await asyncio.gather(
             *[
@@ -322,7 +347,7 @@ async def update_listener(hass, config_entry):
 
 
 async def async_remove_config_entry_device(
-    hass: HomeAssistant, config_entry: ConfigEntry, device_entry: DeviceEntry
+    hass: HomeAssistant, config_entry: ConfigEntry, device_entry: device_registry.DeviceEntry
 ) -> bool:
     """Remove a config entry from a device."""
     dev_id = list(device_entry.identifiers)[0][1].split("_")[-1]

custom_components/localtuya/common.py

@@ -13,6 +13,7 @@
     CONF_ID,
     CONF_PLATFORM,
     CONF_SCAN_INTERVAL,
+    CONF_CLIENT_ID,
 )
 from homeassistant.core import callback
 from homeassistant.helpers.dispatcher import (
@@ -31,6 +32,8 @@
     DATA_CLOUD,
     DOMAIN,
     TUYA_DEVICES,
+    CONF_GATEWAY_DEVICE_ID,
+    CONF_IS_GATEWAY,
 )
 
 _LOGGER = logging.getLogger(__name__)
@@ -128,6 +131,7 @@ def __init__(self, hass, config_entry, dev_id):
         self._hass = hass
         self._config_entry = config_entry
         self._dev_config_entry = config_entry.data[CONF_DEVICES][dev_id].copy()
+        self.device_id = self._dev_config_entry[CONF_DEVICE_ID]
         self._interface = None
         self._status = {}
         self.dps_to_request = {}
@@ -136,7 +140,15 @@ def __init__(self, hass, config_entry, dev_id):
         self._disconnect_task = None
         self._unsub_interval = None
         self._local_key = self._dev_config_entry[CONF_LOCAL_KEY]
-        self.set_logger(_LOGGER, self._dev_config_entry[CONF_DEVICE_ID])
+        self.set_logger(_LOGGER, self.device_id)
+        self.is_gateway = self._dev_config_entry.get(CONF_IS_GATEWAY, False)
+
+        # handling sub-devices
+        self._connected = False
+        self.gateway_device_id = self._dev_config_entry.get(CONF_GATEWAY_DEVICE_ID)
+        self.cid = self._dev_config_entry.get(CONF_CLIENT_ID)
+        self.gateway_device = None
+        self.sub_devices = {}
 
         # This has to be done in case the device type is type_0d
         for entity in self._dev_config_entry[CONF_ENTITIES]:
@@ -145,33 +157,61 @@ def __init__(self, hass, config_entry, dev_id):
     @property
     def connected(self):
         """Return if connected to device."""
-        return self._interface is not None
+        if self.gateway_device:
+            return self.gateway_device.connected
+        else:
+            return self._interface is not None
+
+    def register_in_gateway(self):
+        self.gateway_device = self._hass.data[DOMAIN][TUYA_DEVICES][self.gateway_device_id]
+        self.gateway_device.sub_devices[self.cid] = self
 
     def async_connect(self):
         """Connect to device if not already connected."""
+        if self.gateway_device and not self.gateway_device.connected:
+            # early return in case this is a sub-device,
+            # connect will be triggered by the gateway
+            return
         if not self._is_closing and self._connect_task is None and not self._interface:
             self._connect_task = asyncio.create_task(self._make_connection())
 
+    async def _get_interface(self):
+        if self.gateway_device:
+            return self.gateway_device._interface
+        else:
+            return await pytuya.connect(
+                self._dev_config_entry[CONF_HOST],
+                self.device_id,
+                self._local_key,
+                float(self._dev_config_entry[CONF_PROTOCOL_VERSION]),
+                listener=self,
+                is_gateway=self.is_gateway,
+            )
+
     async def _make_connection(self):
         """Subscribe localtuya entity events."""
         self.debug("Connecting to %s", self._dev_config_entry[CONF_HOST])
 
         try:
-            self._interface = await pytuya.connect(
-                self._dev_config_entry[CONF_HOST],
-                self._dev_config_entry[CONF_DEVICE_ID],
-                self._local_key,
-                float(self._dev_config_entry[CONF_PROTOCOL_VERSION]),
-                self,
-            )
-            self._interface.add_dps_to_request(self.dps_to_request)
+            self._interface = await self._get_interface()
+            if self.cid:
+                self._interface.add_sub_device(self.cid)
 
-            self.debug("Retrieving initial state")
-            status = await self._interface.status()
-            if status is None:
-                raise Exception("Failed to retrieve status")
+            # Query initial state except for gateways
+            if not self.is_gateway:
+                self._interface.add_dps_to_request(self.dps_to_request, cid=self.cid)
+
+                self.debug("Retrieving initial state")
+                status = await self._interface.status(cid=self.cid)
+                if status is None:
+                    raise Exception("Failed to retrieve status")
 
-            self.status_updated(status)
+                self.status_updated(status)
+
+            self._connected = True
+
+            if self._disconnect_task is not None:
+                self._disconnect_task()
 
             def _new_entity_handler(entity_id):
                 self.debug(
@@ -181,7 +221,7 @@ def _new_entity_handler(entity_id):
                 )
                 self._dispatch_status()
 
-            signal = f"localtuya_entity_{self._dev_config_entry[CONF_DEVICE_ID]}"
+            signal = f"localtuya_entity_{self.device_id}"
             self._disconnect_task = async_dispatcher_connect(
                 self._hass, signal, _new_entity_handler
             )
@@ -199,23 +239,31 @@ def _new_entity_handler(entity_id):
             self.exception(
                 f"Connect to {self._dev_config_entry[CONF_HOST]} failed: %s", type(e)
             )
+            self._connected = False
             if self._interface is not None:
-                await self._interface.close()
+                if not self.gateway_device:
+                    await self._interface.close()
                 self._interface = None
 
         except Exception as e:  # pylint: disable=broad-except
             self.exception(f"Connect to {self._dev_config_entry[CONF_HOST]} failed")
+            self._connected = False
             if "json.decode" in str(type(e)):
                 await self.update_local_key()
 
             if self._interface is not None:
-                await self._interface.close()
+                if not self.gateway_device:
+                    await self._interface.close()
                 self._interface = None
         self._connect_task = None
 
+        # if there are sub-devices, we have to connect them as well
+        for sub_device in self.sub_devices.values():
+            sub_device._connect_task = asyncio.create_task(sub_device._make_connection())
+
     async def update_local_key(self):
         """Retrieve updated local_key from Cloud API and update the config_entry."""
-        dev_id = self._dev_config_entry[CONF_DEVICE_ID]
+        dev_id = self.device_id
         await self._hass.data[DOMAIN][DATA_CLOUD].async_get_devices_list()
         cloud_devs = self._hass.data[DOMAIN][DATA_CLOUD].device_list
         if dev_id in cloud_devs:
@@ -239,10 +287,11 @@ async def close(self):
         if self._connect_task is not None:
             self._connect_task.cancel()
             await self._connect_task
-        if self._interface is not None:
+        if self._interface is not None and not self.gateway_device:
             await self._interface.close()
         if self._disconnect_task is not None:
             self._disconnect_task()
+        self._connected = False
         self.debug(
             "Closed connection with device %s.",
             self._dev_config_entry[CONF_FRIENDLY_NAME],
@@ -252,7 +301,7 @@ async def set_dp(self, state, dp_index):
         """Change value of a DP of the Tuya device."""
         if self._interface is not None:
             try:
-                await self._interface.set_dp(state, dp_index)
+                await self._interface.set_dp(state, dp_index, cid=self._dev_config_entry.get(CONF_CLIENT_ID))
             except Exception:  # pylint: disable=broad-except
                 self.exception("Failed to set DP %d to %d", dp_index, state)
         else:
@@ -264,7 +313,7 @@ async def set_dps(self, states):
         """Change value of a DPs of the Tuya device."""
         if self._interface is not None:
             try:
-                await self._interface.set_dps(states)
+                await self._interface.set_dps(states, cid=self._dev_config_entry.get(CONF_CLIENT_ID))
             except Exception:  # pylint: disable=broad-except
                 self.exception("Failed to set DPs %r", states)
         else:
@@ -275,17 +324,35 @@ async def set_dps(self, states):
     @callback
     def status_updated(self, status):
         """Device updated status."""
-        self._status.update(status)
-        self._dispatch_status()
+        for device, device_status in status.items():
+            if device == '_default':
+                # usual case for devices that are not using a gateway
+                self._status.update(device_status)
+                self._dispatch_status()
+            elif self.cid:
+                # check if this is a sub-device, and if so, update based on the cid
+                if self.cid == device:
+                    self._status.update(device_status)
+                    self._dispatch_status()
+                else:
+                    _LOGGER.warning('Sub-device received a status update for a unknown cid: %s', device)
+            else:
+                # otherwise, we are dealing with a status for a sub-device in a gateway instance
+                sub_device = self.sub_devices.get(device)
+                if not sub_device:
+                    _LOGGER.warning('Gateway received a status update for a unknown cid: %s', device)
+                    return
+                sub_device._status.update(device_status)
+                sub_device._dispatch_status()
 
     def _dispatch_status(self):
-        signal = f"localtuya_{self._dev_config_entry[CONF_DEVICE_ID]}"
+        signal = f"localtuya_{self.device_id}"
         async_dispatcher_send(self._hass, signal, self._status)
 
     @callback
     def disconnected(self):
         """Device disconnected."""
-        signal = f"localtuya_{self._dev_config_entry[CONF_DEVICE_ID]}"
+        signal = f"localtuya_{self.device_id}"
         async_dispatcher_send(self._hass, signal, None)
         if self._unsub_interval is not None:
             self._unsub_interval()
@@ -301,11 +368,12 @@ def __init__(self, device, config_entry, dp_id, logger, **kwargs):
         """Initialize the Tuya entity."""
         super().__init__()
         self._device = device
+        self.device_id = device.device_id
         self._dev_config_entry = config_entry
         self._config = get_entity_config(config_entry, dp_id)
         self._dp_id = dp_id
         self._status = {}
-        self.set_logger(logger, self._dev_config_entry[CONF_DEVICE_ID])
+        self.set_logger(logger, self.device_id)
 
     async def async_added_to_hass(self):
         """Subscribe localtuya events."""
@@ -327,13 +395,13 @@ def _update_handler(status):
                     self.status_updated()
                 self.schedule_update_ha_state()
 
-        signal = f"localtuya_{self._dev_config_entry[CONF_DEVICE_ID]}"
+        signal = f"localtuya_{self.device_id}"
 
         self.async_on_remove(
             async_dispatcher_connect(self.hass, signal, _update_handler)
         )
 
-        signal = f"localtuya_entity_{self._dev_config_entry[CONF_DEVICE_ID]}"
+        signal = f"localtuya_entity_{self.device_id}"
         async_dispatcher_send(self.hass, signal, self.entity_id)
 
     @property
@@ -343,11 +411,11 @@ def device_info(self):
         return {
             "identifiers": {
                 # Serial numbers are unique identifiers within a specific domain
-                (DOMAIN, f"local_{self._dev_config_entry[CONF_DEVICE_ID]}")
+                (DOMAIN, f"local_{self.device_id}")
             },
             "name": self._dev_config_entry[CONF_FRIENDLY_NAME],
             "manufacturer": "Tuya",
-            "model": f"{model} ({self._dev_config_entry[CONF_DEVICE_ID]})",
+            "model": f"{model} ({self.device_id})",
             "sw_version": self._dev_config_entry[CONF_PROTOCOL_VERSION],
         }
 
@@ -364,7 +432,7 @@ def should_poll(self):
     @property
     def unique_id(self):
         """Return unique device identifier."""
-        return f"local_{self._dev_config_entry[CONF_DEVICE_ID]}_{self._dp_id}"
+        return f"local_{self.device_id}_{self._dp_id}"
 
     def has_config(self, attr):
         """Return if a config parameter has a valid value."""