Fix lutron caseta triggers when device fails to setup before startup …

…finishes

Refactor the trigger attach so its not dependant on the intergration
being already setup

fixes #82495 & fixes #81999

homeassistant/components/lutron_caseta/__init__.py

@@ -6,7 +6,7 @@
 from itertools import chain
 import logging
 import ssl
-from typing import Any
+from typing import Any, cast
 
 import async_timeout
 from pylutron_caseta import BUTTON_STATUS_PRESSED
@@ -27,6 +27,7 @@
     ACTION_RELEASE,
     ATTR_ACTION,
     ATTR_AREA_NAME,
+    ATTR_BUTTON_NAME,
     ATTR_BUTTON_NUMBER,
     ATTR_DEVICE_NAME,
     ATTR_LEAP_BUTTON_NUMBER,
@@ -50,7 +51,21 @@
     LEAP_TO_DEVICE_TYPE_SUBTYPE_MAP,
     LUTRON_BUTTON_TRIGGER_SCHEMA,
 )
-from .models import LutronButton, LutronCasetaData, LutronKeypad, LutronKeypadData
+from .models import (
+    LUTRON_BUTTON_LEAP_BUTTON_NUMBER,
+    LUTRON_KEYPAD_AREA_NAME,
+    LUTRON_KEYPAD_BUTTONS,
+    LUTRON_KEYPAD_DEVICE_REGISTRY_DEVICE_ID,
+    LUTRON_KEYPAD_LUTRON_DEVICE_ID,
+    LUTRON_KEYPAD_MODEL,
+    LUTRON_KEYPAD_NAME,
+    LUTRON_KEYPAD_SERIAL,
+    LUTRON_KEYPAD_TYPE,
+    LutronButton,
+    LutronCasetaData,
+    LutronKeypad,
+    LutronKeypadData,
+)
 from .util import serial_to_unique_id
 
 _LOGGER = logging.getLogger(__name__)
@@ -225,57 +240,77 @@ def _async_setup_keypads(
     hass: HomeAssistant,
     config_entry_id: str,
     bridge: Smartbridge,
-    bridge_device: dict[str, Any],
+    bridge_device: dict[str, str | int],
 ) -> LutronKeypadData:
     """Register keypad devices (Keypads and Pico Remotes) in the device registry."""
 
     device_registry = dr.async_get(hass)
 
-    bridge_devices = bridge.get_devices()
-    bridge_buttons = bridge.buttons
+    bridge_devices: dict[str, dict[str, str | int]] = bridge.get_devices()
+    bridge_buttons: dict[str, dict[str, str | int]] = bridge.buttons
 
     dr_device_id_to_keypad: dict[str, LutronKeypad] = {}
     keypads: dict[int, LutronKeypad] = {}
     keypad_buttons: dict[int, LutronButton] = {}
     keypad_button_names_to_leap: dict[int, dict[str, int]] = {}
+    leap_to_keypad_button_names: dict[int, dict[int, str]] = {}
 
     for bridge_button in bridge_buttons.values():
 
-        bridge_keypad = bridge_devices[bridge_button["parent_device"]]
-        keypad_device_id = bridge_keypad["device_id"]
-        button_device_id = bridge_button["device_id"]
+        parent_device = cast(str, bridge_button["parent_device"])
+        bridge_keypad = bridge_devices[parent_device]
+        keypad_lutron_device_id = cast(int, bridge_keypad["device_id"])
+        button_lutron_device_id = cast(int, bridge_button["device_id"])
+        leap_button_number = cast(int, bridge_button["button_number"])
+        button_led_device_id = None
+        if "button_led" in bridge_button:
+            button_led_device_id = cast(str, bridge_button["button_led"])
 
-        if not (keypad := keypads.get(keypad_device_id)):
+        if not (keypad := keypads.get(keypad_lutron_device_id)):
             # First time seeing this keypad, build keypad data and store in keypads
