Fix active power control services. Add Active power control sensor.

Fixes #736

__init__.py

@@ -28,7 +28,6 @@
     CONF_ENABLE_PARAMETER_CONFIGURATION,
     CONF_SLAVE_IDS,
     CONFIGURATION_UPDATE_INTERVAL,
-    DATA_BRIDGES_WITH_DEVICEINFOS,
     DATA_UPDATE_COORDINATORS,
     DOMAIN,
     ENERGY_STORAGE_UPDATE_INTERVAL,
@@ -48,10 +47,10 @@
 T = TypeVar("T")
 
 PLATFORMS: list[Platform] = [
-    Platform.SENSOR,
     Platform.NUMBER,
-    Platform.SWITCH,
     Platform.SELECT,
+    Platform.SENSOR,
+    Platform.SWITCH,
 ]
 
 
@@ -222,7 +221,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
             )
 
         hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
-            DATA_BRIDGES_WITH_DEVICEINFOS: bridges_with_device_infos,
             DATA_UPDATE_COORDINATORS: update_coordinators,
         }
     except (HuaweiSolarException, TimeoutError) as err:
@@ -317,16 +315,14 @@ def _battery_product_model_to_manufacturer(spm: rv.StorageProductModel):
             return "Huawei"
         if spm == rv.StorageProductModel.LG_RESU:
             return "LG Chem"
-        else:
-            return None
+        return None
 
     def _battery_product_model_to_model(spm: rv.StorageProductModel):
         if spm == rv.StorageProductModel.HUAWEI_LUNA2000:
             return "LUNA 2000"
         if spm == rv.StorageProductModel.LG_RESU:
             return "RESU"
-        else:
-            return None
+        return None
 
     battery_1_device_info = None
     if bridge.battery_1_type != rv.StorageProductModel.NONE:

const.py

@@ -12,8 +12,6 @@
 CONF_SLAVE_IDS = "slave_ids"
 CONF_ENABLE_PARAMETER_CONFIGURATION = "enable_parameter_configuration"
 
-
-DATA_BRIDGES_WITH_DEVICEINFOS = "bridges"
 DATA_UPDATE_COORDINATORS = "update_coordinators"
 
 INVERTER_UPDATE_INTERVAL = timedelta(seconds=30)

sensor.py

@@ -340,6 +340,7 @@ def context(self):
         value_conversion_function=lambda alarms: ", ".join(alarms)
         if len(alarms)
         else "None",
+        icon="mdi:alarm-light",
     ),
 )
 
@@ -828,6 +829,15 @@ async def async_setup_entry(
                 for entity_description in THREE_PHASE_METER_ENTITY_DESCRIPTIONS
             )
 
+        if ucs.bridge.has_write_permission and ucs.configuration_update_coordinator:
+            entities_to_add.append(
+                HuaweiSolarActivePowerControlModeEntity(
+                    ucs.configuration_update_coordinator,
+                    ucs.bridge,
+                    ucs.device_infos["inverter"],
+                )
+            )
+
         if ucs.bridge.battery_type != rv.StorageProductModel.NONE:
             assert ucs.energy_storage_update_coordinator
             assert ucs.device_infos["connected_energy_storage"]
@@ -1338,6 +1348,86 @@ def _handle_coordinator_update(self) -> None:
         self.async_write_ha_state()
 
 
+class HuaweiSolarActivePowerControlModeEntity(
+    CoordinatorEntity, HuaweiSolarEntity, SensorEntity
+):
+    """Huawei Solar Sensor for the current forcible charge status."""
+
+    REGISTER_NAMES = [
+        rn.ACTIVE_POWER_CONTROL_MODE,
+        rn.MAXIMUM_FEED_GRID_POWER_WATT,
+        rn.MAXIMUM_FEED_GRID_POWER_PERCENT,
+    ]
+
+    def __init__(
+        self,
+        coordinator: HuaweiSolarUpdateCoordinator,
+        bridge: HuaweiSolarBridge,
+        device_info: DeviceInfo,
+    ) -> None:
+        """Create HuaweiSolarForcibleChargeEntity."""
+        super().__init__(
+            coordinator,
+            {"register_names": self.REGISTER_NAMES},
+        )
+        self.coordinator = coordinator
+
+        self.entity_description = HuaweiSolarSensorEntityDescription(
+            key=rn.ACTIVE_POWER_CONTROL_MODE,
+            translation_key="active_power_control",
+            icon="mdi:transmission-tower",
+            entity_category=EntityCategory.DIAGNOSTIC,
+            entity_registry_enabled_default=False,
+        )
+
+        self._bridge = bridge
+        self._attr_device_info = device_info
+        self._attr_unique_id = f"{bridge.serial_number}_{self.entity_description.key}"
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""
+        if (
+            self.coordinator.data
+            and set(self.REGISTER_NAMES) <= self.coordinator.data.keys()
+        ):
+            mode = self.coordinator.data[rn.ACTIVE_POWER_CONTROL_MODE].value
+            maximum_power_watt = self.coordinator.data[
+                rn.MAXIMUM_FEED_GRID_POWER_WATT
+            ].value
+            maximum_power_percent = self.coordinator.data[
+                rn.MAXIMUM_FEED_GRID_POWER_PERCENT
+            ].value
+
+            if mode == rv.ActivePowerControlMode.UNLIMITED:
+                value = "Unlimited"
+            elif (
+                mode == rv.ActivePowerControlMode.POWER_LIMITED_GRID_CONNECTION_PERCENT
+            ):
+                value = f"Limited to {maximum_power_percent}%"
+            elif mode == rv.ActivePowerControlMode.POWER_LIMITED_GRID_CONNECTION_WATT:
+                value = f"Limited to {maximum_power_watt}W"
+            elif mode == rv.ActivePowerControlMode.ZERO_POWER_GRID_CONNECTION:
+                value = "Zero Power"
+            elif mode == rv.ActivePowerControlMode.DI_ACTIVE_SCHEDULING:
+                value = "DI Active Scheduling"
+            else:
+                value = "Unknown"
+
+            self._attr_available = True
+            self._attr_native_value = value
+            self._attr_extra_state_attributes = {
+                "mode": str(mode),
+                "maximum_power_watt": maximum_power_watt,
+                "maximum_power_percent": maximum_power_percent,
+            }
+        else:
+            self._attr_available = False
+            self._attr_native_value = None
+            self._attr_extra_state_attributes.clear()
+        self.async_write_ha_state()
+
+
 class HuaweiSolarOptimizerSensorEntity(
     CoordinatorEntity, HuaweiSolarEntity, SensorEntity
 ):

