Add textual representation entities for Fronius status codes

homeassistant/components/fronius/const.py

@@ -1,7 +1,9 @@
 """Constants for the Fronius integration."""
+from enum import StrEnum
 from typing import Final, NamedTuple, TypedDict
 
 from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.typing import StateType
 
 DOMAIN: Final = "fronius"
 
@@ -25,3 +27,97 @@ class FroniusDeviceInfo(NamedTuple):
     device_info: DeviceInfo
     solar_net_id: SolarNetId
     unique_id: str
+
+
+class InverterStatusCodeOption(StrEnum):
+    """Status codes for Fronius inverters."""
+
+    # these are keys for state translations - so snake_case is used
+    STARTUP = "startup"
+    RUNNING = "running"
+    STANDBY = "standby"
+    BOOTLOADING = "bootloading"
+    ERROR = "error"
+    IDLE = "idle"
+    READY = "ready"
+    SLEEPING = "sleeping"
+    UNKNOWN = "unknown"
+    INVALID = "invalid"
+
+
+_INVERTER_STATUS_CODES: Final[dict[int, InverterStatusCodeOption]] = {
+    0: InverterStatusCodeOption.STARTUP,
+    1: InverterStatusCodeOption.STARTUP,
+    2: InverterStatusCodeOption.STARTUP,
+    3: InverterStatusCodeOption.STARTUP,
+    4: InverterStatusCodeOption.STARTUP,
+    5: InverterStatusCodeOption.STARTUP,
+    6: InverterStatusCodeOption.STARTUP,
+    7: InverterStatusCodeOption.RUNNING,
+    8: InverterStatusCodeOption.STANDBY,
+    9: InverterStatusCodeOption.BOOTLOADING,
+    10: InverterStatusCodeOption.ERROR,
+    11: InverterStatusCodeOption.IDLE,
+    12: InverterStatusCodeOption.READY,
+    13: InverterStatusCodeOption.SLEEPING,
+    255: InverterStatusCodeOption.UNKNOWN,
+}
+
+
+def get_inverter_status_message(code: StateType) -> InverterStatusCodeOption:
+    """Return a status message for a given status code."""
+    return _INVERTER_STATUS_CODES.get(code, InverterStatusCodeOption.INVALID)  # type: ignore[arg-type]
+
+
+class MeterLocationCodeOption(StrEnum):
+    """Meter location codes for Fronius meters."""
+
+    # these are keys for state translations - so snake_case is used
+    FEED_IN = "feed_in"
+    CONSUMPTION_PATH = "consumption_path"
+    GENERATOR = "external_generator"
+    EXT_BATTERY = "external_battery"
+    SUBLOAD = "subload"
+
+
+def get_meter_location_description(code: StateType) -> MeterLocationCodeOption | None:
+    """Return a location_description for a given location code."""
+    match int(code):  # type: ignore[arg-type]
+        case 0:
+            return MeterLocationCodeOption.FEED_IN
+        case 1:
+            return MeterLocationCodeOption.CONSUMPTION_PATH
+        case 3:
+            return MeterLocationCodeOption.GENERATOR
+        case 4:
+            return MeterLocationCodeOption.EXT_BATTERY
+        case _ as _code if 256 <= _code <= 511:
+            return MeterLocationCodeOption.SUBLOAD
+    return None
+
+
+class OhmPilotStateCodeOption(StrEnum):
+    """OhmPilot state codes for Fronius inverters."""
+
+    # these are keys for state translations - so snake_case is used
+    UP_AND_RUNNING = "up_and_running"
+    KEEP_MINIMUM_TEMPERATURE = "keep_minimum_temperature"
+    LEGIONELLA_PROTECTION = "legionella_protection"
+    CRITICAL_FAULT = "critical_fault"
+    FAULT = "fault"
+    BOOST_MODE = "boost_mode"
+
+
+_OHMPILOT_STATE_CODES: Final[dict[int, OhmPilotStateCodeOption]] = {
+    0: OhmPilotStateCodeOption.UP_AND_RUNNING,
+    1: OhmPilotStateCodeOption.KEEP_MINIMUM_TEMPERATURE,
+    2: OhmPilotStateCodeOption.LEGIONELLA_PROTECTION,
+    3: OhmPilotStateCodeOption.CRITICAL_FAULT,
+    4: OhmPilotStateCodeOption.FAULT,
+    5: OhmPilotStateCodeOption.BOOST_MODE,
+}
+
+
+def get_ohmpilot_state_message(code: StateType) -> OhmPilotStateCodeOption | None:
+    """Return a status message for a given status code."""
+    return _OHMPILOT_STATE_CODES.get(code)  # type: ignore[arg-type]

homeassistant/components/fronius/coordinator.py

@@ -49,8 +49,10 @@ def __init__(self, *args: Any, solar_net: FroniusSolarNet, **kwargs: Any) -> Non
         """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 @@ async def _async_update_data(self) -> dict[SolarNetId, Any]:
                 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 @@ def add_entities_for_seen_keys(
         """
 
         @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
-                ):
+                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
                     if device_data[key]["value"] is None:
+                        remaining_unregistered_descriptors.append(description)
                         continue
-                    new_entities.append(entity_constructor(self, key, solar_net_id))
-                    self.unregistered_keys[solar_net_id].remove(key)
+                    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)
         )