Add calculated entities for power generation (#127)

custom_components/rct_power/__init__.py

@@ -49,47 +49,56 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
         hostname=config_entry_data.hostname, port=config_entry_data.port
     )
 
+    frequently_updated_object_ids = list(
+        {
+            object_info.object_id
+            for entity_description in all_entity_descriptions
+            if entity_description.update_priority == EntityUpdatePriority.FREQUENT
+            for object_info in entity_description.object_infos
+        }
+    )
     frequent_update_coordinator = RctPowerDataUpdateCoordinator(
         hass=hass,
         logger=_LOGGER,
         name=f"{DOMAIN} {entry.unique_id} frequent",
         update_interval=timedelta(seconds=config_entry_options.frequent_scan_interval),
-        object_ids=[
+        object_ids=frequently_updated_object_ids,
+        client=client,
+    )
+
+    infrequently_updated_object_ids = list(
+        {
             object_info.object_id
             for entity_description in all_entity_descriptions
-            if entity_description.update_priority == EntityUpdatePriority.FREQUENT
+            if entity_description.update_priority == EntityUpdatePriority.INFREQUENT
             for object_info in entity_description.object_infos
-        ],
-        client=client,
+        }
     )
-
     infrequent_update_coordinator = RctPowerDataUpdateCoordinator(
         hass=hass,
         logger=_LOGGER,
         name=f"{DOMAIN} {entry.unique_id} infrequent",
         update_interval=timedelta(
             seconds=config_entry_options.infrequent_scan_interval
         ),
-        object_ids=[
+        object_ids=infrequently_updated_object_ids,
+        client=client,
+    )
+
+    static_object_ids = list(
+        {
             object_info.object_id
             for entity_description in all_entity_descriptions
-            if entity_description.update_priority == EntityUpdatePriority.INFREQUENT
+            if entity_description.update_priority == EntityUpdatePriority.STATIC
             for object_info in entity_description.object_infos
-        ],
-        client=client,
+        }
     )
-
     static_update_coordinator = RctPowerDataUpdateCoordinator(
         hass=hass,
         logger=_LOGGER,
         name=f"{DOMAIN} {entry.unique_id} static",
         update_interval=timedelta(seconds=config_entry_options.static_scan_interval),
-        object_ids=[
-            object_info.object_id
-            for entity_description in all_entity_descriptions
-            if entity_description.update_priority == EntityUpdatePriority.STATIC
-            for object_info in entity_description.object_infos
-        ],
+        object_ids=static_object_ids,
         client=client,
     )
 

custom_components/rct_power/lib/entities.py

@@ -10,6 +10,10 @@
 
 from .device_info_helpers import get_battery_device_info, get_inverter_device_info
 from .entity import EntityUpdatePriority, RctPowerSensorEntityDescription
+from .state_helpers import (
+    get_first_api_reponse_value_as_absolute_state,
+    sum_api_response_values_as_state,
+)
 
 
 def get_matching_names(expression: str):
@@ -296,6 +300,17 @@ def get_matching_names(expression: str):
         name="Generator B Voltage",
         state_class=STATE_CLASS_MEASUREMENT,
     ),
+    RctPowerSensorEntityDescription(
+        get_device_info=get_inverter_device_info,
+        key="dc_conv.dc_conv_struct.p_dc",
+        object_names=[
+            "dc_conv.dc_conv_struct[0].p_dc",
+            "dc_conv.dc_conv_struct[1].p_dc",
+        ],
+        name="All Generators Power",
+        state_class=STATE_CLASS_MEASUREMENT,
+        get_native_value=sum_api_response_values_as_state,
+    ),
     RctPowerSensorEntityDescription(
         get_device_info=get_inverter_device_info,
         key="dc_conv.start_voltage",
@@ -546,6 +561,16 @@ def get_matching_names(expression: str):
         update_priority=EntityUpdatePriority.INFREQUENT,
         state_class=STATE_CLASS_TOTAL_INCREASING,
     ),
