Use entity_description in shelly rpc sensors (#64895)

* Use entity_description in shelly rpc sensors

* Enable None for binary sensor

* Adjust use_polling_wrapper

Co-authored-by: epenet <epenet@users.noreply.github.com>
This commit is contained in:
epenet 2022-01-25 14:48:51 +01:00 committed by GitHub
parent da316e1547
commit 076bc976ee
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 71 additions and 71 deletions

View file

@ -19,7 +19,7 @@ from .const import CONF_SLEEP_PERIOD
from .entity import ( from .entity import (
BlockAttributeDescription, BlockAttributeDescription,
RestEntityDescription, RestEntityDescription,
RpcAttributeDescription, RpcEntityDescription,
ShellyBlockAttributeEntity, ShellyBlockAttributeEntity,
ShellyRestAttributeEntity, ShellyRestAttributeEntity,
ShellyRpcAttributeEntity, ShellyRpcAttributeEntity,
@ -35,6 +35,11 @@ from .utils import (
) )
@dataclass
class RpcBinarySensorDescription(RpcEntityDescription, BinarySensorEntityDescription):
"""Class to describe a RPC binary sensor."""
@dataclass @dataclass
class RestBinarySensorDescription(RestEntityDescription, BinarySensorEntityDescription): class RestBinarySensorDescription(RestEntityDescription, BinarySensorEntityDescription):
"""Class to describe a REST binary sensor.""" """Class to describe a REST binary sensor."""
@ -134,28 +139,28 @@ REST_SENSORS: Final = {
} }
RPC_SENSORS: Final = { RPC_SENSORS: Final = {
"input": RpcAttributeDescription( "input": RpcBinarySensorDescription(
key="input", key="input",
sub_key="state", sub_key="state",
name="Input", name="Input",
device_class=BinarySensorDeviceClass.POWER, device_class=BinarySensorDeviceClass.POWER,
default_enabled=False, entity_registry_enabled_default=False,
removal_condition=is_rpc_momentary_input, removal_condition=is_rpc_momentary_input,
), ),
"cloud": RpcAttributeDescription( "cloud": RpcBinarySensorDescription(
key="cloud", key="cloud",
sub_key="connected", sub_key="connected",
name="Cloud", name="Cloud",
device_class=BinarySensorDeviceClass.CONNECTIVITY, device_class=BinarySensorDeviceClass.CONNECTIVITY,
default_enabled=False, entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC, entity_category=EntityCategory.DIAGNOSTIC,
), ),
"fwupdate": RpcAttributeDescription( "fwupdate": RpcBinarySensorDescription(
key="sys", key="sys",
sub_key="available_updates", sub_key="available_updates",
name="Firmware Update", name="Firmware Update",
device_class=BinarySensorDeviceClass.UPDATE, device_class=BinarySensorDeviceClass.UPDATE,
default_enabled=False, entity_registry_enabled_default=False,
extra_state_attributes=lambda status, shelly: { extra_state_attributes=lambda status, shelly: {
"latest_stable_version": status.get("stable", {"version": ""})["version"], "latest_stable_version": status.get("stable", {"version": ""})["version"],
"installed_version": shelly["ver"], "installed_version": shelly["ver"],
@ -223,9 +228,13 @@ class RestBinarySensor(ShellyRestAttributeEntity, BinarySensorEntity):
class RpcBinarySensor(ShellyRpcAttributeEntity, BinarySensorEntity): class RpcBinarySensor(ShellyRpcAttributeEntity, BinarySensorEntity):
"""Represent a RPC binary sensor entity.""" """Represent a RPC binary sensor entity."""
entity_description: RpcBinarySensorDescription
@property @property
def is_on(self) -> bool: def is_on(self) -> bool | None:
"""Return true if RPC sensor state is on.""" """Return true if RPC sensor state is on."""
if self.attribute_value is None:
return None
return bool(self.attribute_value) return bool(self.attribute_value)

View file

@ -158,7 +158,7 @@ async def async_setup_entry_rpc(
hass: HomeAssistant, hass: HomeAssistant,
config_entry: ConfigEntry, config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
sensors: dict[str, RpcAttributeDescription], sensors: Mapping[str, RpcEntityDescription],
sensor_class: Callable, sensor_class: Callable,
) -> None: ) -> None:
"""Set up entities for REST sensors.""" """Set up entities for REST sensors."""
@ -188,7 +188,7 @@ async def async_setup_entry_rpc(
unique_id = f"{wrapper.mac}-{key}-{sensor_id}" unique_id = f"{wrapper.mac}-{key}-{sensor_id}"
await async_remove_shelly_entity(hass, domain, unique_id) await async_remove_shelly_entity(hass, domain, unique_id)
else: else:
if description.should_poll: if description.use_polling_wrapper:
entities.append( entities.append(
sensor_class(polling_wrapper, key, sensor_id, description) sensor_class(polling_wrapper, key, sensor_id, description)
) )
@ -251,23 +251,21 @@ class BlockAttributeDescription:
@dataclass @dataclass
class RpcAttributeDescription: class RpcEntityRequiredKeysMixin:
"""Class to describe a RPC sensor.""" """Class for RPC entity required keys."""
key: str
sub_key: str sub_key: str
name: str
icon: str | None = None
unit: str | None = None @dataclass
class RpcEntityDescription(EntityDescription, RpcEntityRequiredKeysMixin):
"""Class to describe a RPC entity."""
value: Callable[[Any, Any], Any] | None = None value: Callable[[Any, Any], Any] | None = None
device_class: str | None = None
state_class: str | None = None
default_enabled: bool = True
available: Callable[[dict], bool] | None = None available: Callable[[dict], bool] | None = None
removal_condition: Callable[[dict, str], bool] | None = None removal_condition: Callable[[dict, str], bool] | None = None
extra_state_attributes: Callable[[dict, dict], dict | None] | None = None extra_state_attributes: Callable[[dict, dict], dict | None] | None = None
entity_category: EntityCategory | None = None use_polling_wrapper: bool = False
should_poll: bool = False
@dataclass @dataclass
@ -535,35 +533,36 @@ class ShellyRestAttributeEntity(update_coordinator.CoordinatorEntity):
class ShellyRpcAttributeEntity(ShellyRpcEntity, entity.Entity): class ShellyRpcAttributeEntity(ShellyRpcEntity, entity.Entity):
"""Helper class to represent a rpc attribute.""" """Helper class to represent a rpc attribute."""
entity_description: RpcEntityDescription
def __init__( def __init__(
self, self,
wrapper: RpcDeviceWrapper, wrapper: RpcDeviceWrapper,
key: str, key: str,
attribute: str, attribute: str,
description: RpcAttributeDescription, description: RpcEntityDescription,
) -> None: ) -> None:
"""Initialize sensor.""" """Initialize sensor."""
super().__init__(wrapper, key) super().__init__(wrapper, key)
self.sub_key = description.sub_key
self.attribute = attribute self.attribute = attribute
self.description = description self.entity_description = description
self._attr_unique_id = f"{super().unique_id}-{attribute}" self._attr_unique_id = f"{super().unique_id}-{attribute}"
self._attr_name = get_rpc_entity_name(wrapper.device, key, description.name) self._attr_name = get_rpc_entity_name(wrapper.device, key, description.name)
self._attr_entity_registry_enabled_default = description.default_enabled
self._attr_device_class = description.device_class
self._attr_icon = description.icon
self._last_value = None self._last_value = None
@property @property
def attribute_value(self) -> StateType: def attribute_value(self) -> StateType:
"""Value of sensor.""" """Value of sensor."""
if callable(self.description.value): if callable(self.entity_description.value):
self._last_value = self.description.value( self._last_value = self.entity_description.value(
self.wrapper.device.status[self.key][self.sub_key], self._last_value self.wrapper.device.status[self.key][self.entity_description.sub_key],
self._last_value,
) )
else: else:
self._last_value = self.wrapper.device.status[self.key][self.sub_key] self._last_value = self.wrapper.device.status[self.key][
self.entity_description.sub_key
]
return self._last_value return self._last_value
@ -572,31 +571,26 @@ class ShellyRpcAttributeEntity(ShellyRpcEntity, entity.Entity):
"""Available.""" """Available."""
available = super().available available = super().available
if not available or not self.description.available: if not available or not self.entity_description.available:
return available return available
return self.description.available( return self.entity_description.available(
self.wrapper.device.status[self.key][self.sub_key] self.wrapper.device.status[self.key][self.entity_description.sub_key]
) )
@property @property
def extra_state_attributes(self) -> dict[str, Any] | None: def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return the state attributes.""" """Return the state attributes."""
if self.description.extra_state_attributes is None: if self.entity_description.extra_state_attributes is None:
return None return None
assert self.wrapper.device.shelly assert self.wrapper.device.shelly
return self.description.extra_state_attributes( return self.entity_description.extra_state_attributes(
self.wrapper.device.status[self.key][self.sub_key], self.wrapper.device.status[self.key][self.entity_description.sub_key],
self.wrapper.device.shelly, self.wrapper.device.shelly,
) )
@property
def entity_category(self) -> EntityCategory | None:
"""Return category of entity."""
return self.description.entity_category
class ShellySleepingBlockAttributeEntity(ShellyBlockAttributeEntity, RestoreEntity): class ShellySleepingBlockAttributeEntity(ShellyBlockAttributeEntity, RestoreEntity):
"""Represent a shelly sleeping block attribute entity.""" """Represent a shelly sleeping block attribute entity."""

View file

@ -32,7 +32,7 @@ from .const import CONF_SLEEP_PERIOD, SHAIR_MAX_WORK_HOURS
from .entity import ( from .entity import (
BlockAttributeDescription, BlockAttributeDescription,
RestEntityDescription, RestEntityDescription,
RpcAttributeDescription, RpcEntityDescription,
ShellyBlockAttributeEntity, ShellyBlockAttributeEntity,
ShellyRestAttributeEntity, ShellyRestAttributeEntity,
ShellyRpcAttributeEntity, ShellyRpcAttributeEntity,
@ -44,6 +44,11 @@ from .entity import (
from .utils import get_device_entry_gen, get_device_uptime, temperature_unit from .utils import get_device_entry_gen, get_device_uptime, temperature_unit
@dataclass
class RpcSensorDescription(RpcEntityDescription, SensorEntityDescription):
"""Class to describe a RPC sensor."""
@dataclass @dataclass
class RestSensorDescription(RestEntityDescription, SensorEntityDescription): class RestSensorDescription(RestEntityDescription, SensorEntityDescription):
"""Class to describe a REST sensor.""" """Class to describe a REST sensor."""
@ -259,66 +264,66 @@ REST_SENSORS: Final = {
RPC_SENSORS: Final = { RPC_SENSORS: Final = {
"power": RpcAttributeDescription( "power": RpcSensorDescription(
key="switch", key="switch",
sub_key="apower", sub_key="apower",
name="Power", name="Power",
unit=POWER_WATT, native_unit_of_measurement=POWER_WATT,
value=lambda status, _: round(float(status), 1), value=lambda status, _: round(float(status), 1),
device_class=SensorDeviceClass.POWER, device_class=SensorDeviceClass.POWER,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
), ),
"voltage": RpcAttributeDescription( "voltage": RpcSensorDescription(
key="switch", key="switch",
sub_key="voltage", sub_key="voltage",
name="Voltage", name="Voltage",
unit=ELECTRIC_POTENTIAL_VOLT, native_unit_of_measurement=ELECTRIC_POTENTIAL_VOLT,
value=lambda status, _: round(float(status), 1), value=lambda status, _: round(float(status), 1),
device_class=SensorDeviceClass.VOLTAGE, device_class=SensorDeviceClass.VOLTAGE,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
default_enabled=False, entity_registry_enabled_default=False,
), ),
"energy": RpcAttributeDescription( "energy": RpcSensorDescription(
key="switch", key="switch",
sub_key="aenergy", sub_key="aenergy",
name="Energy", name="Energy",
unit=ENERGY_KILO_WATT_HOUR, native_unit_of_measurement=ENERGY_KILO_WATT_HOUR,
value=lambda status, _: round(status["total"] / 1000, 2), value=lambda status, _: round(status["total"] / 1000, 2),
device_class=SensorDeviceClass.ENERGY, device_class=SensorDeviceClass.ENERGY,
state_class=SensorStateClass.TOTAL_INCREASING, state_class=SensorStateClass.TOTAL_INCREASING,
), ),
"temperature": RpcAttributeDescription( "temperature": RpcSensorDescription(
key="switch", key="switch",
sub_key="temperature", sub_key="temperature",
name="Temperature", name="Temperature",
unit=TEMP_CELSIUS, native_unit_of_measurement=TEMP_CELSIUS,
value=lambda status, _: round(status["tC"], 1), value=lambda status, _: round(status["tC"], 1),
device_class=SensorDeviceClass.TEMPERATURE, device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
default_enabled=False, entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC, entity_category=EntityCategory.DIAGNOSTIC,
should_poll=True, use_polling_wrapper=True,
), ),
"rssi": RpcAttributeDescription( "rssi": RpcSensorDescription(
key="wifi", key="wifi",
sub_key="rssi", sub_key="rssi",
name="RSSI", name="RSSI",
unit=SIGNAL_STRENGTH_DECIBELS_MILLIWATT, native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
device_class=SensorDeviceClass.SIGNAL_STRENGTH, device_class=SensorDeviceClass.SIGNAL_STRENGTH,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
default_enabled=False, entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC, entity_category=EntityCategory.DIAGNOSTIC,
should_poll=True, use_polling_wrapper=True,
), ),
"uptime": RpcAttributeDescription( "uptime": RpcSensorDescription(
key="sys", key="sys",
sub_key="uptime", sub_key="uptime",
name="Uptime", name="Uptime",
value=get_device_uptime, value=get_device_uptime,
device_class=SensorDeviceClass.TIMESTAMP, device_class=SensorDeviceClass.TIMESTAMP,
default_enabled=False, entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC, entity_category=EntityCategory.DIAGNOSTIC,
should_poll=True, use_polling_wrapper=True,
), ),
} }
@ -380,21 +385,13 @@ class RestSensor(ShellyRestAttributeEntity, SensorEntity):
class RpcSensor(ShellyRpcAttributeEntity, SensorEntity): class RpcSensor(ShellyRpcAttributeEntity, SensorEntity):
"""Represent a RPC sensor.""" """Represent a RPC sensor."""
entity_description: RpcSensorDescription
@property @property
def native_value(self) -> StateType: def native_value(self) -> StateType:
"""Return value of sensor.""" """Return value of sensor."""
return self.attribute_value return self.attribute_value
@property
def state_class(self) -> str | None:
"""State class of sensor."""
return self.description.state_class
@property
def native_unit_of_measurement(self) -> str | None:
"""Return unit of sensor."""
return self.description.unit
class BlockSleepingSensor(ShellySleepingBlockAttributeEntity, SensorEntity): class BlockSleepingSensor(ShellySleepingBlockAttributeEntity, SensorEntity):
"""Represent a block sleeping sensor.""" """Represent a block sleeping sensor."""