Move rfxtrx base entity to separate module (#126521)

homeassistant/components/rfxtrx/__init__.py

@@ -25,21 +25,16 @@ from homeassistant.const import (
 from homeassistant.core import Event, HomeAssistant, ServiceCall, callback
 from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import config_validation as cv, device_registry as dr
-from homeassistant.helpers.device_registry import (
-    DeviceInfo,
-    EventDeviceRegistryUpdatedData,
-)
+from homeassistant.helpers.device_registry import EventDeviceRegistryUpdatedData
 from homeassistant.helpers.dispatcher import (
     async_dispatcher_connect,
     async_dispatcher_send,
 )
 from homeassistant.helpers.entity import Entity
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
-from homeassistant.helpers.restore_state import RestoreEntity
 
 from .const import (
     ATTR_EVENT,
-    COMMAND_GROUP_LIST,
     CONF_AUTOMATIC_ADD,
     CONF_DATA_BITS,
     CONF_PROTOCOLS,
@@ -48,11 +43,11 @@ from .const import (
     DOMAIN,
     EVENT_RFXTRX_EVENT,
     SERVICE_SEND,
+    SIGNAL_EVENT,
 )
 
 DEFAULT_OFF_DELAY = 2.0
 
-SIGNAL_EVENT = f"{DOMAIN}_event"
 CONNECT_TIMEOUT = 30.0
 
 _LOGGER = logging.getLogger(__name__)
@@ -461,14 +456,6 @@ def get_device_tuple_from_identifiers(
     return DeviceTuple(identifier2[1], identifier2[2], identifier2[3])
 
 
-def get_identifiers_from_device_tuple(
-    device_tuple: DeviceTuple,
-) -> set[tuple[str, str]]:
-    """Calculate the device identifier from a device tuple."""
-    # work around legacy identifier, being a multi tuple value
-    return {(DOMAIN, *device_tuple)}  # type: ignore[arg-type]
-
-
 async def async_remove_config_entry_device(
     hass: HomeAssistant, config_entry: ConfigEntry, device_entry: dr.DeviceEntry
 ) -> bool:
@@ -477,102 +464,3 @@ async def async_remove_config_entry_device(
     The actual cleanup is done in the device registry event
     """
     return True
-
-
-class RfxtrxEntity(RestoreEntity):
-    """Represents a Rfxtrx device.
-
-    Contains the common logic for Rfxtrx lights and switches.
-    """
-
-    _attr_assumed_state = True
-    _attr_has_entity_name = True
-    _attr_should_poll = False
-    _device: rfxtrxmod.RFXtrxDevice
-    _event: rfxtrxmod.RFXtrxEvent | None
-
-    def __init__(
-        self,
-        device: rfxtrxmod.RFXtrxDevice,
-        device_id: DeviceTuple,
-        event: rfxtrxmod.RFXtrxEvent | None = None,
-    ) -> None:
-        """Initialize the device."""
-        self._attr_device_info = DeviceInfo(
-            identifiers=get_identifiers_from_device_tuple(device_id),
-            model=device.type_string,
-            name=f"{device.type_string} {device.id_string}",
-        )
-        self._attr_unique_id = "_".join(x for x in device_id)
-        self._device = device
-        self._event = event
-        self._device_id = device_id
-        # If id_string is 213c7f2:1, the group_id is 213c7f2, and the device will respond to
-        # group events regardless of their group indices.
-        (self._group_id, _, _) = cast(str, device.id_string).partition(":")
-
-    async def async_added_to_hass(self) -> None:
-        """Restore RFXtrx device state (ON/OFF)."""
-        if self._event:
-            self._apply_event(self._event)
-
-        self.async_on_remove(
-            async_dispatcher_connect(self.hass, SIGNAL_EVENT, self._handle_event)
-        )
-
-    @property
-    def extra_state_attributes(self) -> dict[str, str] | None:
-        """Return the device state attributes."""
-        if not self._event:
-            return None
-        return {ATTR_EVENT: "".join(f"{x:02x}" for x in self._event.data)}
-
-    def _event_applies(
-        self, event: rfxtrxmod.RFXtrxEvent, device_id: DeviceTuple
-    ) -> bool:
-        """Check if event applies to me."""
-        if isinstance(event, rfxtrxmod.ControlEvent):
-            if (
-                "Command" in event.values
-                and event.values["Command"] in COMMAND_GROUP_LIST
-            ):
-                device: rfxtrxmod.RFXtrxDevice = event.device
-                (group_id, _, _) = cast(str, device.id_string).partition(":")
-                return group_id == self._group_id
-
-        # Otherwise, the event only applies to the matching device.
-        return device_id == self._device_id
-
-    def _apply_event(self, event: rfxtrxmod.RFXtrxEvent) -> None:
-        """Apply a received event."""
-        self._event = event
-
-    @callback
-    def _handle_event(
-        self, event: rfxtrxmod.RFXtrxEvent, device_id: DeviceTuple
-    ) -> None:
-        """Handle a reception of data, overridden by other classes."""
-
-
-class RfxtrxCommandEntity(RfxtrxEntity):
-    """Represents a Rfxtrx device.
-
-    Contains the common logic for Rfxtrx lights and switches.
-    """
-
-    _attr_name = None
-
-    def __init__(
-        self,
-        device: rfxtrxmod.RFXtrxDevice,
-        device_id: DeviceTuple,
-        event: rfxtrxmod.RFXtrxEvent | None = None,
-    ) -> None:
-        """Initialzie a switch or light device."""
-        super().__init__(device, device_id, event=event)
-
-    async def _async_send[*_Ts](
-        self, fun: Callable[[rfxtrxmod.PySerialTransport, *_Ts], None], *args: *_Ts
-    ) -> None:
-        rfx_object: rfxtrxmod.Connect = self.hass.data[DOMAIN][DATA_RFXOBJECT]
-        await self.hass.async_add_executor_job(fun, rfx_object.transport, *args)

homeassistant/components/rfxtrx/binary_sensor.py

@@ -19,7 +19,7 @@ from homeassistant.helpers import event as evt
 from homeassistant.helpers.entity import Entity
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from . import DeviceTuple, RfxtrxEntity, async_setup_platform_entry, get_pt2262_cmd
+from . import DeviceTuple, async_setup_platform_entry, get_pt2262_cmd
 from .const import (
     COMMAND_OFF_LIST,
     COMMAND_ON_LIST,
@@ -27,6 +27,7 @@ from .const import (
     CONF_OFF_DELAY,
     DEVICE_PACKET_TYPE_LIGHTING4,
 )
+from .entity import RfxtrxEntity
 
 _LOGGER = logging.getLogger(__name__)
 

homeassistant/components/rfxtrx/const.py

@@ -46,3 +46,5 @@ EVENT_RFXTRX_EVENT = "rfxtrx_event"
 DATA_RFXOBJECT = "rfxobject"
 
 DOMAIN = "rfxtrx"
+
+SIGNAL_EVENT = f"{DOMAIN}_event"

homeassistant/components/rfxtrx/cover.py

@@ -14,7 +14,7 @@ from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.entity import Entity
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from . import DeviceTuple, RfxtrxCommandEntity, async_setup_platform_entry
+from . import DeviceTuple, async_setup_platform_entry
 from .const import (
     COMMAND_OFF_LIST,
     COMMAND_ON_LIST,
@@ -22,6 +22,7 @@ from .const import (
     CONST_VENETIAN_BLIND_MODE_EU,
     CONST_VENETIAN_BLIND_MODE_US,
 )
+from .entity import RfxtrxCommandEntity
 
 _LOGGER = logging.getLogger(__name__)
 

homeassistant/components/rfxtrx/entity.py

@@ -0,0 +1,123 @@
+"""Support for RFXtrx devices."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import cast
+
+import RFXtrx as rfxtrxmod
+
+from homeassistant.core import callback
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.dispatcher import async_dispatcher_connect
+from homeassistant.helpers.restore_state import RestoreEntity
+
+from . import DeviceTuple
+from .const import ATTR_EVENT, COMMAND_GROUP_LIST, DATA_RFXOBJECT, DOMAIN, SIGNAL_EVENT
+
+
+def _get_identifiers_from_device_tuple(
+    device_tuple: DeviceTuple,
+) -> set[tuple[str, str]]:
+    """Calculate the device identifier from a device tuple."""
+    # work around legacy identifier, being a multi tuple value
+    return {(DOMAIN, *device_tuple)}  # type: ignore[arg-type]
+
+
+class RfxtrxEntity(RestoreEntity):
+    """Represents a Rfxtrx device.
+
+    Contains the common logic for Rfxtrx lights and switches.
+    """
+
+    _attr_assumed_state = True
+    _attr_has_entity_name = True
+    _attr_should_poll = False
+    _device: rfxtrxmod.RFXtrxDevice
+    _event: rfxtrxmod.RFXtrxEvent | None
+
+    def __init__(
+        self,
+        device: rfxtrxmod.RFXtrxDevice,
+        device_id: DeviceTuple,
+        event: rfxtrxmod.RFXtrxEvent | None = None,
+    ) -> None:
+        """Initialize the device."""
+        self._attr_device_info = DeviceInfo(
+            identifiers=_get_identifiers_from_device_tuple(device_id),
+            model=device.type_string,
+            name=f"{device.type_string} {device.id_string}",
+        )
+        self._attr_unique_id = "_".join(x for x in device_id)
+        self._device = device
+        self._event = event
+        self._device_id = device_id
+        # If id_string is 213c7f2:1, the group_id is 213c7f2, and the device will respond to
+        # group events regardless of their group indices.
+        (self._group_id, _, _) = cast(str, device.id_string).partition(":")
+
+    async def async_added_to_hass(self) -> None:
+        """Restore RFXtrx device state (ON/OFF)."""
+        if self._event:
+            self._apply_event(self._event)
+
+        self.async_on_remove(
+            async_dispatcher_connect(self.hass, SIGNAL_EVENT, self._handle_event)
+        )
+
+    @property
+    def extra_state_attributes(self) -> dict[str, str] | None:
+        """Return the device state attributes."""
+        if not self._event:
+            return None
+        return {ATTR_EVENT: "".join(f"{x:02x}" for x in self._event.data)}
+
+    def _event_applies(
+        self, event: rfxtrxmod.RFXtrxEvent, device_id: DeviceTuple
+    ) -> bool:
+        """Check if event applies to me."""
+        if isinstance(event, rfxtrxmod.ControlEvent):
+            if (
+                "Command" in event.values
+                and event.values["Command"] in COMMAND_GROUP_LIST
+            ):
+                device: rfxtrxmod.RFXtrxDevice = event.device
+                (group_id, _, _) = cast(str, device.id_string).partition(":")
+                return group_id == self._group_id
+
+        # Otherwise, the event only applies to the matching device.
+        return device_id == self._device_id
+
+    def _apply_event(self, event: rfxtrxmod.RFXtrxEvent) -> None:
+        """Apply a received event."""
+        self._event = event
+
+    @callback
+    def _handle_event(
+        self, event: rfxtrxmod.RFXtrxEvent, device_id: DeviceTuple
+    ) -> None:
+        """Handle a reception of data, overridden by other classes."""
+
+
+class RfxtrxCommandEntity(RfxtrxEntity):
+    """Represents a Rfxtrx device.
+
+    Contains the common logic for Rfxtrx lights and switches.
+    """
+
+    _attr_name = None
+
+    def __init__(
+        self,
+        device: rfxtrxmod.RFXtrxDevice,
+        device_id: DeviceTuple,
+        event: rfxtrxmod.RFXtrxEvent | None = None,
+    ) -> None:
+        """Initialzie a switch or light device."""
+        super().__init__(device, device_id, event=event)
+
+    async def _async_send[*_Ts](
+        self, fun: Callable[[rfxtrxmod.PySerialTransport, *_Ts], None], *args: *_Ts
+    ) -> None:
+        rfx_object: rfxtrxmod.Connect = self.hass.data[DOMAIN][DATA_RFXOBJECT]
+        await self.hass.async_add_executor_job(fun, rfx_object.transport, *args)

homeassistant/components/rfxtrx/event.py

@@ -14,8 +14,9 @@ from homeassistant.helpers.entity import Entity
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.util import slugify
 
-from . import DeviceTuple, RfxtrxEntity, async_setup_platform_entry
+from . import DeviceTuple, async_setup_platform_entry
 from .const import DEVICE_PACKET_TYPE_LIGHTING4
+from .entity import RfxtrxEntity
 
 _LOGGER = logging.getLogger(__name__)
 

homeassistant/components/rfxtrx/light.py

@@ -14,8 +14,9 @@ from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.entity import Entity
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from . import DeviceTuple, RfxtrxCommandEntity, async_setup_platform_entry
+from . import DeviceTuple, async_setup_platform_entry
 from .const import COMMAND_OFF_LIST, COMMAND_ON_LIST
+from .entity import RfxtrxCommandEntity
 
 _LOGGER = logging.getLogger(__name__)
 

homeassistant/components/rfxtrx/sensor.py

@@ -39,8 +39,9 @@ from homeassistant.helpers.entity import Entity
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.typing import StateType
 
-from . import DeviceTuple, RfxtrxEntity, async_setup_platform_entry, get_rfx_object
+from . import DeviceTuple, async_setup_platform_entry, get_rfx_object
 from .const import ATTR_EVENT
+from .entity import RfxtrxEntity
 
 _LOGGER = logging.getLogger(__name__)
 

homeassistant/components/rfxtrx/siren.py

@@ -14,13 +14,9 @@ from homeassistant.helpers.entity import Entity
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.event import async_call_later
 
-from . import (
-    DEFAULT_OFF_DELAY,
-    DeviceTuple,
-    RfxtrxCommandEntity,
-    async_setup_platform_entry,
-)
+from . import DEFAULT_OFF_DELAY, DeviceTuple, async_setup_platform_entry
 from .const import CONF_OFF_DELAY
+from .entity import RfxtrxCommandEntity
 
 SECURITY_PANIC_ON = "Panic"
 SECURITY_PANIC_OFF = "End Panic"

homeassistant/components/rfxtrx/switch.py

@@ -14,19 +14,15 @@ from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.entity import Entity
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from . import (
-    DOMAIN,
-    DeviceTuple,
-    RfxtrxCommandEntity,
-    async_setup_platform_entry,
-    get_pt2262_cmd,
-)
+from . import DeviceTuple, async_setup_platform_entry, get_pt2262_cmd
 from .const import (
     COMMAND_OFF_LIST,
     COMMAND_ON_LIST,
     CONF_DATA_BITS,
     DEVICE_PACKET_TYPE_LIGHTING4,
+    DOMAIN,
 )
+from .entity import RfxtrxCommandEntity
 
 DATA_SWITCH = f"{DOMAIN}_switch"