+    RctPowerSensorEntityDescription(
+        get_device_info=get_inverter_device_info,
+        key="energy.e_grid_feed_absolute_total",
+        unique_id="energy.e_grid_feed_absolute_total",  # to avoid collision
+        object_names=["energy.e_grid_feed_total"],
+        name="Grid Energy Production Absolute Total",
+        update_priority=EntityUpdatePriority.INFREQUENT,
+        state_class=STATE_CLASS_TOTAL_INCREASING,
+        get_native_value=get_first_api_reponse_value_as_absolute_state,
+    ),
     RctPowerSensorEntityDescription(
         get_device_info=get_inverter_device_info,
         key="energy.e_grid_load_day",
@@ -658,6 +683,15 @@ def get_matching_names(expression: str):
         update_priority=EntityUpdatePriority.INFREQUENT,
         state_class=STATE_CLASS_TOTAL_INCREASING,
     ),
+    RctPowerSensorEntityDescription(
+        get_device_info=get_inverter_device_info,
+        key="energy.e_dc_total",
+        object_names=["energy.e_dc_total[0]", "energy.e_dc_total[1]"],
+        name="All Generators Energy Production Total",
+        update_priority=EntityUpdatePriority.INFREQUENT,
+        state_class=STATE_CLASS_TOTAL_INCREASING,
+        get_native_value=sum_api_response_values_as_state,
+    ),
 ]
 
 fault_sensor_entity_descriptions: List[RctPowerSensorEntityDescription] = [
@@ -671,6 +705,7 @@ def get_matching_names(expression: str):
             "fault[3].flt",
         ],
         name="Faults",
+        unique_id="fault[0].flt",  # for backwards-compatibility
     ),
 ]
 

custom_components/rct_power/lib/entity.py

@@ -4,6 +4,7 @@
 from homeassistant.components.sensor import SensorEntity, SensorEntityDescription
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.helpers.entity import DeviceInfo, EntityDescription
+from homeassistant.helpers.typing import StateType
 from rctclient.registry import REGISTRY, ObjectInfo
 
 from .api import (
@@ -12,10 +13,11 @@
     ValidApiResponse,
     get_valid_response_value_or,
 )
-from .const import ICON, NUMERIC_STATE_DECIMAL_DIGITS, EntityUpdatePriority
+from .const import ICON, EntityUpdatePriority
 from .device_class_helpers import guess_device_class_from_unit
 from .entry import RctPowerConfigEntryData
 from .multi_coordinator_entity import MultiCoordinatorEntity
+from .state_helpers import get_first_api_response_value_as_state
 from .update_coordinator import RctPowerDataUpdateCoordinator
 
 
@@ -79,7 +81,13 @@ def config_entry_data(self):
     @property
     def unique_id(self):
         """Return a unique ID to use for this entity."""
-        return f"{self.config_entry.entry_id}-{self.object_infos[0].object_id}"
+        # this allows for keeping the entity identity stable for existing
+        # sensors when the algorithm below changes
+        if uid := self.entity_description.unique_id:
+            return uid
+
+        object_ids = [str(object_id) for object_id in self.object_ids]
+        return "-".join([self.config_entry.entry_id, *object_ids])
 
     @property
     def name(self):
@@ -129,22 +137,11 @@ class RctPowerSensorEntity(SensorEntity, RctPowerEntity):
 
     @property
     def native_value(self):
-        """Return the state of the sensor."""
-        value = self.get_valid_api_response_value_by_id(self.object_ids[0], None)
-
-        if isinstance(value, bytes):
-            return value.hex()
-
-        if isinstance(value, tuple):
-            return None
-
-        if isinstance(value, (int, float)) and self.native_unit_of_measurement == "%":
-            return round(value * 100, NUMERIC_STATE_DECIMAL_DIGITS)
-
-        if isinstance(value, (int, float)):
-            return round(value, NUMERIC_STATE_DECIMAL_DIGITS)
-
-        return value
+        values = [
+            self.get_valid_api_response_value_by_id(object_id, None)
+            for object_id in self.object_ids
+        ]
+        return self.entity_description.get_native_value(self, values)
 
     @property
     def native_unit_of_measurement(self):
