Generalize a base ReCollect Waste entity (#85166)

This commit is contained in:
Aaron Bach 2023-01-05 03:26:59 -07:00 committed by GitHub
parent 280f6e4752
commit 6b68d3d365
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 53 additions and 31 deletions

View File

@ -0,0 +1,42 @@
"""Define a base ReCollect Waste entity."""
from aiorecollect.client import PickupEvent
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.device_registry import DeviceEntryType
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from .const import CONF_PLACE_ID, CONF_SERVICE_ID, DOMAIN
class ReCollectWasteEntity(CoordinatorEntity[DataUpdateCoordinator[list[PickupEvent]]]):
"""Define a base ReCollect Waste entity."""
_attr_has_entity_name = True
def __init__(
self,
coordinator: DataUpdateCoordinator[list[PickupEvent]],
entry: ConfigEntry,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self._identifier = f"{entry.data[CONF_PLACE_ID]}_{entry.data[CONF_SERVICE_ID]}"
self._attr_device_info = DeviceInfo(
entry_type=DeviceEntryType.SERVICE,
identifiers={(DOMAIN, self._identifier)},
manufacturer="ReCollect Waste",
name="ReCollect Waste",
)
self._attr_extra_state_attributes = {}
self._entry = entry
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
self._handle_coordinator_update()

View File

@ -12,12 +12,10 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_FRIENDLY_NAME from homeassistant.const import CONF_FRIENDLY_NAME
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import ( from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
CoordinatorEntity,
DataUpdateCoordinator,
)
from .const import CONF_PLACE_ID, CONF_SERVICE_ID, DOMAIN, LOGGER from .const import DOMAIN, LOGGER
from .entity import ReCollectWasteEntity
ATTR_PICKUP_TYPES = "pickup_types" ATTR_PICKUP_TYPES = "pickup_types"
ATTR_AREA_NAME = "area_name" ATTR_AREA_NAME = "area_name"
@ -59,20 +57,15 @@ async def async_setup_entry(
] ]
async_add_entities( async_add_entities(
[
ReCollectWasteSensor(coordinator, entry, description) ReCollectWasteSensor(coordinator, entry, description)
for description in SENSOR_DESCRIPTIONS for description in SENSOR_DESCRIPTIONS
]
) )
class ReCollectWasteSensor( class ReCollectWasteSensor(ReCollectWasteEntity, SensorEntity):
CoordinatorEntity[DataUpdateCoordinator[list[PickupEvent]]], SensorEntity """Define a ReCollect Waste sensor."""
):
"""ReCollect Waste Sensor."""
_attr_device_class = SensorDeviceClass.DATE _attr_device_class = SensorDeviceClass.DATE
_attr_has_entity_name = True
def __init__( def __init__(
self, self,
@ -80,28 +73,15 @@ class ReCollectWasteSensor(
entry: ConfigEntry, entry: ConfigEntry,
description: SensorEntityDescription, description: SensorEntityDescription,
) -> None: ) -> None:
"""Initialize the sensor.""" """Initialize."""
super().__init__(coordinator) super().__init__(coordinator, entry)
self._attr_extra_state_attributes = {} self._attr_unique_id = f"{self._identifier}_{description.key}"
self._attr_unique_id = f"{entry.data[CONF_PLACE_ID]}_{entry.data[CONF_SERVICE_ID]}_{description.key}"
self._entry = entry
self.entity_description = description self.entity_description = description
@callback @callback
def _handle_coordinator_update(self) -> None: def _handle_coordinator_update(self) -> None:
"""Respond to a DataUpdateCoordinator update.""" """Handle updated data from the coordinator."""
self.update_from_latest_data()
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
await super().async_added_to_hass()
self.update_from_latest_data()
@callback
def update_from_latest_data(self) -> None:
"""Update the state."""
if self.entity_description.key == SENSOR_TYPE_CURRENT_PICKUP: if self.entity_description.key == SENSOR_TYPE_CURRENT_PICKUP:
try: try:
event = self.coordinator.data[0] event = self.coordinator.data[0]