Merge pull request #364 from nathanmarlor/feature/charge-period-card

Add support for the charge period card

custom_components/foxess_modbus/__init__.py

@@ -41,6 +41,7 @@
 from .modbus_client import ModbusClient
 from .modbus_controller import ModbusController
 from .services import update_charge_period_service
+from .services import websocket_api
 from .services import write_registers_service
 
 _LOGGER: logging.Logger = logging.getLogger(__package__)
@@ -75,6 +76,7 @@ def create_controller(client: ModbusClient, inverter: dict[str, Any]) -> None:
             hass,
             client,
             inverter_connection_type_profile_from_config(inverter),
+            inverter,
             inverter[MODBUS_SLAVE],
             inverter[POLL_RATE],
             inverter[MAX_READ],
@@ -118,6 +120,7 @@ def create_controller(client: ModbusClient, inverter: dict[str, Any]) -> None:
 
     write_registers_service.register(hass, inverter_controllers)
     update_charge_period_service.register(hass, inverter_controllers)
+    websocket_api.register(hass)
 
     hass.data[DOMAIN][entry.entry_id][INVERTERS] = inverter_controllers
     hass.data[DOMAIN][entry.entry_id][MODBUS_CLIENTS] = clients.values()

custom_components/foxess_modbus/entities/charge_periods.py

@@ -5,7 +5,7 @@
 from ..const import AIO_H1
 from ..const import H1
 from .modbus_charge_period_config import ChargePeriodAddressSpec
-from .modbus_charge_period_config import ModbusChargePeriodConfig
+from .modbus_charge_period_config import ModbusChargePeriodAddressConfig
 from .modbus_charge_period_config import ModbusChargePeriodFactory
 
 _LOGGER: logging.Logger = logging.getLogger(__package__)