@@ -188,6 +185,8 @@ class RctPowerEntityDescription(EntityDescription):
     icon: Optional[str] = ICON
     object_infos: List[ObjectInfo] = field(init=False)
     object_names: List[str] = field(default_factory=list)
+    # to allow for stable enitity identities even if the object ids change
+    unique_id: Optional[str] = None
     update_priority: EntityUpdatePriority = EntityUpdatePriority.FREQUENT
     get_device_info: Callable[[RctPowerEntity], Optional[DeviceInfo]] = lambda e: None
 
@@ -203,7 +202,9 @@ def __post_init__(self):
 class RctPowerSensorEntityDescription(
     RctPowerEntityDescription, SensorEntityDescription
 ):
-    pass
+    get_native_value: Callable[
+        [RctPowerSensorEntity, list[Optional[ApiResponseValue]]], StateType
+    ] = get_first_api_response_value_as_state
 
 
 def slugify_entity_name(name: str):

custom_components/rct_power/lib/entry.py

@@ -22,7 +22,7 @@ def from_config_entry(cls, config_entry: ConfigEntry):
         return cls(**config_entry.data)
 
     @classmethod
-    def from_user_input(cls, user_input):
+    def from_user_input(cls, user_input: object):
         valid_user_input = get_schema_for_dataclass(cls)(user_input)
 
         return cls(**valid_user_input)

custom_components/rct_power/lib/schema_helpers.py

@@ -1,21 +1,21 @@
 from dataclasses import Field, MISSING, fields
-from typing import List, Optional
+from typing import Any, List, Optional
 
 from voluptuous import Optional as OptionalField, Required as RequiredField, Schema
 
 
-def get_key_for_field(field: Field):
+def get_key_for_field(field: Field[Any]):
     if field.default == MISSING:
         return RequiredField(field.name)
 
     return OptionalField(field.name, default=field.default)
 
 
-def get_schema_for_field(field: Field):
+def get_schema_for_field(field: Field[Any]):
     return field.metadata.get("schema_type", field.type)
 
 
-def get_schema_for_dataclass(cls, allow_fields: Optional[List[str]] = None):
+def get_schema_for_dataclass(cls: type, allow_fields: Optional[List[str]] = None):
     return Schema(
         {
             get_key_for_field(field): get_schema_for_field(field)

custom_components/rct_power/lib/state_helpers.py

@@ -0,0 +1,55 @@
+from typing import Optional
+
+from homeassistant.components.sensor import SensorEntity
+from homeassistant.helpers.typing import StateType
+
+from .api import ApiResponseValue
+from .const import NUMERIC_STATE_DECIMAL_DIGITS
+
+
+def get_first_api_response_value_as_state(
+    entity: SensorEntity,
+    values: list[Optional[ApiResponseValue]],
+) -> StateType:
+    if len(values) <= 0:
+        return None
+
+    return get_api_response_value_as_state(entity=entity, value=values[0])
+
+
+def get_api_response_value_as_state(
+    entity: SensorEntity,
+    value: Optional[ApiResponseValue],
+) -> StateType:
+    if isinstance(value, bytes):
+        return value.hex()
+
+    if isinstance(value, tuple):
+        return None
+
+    if isinstance(value, (int, float)) and entity.native_unit_of_measurement == "%":
+        return round(value * 100, NUMERIC_STATE_DECIMAL_DIGITS)
+
+    if isinstance(value, (int, float)):
+        return round(value, NUMERIC_STATE_DECIMAL_DIGITS)
+
+    return value
+
+
+def get_first_api_reponse_value_as_absolute_state(
+    entity: SensorEntity,
+    values: list[Optional[ApiResponseValue]],
+) -> StateType:
+    value = get_first_api_response_value_as_state(entity=entity, values=values)
+
+    if isinstance(value, (int, float)):
+        return abs(value)
+
+    return value
+
+
+def sum_api_response_values_as_state(
+    entity: SensorEntity,
+    values: list[Optional[ApiResponseValue]],
+) -> StateType:
+    return sum(value for value in values if isinstance(value, (int, float)))