1
0
mirror of https://github.com/elisspace/core.git synced 2026-08-29 15:43:55 +00:00

optionally decouple EntityDescription.key from API response key

this makes it possible to have multiple entities for a single API response field
This commit is contained in:
farmio
2023-06-07 09:26:55 +02:00
parent d8ceb6463e
commit 76920037e6
2 changed files with 61 additions and 56 deletions

View File

@@ -49,8 +49,10 @@ class FroniusCoordinatorBase(
"""Set up the FroniusCoordinatorBase class."""
self._failed_update_count = 0
self.solar_net = solar_net
# unregistered_keys are used to create entities in platform module
self.unregistered_keys: dict[SolarNetId, set[str]] = {}
# unregistered_descriptors are used to create entities in platform module
self.unregistered_descriptors: dict[
SolarNetId, list[FroniusSensorEntityDescription]
] = {}
super().__init__(*args, update_interval=self.default_interval, **kwargs)
@abstractmethod
@@ -73,11 +75,11 @@ class FroniusCoordinatorBase(
self.update_interval = self.default_interval
for solar_net_id in data:
if solar_net_id not in self.unregistered_keys:
if solar_net_id not in self.unregistered_descriptors:
# id seen for the first time
self.unregistered_keys[solar_net_id] = {
desc.key for desc in self.valid_descriptions
}
self.unregistered_descriptors[
solar_net_id
] = self.valid_descriptions.copy()
return data
@callback
@@ -92,22 +94,34 @@ class FroniusCoordinatorBase(
"""
@callback
def _add_entities_for_unregistered_keys() -> None:
def _add_entities_for_unregistered_descriptors() -> None:
"""Add entities for keys seen for the first time."""
new_entities: list = []
new_entities: list[_FroniusEntityT] = []
for solar_net_id, device_data in self.data.items():
for key in self.unregistered_keys[solar_net_id].intersection(
device_data
):
if device_data[key]["value"] is None:
remaining_unregistered_descriptors = []
for description in self.unregistered_descriptors[solar_net_id]:
key = description.response_key or description.key
if key not in device_data:
remaining_unregistered_descriptors.append(description)
continue
new_entities.append(entity_constructor(self, key, solar_net_id))
self.unregistered_keys[solar_net_id].remove(key)
if device_data[key]["value"] is None:
remaining_unregistered_descriptors.append(description)
continue
new_entities.append(
entity_constructor(
coordinator=self,
description=description,
solar_net_id=solar_net_id,
)
)
self.unregistered_descriptors[
solar_net_id
] = remaining_unregistered_descriptors
async_add_entities(new_entities)
_add_entities_for_unregistered_keys()
_add_entities_for_unregistered_descriptors()
self.solar_net.cleanup_callbacks.append(
self.async_add_listener(_add_entities_for_unregistered_keys)
self.async_add_listener(_add_entities_for_unregistered_descriptors)
)

View File

@@ -84,6 +84,7 @@ class FroniusSensorEntityDescription(SensorEntityDescription):
"""Describes Fronius sensor entity."""
default_value: StateType | None = None
response_key: str | None = None
INVERTER_ENTITY_DESCRIPTIONS: list[FroniusSensorEntityDescription] = [
@@ -603,24 +604,22 @@ class _FroniusSensorEntity(CoordinatorEntity["FroniusCoordinatorBase"], SensorEn
"""Defines a Fronius coordinator entity."""
entity_description: FroniusSensorEntityDescription
entity_descriptions: list[FroniusSensorEntityDescription]
_attr_has_entity_name = True
def __init__(
self,
coordinator: FroniusCoordinatorBase,
key: str,
description: FroniusSensorEntityDescription,
solar_net_id: str,
) -> None:
"""Set up an individual Fronius meter sensor."""
super().__init__(coordinator)
self.entity_description = next(
desc for desc in self.entity_descriptions if desc.key == key
)
self.entity_description = description
self.response_key = description.response_key or description.key
self.solar_net_id = solar_net_id
self._attr_native_value = self._get_entity_value()
self._attr_translation_key = self.entity_description.key
self._attr_translation_key = description.key
def _device_data(self) -> dict[str, Any]:
"""Extract information for SolarNet device from coordinator data."""
@@ -628,9 +627,7 @@ class _FroniusSensorEntity(CoordinatorEntity["FroniusCoordinatorBase"], SensorEn
def _get_entity_value(self) -> Any:
"""Extract entity value from coordinator. Raises KeyError if not included in latest update."""
new_value = self.coordinator.data[self.solar_net_id][
self.entity_description.key
]["value"]
new_value = self.coordinator.data[self.solar_net_id][self.response_key]["value"]
if new_value is None:
return self.entity_description.default_value
if isinstance(new_value, float):
@@ -651,54 +648,54 @@ class _FroniusSensorEntity(CoordinatorEntity["FroniusCoordinatorBase"], SensorEn
class InverterSensor(_FroniusSensorEntity):
"""Defines a Fronius inverter device sensor entity."""
entity_descriptions = INVERTER_ENTITY_DESCRIPTIONS
def __init__(
self,
coordinator: FroniusInverterUpdateCoordinator,
key: str,
description: FroniusSensorEntityDescription,
solar_net_id: str,
) -> None:
"""Set up an individual Fronius inverter sensor."""
super().__init__(coordinator, key, solar_net_id)
super().__init__(coordinator, description, solar_net_id)
# device_info created in __init__ from a `GetInverterInfo` request
self._attr_device_info = coordinator.inverter_info.device_info
self._attr_unique_id = f"{coordinator.inverter_info.unique_id}-{key}"
self._attr_unique_id = (
f"{coordinator.inverter_info.unique_id}-{description.key}"
)
class LoggerSensor(_FroniusSensorEntity):
"""Defines a Fronius logger device sensor entity."""
entity_descriptions = LOGGER_ENTITY_DESCRIPTIONS
def __init__(
self,
coordinator: FroniusLoggerUpdateCoordinator,
key: str,
description: FroniusSensorEntityDescription,
solar_net_id: str,
) -> None:
"""Set up an individual Fronius meter sensor."""
super().__init__(coordinator, key, solar_net_id)
super().__init__(coordinator, description, solar_net_id)
logger_data = self._device_data()
# Logger device is already created in FroniusSolarNet._create_solar_net_device
self._attr_device_info = coordinator.solar_net.system_device_info
self._attr_native_unit_of_measurement = logger_data[key].get("unit")
self._attr_unique_id = f'{logger_data["unique_identifier"]["value"]}-{key}'
self._attr_native_unit_of_measurement = logger_data[self.response_key].get(
"unit"
)
self._attr_unique_id = (
f'{logger_data["unique_identifier"]["value"]}-{description.key}'
)
class MeterSensor(_FroniusSensorEntity):
"""Defines a Fronius meter device sensor entity."""
entity_descriptions = METER_ENTITY_DESCRIPTIONS
def __init__(
self,
coordinator: FroniusMeterUpdateCoordinator,
key: str,
description: FroniusSensorEntityDescription,
solar_net_id: str,
) -> None:
"""Set up an individual Fronius meter sensor."""
super().__init__(coordinator, key, solar_net_id)
super().__init__(coordinator, description, solar_net_id)
meter_data = self._device_data()
# S0 meters connected directly to inverters respond "n.a." as serial number
# `model` contains the inverter id: "S0 Meter at inverter 1"
@@ -715,22 +712,20 @@ class MeterSensor(_FroniusSensorEntity):
name=meter_data["model"]["value"],
via_device=(DOMAIN, coordinator.solar_net.solar_net_device_id),
)
self._attr_unique_id = f"{meter_uid}-{key}"
self._attr_unique_id = f"{meter_uid}-{description.key}"
class OhmpilotSensor(_FroniusSensorEntity):
"""Defines a Fronius Ohmpilot sensor entity."""
entity_descriptions = OHMPILOT_ENTITY_DESCRIPTIONS
def __init__(
self,
coordinator: FroniusOhmpilotUpdateCoordinator,
key: str,
description: FroniusSensorEntityDescription,
solar_net_id: str,
) -> None:
"""Set up an individual Fronius meter sensor."""
super().__init__(coordinator, key, solar_net_id)
super().__init__(coordinator, description, solar_net_id)
device_data = self._device_data()
self._attr_device_info = DeviceInfo(
@@ -741,45 +736,41 @@ class OhmpilotSensor(_FroniusSensorEntity):
sw_version=device_data["software"]["value"],
via_device=(DOMAIN, coordinator.solar_net.solar_net_device_id),
)
self._attr_unique_id = f'{device_data["serial"]["value"]}-{key}'
self._attr_unique_id = f'{device_data["serial"]["value"]}-{description.key}'
class PowerFlowSensor(_FroniusSensorEntity):
"""Defines a Fronius power flow sensor entity."""
entity_descriptions = POWER_FLOW_ENTITY_DESCRIPTIONS
def __init__(
self,
coordinator: FroniusPowerFlowUpdateCoordinator,
key: str,
description: FroniusSensorEntityDescription,
solar_net_id: str,
) -> None:
"""Set up an individual Fronius power flow sensor."""
super().__init__(coordinator, key, solar_net_id)
super().__init__(coordinator, description, solar_net_id)
# SolarNet device is already created in FroniusSolarNet._create_solar_net_device
self._attr_device_info = coordinator.solar_net.system_device_info
self._attr_unique_id = (
f"{coordinator.solar_net.solar_net_device_id}-power_flow-{key}"
f"{coordinator.solar_net.solar_net_device_id}-power_flow-{description.key}"
)
class StorageSensor(_FroniusSensorEntity):
"""Defines a Fronius storage device sensor entity."""
entity_descriptions = STORAGE_ENTITY_DESCRIPTIONS
def __init__(
self,
coordinator: FroniusStorageUpdateCoordinator,
key: str,
description: FroniusSensorEntityDescription,
solar_net_id: str,
) -> None:
"""Set up an individual Fronius storage sensor."""
super().__init__(coordinator, key, solar_net_id)
super().__init__(coordinator, description, solar_net_id)
storage_data = self._device_data()
self._attr_unique_id = f'{storage_data["serial"]["value"]}-{key}'
self._attr_unique_id = f'{storage_data["serial"]["value"]}-{description.key}'
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, storage_data["serial"]["value"])},
manufacturer=storage_data["manufacturer"]["value"],