rfxtrx: Add command_on/command_off support for pt2262 switch entities (#65798)

homeassistant/components/rfxtrx/switch.py

@@ -7,7 +7,7 @@ import RFXtrx as rfxtrxmod
 
 from homeassistant.components.switch import SwitchEntity
 from homeassistant.config_entries import ConfigEntry
-from homeassistant.const import STATE_ON
+from homeassistant.const import CONF_COMMAND_OFF, CONF_COMMAND_ON, STATE_ON
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
@@ -17,8 +17,15 @@ from . import (
     DeviceTuple,
     RfxtrxCommandEntity,
     async_setup_platform_entry,
+    get_pt2262_cmd,
+)
+from .const import (
+    COMMAND_OFF_LIST,
+    COMMAND_ON_LIST,
+    CONF_DATA_BITS,
+    CONF_SIGNAL_REPETITIONS,
+    DEVICE_PACKET_TYPE_LIGHTING4,
 )
-from .const import COMMAND_OFF_LIST, COMMAND_ON_LIST, CONF_SIGNAL_REPETITIONS
 
 DATA_SWITCH = f"{DOMAIN}_switch"
 
@@ -53,6 +60,9 @@ async def async_setup_entry(
                 event.device,
                 device_id,
                 entity_info.get(CONF_SIGNAL_REPETITIONS, DEFAULT_SIGNAL_REPETITIONS),
+                entity_info.get(CONF_DATA_BITS),
+                entity_info.get(CONF_COMMAND_ON),
+                entity_info.get(CONF_COMMAND_OFF),
                 event=event if auto else None,
             )
         ]
@@ -65,6 +75,22 @@ async def async_setup_entry(
 class RfxtrxSwitch(RfxtrxCommandEntity, SwitchEntity):
     """Representation of a RFXtrx switch."""
 
+    def __init__(
+        self,
+        device: rfxtrxmod.RFXtrxDevice,
+        device_id: DeviceTuple,
+        signal_repetitions: int = 1,
+        data_bits: int | None = None,
+        cmd_on: int | None = None,
+        cmd_off: int | None = None,
+        event: rfxtrxmod.RFXtrxEvent | None = None,
+    ) -> None:
+        """Initialize the RFXtrx switch."""
+        super().__init__(device, device_id, signal_repetitions, event=event)
+        self._data_bits = data_bits
+        self._cmd_on = cmd_on
+        self._cmd_off = cmd_off
+
     async def async_added_to_hass(self):
         """Restore device state."""
         await super().async_added_to_hass()
@@ -74,15 +100,34 @@ class RfxtrxSwitch(RfxtrxCommandEntity, SwitchEntity):
             if old_state is not None:
                 self._state = old_state.state == STATE_ON
 
-    def _apply_event(self, event: rfxtrxmod.RFXtrxEvent) -> None:
-        """Apply command from rfxtrx."""
+    def _apply_event_lighting4(self, event: rfxtrxmod.RFXtrxEvent):
+        """Apply event for a lighting 4 device."""
+        if self._data_bits is not None:
+            cmdstr = get_pt2262_cmd(event.device.id_string, self._data_bits)
+            assert cmdstr
+            cmd = int(cmdstr, 16)
+            if cmd == self._cmd_on:
+                self._state = True
+            elif cmd == self._cmd_off:
+                self._state = False
+        else:
+            self._state = True
+
+    def _apply_event_standard(self, event: rfxtrxmod.RFXtrxEvent) -> None:
         assert isinstance(event, rfxtrxmod.ControlEvent)
-        super()._apply_event(event)
         if event.values["Command"] in COMMAND_ON_LIST:
             self._state = True
         elif event.values["Command"] in COMMAND_OFF_LIST:
             self._state = False
 
+    def _apply_event(self, event: rfxtrxmod.RFXtrxEvent) -> None:
+        """Apply command from rfxtrx."""
+        super()._apply_event(event)
+        if event.device.packettype == DEVICE_PACKET_TYPE_LIGHTING4:
+            self._apply_event_lighting4(event)
+        else:
+            self._apply_event_standard(event)
+
     @callback
     def _handle_event(
         self, event: rfxtrxmod.RFXtrxEvent, device_id: DeviceTuple
@@ -100,12 +145,18 @@ class RfxtrxSwitch(RfxtrxCommandEntity, SwitchEntity):
 
     async def async_turn_on(self, **kwargs):
         """Turn the device on."""
-        await self._async_send(self._device.send_on)
+        if self._cmd_on is not None:
+            await self._async_send(self._device.send_command, self._cmd_on)
+        else:
+            await self._async_send(self._device.send_on)
         self._state = True
         self.async_write_ha_state()
 
     async def async_turn_off(self, **kwargs):
         """Turn the device off."""
-        await self._async_send(self._device.send_off)
+        if self._cmd_off is not None:
+            await self._async_send(self._device.send_command, self._cmd_off)
+        else:
+            await self._async_send(self._device.send_off)
         self._state = False
         self.async_write_ha_state()