Support for Shelly Input Events (#43479)

homeassistant/components/shelly/__init__.py

@@ -27,6 +27,7 @@ from .const import (
     COAP,
     DATA_CONFIG_ENTRY,
     DOMAIN,
+    INPUTS_EVENTS_DICT,
     POLLING_TIMEOUT_MULTIPLIER,
     REST,
     REST_SENSORS_UPDATE_INTERVAL,
@@ -34,6 +35,7 @@ from .const import (
     SLEEP_PERIOD_MULTIPLIER,
     UPDATE_PERIOD_MULTIPLIER,
 )
+from .utils import get_device_name
 
 PLATFORMS = ["binary_sensor", "cover", "light", "sensor", "switch"]
 _LOGGER = logging.getLogger(__name__)
@@ -54,11 +56,6 @@ async def get_coap_context(hass):
     return context
 
 
-def get_device_name(device):
-    """Naming for device."""
-    return device.settings["name"] or device.settings["device"]["hostname"]
-
-
 async def async_setup(hass: HomeAssistant, config: dict):
     """Set up the Shelly component."""
     hass.data[DOMAIN] = {DATA_CONFIG_ENTRY: {}}
@@ -113,6 +110,7 @@ class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
 
     def __init__(self, hass, entry, device: aioshelly.Device):
         """Initialize the Shelly device wrapper."""
+        self.device_id = None
         sleep_mode = device.settings.get("sleep_mode")
 
         if sleep_mode:
@@ -140,6 +138,46 @@ class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
 
         self.device.subscribe_updates(self.async_set_updated_data)
 
+        self._async_remove_input_events_handler = self.async_add_listener(
+            self._async_input_events_handler
+        )
+        self._last_input_events_count = dict()
+
+    @callback
+    def _async_input_events_handler(self):
+        """Handle device input events."""
+        for block in self.device.blocks:
+            if (
+                "inputEvent" not in block.sensor_ids
+                or "inputEventCnt" not in block.sensor_ids
+            ):
+                continue
+
+            channel = int(block.channel or 0) + 1
+            event_type = block.inputEvent
+            last_event_count = self._last_input_events_count.get(channel)
+            self._last_input_events_count[channel] = block.inputEventCnt
+
+            if last_event_count == block.inputEventCnt or event_type == "":
+                continue
+
+            if event_type in INPUTS_EVENTS_DICT:
+                self.hass.bus.async_fire(
+                    "shelly.click",
+                    {
+                        "device_id": self.device_id,
+                        "device": self.device.settings["device"]["hostname"],
+                        "channel": channel,
+                        "click_type": INPUTS_EVENTS_DICT[event_type],
+                    },
+                )
+            else:
+                _LOGGER.warning(
+                    "Shelly input event %s for device %s is not supported, please open issue",
+                    event_type,
+                    self.name,
+                )
+
     async def _async_update_data(self):
         """Fetch data."""
         _LOGGER.debug("Polling Shelly Device - %s", self.name)
@@ -166,7 +204,7 @@ class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
         """Set up the wrapper."""
         dev_reg = await device_registry.async_get_registry(self.hass)
         model_type = self.device.settings["device"]["type"]
-        dev_reg.async_get_or_create(
+        entry = dev_reg.async_get_or_create(
             config_entry_id=self.entry.entry_id,
             name=self.name,
             connections={(device_registry.CONNECTION_NETWORK_MAC, self.mac)},
@@ -176,10 +214,12 @@ class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
             model=aioshelly.MODEL_NAMES.get(model_type, model_type),
             sw_version=self.device.settings["fw"],
         )
+        self.device_id = entry.id
 
     def shutdown(self):
         """Shutdown the wrapper."""
         self.device.shutdown()
+        self._async_remove_input_events_handler()
 
 
 class ShellyDeviceRestWrapper(update_coordinator.DataUpdateCoordinator):
@@ -200,7 +240,7 @@ class ShellyDeviceRestWrapper(update_coordinator.DataUpdateCoordinator):
         """Fetch data."""
         try:
             async with async_timeout.timeout(5):
-                _LOGGER.debug("REST update for %s", get_device_name(self.device))
+                _LOGGER.debug("REST update for %s", self.name)
                 return await self.device.update_status()
         except OSError as err:
             raise update_coordinator.UpdateFailed("Error fetching data") from err

homeassistant/components/shelly/const.py

@@ -22,3 +22,13 @@ UPDATE_PERIOD_MULTIPLIER = 2.2
 
 # Shelly Air - Maximum work hours before lamp replacement
 SHAIR_MAX_WORK_HOURS = 9000
+
+# Map Shelly input events
+INPUTS_EVENTS_DICT = {
+    "S": "single",
+    "SS": "double",
+    "SSS": "triple",
+    "L": "long",
+    "SL": "single_long",
+    "LS": "long_single",
+}

homeassistant/components/shelly/entity.py

@@ -116,7 +116,7 @@ class ShellyBlockEntity(entity.Entity):
         """Initialize Shelly entity."""
         self.wrapper = wrapper
         self.block = block
-        self._name = get_entity_name(wrapper, block)
+        self._name = get_entity_name(wrapper.device, block)
 
     @property
     def name(self):
@@ -182,7 +182,7 @@ class ShellyBlockAttributeEntity(ShellyBlockEntity, entity.Entity):
 
         self._unit = unit
         self._unique_id = f"{super().unique_id}-{self.attribute}"
-        self._name = get_entity_name(wrapper, block, self.description.name)
+        self._name = get_entity_name(wrapper.device, block, self.description.name)
 
     @property
     def unique_id(self):
@@ -255,7 +255,7 @@ class ShellyRestAttributeEntity(update_coordinator.CoordinatorEntity):
         self.description = description
 
         self._unit = self.description.unit
-        self._name = get_entity_name(wrapper, None, self.description.name)
+        self._name = get_entity_name(wrapper.device, None, self.description.name)
         self.path = self.description.path
         self._attributes = self.description.attributes
 

homeassistant/components/shelly/utils.py

@@ -9,7 +9,6 @@ import aioshelly
 from homeassistant.components.sensor import DEVICE_CLASS_TIMESTAMP
 from homeassistant.const import TEMP_CELSIUS, TEMP_FAHRENHEIT
 
-from . import ShellyDeviceWrapper
 from .const import DOMAIN
 
 _LOGGER = logging.getLogger(__name__)
@@ -31,22 +30,31 @@ def temperature_unit(block_info: dict) -> str:
     return TEMP_CELSIUS
 
 
+def get_device_name(device: aioshelly.Device) -> str:
+    """Naming for device."""
+    return device.settings["name"] or device.settings["device"]["hostname"]
+
+
 def get_entity_name(
-    wrapper: ShellyDeviceWrapper,
+    device: aioshelly.Device,
     block: aioshelly.Block,
     description: Optional[str] = None,
-):
+) -> str:
     """Naming for switch and sensors."""
-    entity_name = wrapper.name
+    entity_name = get_device_name(device)
 
     if block:
         channels = None
         if block.type == "input":
-            channels = wrapper.device.shelly.get("num_inputs")
+            # Shelly Dimmer/1L has two input channels and missing "num_inputs"
+            if device.settings["device"]["type"] in ["SHDM-1", "SHDM-2", "SHSW-L"]:
+                channels = 2
+            else:
+                channels = device.shelly.get("num_inputs")
         elif block.type == "emeter":
-            channels = wrapper.device.shelly.get("num_emeters")
+            channels = device.shelly.get("num_emeters")
         elif block.type in ["relay", "light"]:
-            channels = wrapper.device.shelly.get("num_outputs")
+            channels = device.shelly.get("num_outputs")
         elif block.type in ["roller", "device"]:
             channels = 1
 
@@ -55,21 +63,17 @@ def get_entity_name(
         if channels > 1 and block.type != "device":
             entity_name = None
             mode = block.type + "s"
-            if mode in wrapper.device.settings:
+            if mode in device.settings:
-                entity_name = wrapper.device.settings[mode][int(block.channel)].get(
+                entity_name = device.settings[mode][int(block.channel)].get("name")
-                    "name"
-                )
 
             if not entity_name:
-                if wrapper.model == "SHEM-3":
+                if device.settings["device"]["type"] == "SHEM-3":
                     base = ord("A")
                 else:
                     base = ord("1")
-                entity_name = f"{wrapper.name} channel {chr(int(block.channel)+base)}"
+                entity_name = (
-
+                    f"{get_device_name(device)} channel {chr(int(block.channel)+base)}"
-        # Shelly Dimmer has two input channels and missing "num_inputs"
+                )
-        if wrapper.model in ["SHDM-1", "SHDM-2"] and block.type == "input":
-            entity_name = f"{entity_name} channel {int(block.channel)+1}"
 
     if description:
         entity_name = f"{entity_name} {description}"