evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Reduce boilerplate code in entry init of rfxtrx (#58844)

Browse source 
* Reduce boilerplate code for rfxtrx

* Use rfxtrx built in to construct event

* Fixup mypy after rebase

* Also fix callable import
This commit is contained in:
Joakim Plate 2021-12-22 22:38:55 +01:00 • committed by GitHub
parent 91a8b1e7b3
commit bda1f02371
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 180 additions and 358 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

110
homeassistant/components/rfxtrx/__init__.py
View file

@ -3,6 +3,7 @@ from __future__ import annotations
import asyncio
import binascii
from collections.abc import Callable
import copy
import functools
import logging
@ -23,10 +24,11 @@ from homeassistant.const import (
EVENT_HOMEASSISTANT_STOP,
Platform,
)
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.device_registry import DeviceRegistry
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity import DeviceInfo, Entity
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
from .const import (
@ -35,8 +37,6 @@ from .const import (
CONF_AUTOMATIC_ADD,
CONF_DATA_BITS,
CONF_REMOVE_DEVICE,
DATA_CLEANUP_CALLBACKS,
DATA_LISTENER,
DATA_RFXOBJECT,
DEVICE_PACKET_TYPE_LIGHTING4,
EVENT_RFXTRX_EVENT,
@ -85,8 +85,6 @@ async def async_setup_entry(hass, entry: config_entries.ConfigEntry):
"""Set up the RFXtrx component."""
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][DATA_CLEANUP_CALLBACKS] = []
try:
await async_setup_internal(hass, entry)
except asyncio.TimeoutError:
@ -108,12 +106,6 @@ async def async_unload_entry(hass, entry: config_entries.ConfigEntry):
hass.services.async_remove(DOMAIN, SERVICE_SEND)
for cleanup_callback in hass.data[DOMAIN][DATA_CLEANUP_CALLBACKS]:
cleanup_callback()
listener = hass.data[DOMAIN][DATA_LISTENER]
listener()
rfx_object = hass.data[DOMAIN][DATA_RFXOBJECT]
await hass.async_add_executor_job(rfx_object.close_connection)
@ -160,6 +152,7 @@ async def async_setup_internal(hass, entry: config_entries.ConfigEntry):
# Setup some per device config
devices = _get_device_lookup(config[CONF_DEVICES])
pt2262_devices: list[str] = []
device_registry: DeviceRegistry = (
await hass.helpers.device_registry.async_get_registry()
@ -193,6 +186,10 @@ async def async_setup_internal(hass, entry: config_entries.ConfigEntry):
else:
return
if event.device.packettype == DEVICE_PACKET_TYPE_LIGHTING4:
find_possible_pt2262_device(pt2262_devices, event.device.id_string)
pt2262_devices.append(event.device.id_string)
device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, *device_id)},
)
@ -211,6 +208,14 @@ async def async_setup_internal(hass, entry: config_entries.ConfigEntry):
config = {}
config[CONF_DEVICE_ID] = device_id
_LOGGER.info(
"Added device (Device ID: %s Class: %s Sub: %s, Event: %s)",
event.device.id_string.lower(),
event.device.__class__.__name__,
event.device.subtype,
"".join(f"{x:02x}" for x in event.data),
)
data = entry.data.copy()
data[CONF_DEVICES] = copy.deepcopy(entry.data[CONF_DEVICES])
event_code = binascii.hexlify(event.data).decode("ASCII")
@ -222,9 +227,9 @@ async def async_setup_internal(hass, entry: config_entries.ConfigEntry):
"""Close connection with RFXtrx."""
rfx_object.close_connection()
listener = hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _shutdown_rfxtrx)
hass.data[DOMAIN][DATA_LISTENER] = listener
entry.async_on_unload(
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _shutdown_rfxtrx)
)
hass.data[DOMAIN][DATA_RFXOBJECT] = rfx_object
rfx_object.event_callback = lambda event: hass.add_job(async_handle_receive, event)
@ -236,25 +241,66 @@ async def async_setup_internal(hass, entry: config_entries.ConfigEntry):
hass.services.async_register(DOMAIN, SERVICE_SEND, send, schema=SERVICE_SEND_SCHEMA)
async def async_setup_platform_entry(
hass: HomeAssistant,
config_entry: config_entries.ConfigEntry,
async_add_entities: AddEntitiesCallback,
supported: Callable[[rfxtrxmod.RFXtrxEvent], bool],
constructor: Callable[
[rfxtrxmod.RFXtrxEvent, rfxtrxmod.RFXtrxEvent | None, DeviceTuple, dict],
list[Entity],
],
):
"""Set up config entry."""
entry_data = config_entry.data
device_ids: set[DeviceTuple] = set()
# Add entities from config
entities = []
for packet_id, entity_info in entry_data[CONF_DEVICES].items():
if (event := get_rfx_object(packet_id)) is None:
_LOGGER.error("Invalid device: %s", packet_id)
continue
if not supported(event):
continue
device_id = get_device_id(
event.device, data_bits=entity_info.get(CONF_DATA_BITS)
)
if device_id in device_ids:
continue
device_ids.add(device_id)
entities.extend(constructor(event, None, device_id, entity_info))
async_add_entities(entities)
# If automatic add is on, hookup listener
if entry_data[CONF_AUTOMATIC_ADD]:
@callback
def _update(event: rfxtrxmod.RFXtrxEvent, device_id: DeviceTuple):
"""Handle light updates from the RFXtrx gateway."""
if not supported(event):
return
if device_id in device_ids:
return
device_ids.add(device_id)
async_add_entities(constructor(event, event, device_id, {}))
config_entry.async_on_unload(
hass.helpers.dispatcher.async_dispatcher_connect(SIGNAL_EVENT, _update)
)
def get_rfx_object(packetid: str) -> rfxtrxmod.RFXtrxEvent | None:
"""Return the RFXObject with the packetid."""
try:
binarypacket = bytearray.fromhex(packetid)
except ValueError:
return None
pkt = rfxtrxmod.lowlevel.parse(binarypacket)
if pkt is None:
return None
if isinstance(pkt, rfxtrxmod.lowlevel.SensorPacket):
obj = rfxtrxmod.SensorEvent(pkt)
elif isinstance(pkt, rfxtrxmod.lowlevel.Status):
obj = rfxtrxmod.StatusEvent(pkt)
else:
obj = rfxtrxmod.ControlEvent(pkt)
obj.data = binarypacket
return obj
return rfxtrxmod.RFXtrxTransport.parse(binarypacket)
def get_pt2262_deviceid(device_id: str, nb_data_bits: int | None) -> bytes | None:
@ -341,14 +387,6 @@ def get_device_id(
return DeviceTuple(f"{device.packettype:x}", f"{device.subtype:x}", id_string)
def connect_auto_add(hass, entry_data, callback_fun):
"""Connect to dispatcher for automatic add."""
if entry_data[CONF_AUTOMATIC_ADD]:
hass.data[DOMAIN][DATA_CLEANUP_CALLBACKS].append(
hass.helpers.dispatcher.async_dispatcher_connect(SIGNAL_EVENT, callback_fun)
)
class RfxtrxEntity(RestoreEntity):
"""Represents a Rfxtrx device.

107
homeassistant/components/rfxtrx/binary_sensor.py
View file

@ -10,24 +10,11 @@ from homeassistant.components.binary_sensor import (
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.const import (
CONF_COMMAND_OFF,
CONF_COMMAND_ON,
CONF_DEVICES,
STATE_ON,
)
from homeassistant.const import CONF_COMMAND_OFF, CONF_COMMAND_ON, STATE_ON
from homeassistant.core import CALLBACK_TYPE, callback
from homeassistant.helpers import event as evt
from . import (
DeviceTuple,
RfxtrxEntity,
connect_auto_add,
find_possible_pt2262_device,
get_device_id,
get_pt2262_cmd,
get_rfx_object,
)
from . import DeviceTuple, RfxtrxEntity, async_setup_platform_entry, get_pt2262_cmd
from .const import (
COMMAND_OFF_LIST,
COMMAND_ON_LIST,
@ -100,82 +87,36 @@ async def async_setup_entry(
config_entry,
async_add_entities,
):
"""Set up platform."""
sensors = []
device_ids: set[DeviceTuple] = set()
pt2262_devices: list[str] = []
discovery_info = config_entry.data
"""Set up config entry."""
def get_sensor_description(type_string: str):
if (description := SENSOR_TYPES_DICT.get(type_string)) is None:
return BinarySensorEntityDescription(key=type_string)
return description
for packet_id, entity_info in discovery_info[CONF_DEVICES].items():
if (event := get_rfx_object(packet_id)) is None:
_LOGGER.error("Invalid device: %s", packet_id)
continue
if not supported(event):
continue
def _constructor(
event: rfxtrxmod.RFXtrxEvent,
auto: bool,
device_id: DeviceTuple,
entity_info: dict,
):
device_id = get_device_id(
event.device, data_bits=entity_info.get(CONF_DATA_BITS)
)
if device_id in device_ids:
continue
device_ids.add(device_id)
return [
RfxtrxBinarySensor(
event.device,
device_id,
get_sensor_description(event.device.type_string),
entity_info.get(CONF_OFF_DELAY),
entity_info.get(CONF_DATA_BITS),
entity_info.get(CONF_COMMAND_ON),
entity_info.get(CONF_COMMAND_OFF),
event=event if auto else None,
)
]
device: rfxtrxmod.RFXtrxDevice = event.device
if device.packettype == DEVICE_PACKET_TYPE_LIGHTING4:
find_possible_pt2262_device(pt2262_devices, device.id_string)
pt2262_devices.append(device.id_string)
entity = RfxtrxBinarySensor(
device,
device_id,
get_sensor_description(device.type_string),
entity_info.get(CONF_OFF_DELAY),
entity_info.get(CONF_DATA_BITS),
entity_info.get(CONF_COMMAND_ON),
entity_info.get(CONF_COMMAND_OFF),
)
sensors.append(entity)
async_add_entities(sensors)
@callback
def binary_sensor_update(
event: rfxtrxmod.RFXtrxEvent, device_id: DeviceTuple
) -> None:
"""Call for control updates from the RFXtrx gateway."""
if not supported(event):
return
if device_id in device_ids:
return
device_ids.add(device_id)
_LOGGER.info(
"Added binary sensor (Device ID: %s Class: %s Sub: %s Event: %s)",
event.device.id_string.lower(),
event.device.__class__.__name__,
event.device.subtype,
"".join(f"{x:02x}" for x in event.data),
)
sensor = RfxtrxBinarySensor(
event.device,
device_id,
event=event,
entity_description=get_sensor_description(event.device.type_string),
)
async_add_entities([sensor])
# Subscribe to main RFXtrx events
connect_auto_add(hass, discovery_info, binary_sensor_update)
await async_setup_platform_entry(
hass, config_entry, async_add_entities, supported, _constructor
)
class RfxtrxBinarySensor(RfxtrxEntity, BinarySensorEntity):

2
homeassistant/components/rfxtrx/const.py
View file

@ -45,5 +45,3 @@ DEVICE_PACKET_TYPE_LIGHTING4 = 0x13
EVENT_RFXTRX_EVENT = "rfxtrx_event"
DATA_RFXOBJECT = "rfxobject"
DATA_LISTENER = "ha_stop"
DATA_CLEANUP_CALLBACKS = "cleanup_callbacks"

77
homeassistant/components/rfxtrx/cover.py
View file

@ -14,21 +14,18 @@ from homeassistant.components.cover import (
SUPPORT_STOP_TILT,
CoverEntity,
)
from homeassistant.const import CONF_DEVICES, STATE_OPEN
from homeassistant.const import STATE_OPEN
from homeassistant.core import callback
from . import (
DEFAULT_SIGNAL_REPETITIONS,
DeviceTuple,
RfxtrxCommandEntity,
connect_auto_add,
get_device_id,
get_rfx_object,
async_setup_platform_entry,
)
from .const import (
COMMAND_OFF_LIST,
COMMAND_ON_LIST,
CONF_DATA_BITS,
CONF_SIGNAL_REPETITIONS,
CONF_VENETIAN_BLIND_MODE,
CONST_VENETIAN_BLIND_MODE_EU,
@ -49,60 +46,26 @@ async def async_setup_entry(
async_add_entities,
):
"""Set up config entry."""
discovery_info = config_entry.data
device_ids: set[DeviceTuple] = set()
entities = []
for packet_id, entity_info in discovery_info[CONF_DEVICES].items():
if (event := get_rfx_object(packet_id)) is None:
_LOGGER.error("Invalid device: %s", packet_id)
continue
if not supported(event):
continue
def _constructor(
event: rfxtrxmod.RFXtrxEvent,
auto: bool,
device_id: DeviceTuple,
entity_info: dict,
):
return [
RfxtrxCover(
event.device,
device_id,
entity_info.get(CONF_SIGNAL_REPETITIONS, DEFAULT_SIGNAL_REPETITIONS),
venetian_blind_mode=entity_info.get(CONF_VENETIAN_BLIND_MODE),
event=event if auto else None,
)
]
device_id = get_device_id(
event.device, data_bits=entity_info.get(CONF_DATA_BITS)
)
if device_id in device_ids:
continue
device_ids.add(device_id)
entity = RfxtrxCover(
event.device,
device_id,
signal_repetitions=entity_info.get(CONF_SIGNAL_REPETITIONS, 1),
venetian_blind_mode=entity_info.get(CONF_VENETIAN_BLIND_MODE),
)
entities.append(entity)
async_add_entities(entities)
@callback
def cover_update(event: rfxtrxmod.RFXtrxEvent, device_id: DeviceTuple) -> None:
"""Handle cover updates from the RFXtrx gateway."""
if not supported(event):
return
if device_id in device_ids:
return
device_ids.add(device_id)
device: rfxtrxmod.RFXtrxDevice = event.device
_LOGGER.info(
"Added cover (Device ID: %s Class: %s Sub: %s, Event: %s)",
device.id_string.lower(),
device.__class__.__name__,
device.subtype,
"".join(f"{x:02x}" for x in event.data),
)
entity = RfxtrxCover(
event.device, device_id, DEFAULT_SIGNAL_REPETITIONS, event=event
)
async_add_entities([entity])
# Subscribe to main RFXtrx events
connect_auto_add(hass, discovery_info, cover_update)
await async_setup_platform_entry(
hass, config_entry, async_add_entities, supported, _constructor
)
class RfxtrxCover(RfxtrxCommandEntity, CoverEntity):

81
homeassistant/components/rfxtrx/light.py
View file

@ -10,23 +10,16 @@ from homeassistant.components.light import (
SUPPORT_BRIGHTNESS,
LightEntity,
)
from homeassistant.const import CONF_DEVICES, STATE_ON
from homeassistant.const import STATE_ON
from homeassistant.core import callback
from . import (
DEFAULT_SIGNAL_REPETITIONS,
DeviceTuple,
RfxtrxCommandEntity,
connect_auto_add,
get_device_id,
get_rfx_object,
)
from .const import (
COMMAND_OFF_LIST,
COMMAND_ON_LIST,
CONF_DATA_BITS,
CONF_SIGNAL_REPETITIONS,
async_setup_platform_entry,
)
from .const import COMMAND_OFF_LIST, COMMAND_ON_LIST, CONF_SIGNAL_REPETITIONS
_LOGGER = logging.getLogger(__name__)
@ -47,59 +40,25 @@ async def async_setup_entry(
async_add_entities,
):
"""Set up config entry."""
discovery_info = config_entry.data
device_ids: set[DeviceTuple] = set()
# Add switch from config file
entities = []
for packet_id, entity_info in discovery_info[CONF_DEVICES].items():
if (event := get_rfx_object(packet_id)) is None:
_LOGGER.error("Invalid device: %s", packet_id)
continue
if not supported(event):
continue
def _constructor(
event: rfxtrxmod.RFXtrxEvent,
auto: bool,
device_id: DeviceTuple,
entity_info: dict,
):
return [
RfxtrxLight(
event.device,
device_id,
entity_info.get(CONF_SIGNAL_REPETITIONS, DEFAULT_SIGNAL_REPETITIONS),
event=event if auto else None,
)
]
device_id = get_device_id(
event.device, data_bits=entity_info.get(CONF_DATA_BITS)
)
if device_id in device_ids:
continue
device_ids.add(device_id)
entity = RfxtrxLight(
event.device, device_id, entity_info.get(CONF_SIGNAL_REPETITIONS, 1)
)
entities.append(entity)
async_add_entities(entities)
@callback
def light_update(event: rfxtrxmod.RFXtrxEvent, device_id: DeviceTuple):
"""Handle light updates from the RFXtrx gateway."""
if not supported(event):
return
if device_id in device_ids:
return
device_ids.add(device_id)
_LOGGER.info(
"Added light (Device ID: %s Class: %s Sub: %s, Event: %s)",
event.device.id_string.lower(),
event.device.__class__.__name__,
event.device.subtype,
"".join(f"{x:02x}" for x in event.data),
)
entity = RfxtrxLight(
event.device, device_id, DEFAULT_SIGNAL_REPETITIONS, event=event
)
async_add_entities([entity])
# Subscribe to main RFXtrx events
connect_auto_add(hass, discovery_info, light_update)
await async_setup_platform_entry(
hass, config_entry, async_add_entities, supported, _constructor
)
class RfxtrxLight(RfxtrxCommandEntity, LightEntity):

80
homeassistant/components/rfxtrx/sensor.py
View file

@ -5,7 +5,7 @@ from collections.abc import Callable
from dataclasses import dataclass
import logging
from RFXtrx import ControlEvent, SensorEvent
from RFXtrx import ControlEvent, RFXtrxEvent, SensorEvent
from homeassistant.components.sensor import (
SensorDeviceClass,
@ -14,7 +14,6 @@ from homeassistant.components.sensor import (
SensorStateClass,
)
from homeassistant.const import (
CONF_DEVICES,
DEGREE,
ELECTRIC_CURRENT_AMPERE,
ELECTRIC_POTENTIAL_VOLT,
@ -32,13 +31,7 @@ from homeassistant.const import (
from homeassistant.core import callback
from homeassistant.helpers.entity import EntityCategory
from . import (
CONF_DATA_BITS,
RfxtrxEntity,
connect_auto_add,
get_device_id,
get_rfx_object,
)
from . import DeviceTuple, RfxtrxEntity, async_setup_platform_entry, get_rfx_object
from .const import ATTR_EVENT
_LOGGER = logging.getLogger(__name__)
@ -219,62 +212,33 @@ async def async_setup_entry(
config_entry,
async_add_entities,
):
"""Set up platform."""
discovery_info = config_entry.data
data_ids = set()
"""Set up config entry."""
def supported(event):
def _supported(event):
return isinstance(event, (ControlEvent, SensorEvent))
entities = []
for packet_id, entity_info in discovery_info[CONF_DEVICES].items():
if (event := get_rfx_object(packet_id)) is None:
_LOGGER.error("Invalid device: %s", packet_id)
continue
if not supported(event):
continue
device_id = get_device_id(
event.device, data_bits=entity_info.get(CONF_DATA_BITS)
)
def _constructor(
event: RFXtrxEvent,
auto: bool,
device_id: DeviceTuple,
entity_info: dict,
):
entities: list[RfxtrxSensor] = []
for data_type in set(event.values) & set(SENSOR_TYPES_DICT):
data_id = (*device_id, str(data_type))
if data_id in data_ids:
continue
data_ids.add(data_id)
entity = RfxtrxSensor(event.device, device_id, SENSOR_TYPES_DICT[data_type])
entities.append(entity)
async_add_entities(entities)
@callback
def sensor_update(event, device_id):
"""Handle sensor updates from the RFXtrx gateway."""
if not supported(event):
return
for data_type in set(event.values) & set(SENSOR_TYPES_DICT):
data_id = (*device_id, data_type)
if data_id in data_ids:
continue
data_ids.add(data_id)
_LOGGER.info(
"Added sensor (Device ID: %s Class: %s Sub: %s, Event: %s)",
event.device.id_string.lower(),
event.device.__class__.__name__,
event.device.subtype,
"".join(f"{x:02x}" for x in event.data),
entities.append(
RfxtrxSensor(
event.device,
device_id,
SENSOR_TYPES_DICT[data_type],
event=event if auto else None,
)
)
entity = RfxtrxSensor(
event.device, device_id, SENSOR_TYPES_DICT[data_type], event=event
)
async_add_entities([entity])
return entities
# Subscribe to main RFXtrx events
connect_auto_add(hass, discovery_info, sensor_update)
await async_setup_platform_entry(
hass, config_entry, async_add_entities, _supported, _constructor
)
class RfxtrxSensor(RfxtrxEntity, SensorEntity):

81
homeassistant/components/rfxtrx/switch.py
View file

@ -6,7 +6,7 @@ import logging
import RFXtrx as rfxtrxmod
from homeassistant.components.switch import SwitchEntity
from homeassistant.const import CONF_DEVICES, STATE_ON
from homeassistant.const import STATE_ON
from homeassistant.core import callback
from . import (
@ -14,16 +14,9 @@ from . import (
DOMAIN,
DeviceTuple,
RfxtrxCommandEntity,
connect_auto_add,
get_device_id,
get_rfx_object,
)
from .const import (
COMMAND_OFF_LIST,
COMMAND_ON_LIST,
CONF_DATA_BITS,
CONF_SIGNAL_REPETITIONS,
async_setup_platform_entry,
)
from .const import COMMAND_OFF_LIST, COMMAND_ON_LIST, CONF_SIGNAL_REPETITIONS
DATA_SWITCH = f"{DOMAIN}_switch"
@ -46,59 +39,25 @@ async def async_setup_entry(
async_add_entities,
):
"""Set up config entry."""
discovery_info = config_entry.data
device_ids: set[DeviceTuple] = set()
# Add switch from config file
entities = []
for packet_id, entity_info in discovery_info[CONF_DEVICES].items():
if (event := get_rfx_object(packet_id)) is None:
_LOGGER.error("Invalid device: %s", packet_id)
continue
if not supported(event):
continue
def _constructor(
event: rfxtrxmod.RFXtrxEvent,
auto: bool,
device_id: DeviceTuple,
entity_info: dict,
):
return [
RfxtrxSwitch(
event.device,
device_id,
entity_info.get(CONF_SIGNAL_REPETITIONS, DEFAULT_SIGNAL_REPETITIONS),
event=event if auto else None,
)
]
device_id = get_device_id(
event.device, data_bits=entity_info.get(CONF_DATA_BITS)
)
if device_id in device_ids:
continue
device_ids.add(device_id)
entity = RfxtrxSwitch(
event.device, device_id, entity_info.get(CONF_SIGNAL_REPETITIONS, 1)
)
entities.append(entity)
async_add_entities(entities)
@callback
def switch_update(event, device_id):
"""Handle sensor updates from the RFXtrx gateway."""
if not supported(event):
return
if device_id in device_ids:
return
device_ids.add(device_id)
device: rfxtrxmod.RFXtrxDevice = event.device
_LOGGER.info(
"Added switch (Device ID: %s Class: %s Sub: %s, Event: %s)",
device.id_string.lower(),
device.__class__.__name__,
device.subtype,
"".join(f"{x:02x}" for x in event.data),
)
entity = RfxtrxSwitch(
device, device_id, DEFAULT_SIGNAL_REPETITIONS, event=event
)
async_add_entities([entity])
# Subscribe to main RFXtrx events
connect_auto_add(hass, discovery_info, switch_update)
await async_setup_platform_entry(
hass, config_entry, async_add_entities, supported, _constructor
)
class RfxtrxSwitch(RfxtrxCommandEntity, SwitchEntity):