services.py

@@ -5,7 +5,7 @@
 from functools import partial
 import logging
 import re
-from typing import Any, TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 import voluptuous as vol
 
@@ -24,7 +24,6 @@
 
 from .const import (
     CONF_ENABLE_PARAMETER_CONFIGURATION,
-    DATA_BRIDGES_WITH_DEVICEINFOS,
     DATA_UPDATE_COORDINATORS,
     DOMAIN,
     SERVICE_FORCIBLE_CHARGE,
@@ -232,26 +231,32 @@ def get_battery_bridge(
 
 
 @callback
-def _get_inverter_bridge(hass: HomeAssistant, device_id: str):
+def _get_inverter_bridge(
+    hass: HomeAssistant, device_id: str
+) -> tuple[HuaweiSolarBridge, HuaweiSolarUpdateCoordinator]:
     dev_reg = dr.async_get(hass)
     device_entry = dev_reg.async_get(device_id)
 
     if not device_entry:
         raise HuaweiSolarServiceException("No such device found")
     for entry_data in hass.data[DOMAIN].values():
-        bridges_with_device_infos = entry_data[DATA_BRIDGES_WITH_DEVICEINFOS]
-        for bridge, device_infos in bridges_with_device_infos:
-            for identifier in device_infos["inverter"]["identifiers"]:
+        hsucs: list[HuaweiSolarUpdateCoordinators] = entry_data[
+            DATA_UPDATE_COORDINATORS
+        ]
+        for uc in hsucs:
+            for identifier in uc.device_infos["inverter"]["identifiers"]:
                 for device_identifier in device_entry.identifiers:
                     if identifier == device_identifier:
-                        return bridge
+                        return uc.bridge, uc.configuration_update_coordinator
 
     _LOGGER.error("The provided device is not an inverter")
     raise HuaweiSolarServiceException("Not a valid 'Inverter' device")
 
 
 @callback
-def get_inverter_bridge(hass: HomeAssistant, service_call: ServiceCall):
+def get_inverter_bridge(
+    hass: HomeAssistant, service_call: ServiceCall
+) -> tuple[HuaweiSolarBridge, HuaweiSolarUpdateCoordinator]:
     """Return the HuaweiSolarBridge associated with the inverter device_id in the service call."""
     device_id = service_call.data[DATA_DEVICE_ID]
     return _get_inverter_bridge(hass, device_id)
@@ -649,14 +654,11 @@ async def async_setup_services(  # noqa: C901
         schema=MAXIMUM_FEED_GRID_POWER_PERCENTAGE_SCHEMA,
     )
 
-    bridges_with_device_infos = hass.data[DOMAIN][entry.entry_id][
-        DATA_BRIDGES_WITH_DEVICEINFOS
+    hsucs: list[HuaweiSolarUpdateCoordinators] = hass.data[DOMAIN][entry.entry_id][
+        DATA_UPDATE_COORDINATORS
     ]
 
-    if any(
-        bridge.battery_type != rv.StorageProductModel.NONE
-        for bridge, _ in bridges_with_device_infos
-    ):
+    if any(uc.bridge.battery_type != rv.StorageProductModel.NONE for uc in hsucs):
         hass.services.async_register(
             DOMAIN,
             SERVICE_FORCIBLE_CHARGE,
@@ -702,9 +704,7 @@ async def async_setup_services(  # noqa: C901
             schema=FIXED_CHARGE_PERIODS_SCHEMA,
         )
 
-        if any(
-            bridge.supports_capacity_control for bridge, _ in bridges_with_device_infos
-        ):
+        if any(uc.bridge.supports_capacity_control for uc in hsucs):
             hass.services.async_register(
                 DOMAIN,
                 SERVICE_SET_CAPACITY_CONTROL_PERIODS,

strings.json

@@ -469,7 +469,10 @@
                 "name": "Remaining charge/discharge time"
             },
             "forcible_charge_summary": {
-                "name": "Forcible Charge"
+                "name": "Forcible charge"
+            },
+            "active_power_control": {
+                "name": "Active power control"
             }
         },
         "select": {

translations/en.json

@@ -145,6 +145,9 @@
             "active_power": {
                 "name": "Active power"
             },
+            "active_power_control": {
+                "name": "Active power control"
+            },
             "alarm": {
                 "name": "Alarm"
             },