Move and rename lutron caseta base entity to separate module (#126103)

pull/126128/head
epenet 2024-09-17 15:39:11 +02:00 committed by GitHub
parent 93f2b7c8a3
commit 3a55cbc818
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 151 additions and 133 deletions

View File

@ -14,13 +14,12 @@ from pylutron_caseta.smartbridge import Smartbridge
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import ATTR_DEVICE_ID, ATTR_SUGGESTED_AREA, CONF_HOST, Platform
from homeassistant.const import ATTR_DEVICE_ID, CONF_HOST, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr, entity_registry as er
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.typing import ConfigType
from .const import (
@ -40,7 +39,6 @@ from .const import (
CONF_CERTFILE,
CONF_KEYFILE,
CONF_SUBTYPE,
CONFIG_URL,
DOMAIN,
LUTRON_CASETA_BUTTON_EVENT,
MANUFACTURER,
@ -68,7 +66,7 @@ from .models import (
LutronKeypad,
LutronKeypadData,
)
from .util import serial_to_unique_id
from .util import area_name_from_id, serial_to_unique_id
_LOGGER = logging.getLogger(__name__)
@ -224,7 +222,7 @@ def _async_register_bridge_device(
configuration_url="https://device-login.lutron.com",
)
area = _area_name_from_id(bridge.areas, bridge_device["area"])
area = area_name_from_id(bridge.areas, bridge_device["area"])
if area != UNASSIGNED_AREA:
device_args["suggested_area"] = area
@ -342,7 +340,7 @@ 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"])
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"])
device_info = DeviceInfo(
@ -404,27 +402,6 @@ def _handle_none_keypad_serial(keypad_device: dict, bridge_serial: int) -> str:
return keypad_device["serial"] or f"{bridge_serial}_{keypad_device['device_id']}"
def _area_name_from_id(areas: dict[str, dict], area_id: str | None) -> str:
"""Return the full area name including parent(s)."""
if area_id is None:
return UNASSIGNED_AREA
return _construct_area_name_from_id(areas, area_id, [])
def _construct_area_name_from_id(
areas: dict[str, dict], area_id: str, labels: list[str]
) -> str:
"""Recursively construct the full area name including parent(s)."""
area = areas[area_id]
parent_area_id = area["parent_id"]
if parent_area_id is None:
# This is the root area, return last area
return " ".join(labels)
labels.insert(0, area["name"])
return _construct_area_name_from_id(areas, parent_area_id, labels)
@callback
def async_get_lip_button(device_type: str, leap_button: int) -> int | None:
"""Get the LIP button for a given LEAP button."""
@ -500,98 +477,6 @@ async def async_unload_entry(
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
class LutronCasetaDevice(Entity):
"""Common base class for all Lutron Caseta devices."""
_attr_should_poll = False
def __init__(self, device: dict[str, Any], data: LutronCasetaData) -> None:
"""Set up the base class.
[:param]device the device metadata
[:param]bridge the smartbridge object
[:param]bridge_device a dict with the details of the bridge
"""
self._device = device
self._smartbridge = data.bridge
self._bridge_device = data.bridge_device
self._bridge_unique_id = serial_to_unique_id(data.bridge_device["serial"])
if "serial" not in self._device:
return
if "parent_device" in device:
# This is a child entity, handle the naming in button.py and switch.py
return
area = _area_name_from_id(self._smartbridge.areas, device["area"])
name = device["name"].split("_")[-1]
self._attr_name = full_name = f"{area} {name}"
info = DeviceInfo(
# Historically we used the device serial number for the identifier
# but the serial is usually an integer and a string is expected
# here. Since it would be a breaking change to change the identifier
# we are ignoring the type error here until it can be migrated to
# a string in a future release.
identifiers={
(
DOMAIN,
self._handle_none_serial(self.serial), # type: ignore[arg-type]
)
},
manufacturer=MANUFACTURER,
model=f"{device['model']} ({device['type']})",
name=full_name,
via_device=(DOMAIN, self._bridge_device["serial"]),
configuration_url=CONFIG_URL,
)
if area != UNASSIGNED_AREA:
info[ATTR_SUGGESTED_AREA] = area
self._attr_device_info = info
async def async_added_to_hass(self):
"""Register callbacks."""
self._smartbridge.add_subscriber(self.device_id, self.async_write_ha_state)
def _handle_none_serial(self, serial: str | int | None) -> str | int:
"""Handle None serial returned by RA3 and QSX processors."""
if serial is None:
return f"{self._bridge_unique_id}_{self.device_id}"
return serial
@property
def device_id(self):
"""Return the device ID used for calling pylutron_caseta."""
return self._device["device_id"]
@property
def serial(self) -> int | None:
"""Return the serial number of the device."""
return self._device["serial"]
@property
def unique_id(self) -> str:
"""Return the unique ID of the device (serial)."""
return str(self._handle_none_serial(self.serial))
@property
def extra_state_attributes(self):
"""Return the state attributes."""
attributes = {
"device_id": self.device_id,
}
if zone := self._device.get("zone"):
attributes["zone_id"] = zone
return attributes
class LutronCasetaDeviceUpdatableEntity(LutronCasetaDevice):
"""A lutron_caseta entity that can update by syncing data from the bridge."""
async def async_update(self) -> None:
"""Update when forcing a refresh of the device."""
self._device = self._smartbridge.get_device_by_id(self.device_id)
_LOGGER.debug(self._device)
def _id_to_identifier(lutron_id: str) -> tuple[str, str]:
"""Convert a lutron caseta identifier to a device identifier."""
return (DOMAIN, lutron_id)

View File

@ -11,9 +11,11 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import DOMAIN as CASETA_DOMAIN, LutronCasetaDevice, _area_name_from_id
from . import DOMAIN as CASETA_DOMAIN
from .const import CONFIG_URL, MANUFACTURER, UNASSIGNED_AREA
from .entity import LutronCasetaEntity
from .models import LutronCasetaConfigEntry
from .util import area_name_from_id
async def async_setup_entry(
@ -35,7 +37,7 @@ async def async_setup_entry(
)
class LutronOccupancySensor(LutronCasetaDevice, BinarySensorEntity):
class LutronOccupancySensor(LutronCasetaEntity, BinarySensorEntity):
"""Representation of a Lutron occupancy group."""
_attr_device_class = BinarySensorDeviceClass.OCCUPANCY
@ -43,7 +45,7 @@ class LutronOccupancySensor(LutronCasetaDevice, BinarySensorEntity):
def __init__(self, device, data):
"""Init an occupancy sensor."""
super().__init__(device, data)
area = _area_name_from_id(self._smartbridge.areas, device["area"])
area = area_name_from_id(self._smartbridge.areas, device["area"])
name = f"{area} {device['device_name']}"
self._attr_name = name
self._attr_device_info = DeviceInfo(

View File

@ -9,8 +9,8 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import LutronCasetaDevice
from .device_trigger import LEAP_TO_DEVICE_TYPE_SUBTYPE_MAP
from .entity import LutronCasetaEntity
from .models import LutronCasetaConfigEntry, LutronCasetaData
@ -65,7 +65,7 @@ async def async_setup_entry(
async_add_entities(entities)
class LutronCasetaButton(LutronCasetaDevice, ButtonEntity):
class LutronCasetaButton(LutronCasetaEntity, ButtonEntity):
"""Representation of a Lutron pico and keypad button."""
def __init__(

View File

@ -13,11 +13,11 @@ from homeassistant.components.cover import (
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import LutronCasetaDeviceUpdatableEntity
from .entity import LutronCasetaUpdatableEntity
from .models import LutronCasetaConfigEntry
class LutronCasetaShade(LutronCasetaDeviceUpdatableEntity, CoverEntity):
class LutronCasetaShade(LutronCasetaUpdatableEntity, CoverEntity):
"""Representation of a Lutron shade with open/close functionality."""
_attr_supported_features = (
@ -59,7 +59,7 @@ class LutronCasetaShade(LutronCasetaDeviceUpdatableEntity, CoverEntity):
await self._smartbridge.set_value(self.device_id, kwargs[ATTR_POSITION])
class LutronCasetaTiltOnlyBlind(LutronCasetaDeviceUpdatableEntity, CoverEntity):
class LutronCasetaTiltOnlyBlind(LutronCasetaUpdatableEntity, CoverEntity):
"""Representation of a Lutron tilt only blind."""
_attr_supported_features = (

View File

@ -0,0 +1,108 @@
"""Component for interacting with a Lutron Caseta system."""
from __future__ import annotations
import logging
from typing import Any
from homeassistant.const import ATTR_SUGGESTED_AREA
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import Entity
from .const import CONFIG_URL, DOMAIN, MANUFACTURER, UNASSIGNED_AREA
from .models import LutronCasetaData
from .util import area_name_from_id, serial_to_unique_id
_LOGGER = logging.getLogger(__name__)
class LutronCasetaEntity(Entity):
"""Common base class for all Lutron Caseta devices."""
_attr_should_poll = False
def __init__(self, device: dict[str, Any], data: LutronCasetaData) -> None:
"""Set up the base class.
[:param]device the device metadata
[:param]bridge the smartbridge object
[:param]bridge_device a dict with the details of the bridge
"""
self._device = device
self._smartbridge = data.bridge
self._bridge_device = data.bridge_device
self._bridge_unique_id = serial_to_unique_id(data.bridge_device["serial"])
if "serial" not in self._device:
return
if "parent_device" in device:
# This is a child entity, handle the naming in button.py and switch.py
return
area = area_name_from_id(self._smartbridge.areas, device["area"])
name = device["name"].split("_")[-1]
self._attr_name = full_name = f"{area} {name}"
info = DeviceInfo(
# Historically we used the device serial number for the identifier
# but the serial is usually an integer and a string is expected
# here. Since it would be a breaking change to change the identifier
# we are ignoring the type error here until it can be migrated to
# a string in a future release.
identifiers={
(
DOMAIN,
self._handle_none_serial(self.serial), # type: ignore[arg-type]
)
},
manufacturer=MANUFACTURER,
model=f"{device['model']} ({device['type']})",
name=full_name,
via_device=(DOMAIN, self._bridge_device["serial"]),
configuration_url=CONFIG_URL,
)
if area != UNASSIGNED_AREA:
info[ATTR_SUGGESTED_AREA] = area
self._attr_device_info = info
async def async_added_to_hass(self):
"""Register callbacks."""
self._smartbridge.add_subscriber(self.device_id, self.async_write_ha_state)
def _handle_none_serial(self, serial: str | int | None) -> str | int:
"""Handle None serial returned by RA3 and QSX processors."""
if serial is None:
return f"{self._bridge_unique_id}_{self.device_id}"
return serial
@property
def device_id(self):
"""Return the device ID used for calling pylutron_caseta."""
return self._device["device_id"]
@property
def serial(self) -> int | None:
"""Return the serial number of the device."""
return self._device["serial"]
@property
def unique_id(self) -> str:
"""Return the unique ID of the device (serial)."""
return str(self._handle_none_serial(self.serial))
@property
def extra_state_attributes(self):
"""Return the state attributes."""
attributes = {
"device_id": self.device_id,
}
if zone := self._device.get("zone"):
attributes["zone_id"] = zone
return attributes
class LutronCasetaUpdatableEntity(LutronCasetaEntity):
"""A lutron_caseta entity that can update by syncing data from the bridge."""
async def async_update(self) -> None:
"""Update when forcing a refresh of the device."""
self._device = self._smartbridge.get_device_by_id(self.device_id)
_LOGGER.debug(self._device)

View File

@ -18,7 +18,7 @@ from homeassistant.util.percentage import (
percentage_to_ordered_list_item,
)
from . import LutronCasetaDeviceUpdatableEntity
from .entity import LutronCasetaUpdatableEntity
from .models import LutronCasetaConfigEntry
DEFAULT_ON_PERCENTAGE = 50
@ -41,7 +41,7 @@ async def async_setup_entry(
async_add_entities(LutronCasetaFan(fan_device, data) for fan_device in fan_devices)
class LutronCasetaFan(LutronCasetaDeviceUpdatableEntity, FanEntity):
class LutronCasetaFan(LutronCasetaUpdatableEntity, FanEntity):
"""Representation of a Lutron Caseta fan. Including Fan Speed."""
_attr_supported_features = (

View File

@ -24,8 +24,8 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import LutronCasetaDeviceUpdatableEntity
from .const import DEVICE_TYPE_SPECTRUM_TUNE, DEVICE_TYPE_WHITE_TUNE
from .entity import LutronCasetaUpdatableEntity
from .models import LutronCasetaData
SUPPORTED_COLOR_MODE_DICT = {
@ -68,7 +68,7 @@ async def async_setup_entry(
)
class LutronCasetaLight(LutronCasetaDeviceUpdatableEntity, LightEntity):
class LutronCasetaLight(LutronCasetaUpdatableEntity, LightEntity):
"""Representation of a Lutron Light, including dimmable, white tune, and spectrum tune."""
_attr_supported_features = LightEntityFeature.TRANSITION

View File

@ -7,7 +7,7 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import LutronCasetaDeviceUpdatableEntity
from .entity import LutronCasetaUpdatableEntity
async def async_setup_entry(
@ -28,7 +28,7 @@ async def async_setup_entry(
)
class LutronCasetaLight(LutronCasetaDeviceUpdatableEntity, SwitchEntity):
class LutronCasetaLight(LutronCasetaUpdatableEntity, SwitchEntity):
"""Representation of a Lutron Caseta switch."""
def __init__(self, device, data):

View File

@ -2,7 +2,30 @@
from __future__ import annotations
from .const import UNASSIGNED_AREA
def serial_to_unique_id(serial: int) -> str:
"""Convert a lutron serial number to a unique id."""
return hex(serial)[2:].zfill(8)
def area_name_from_id(areas: dict[str, dict], area_id: str | None) -> str:
"""Return the full area name including parent(s)."""
if area_id is None:
return UNASSIGNED_AREA
return _construct_area_name_from_id(areas, area_id, [])
def _construct_area_name_from_id(
areas: dict[str, dict], area_id: str, labels: list[str]
) -> str:
"""Recursively construct the full area name including parent(s)."""
area = areas[area_id]
parent_area_id = area["parent_id"]
if parent_area_id is None:
# This is the root area, return last area
return " ".join(labels)
labels.insert(0, area["name"])
return _construct_area_name_from_id(areas, parent_area_id, labels)