-
-
Notifications
You must be signed in to change notification settings - Fork 30.5k
/
switch.py
156 lines (131 loc) · 4.91 KB
/
switch.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
"""Support for RFXtrx switches."""
from __future__ import annotations
import logging
from typing import Any
import RFXtrx as rfxtrxmod
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_COMMAND_OFF, CONF_COMMAND_ON, STATE_ON
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 .const import (
COMMAND_OFF_LIST,
COMMAND_ON_LIST,
CONF_DATA_BITS,
DEVICE_PACKET_TYPE_LIGHTING4,
)
DATA_SWITCH = f"{DOMAIN}_switch"
_LOGGER = logging.getLogger(__name__)
def supported(event: rfxtrxmod.RFXtrxEvent) -> bool:
"""Return whether an event supports switch."""
return (
isinstance(event.device, rfxtrxmod.LightingDevice)
and not event.device.known_to_be_dimmable
and not event.device.known_to_be_rollershutter
or isinstance(event.device, rfxtrxmod.RfyDevice)
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up config entry."""
def _constructor(
event: rfxtrxmod.RFXtrxEvent,
auto: rfxtrxmod.RFXtrxEvent | None,
device_id: DeviceTuple,
entity_info: dict[str, Any],
) -> list[Entity]:
return [
RfxtrxSwitch(
event.device,
device_id,
entity_info.get(CONF_DATA_BITS),
entity_info.get(CONF_COMMAND_ON),
entity_info.get(CONF_COMMAND_OFF),
event=event if auto else None,
)
]
await async_setup_platform_entry(
hass, config_entry, async_add_entities, supported, _constructor
)
class RfxtrxSwitch(RfxtrxCommandEntity, SwitchEntity):
"""Representation of a RFXtrx switch."""
def __init__(
self,
device: rfxtrxmod.RFXtrxDevice,
device_id: DeviceTuple,
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, event=event)
self._data_bits = data_bits
self._cmd_on = cmd_on
self._cmd_off = cmd_off
async def async_added_to_hass(self) -> None:
"""Restore device state."""
await super().async_added_to_hass()
if self._event is None:
old_state = await self.async_get_last_state()
if old_state is not None:
self._attr_is_on = old_state.state == STATE_ON
def _apply_event_lighting4(self, event: rfxtrxmod.RFXtrxEvent) -> None:
"""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._attr_is_on = True
elif cmd == self._cmd_off:
self._attr_is_on = False
else:
self._attr_is_on = True
def _apply_event_standard(self, event: rfxtrxmod.RFXtrxEvent) -> None:
assert isinstance(event, rfxtrxmod.ControlEvent)
if event.values["Command"] in COMMAND_ON_LIST:
self._attr_is_on = True
elif event.values["Command"] in COMMAND_OFF_LIST:
self._attr_is_on = 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
) -> None:
"""Check if event applies to me and update."""
if self._event_applies(event, device_id):
self._apply_event(event)
self.async_write_ha_state()
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the device 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._attr_is_on = True
self.async_write_ha_state()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the device 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._attr_is_on = False
self.async_write_ha_state()