-            keypad = keypads[keypad_device_id] = _async_build_lutron_keypad(
-                bridge, bridge_device, bridge_keypad, keypad_device_id
+            keypad = keypads[keypad_lutron_device_id] = _async_build_lutron_keypad(
+                bridge, bridge_device, bridge_keypad, keypad_lutron_device_id
             )
 
             # Register the keypad device
             dr_device = device_registry.async_get_or_create(
                 **keypad["device_info"], config_entry_id=config_entry_id
             )
-            keypad["dr_device_id"] = dr_device.id
+            keypad[LUTRON_KEYPAD_DEVICE_REGISTRY_DEVICE_ID] = dr_device.id
             dr_device_id_to_keypad[dr_device.id] = keypad
 
+        button_name = _get_button_name(keypad, bridge_button)
+        keypad_lutron_device_id = keypad[LUTRON_KEYPAD_LUTRON_DEVICE_ID]
+
         # Add button to parent keypad, and build keypad_buttons and keypad_button_names_to_leap
-        button = keypad_buttons[button_device_id] = LutronButton(
-            lutron_device_id=button_device_id,
-            leap_button_number=bridge_button["button_number"],
-            button_name=_get_button_name(keypad, bridge_button),
-            led_device_id=bridge_button.get("button_led"),
-            parent_keypad=keypad["lutron_device_id"],
+        keypad_buttons[button_lutron_device_id] = LutronButton(
+            lutron_device_id=button_lutron_device_id,
+            leap_button_number=leap_button_number,
+            button_name=button_name,
+            led_device_id=button_led_device_id,
+            parent_keypad=keypad_lutron_device_id,
         )
 
-        keypad["buttons"].append(button["lutron_device_id"])
+        keypad[LUTRON_KEYPAD_BUTTONS].append(button_lutron_device_id)
 
-        keypad_button_names_to_leap.setdefault(keypad["lutron_device_id"], {}).update(
-            {button["button_name"]: int(button["leap_button_number"])}
+        button_name_to_leap = keypad_button_names_to_leap.setdefault(
+            keypad_lutron_device_id, {}
         )
+        button_name_to_leap[button_name] = leap_button_number
+        leap_to_button_name = leap_to_keypad_button_names.setdefault(
+            keypad_lutron_device_id, {}
+        )
+        leap_to_button_name[leap_button_number] = button_name
 
     keypad_trigger_schemas = _async_build_trigger_schemas(keypad_button_names_to_leap)
 
-    _async_subscribe_keypad_events(hass, bridge, keypads, keypad_buttons)
+    _async_subscribe_keypad_events(
+        hass=hass,
+        bridge=bridge,
+        keypads=keypads,
+        keypad_buttons=keypad_buttons,
+        leap_to_keypad_button_names=leap_to_keypad_button_names,
+    )
 
     return LutronKeypadData(
         dr_device_id_to_keypad,
@@ -312,7 +347,6 @@ def _async_build_lutron_keypad(
     keypad_device_id: int,
 ) -> LutronKeypad:
     # First time seeing this keypad, build keypad data and store in keypads
-
     area_name = _area_name_from_id(bridge.areas, bridge_keypad["area"])
     keypad_name = bridge_keypad["name"].split("_")[-1]
     keypad_serial = _handle_none_keypad_serial(bridge_keypad, bridge_device["serial"])
@@ -350,7 +384,7 @@ def _get_button_name(keypad: LutronKeypad, bridge_button: dict[str, Any]) -> str
         # This is a Caseta Button retrieve name from hardcoded trigger definitions.
         return _get_button_name_from_triggers(keypad, button_number)
 
-    keypad_model = keypad["model"]
+    keypad_model = keypad[LUTRON_KEYPAD_MODEL]
     if keypad_model_override := KEYPAD_LEAP_BUTTON_NAME_OVERRIDE.get(keypad_model):
         if alt_button_name := keypad_model_override.get(button_number):
             return alt_button_name
@@ -412,8 +446,9 @@ def async_get_lip_button(device_type: str, leap_button: int) -> int | None:
 def _async_subscribe_keypad_events(
     hass: HomeAssistant,
     bridge: Smartbridge,
-    keypads: dict[int, Any],
-    keypad_buttons: dict[int, Any],
+    keypads: dict[int, LutronKeypad],
+    keypad_buttons: dict[int, LutronButton],
+    leap_to_keypad_button_names: dict[int, dict[int, str]],
 ):
     """Subscribe to lutron events."""
 
@@ -429,20 +464,25 @@ def _async_button_event(button_id, event_type):
         else:
             action = ACTION_RELEASE
 
-        keypad_type = keypad["type"]
-        leap_button_number = button["leap_button_number"]
+        keypad_type = keypad[LUTRON_KEYPAD_TYPE]
+        keypad_device_id = keypad[LUTRON_KEYPAD_LUTRON_DEVICE_ID]
+        leap_button_number = button[LUTRON_BUTTON_LEAP_BUTTON_NUMBER]
         lip_button_number = async_get_lip_button(keypad_type, leap_button_number)
+        button_name = LEAP_TO_DEVICE_TYPE_SUBTYPE_MAP.get(
+            keypad_type, leap_to_keypad_button_names[keypad_device_id]
+        )[leap_button_number]
 
         hass.bus.async_fire(
             LUTRON_CASETA_BUTTON_EVENT,
             {
-                ATTR_SERIAL: keypad["serial"],
+                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["name"],
-                ATTR_DEVICE_ID: keypad["dr_device_id"],
-                ATTR_AREA_NAME: keypad["area_name"],
+                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_NAME: button_name,
                 ATTR_ACTION: action,
             },
         )

homeassistant/components/lutron_caseta/const.py

@@ -18,6 +18,7 @@
 
 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"

homeassistant/components/lutron_caseta/device_trigger.py

@@ -6,9 +6,6 @@
 import voluptuous as vol
 
 from homeassistant.components.device_automation import DEVICE_TRIGGER_BASE_SCHEMA
-from homeassistant.components.device_automation.exceptions import (
-    InvalidDeviceAutomationConfig,
-)
 from homeassistant.components.homeassistant.triggers import event as event_trigger
 from homeassistant.const import (
     CONF_DEVICE_ID,
@@ -18,16 +15,14 @@
     CONF_TYPE,
 )
 from homeassistant.core import CALLBACK_TYPE, HomeAssistant
-from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo
 from homeassistant.helpers.typing import ConfigType
 
 from .const import (
     ACTION_PRESS,
     ACTION_RELEASE,
     ATTR_ACTION,
-    ATTR_LEAP_BUTTON_NUMBER,
-    ATTR_SERIAL,
+    ATTR_BUTTON_TYPE,
     CONF_SUBTYPE,
     DOMAIN,
     LUTRON_CASETA_BUTTON_EVENT,
@@ -317,7 +312,7 @@ def _reverse_dict(forward_dict: dict) -> dict:
     "FourGroupRemote": FOUR_GROUP_REMOTE_BUTTON_TYPES_TO_LEAP,
 }
 
-LEAP_TO_DEVICE_TYPE_SUBTYPE_MAP = {
+LEAP_TO_DEVICE_TYPE_SUBTYPE_MAP: dict[str, dict[int, str]] = {
     k: _reverse_dict(v) for k, v in DEVICE_TYPE_SUBTYPE_MAP_TO_LEAP.items()
 }
 
@@ -421,53 +416,22 @@ async def async_attach_trigger(
     trigger_info: TriggerInfo,
 ) -> CALLBACK_TYPE:
     """Attach a trigger."""
-    device_id = config[CONF_DEVICE_ID]
-    subtype = config[CONF_SUBTYPE]
-    if not (data := get_lutron_data_by_dr_id(hass, device_id)) or not (
-        keypad := data.keypad_data.dr_device_id_to_keypad[device_id]
-    ):
-        raise HomeAssistantError(
-            f"Cannot attach trigger {config} because device with id {device_id} is missing or invalid"
-        )
-
-    keypad_trigger_schemas = data.keypad_data.trigger_schemas
-    keypad_button_names_to_leap = data.keypad_data.button_names_to_leap
-
-    device_type = keypad["type"]
-    serial = keypad["serial"]
-    lutron_device_id = keypad["lutron_device_id"]
-
-    # Retrieve trigger schema, preferring hard-coded triggers from device_trigger.py
-    schema = DEVICE_TYPE_SCHEMA_MAP.get(
-        device_type,
-        keypad_trigger_schemas[lutron_device_id],
-    )
-
-    # Retrieve list of valid buttons, preferring hard-coded triggers from device_trigger.py
-    valid_buttons = DEVICE_TYPE_SUBTYPE_MAP_TO_LEAP.get(
-        device_type,
-        keypad_button_names_to_leap[lutron_device_id],
-    )
-
-    if subtype not in valid_buttons:
-        raise InvalidDeviceAutomationConfig(
-            f"Cannot attach trigger {config} because subtype {subtype} is invalid"
-        )
-
-    config = schema(config)
-    event_config = {
-        event_trigger.CONF_PLATFORM: CONF_EVENT,
-        event_trigger.CONF_EVENT_TYPE: LUTRON_CASETA_BUTTON_EVENT,
-        event_trigger.CONF_EVENT_DATA: {
-            ATTR_SERIAL: serial,
-            ATTR_LEAP_BUTTON_NUMBER: valid_buttons[subtype],
-            ATTR_ACTION: config[CONF_TYPE],
-        },
-    }
-    event_config = event_trigger.TRIGGER_SCHEMA(event_config)
-
     return await event_trigger.async_attach_trigger(
-        hass, event_config, action, trigger_info, platform_type="device"
+        hass,
+        event_trigger.TRIGGER_SCHEMA(
+            {
+                event_trigger.CONF_PLATFORM: CONF_EVENT,
+                event_trigger.CONF_EVENT_TYPE: LUTRON_CASETA_BUTTON_EVENT,
+                event_trigger.CONF_EVENT_DATA: {
+                    CONF_DEVICE_ID: config[CONF_DEVICE_ID],
+                    ATTR_ACTION: config[CONF_TYPE],
+                    ATTR_BUTTON_TYPE: config[CONF_SUBTYPE],
+                },
+            }
+        ),
+        action,
+        trigger_info,
+        platform_type="device",
     )
 
 

homeassistant/components/lutron_caseta/models.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Any, TypedDict
+from typing import Any, Final, TypedDict
 
 from pylutron_caseta.smartbridge import Smartbridge
 import voluptuous as vol
@@ -45,11 +45,30 @@ class LutronKeypad(TypedDict):
     buttons: list[int]
 
 
+LUTRON_KEYPAD_LUTRON_DEVICE_ID: Final = "lutron_device_id"
+LUTRON_KEYPAD_DEVICE_REGISTRY_DEVICE_ID: Final = "dr_device_id"
+LUTRON_KEYPAD_AREA_ID: Final = "area_id"
+LUTRON_KEYPAD_AREA_NAME: Final = "area_name"
+LUTRON_KEYPAD_NAME: Final = "name"
+LUTRON_KEYPAD_SERIAL: Final = "serial"
+LUTRON_KEYPAD_DEVICE_INFO: Final = "device_info"
+LUTRON_KEYPAD_MODEL: Final = "model"
+LUTRON_KEYPAD_TYPE: Final = "type"
+LUTRON_KEYPAD_BUTTONS: Final = "buttons"
+
+
 class LutronButton(TypedDict):
     """A lutron_caseta button."""
 
     lutron_device_id: int
     leap_button_number: int
     button_name: str
-    led_device_id: int
+    led_device_id: str | None
     parent_keypad: int
+
+
+LUTRON_BUTTON_LUTRON_DEVICE_ID: Final = "lutron_device_id"
+LUTRON_BUTTON_LEAP_BUTTON_NUMBER: Final = "leap_button_number"
+LUTRON_BUTTON_BUTTON_NAME: Final = "button_name"
+LUTRON_BUTTON_LED_DEVICE_ID: Final = "led_device_id"
+LUTRON_BUTTON_PARENT_KEYPAD: Final = "parent_keypad"