merge

homeassistant/components/lutron_caseta/__init__.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import contextlib
+from functools import partial
 from itertools import chain
 import logging
 import ssl
@@ -14,26 +15,19 @@ from pylutron_caseta.smartbridge import Smartbridge
 import voluptuous as vol
 
 from homeassistant import config_entries
-from homeassistant.const import ATTR_DEVICE_ID, ATTR_SUGGESTED_AREA, CONF_HOST, Platform
+from homeassistant.const import ATTR_SUGGESTED_AREA, CONF_HOST, Platform
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import device_registry as dr, entity_registry as er
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.dispatcher import async_dispatcher_send
 from homeassistant.helpers.entity import Entity
 from homeassistant.helpers.typing import ConfigType
 
 from .const import (
     ACTION_PRESS,
     ACTION_RELEASE,
-    ATTR_ACTION,
-    ATTR_AREA_NAME,
-    ATTR_BUTTON_NUMBER,
-    ATTR_BUTTON_TYPE,
-    ATTR_DEVICE_NAME,
-    ATTR_LEAP_BUTTON_NUMBER,
-    ATTR_SERIAL,
-    ATTR_TYPE,
     BRIDGE_DEVICE_ID,
     BRIDGE_TIMEOUT,
     CONF_CA_CERTS,
@@ -63,6 +57,8 @@ from .models import (
     LUTRON_KEYPAD_SERIAL,
     LUTRON_KEYPAD_TYPE,
     LutronButton,
+    LutronCasetaButtonDevice,
+    LutronCasetaButtonEventData,
     LutronCasetaConfigEntry,
     LutronCasetaData,
     LutronKeypad,
@@ -95,6 +91,7 @@ PLATFORMS = [
     Platform.BINARY_SENSOR,
     Platform.BUTTON,
     Platform.COVER,
+    Platform.EVENT,
     Platform.FAN,
     Platform.LIGHT,
     Platform.SCENE,
@@ -200,14 +197,72 @@ async def async_setup_entry(
 
     # Store this bridge (keyed by entry_id) so it can be retrieved by the
     # platforms we're setting up.
+    button_devices = _async_build_button_devices(bridge, keypad_data)
 
-    entry.runtime_data = LutronCasetaData(bridge, bridge_device, keypad_data)
+    entry.runtime_data = LutronCasetaData(
+        bridge, bridge_device, keypad_data, button_devices
+    )
 
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
 
     return True
 
 
+@callback
+def _async_build_button_devices(
+    bridge: Smartbridge, keypad_data: LutronKeypadData
+) -> list[LutronCasetaButtonDevice]:
+    button_devices = bridge.get_buttons()
+    all_devices = bridge.get_devices()
+    keypads = keypad_data.keypads
+    buttons: list[LutronCasetaButtonDevice] = []
+
+    for button_id, device in button_devices.items():
+        parent_keypad = keypads[device["parent_device"]]
+        parent_device_info = parent_keypad["device_info"]
+        parent_name = parent_device_info["name"]
+
+        button_key: str | None = None
+        button_name: str
+        device_name = cast(str | None, device.get("device_name"))
+        user_defined_name = bool(device_name)
+        if device_name:
+            button_name = device_name
+        else:
+            # device name (button name) is missing, probably a caseta pico
+            # try to get the name using the button number from the triggers
+            # disable the button by default
+            keypad_device = all_devices[device["parent_device"]]
+            button_numbers = LEAP_TO_DEVICE_TYPE_SUBTYPE_MAP.get(
+                keypad_device["type"],
+                {},
+            )
+            button_key = button_numbers.get(int(device["button_number"]))
+            button_name = (
+                button_key
+                or f"button {device['button_number']}".replace("_", " ").title()
+            )
+
+        # Append the child device name to the end of the parent keypad
+        # name to create the entity name
+        full_name = f"{parent_name} {button_name}"
+        # Set the device_info to the same as the Parent Keypad
+        # The entities will be nested inside the keypad device
+        buttons.append(
+            LutronCasetaButtonDevice(
+                button_id,
+                device,
+                button_key,
+                button_name,
+                full_name,
+                user_defined_name,
+                parent_device_info,
+            ),
+        )
+
+    return buttons
+
+
 @callback
 def _async_register_bridge_device(
     hass: HomeAssistant, config_entry_id: str, bridge_device: dict, bridge: Smartbridge
@@ -301,6 +356,7 @@ def _async_setup_keypads(
 
     _async_subscribe_keypad_events(
         hass=hass,
+        config_entry_id=config_entry_id,
         bridge=bridge,
         keypads=keypads,
         keypad_buttons=keypad_buttons,
@@ -440,15 +496,16 @@ def async_get_lip_button(device_type: str, leap_button: int) -> int | None:
 @callback
 def _async_subscribe_keypad_events(
     hass: HomeAssistant,
+    config_entry_id: str,
     bridge: Smartbridge,
     keypads: dict[int, LutronKeypad],
     keypad_buttons: dict[int, LutronButton],
     leap_to_keypad_button_names: dict[int, dict[int, str]],
-):
+) -> None:
     """Subscribe to lutron events."""
 
     @callback
-    def _async_button_event(button_id, event_type):
+    def _async_button_event(button_id: int, event_type: str) -> None:
         if not (button := keypad_buttons.get(button_id)) or not (
             keypad := keypads.get(button["parent_keypad"])
         ):
@@ -467,27 +524,24 @@ def _async_subscribe_keypad_events(
             keypad_type, leap_to_keypad_button_names[keypad_device_id]
         )[leap_button_number]
 
-        hass.bus.async_fire(
-            LUTRON_CASETA_BUTTON_EVENT,
-            {
-                ATTR_SERIAL: keypad[LUTRON_KEYPAD_SERIAL],
-                ATTR_TYPE: keypad_type,
-                ATTR_BUTTON_NUMBER: lip_button_number,
-                ATTR_LEAP_BUTTON_NUMBER: leap_button_number,
-                ATTR_DEVICE_NAME: keypad[LUTRON_KEYPAD_NAME],
-                ATTR_DEVICE_ID: keypad[LUTRON_KEYPAD_DEVICE_REGISTRY_DEVICE_ID],
-                ATTR_AREA_NAME: keypad[LUTRON_KEYPAD_AREA_NAME],
-                ATTR_BUTTON_TYPE: button_type,
-                ATTR_ACTION: action,
-            },
+        data = LutronCasetaButtonEventData(
+            serial=keypad[LUTRON_KEYPAD_SERIAL],
+            type=keypad_type,
+            button_number=lip_button_number,
+            leap_button_number=leap_button_number,
+            device_name=keypad[LUTRON_KEYPAD_NAME],
+            device_id=keypad[LUTRON_KEYPAD_DEVICE_REGISTRY_DEVICE_ID],
+            area_name=keypad[LUTRON_KEYPAD_AREA_NAME],
+            button_type=button_type,
+            action=action,
         )
+        signal = f"{DOMAIN}_{config_entry_id}_button_{button_id}"
+        async_dispatcher_send(hass, signal, data)
+        hass.bus.async_fire(LUTRON_CASETA_BUTTON_EVENT, data)
 
     for button_id in keypad_buttons:
         bridge.add_button_subscriber(
-            str(button_id),
-            lambda event_type, button_id=button_id: _async_button_event(
-                button_id, event_type
-            ),
+            str(button_id), partial(_async_button_event, button_id)
         )
 
 

homeassistant/components/lutron_caseta/button.py

@@ -2,16 +2,12 @@
 
 from __future__ import annotations
 
-from typing import Any
-
 from homeassistant.components.button import ButtonEntity
 from homeassistant.core import HomeAssistant
-from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
 from . import LutronCasetaDevice
-from .device_trigger import LEAP_TO_DEVICE_TYPE_SUBTYPE_MAP
-from .models import LutronCasetaConfigEntry, LutronCasetaData
+from .models import LutronCasetaButtonDevice, LutronCasetaConfigEntry, LutronCasetaData
 
 
 async def async_setup_entry(
@@ -21,66 +17,26 @@ async def async_setup_entry(
 ) -> None:
     """Set up Lutron pico and keypad buttons."""
     data = config_entry.runtime_data
-    bridge = data.bridge
-    button_devices = bridge.get_buttons()
-    all_devices = data.bridge.get_devices()
-    keypads = data.keypad_data.keypads
-    entities: list[LutronCasetaButton] = []
-
-    for device in button_devices.values():
-        parent_keypad = keypads[device["parent_device"]]
-        parent_device_info = parent_keypad["device_info"]
-
-        enabled_default = True
-        if not (device_name := device.get("device_name")):
-            # device name (button name) is missing, probably a caseta pico
-            # try to get the name using the button number from the triggers
-            # disable the button by default
-            enabled_default = False
-            keypad_device = all_devices[device["parent_device"]]
-            button_numbers = LEAP_TO_DEVICE_TYPE_SUBTYPE_MAP.get(
-                keypad_device["type"],
-                {},
-            )
-            device_name = (
-                button_numbers.get(
-                    int(device["button_number"]),
-                    f"button {device['button_number']}",
-                )
-                .replace("_", " ")
-                .title()
-            )
-
-        # Append the child device name to the end of the parent keypad
-        # name to create the entity name
-        full_name = f'{parent_device_info.get("name")} {device_name}'
-        # Set the device_info to the same as the Parent Keypad
-        # The entities will be nested inside the keypad device
-        entities.append(
-            LutronCasetaButton(
-                device, data, full_name, enabled_default, parent_device_info
-            ),
-        )
-
-    async_add_entities(entities)
+    async_add_entities(
+        LutronCasetaButton(data, button_device) for button_device in data.button_devices
+    )
 
 
 class LutronCasetaButton(LutronCasetaDevice, ButtonEntity):
     """Representation of a Lutron pico and keypad button."""
 
+    _attr_has_entity_name = True
+
     def __init__(
         self,
-        device: dict[str, Any],
         data: LutronCasetaData,
-        full_name: str,
-        enabled_default: bool,
-        device_info: DeviceInfo,
+        button_device: LutronCasetaButtonDevice,
     ) -> None:
         """Init a button entity."""
-        super().__init__(device, data)
-        self._attr_entity_registry_enabled_default = enabled_default
-        self._attr_name = full_name
-        self._attr_device_info = device_info
+        super().__init__(button_device.device, data)
+        self._attr_entity_registry_enabled_default = button_device.user_defined_name
+        self._attr_name = button_device.button_name
+        self._attr_device_info = button_device.parent_device_info
 
     async def async_press(self) -> None:
         """Send a button press event."""

homeassistant/components/lutron_caseta/const.py

@@ -1,5 +1,7 @@
 """Lutron Caseta constants."""
 
+from typing import Final
+
 DOMAIN = "lutron_caseta"
 
 CONF_KEYFILE = "keyfile"
@@ -19,14 +21,14 @@ DEVICE_TYPE_SPECTRUM_TUNE = "SpectrumTune"
 
 MANUFACTURER = "Lutron Electronics Co., Inc"
 
-ATTR_SERIAL = "serial"
-ATTR_TYPE = "type"
-ATTR_BUTTON_TYPE = "button_type"
-ATTR_LEAP_BUTTON_NUMBER = "leap_button_number"
-ATTR_BUTTON_NUMBER = "button_number"  # LIP button number
-ATTR_DEVICE_NAME = "device_name"
-ATTR_AREA_NAME = "area_name"
-ATTR_ACTION = "action"
+ATTR_SERIAL: Final = "serial"
+ATTR_TYPE: Final = "type"
+ATTR_BUTTON_TYPE: Final = "button_type"
+ATTR_LEAP_BUTTON_NUMBER: Final = "leap_button_number"
+ATTR_BUTTON_NUMBER: Final = "button_number"  # LIP button number
+ATTR_DEVICE_NAME: Final = "device_name"
+ATTR_AREA_NAME: Final = "area_name"
+ATTR_ACTION: Final = "action"
 
 ACTION_PRESS = "press"
 ACTION_RELEASE = "release"

homeassistant/components/lutron_caseta/event.py

@@ -0,0 +1,74 @@
+"""Support for pico and keypad button events."""
+
+from __future__ import annotations
+
+from homeassistant.components.event import EventDeviceClass, EventEntity
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.dispatcher import async_dispatcher_connect
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from . import LutronCasetaDevice
+from .const import ACTION_PRESS, ACTION_RELEASE, DOMAIN
+from .models import (
+    LutronCasetaButtonDevice,
+    LutronCasetaButtonEventData,
+    LutronCasetaConfigEntry,
+    LutronCasetaData,
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: LutronCasetaConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up Lutron pico and keypad buttons."""
+    data = config_entry.runtime_data
+    async_add_entities(
+        LutronCasetaButtonEvent(data, button_device, config_entry.entry_id)
+        for button_device in data.button_devices
+    )
+
+
+class LutronCasetaButtonEvent(LutronCasetaDevice, EventEntity):
+    """Representation of a Lutron pico and keypad button event."""
+
+    _attr_device_class = EventDeviceClass.BUTTON
+    _attr_event_types = [ACTION_PRESS, ACTION_RELEASE]
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        data: LutronCasetaData,
+        button_device: LutronCasetaButtonDevice,
+        entry_id: str,
+    ) -> None:
+        """Init a button event entity."""
+        super().__init__(button_device.device, data)
+        self._attr_name = button_device.button_name
+        self._attr_translation_key = button_device.button_key
+        self._attr_device_info = button_device.parent_device_info
+        self._button_id = button_device.button_id
+        self._entry_id = entry_id
+
+    @property
+    def serial(self):
+        """Buttons shouldn't have serial numbers, Return None."""
+        return None
+
+    @callback
+    def _async_handle_button_event(self, data: LutronCasetaButtonEventData) -> None:
+        """Handle a button event."""
+        self._trigger_event(data["action"])
+        self.async_write_ha_state()
+
+    async def async_added_to_hass(self) -> None:
+        """Subscribe to button events."""
+        self.async_on_remove(
+            async_dispatcher_connect(
+                self.hass,
+                f"{DOMAIN}_{self._entry_id}_button_{self._button_id}",
+                self._async_handle_button_event,
+            )
+        )
+        await super().async_added_to_hass()

homeassistant/components/lutron_caseta/models.py

@@ -14,16 +14,17 @@ from homeassistant.helpers.device_registry import DeviceInfo
 type LutronCasetaConfigEntry = ConfigEntry[LutronCasetaData]
 
 
-@dataclass
+@dataclass(slots=True)
 class LutronCasetaData:
     """Data for the lutron_caseta integration."""
 
     bridge: Smartbridge
     bridge_device: dict[str, Any]
     keypad_data: LutronKeypadData
+    button_devices: list[LutronCasetaButtonDevice]
 
 
-@dataclass
+@dataclass(slots=True)
 class LutronKeypadData:
     """Data for the lutron_caseta integration keypads."""
 
@@ -34,6 +35,33 @@ class LutronKeypadData:
     trigger_schemas: dict[int, vol.Schema]
 
 
+@dataclass(slots=True)
+class LutronCasetaButtonDevice:
+    """A lutron_caseta button device."""
+
+    button_id: int
+    device: dict
+    button_key: str | None
+    button_name: str
+    full_name: str
+    user_defined_name: bool
+    parent_device_info: DeviceInfo
+
+
+class LutronCasetaButtonEventData(TypedDict):
+    """A lutron_caseta button event data."""
+
+    serial: str
+    type: str
+    button_number: int | None
+    leap_button_number: int
+    device_name: str
+    device_id: str
+    area_name: str
+    button_type: str
+    action: str
+
+
 class LutronKeypad(TypedDict):
     """A lutron_caseta keypad device."""
 

tests/components/lutron_caseta/__init__.py

@@ -1,5 +1,6 @@
 """Tests for the Lutron Caseta integration."""
 
+from collections import defaultdict
 from unittest.mock import patch
 
 from homeassistant.components.lutron_caseta import DOMAIN
@@ -84,7 +85,9 @@ _LEAP_DEVICE_TYPES = {
 }
 
 
-async def async_setup_integration(hass: HomeAssistant, mock_bridge) -> MockConfigEntry:
+async def async_setup_integration(
+    hass: HomeAssistant, mock_bridge
+) -> tuple[MockConfigEntry, "MockBridge"]:
     """Set up a mock bridge."""
     mock_entry = MockConfigEntry(domain=DOMAIN, data=ENTRY_MOCK_DATA)
     mock_entry.add_to_hass(hass)
@@ -92,10 +95,12 @@ async def async_setup_integration(hass: HomeAssistant, mock_bridge) -> MockConfi
     with patch(
         "homeassistant.components.lutron_caseta.Smartbridge.create_tls"
     ) as create_tls:
-        create_tls.return_value = mock_bridge(can_connect=True)
+        mocked_bridge = mock_bridge(can_connect=True)
+        create_tls.return_value = mocked_bridge
         await hass.config_entries.async_setup(mock_entry.entry_id)
         await hass.async_block_till_done()
-    return mock_entry
+
+    return mock_entry, mocked_bridge
 
 
 class MockBridge:
@@ -110,6 +115,7 @@ class MockBridge:
         self.scenes = self.get_scenes()
         self.devices = self.load_devices()
         self.buttons = self.load_buttons()
+        self.button_subscribers: defaultdict[str, list] = defaultdict(list)
 
     async def connect(self):
         """Connect the mock bridge."""
@@ -121,6 +127,7 @@ class MockBridge:
 
     def add_button_subscriber(self, button_id: str, callback_):
         """Mock a listener for button presses."""
+        self.button_subscribers[button_id].append(callback_)
 
     def is_connected(self):
         """Return whether the mock bridge is connected."""

tests/components/lutron_caseta/test_device_trigger.py

@@ -7,16 +7,14 @@ from pytest_unordered import unordered
 
 from homeassistant.components import automation
 from homeassistant.components.device_automation import DeviceAutomationType
-from homeassistant.components.lutron_caseta import (
+from homeassistant.components.lutron_caseta.const import (
     ATTR_ACTION,
     ATTR_AREA_NAME,
+    ATTR_BUTTON_TYPE,
     ATTR_DEVICE_NAME,
+    ATTR_LEAP_BUTTON_NUMBER,
     ATTR_SERIAL,
     ATTR_TYPE,
-)
-from homeassistant.components.lutron_caseta.const import (
-    ATTR_BUTTON_TYPE,
-    ATTR_LEAP_BUTTON_NUMBER,
     CONF_CA_CERTS,
     CONF_CERTFILE,
     CONF_KEYFILE,
@@ -159,6 +157,7 @@ async def test_get_triggers(hass: HomeAssistant) -> None:
     triggers = await async_get_device_automations(
         hass, DeviceAutomationType.TRIGGER, device_id
     )
+    triggers = [trigger for trigger in triggers if trigger[CONF_DOMAIN] == DOMAIN]
 
     assert triggers == unordered(expected_triggers)