@@ -15,7 +15,7 @@
         addresses=[
             ChargePeriodAddressSpec(
                 models=[H1, AIO_H1, AC1],
-                input=ModbusChargePeriodConfig(
+                input=ModbusChargePeriodAddressConfig(
                     period_start_address=41002,
                     period_end_address=41003,
                     enable_charge_from_grid_address=41001,
@@ -35,7 +35,7 @@
         addresses=[
             ChargePeriodAddressSpec(
                 models=[H1, AIO_H1, AC1],
-                input=ModbusChargePeriodConfig(
+                input=ModbusChargePeriodAddressConfig(
                     period_start_address=41005,
                     period_end_address=41006,
                     enable_charge_from_grid_address=41004,

custom_components/foxess_modbus/entities/modbus_charge_period_config.py

@@ -1,6 +1,7 @@
 """Time period config"""
 import logging
 from dataclasses import dataclass
+from typing import Any
 
 from homeassistant.components.binary_sensor import BinarySensorDeviceClass
 
@@ -10,6 +11,7 @@
 from .modbus_binary_sensor import ModbusBinarySensorDescription
 from .modbus_charge_period_sensors import ModbusChargePeriodStartEndSensorDescription
 from .modbus_charge_period_sensors import ModbusEnableForceChargeSensorDescription
+from .modbus_entity_mixin import add_entity_id_prefix
 from .validation import Time
 
 _LOGGER = logging.getLogger(__name__)
@@ -19,33 +21,42 @@
 
 
 @dataclass
-class ModbusChargePeriodConfig:
+class ModbusChargePeriodAddressConfig:
     """Defines the set of registers which are used to define a charge period"""
 
     period_start_address: int
     period_end_address: int
     enable_charge_from_grid_address: int
 
 
+@dataclass
+class ModbusChargePeriodInfo:
+    addresses: ModbusChargePeriodAddressConfig
+    period_start_entity_id: str
+    period_end_entity_id: str
+    enable_force_charge_entity_id: str
+    enable_charge_from_grid_entity_id: str
+
+
 class ChargePeriodAddressSpec:
     """
     Specifies the addresses involved in a charge period, for a given list of inverter models
 
     For example:
     addrseses=[
-        ChargePeriodAddressSpec([H1, AC1], aux=ModbusChargePeriodConfig(period_start_address=...))
-        ChargePeriodAddressSpec([H3], aux=ModbusChargePeriodConfig(period_start_address=...))
+        ChargePeriodAddressSpec([H1, AC1], aux=ModbusChargePeriodAddressConfig(period_start_address=...))
+        ChargePeriodAddressSpec([H3], aux=ModbusChargePeriodAddressConfig(period_start_address=...))
     ]
     """
 
     def __init__(
         self,
         models: list[str],
-        input: ModbusChargePeriodConfig | None = None,  # noqa: A002
-        holding: ModbusChargePeriodConfig | None = None,
+        input: ModbusChargePeriodAddressConfig | None = None,  # noqa
+        holding: ModbusChargePeriodAddressConfig | None = None,
     ) -> None:
         self.models = models
-        self.register_types: dict[RegisterType, ModbusChargePeriodConfig] = {}
+        self.register_types: dict[RegisterType, ModbusChargePeriodAddressConfig] = {}
         if input is not None:
             self.register_types[RegisterType.INPUT] = input
         if holding is not None:
@@ -82,7 +93,7 @@ class ModbusChargePeriodFactory:
     Factory which creates various things required to define and specify a charge period
 
     This is used to create the entities which visualise the various bits of charge period start
-    (start time, etc), and also the ModbusChargePeriodConfig which is used internally when
+    (start time, etc), and also the ModbusChargePeriodAddressConfig which is used internally when
     interacting with charge periods
     """
 
@@ -100,6 +111,11 @@ def __init__(
     ) -> None:
         self.address_specs = addresses
 
+        self._period_start_key = period_start_key
+        self._period_end_key = period_end_key
+        self._enable_force_charge_key = enable_force_charge_key
+        self._enable_charge_from_grid_key = enable_charge_from_grid_key
+
         period_start_address = [x.get_start_address() for x in addresses]
         period_end_address = [x.get_end_address() for x in addresses]
         enable_charge_from_grid_address = [x.get_enable_charge_from_grid_address() for x in addresses]
@@ -146,20 +162,36 @@ def __init__(
         ]
 
     def create_charge_period_config_if_supported(
-        self, inverter_model: str, register_type: RegisterType
-    ) -> ModbusChargePeriodConfig | None:
+        self, inverter_model: str, register_type: RegisterType, inv_details: dict[str, Any]
+    ) -> ModbusChargePeriodInfo | None:
         """
-        If the inverter model / connection type supports a charge period, fetches a ModbusChargePeriodConfig containing
-        the register addresses involved. If not supported, returns None.
+        If the inverter model / connection type supports a charge period, fetches a ModbusChargePeriodAddressConfig
+        containing the register addresses involved. If not supported, returns None.
         """
 
-        result: ModbusChargePeriodConfig | None = None
+        result: ModbusChargePeriodInfo | None = None
         for address_spec in self.address_specs:
             if inverter_model in address_spec.models:
-                config = address_spec.register_types.get(register_type)
-                if config is not None:
+                address_config = address_spec.register_types.get(register_type)
+                if address_config is not None:
                     assert (
                         result is None
                     ), f"{self}: multiple charge periods defined for ({inverter_model}, {register_type})"
-                    result = config
+
+                    # TODO: It isn't great that this logic is duplicated between this and the entities
+                    start_id = f"sensor.{add_entity_id_prefix(self._period_start_key, inv_details)}"
+                    end_id = f"sensor.{add_entity_id_prefix(self._period_end_key, inv_details)}"
+                    enable_force_charge_id = (
+                        f"binary_sensor.{add_entity_id_prefix(self._enable_force_charge_key, inv_details)}"
+                    )
+                    enable_charge_from_grid_id = (
+                        f"binary_sensor.{add_entity_id_prefix(self._enable_charge_from_grid_key, inv_details)}"
+                    )
+                    result = ModbusChargePeriodInfo(
+                        addresses=address_config,
+                        period_start_entity_id=start_id,
+                        period_end_entity_id=end_id,
+                        enable_force_charge_entity_id=enable_force_charge_id,
+                        enable_charge_from_grid_entity_id=enable_charge_from_grid_id,
+                    )
         return result

custom_components/foxess_modbus/entities/modbus_entity_mixin.py

@@ -20,6 +20,16 @@
 _LOGGER = logging.getLogger(__name__)
 
 
+def add_entity_id_prefix(entity_id: str, inv_details: dict[str, Any]) -> str:
+    """Add the entity ID prefix to the beginning of the given input string"""
+    entity_id_prefix = inv_details[ENTITY_ID_PREFIX]
+
+    if entity_id_prefix:
+        entity_id = f"{entity_id_prefix}_{entity_id}"
+
+    return entity_id
+
+
 class ModbusEntityProtocol(Protocol):
     """Protocol which types including ModbusEntityMixin must implement"""
 
@@ -102,14 +112,9 @@ def _get_unique_id(self) -> str:
         """Get unique ID"""
         return self._add_entity_id_prefix(self.entity_description.key)
 
-    def _add_entity_id_prefix(self, value: str) -> str:
+    def _add_entity_id_prefix(self, entity_id: str) -> str:
         """Add the entity ID prefix to the beginning of the given input string"""
-        entity_id_prefix = self._inv_details[ENTITY_ID_PREFIX]
-
-        if entity_id_prefix:
-            value = f"{entity_id_prefix}_{value}"
-
-        return value
+        return add_entity_id_prefix(entity_id, self._inv_details)
 
     def _validate(
         self,

custom_components/foxess_modbus/inverter_profiles.py

@@ -20,7 +20,7 @@
 from .entities import invalid_ranges
 from .entities.charge_periods import CHARGE_PERIODS
 from .entities.entity_descriptions import ENTITIES
-from .entities.modbus_charge_period_config import ModbusChargePeriodConfig
+from .entities.modbus_charge_period_config import ModbusChargePeriodInfo
 
 _LOGGER = logging.getLogger(__package__)
 
@@ -69,14 +69,14 @@ def create_entities(
 
         return result
 
-    def create_charge_periods(self) -> list[ModbusChargePeriodConfig]:
+    def create_charge_periods(self, inverter_details: dict[str, Any]) -> list[ModbusChargePeriodInfo]:
         """Create all of the charge periods which support this inverter/connection combination"""
 
         result = []
 
         for charge_period_factory in CHARGE_PERIODS:
             charge_period = charge_period_factory.create_charge_period_config_if_supported(
-                self.inverter_model, self.register_type
+                self.inverter_model, self.register_type, inverter_details
             )
             if charge_period is not None:
                 result.append(charge_period)

custom_components/foxess_modbus/modbus_controller.py

@@ -51,6 +51,7 @@ def __init__(
         hass: HomeAssistant,
         client: ModbusClient,
         connection_type_profile: InverterModelConnectionTypeProfile,
+        inverter_details: dict[str, Any],
         slave: int,
         poll_rate: int,
         max_read: int,
@@ -61,7 +62,8 @@ def __init__(
         self._data: dict[int, int | None] = {}
         self._client = client
         self._connection_type_profile = connection_type_profile
-        self.charge_periods = connection_type_profile.create_charge_periods()
+        self.inverter_details = inverter_details
+        self.charge_periods = connection_type_profile.create_charge_periods(inverter_details)
         self._slave = slave
         self._poll_rate = poll_rate
         self._max_read = max_read

custom_components/foxess_modbus/services/update_charge_period_service.py

@@ -188,9 +188,9 @@ async def _update_charge_period(
         if i == charge_period_index:
             continue
 
-        period_start_time_value = controller.read(charge_period.period_start_address)
-        period_end_time_value = controller.read(charge_period.period_end_address)
-        period_enable_charge_from_grid_value = controller.read(charge_period.enable_charge_from_grid_address)
+        period_start_time_value = controller.read(charge_period.addresses.period_start_address)
+        period_end_time_value = controller.read(charge_period.addresses.period_end_address)
+        period_enable_charge_from_grid_value = controller.read(charge_period.addresses.enable_charge_from_grid_address)
 
         if (
             period_start_time_value is None
@@ -245,19 +245,19 @@ async def _set_charge_periods(controller: ModbusController, charge_periods: list
     for charge_period, config in zip(charge_periods, controller.charge_periods, strict=True):
         writes.append(
             (
-                config.period_start_address,
+                config.addresses.period_start_address,
                 serialize_time_to_value(charge_period.start) if charge_period.enable_force_charge else 0,
             )
         )
         writes.append(
             (
-                config.period_end_address,
+                config.addresses.period_end_address,
                 serialize_time_to_value(charge_period.end) if charge_period.enable_force_charge else 0,
             )
         )
         writes.append(
             (
-                config.enable_charge_from_grid_address,
+                config.addresses.enable_charge_from_grid_address,
                 1 if charge_period.enable_charge_from_grid else 0,
             )
         )

custom_components/foxess_modbus/services/websocket_api.py

@@ -0,0 +1,45 @@
+from typing import Any
+
+import voluptuous as vol
+from homeassistant.components import websocket_api
+from homeassistant.core import HomeAssistant
+from homeassistant.core import callback
+from homeassistant.helpers import config_validation as cv
+
+from ..const import DOMAIN
+from ..const import FRIENDLY_NAME
+from ..const import INVERTERS
+from .utils import get_controller_from_friendly_name_or_device_id
+
+
+def register(hass: HomeAssistant) -> None:
+    hass.components.websocket_api.async_register_command(get_charge_periods)
+
+
+@websocket_api.websocket_command(
+    {
+        vol.Required("type"): "foxess_modbus/get_charge_periods",
+        vol.Required("inverter"): vol.Any(cv.string, None),
+    }
+)
+@callback
+def get_charge_periods(hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]) -> None:
+    inverter_controllers = [x for entry in hass.data[DOMAIN].values() for x in entry[INVERTERS]]
+    controller = get_controller_from_friendly_name_or_device_id(msg["inverter"], inverter_controllers, hass)
+    charge_periods = []
+    for charge_period in controller.charge_periods:
+        charge_periods.append(
+            {
+                "period_start_entity_id": charge_period.period_start_entity_id,
+                "period_end_entity_id": charge_period.period_end_entity_id,
+                "enable_force_charge_entity_id": charge_period.enable_force_charge_entity_id,
+                "enable_charge_from_grid_entity_id": charge_period.enable_charge_from_grid_entity_id,
+            }
+        )
+    connection.send_result(
+        msg["id"],
+        {
+            "friendly_name": controller.inverter_details[FRIENDLY_NAME],
+            "charge_periods": charge_periods,
+        },
+    )