Add ability to change Inv based on selected inverter manager version

We don't (yet?) have the option to automatically switch based on the
version reported by the inverter -- that's something I've tried and failed
to do a couple of times. But it adds an option to the settings to select
your manager version, which at least means that people have the option
of not being broken.

custom_components/foxess_modbus/const.py

@@ -47,6 +47,9 @@
 INVERTER_MODEL = "inverter_model"
 INVERTER_BASE = "inverter_base"
 INVERTER_CONN = "inverter_conn"
+# The inverter manager version to use. This is the version corresponding to InverterModelConnectionTypeProfile.versions,
+# i.e. the upper bound of a range of versions we support. None means use the latest.
+INVERTER_VERSION = "inverter_version"
 INVERTERS = "inverters"
 
 CONFIG_SAVE_TIME = "save_time"

custom_components/foxess_modbus/flow/options_handler.py

@@ -8,12 +8,14 @@
 
 from ..const import ADAPTER_ID
 from ..const import CONFIG_ENTRY_TITLE
+from ..const import INVERTER_VERSION
 from ..const import INVERTERS
 from ..const import MAX_READ
 from ..const import MODBUS_TYPE
 from ..const import POLL_RATE
 from ..const import ROUND_SENSOR_VALUES
 from ..inverter_adapters import ADAPTERS
+from ..inverter_profiles import inverter_connection_type_profile_from_config
 from .adapter_flow_segment import AdapterFlowSegment
 from .flow_handler_mixin import FlowHandlerMixin
 
@@ -117,15 +119,23 @@ async def async_step_inverter_advanced_options(self, user_input: dict[str, Any]
         current_adapter = ADAPTERS[combined_config_options[ADAPTER_ID]]
 
         async def body(user_input: dict[str, Any]) -> FlowResult:
+            version = user_input.get("version")
+            if version is None or version == "latest":
+                options.pop(INVERTER_VERSION, None)
+            else:
+                options[INVERTER_VERSION] = version
+
             poll_rate = user_input.get("poll_rate")
             if poll_rate is not None:
                 options[POLL_RATE] = poll_rate
             else:
                 options.pop(POLL_RATE, None)
+
             if user_input.get("round_sensor_values", False):
                 options[ROUND_SENSOR_VALUES] = True
             else:
                 options.pop(ROUND_SENSOR_VALUES, None)
+
             max_read = user_input.get("max_read")
             if max_read is not None:
                 options[MAX_READ] = max_read
@@ -136,6 +146,28 @@ async def body(user_input: dict[str, Any]) -> FlowResult:
 
         schema_parts: dict[Any, Any] = {}
 
+        versions = sorted(inverter_connection_type_profile_from_config(combined_config_options).versions.keys())
+        if len(versions) > 1:
+            version_options = []
+            prev_version = None
+            # The last element will be None, which means "latest"
+            for version in versions[:-1]:
+                label = f"Up to {version}" if prev_version is None else f"{prev_version} - {version}"
+                version_options.append({"label": label, "value": str(version)})
+                prev_version = version
+            version_options.append(
+                {"label": f"{versions[-2]} and higher", "value": "latest"}
+            )  # hass can't cope with None
+
+            schema_parts[vol.Required("version", default=options.get(INVERTER_VERSION, "latest"))] = selector(
+                {
+                    "select": {
+                        "options": list(reversed(version_options)),
+                        "mode": "dropdown",
+                    }
+                }
+            )
+
         schema_parts[vol.Required("round_sensor_values", default=options.get(ROUND_SENSOR_VALUES, False))] = selector(
             {"boolean": {}}
         )

custom_components/foxess_modbus/inverter_profiles.py

@@ -1,5 +1,6 @@
 """Defines the different inverter models and connection types"""
 
+import functools
 import logging
 import re
 from dataclasses import dataclass
@@ -15,6 +16,7 @@
 from .common.types import RegisterType
 from .const import INVERTER_BASE
 from .const import INVERTER_CONN
+from .const import INVERTER_VERSION
 from .entities.charge_period_descriptions import CHARGE_PERIODS
 from .entities.entity_descriptions import ENTITIES
 from .entities.modbus_charge_period_config import ModbusChargePeriodInfo
@@ -24,6 +26,40 @@
 _LOGGER = logging.getLogger(__package__)
 
 
+@functools.total_ordering
+class Version:
+    def __init__(self, major: int, minor: int) -> None:
+        self.major = major
+        self.minor = minor
+
+    @staticmethod
+    def parse(version: str) -> "Version":
+        match = re.fullmatch(r"(\d+)\.(\d+)", version)
+        if match is None:
+            raise ValueError(f"Version {version} is not a valid version")
+        return Version(int(match[1]), int(match[2]))
+
+    def __eq__(self, other: Any) -> bool:
+        return isinstance(other, Version) and self.major == other.major and self.minor == other.minor
+
+    def __hash__(self) -> int:
+        return hash((self.major, self.minor))
+
+    def __lt__(self, other: Any) -> bool:
+        # None means "the latest", and so sorts higher than anything (except None)
+        if other is None:
+            return True
+        if self.major != other.major:
+            return self.major < other.major
+        return self.minor < other.minor
+
+    def __str__(self) -> str:
+        return f"{self.major}.{self.minor}"
+
+    def __repr__(self) -> str:
+        return f"Version({self.major}, {self.minor})"
+
+
 class SpecialRegisterConfig:
     def __init__(
         self,
@@ -90,17 +126,31 @@ class InverterModelConnectionTypeProfile:
     def __init__(
         self,
         inverter_model_profile: "InverterModelProfile",
-        inv: Inv,
         connection_type: ConnectionType,
         register_type: RegisterType,
+        versions: dict[Version | None, Inv],
         special_registers: SpecialRegisterConfig,
     ) -> None:
-        self._inv = inv
         self.inverter_model_profile = inverter_model_profile
         self.connection_type = connection_type
         self.register_type = register_type
+        self.versions = versions
         self.special_registers = special_registers
 
+        assert None in versions
+
+    def _get_inv(self, controller: EntityController) -> Inv:
+        version_from_config = controller.inverter_details.get(INVERTER_VERSION)
+        # Remember that self._versions is a map of maximum supported manager version (or None to support the max
+        # firmware version) -> Inv for that version
+        if version_from_config is None:
+            return self.versions[None]
+
+        inverter_version = Version.parse(version_from_config)
+        versions = sorted(self.versions.items(), reverse=True)
+        matched_version = next((x for x in versions if x[0] <= inverter_version), versions[0])
+        return matched_version[1]
+
     def overlaps_invalid_range(self, start_address: int, end_address: int) -> bool:
         """Determines whether the given inclusive address range overlaps any invalid address ranges"""
         return any(
@@ -122,9 +172,7 @@ def create_entities(
         for entity_factory in ENTITIES:
             if entity_factory.entity_type == entity_type:
                 entity = entity_factory.create_entity_if_supported(
-                    controller,
-                    self._inv,
-                    self.register_type,
+                    controller, self._get_inv(controller), self.register_type
                 )
                 if entity is not None:
                     result.append(entity)
@@ -138,15 +186,15 @@ def create_charge_periods(self, controller: EntityController) -> list[ModbusChar
 
         for charge_period_factory in CHARGE_PERIODS:
             charge_period = charge_period_factory.create_charge_period_config_if_supported(
-                controller, self._inv, self.register_type
+                controller, self._get_inv(controller), self.register_type
             )
             if charge_period is not None:
                 result.append(charge_period)
 
         return result
 
     def create_remote_control_config(self, controller: EntityController) -> ModbusRemoteControlAddressConfig | None:
-        return REMOTE_CONTROL_DESCRIPTION.create_if_supported(controller, self._inv, self.register_type)
+        return REMOTE_CONTROL_DESCRIPTION.create_if_supported(controller, self._get_inv(controller), self.register_type)
 
 
 class InverterModelProfile:
@@ -160,9 +208,9 @@ def __init__(self, model: InverterModel, model_pattern: str, capacity_parser: Ca
 
     def add_connection_type(
         self,
-        inv: Inv,
         connection_type: ConnectionType,
         register_type: RegisterType,
+        versions: dict[Version | None, Inv],  # Map of maximum supported manager versions -> Inv for that
         special_registers: SpecialRegisterConfig | None = None,
     ) -> "InverterModelProfile":
         """Add the given connection type to the profile"""
@@ -173,9 +221,9 @@ def add_connection_type(
 
         self.connection_types[connection_type] = InverterModelConnectionTypeProfile(
             self,
-            inv,
             connection_type,
             register_type,
+            versions,
             special_registers,
         )
         return self
@@ -196,102 +244,102 @@ def inverter_capacity(self, inverter_model: str) -> int:
         InverterModelProfile(
             InverterModel.H1_G2, r"^H1-([\d\.]+)-E-G2", capacity_parser=CapacityParser.H1
         ).add_connection_type(
-            Inv.H1_G2,
             ConnectionType.AUX,
             RegisterType.HOLDING,
+            versions={None: Inv.H1_G2},
             special_registers=H1_G2_REGISTERS,
         ),
         # Can be both e.g. H1-5.0 and H1-5.0-E, but not H1-5.0-E-G2
         InverterModelProfile(InverterModel.H1_G1, r"^H1-([\d\.]+)", capacity_parser=CapacityParser.H1)
         .add_connection_type(
-            Inv.H1_G1,
             ConnectionType.AUX,
             RegisterType.INPUT,
+            versions={None: Inv.H1_G1},
             special_registers=H1_AC1_REGISTERS,
         )
         .add_connection_type(
-            Inv.H1_LAN,
             ConnectionType.LAN,
             RegisterType.HOLDING,
+            versions={None: Inv.H1_LAN},
         ),
         InverterModelProfile(InverterModel.AC1, r"^AC1-([\d\.]+)", capacity_parser=CapacityParser.H1)
         .add_connection_type(
-            Inv.H1_G1,
             ConnectionType.AUX,
             RegisterType.INPUT,
+            versions={None: Inv.H1_G1},
             special_registers=H1_AC1_REGISTERS,
         )
         .add_connection_type(
-            Inv.H1_LAN,
             ConnectionType.LAN,
             RegisterType.HOLDING,
+            versions={None: Inv.H1_LAN},
         ),
         InverterModelProfile(InverterModel.AIO_H1, r"^AIO-H1-([\d\.]+)", capacity_parser=CapacityParser.H1)
         .add_connection_type(
-            Inv.H1_G1,
             ConnectionType.AUX,
             RegisterType.INPUT,
+            versions={None: Inv.H1_G1},
             special_registers=H1_AC1_REGISTERS,
         )
         .add_connection_type(
-            Inv.H1_LAN,
             ConnectionType.LAN,
             RegisterType.HOLDING,
+            versions={None: Inv.H1_LAN},
         ),
         InverterModelProfile(
             InverterModel.AIO_AC1, r"^AIO-AC1-([\d\.]+)", capacity_parser=CapacityParser.H1
         ).add_connection_type(
-            Inv.H1_G1,
             ConnectionType.AUX,
             RegisterType.INPUT,
+            versions={None: Inv.H1_G1},
             special_registers=H1_AC1_REGISTERS,
         ),
         # The KH doesn't have a LAN port. It supports both input and holding over RS485
         # Some models start with KH-, but some are just e.g. KH10.5
         InverterModelProfile(InverterModel.KH, r"^KH([\d\.]+)").add_connection_type(
-            Inv.KH_119,
             ConnectionType.AUX,
             RegisterType.HOLDING,
+            versions={Version(1, 19): Inv.KH_PRE119, None: Inv.KH_119},
             special_registers=KH_REGISTERS,
         ),
         # The H3 seems to use holding registers for everything
         InverterModelProfile(InverterModel.H3, r"^H3-([\d\.]+)")
         .add_connection_type(
-            Inv.H3,
             ConnectionType.LAN,
             RegisterType.HOLDING,
+            versions={None: Inv.H3},
             special_registers=H3_REGISTERS,
         )
         .add_connection_type(
-            Inv.H3,
             ConnectionType.AUX,
             RegisterType.HOLDING,
+            versions={None: Inv.H3},
             special_registers=H3_REGISTERS,
         ),
         InverterModelProfile(InverterModel.AC3, r"^AC3-([\d\.]+)")
         .add_connection_type(
-            Inv.H3,
             ConnectionType.LAN,
             RegisterType.HOLDING,
+            versions={None: Inv.H3},
             special_registers=H3_REGISTERS,
         )
         .add_connection_type(
-            Inv.H3,
             ConnectionType.AUX,
             RegisterType.HOLDING,
+            versions={None: Inv.H3},
             special_registers=H3_REGISTERS,
         ),
         InverterModelProfile(InverterModel.AIO_H3, r"^AIO-H3-([\d\.]+)")
         .add_connection_type(
-            Inv.AIO_H3,
             ConnectionType.AUX,
             RegisterType.HOLDING,
+            versions={None: Inv.AIO_H3},
             special_registers=H3_REGISTERS,
         )
         .add_connection_type(
-            Inv.H3,
             ConnectionType.LAN,
             RegisterType.HOLDING,
+            versions={None: Inv.AIO_H3},
             special_registers=H3_REGISTERS,
         ),
         # Kuara 6.0-3-H: H3-6.0-E
@@ -300,27 +348,27 @@ def inverter_capacity(self, inverter_model: str) -> int:
         # Kuara 12.0-3-H: H3-12.0-E
         # I haven't seen any indication that these support a direct LAN connection
         InverterModelProfile(InverterModel.KUARA_H3, r"^Kuara ([\d\.]+)-3-H$").add_connection_type(
-            Inv.KUARA_H3,
             ConnectionType.AUX,
             RegisterType.HOLDING,
+            versions={None: Inv.KUARA_H3},
             special_registers=H3_REGISTERS,
         ),
         # Sonnenkraft:
         # SK-HWR-8: H3-8.0-E
         # (presumably there are other sizes also)
         InverterModelProfile(InverterModel.SK_HWR, r"^SK-HWR-([\d\.]+)").add_connection_type(
-            Inv.H3,
             ConnectionType.AUX,
             RegisterType.HOLDING,
+            versions={None: Inv.H3},
             special_registers=H3_REGISTERS,
         ),
         # STAR
         # STAR-H3-12.0-E: H3-12.0-E
         # (presumably there are other sizes also)
         InverterModelProfile(InverterModel.STAR_H3, r"^STAR-H3-([\d\.]+)").add_connection_type(
-            Inv.H3,
             ConnectionType.AUX,
             RegisterType.HOLDING,
+            versions={None: Inv.H3},
             special_registers=H3_REGISTERS,
         ),
         # Solavita SP
@@ -338,26 +386,23 @@ def inverter_capacity(self, inverter_model: str) -> int:
                 fallback_to_kw=False,
             ),
         ).add_connection_type(
-            Inv.H3,
             ConnectionType.AUX,
             RegisterType.HOLDING,
+            versions={None: Inv.H3},
             special_registers=H3_REGISTERS,
         ),
         # E.g. H3-Pro-20.0
         InverterModelProfile(InverterModel.H3_PRO, r"^H3-Pro-([\d\.]+)").add_connection_type(
-            Inv.H3_PRO,
             ConnectionType.AUX,
             RegisterType.HOLDING,
+            versions={None: Inv.H3_PRO},
             special_registers=H3_REGISTERS,
         ),
     ]
 }
 
 
-def create_entities(
-    entity_type: type[Entity],
-    controller: EntityController,
-) -> list[Entity]:
+def create_entities(entity_type: type[Entity], controller: EntityController) -> list[Entity]:
     """Create all of the entities which support the inverter described by the given configuration object"""
 
     return inverter_connection_type_profile_from_config(controller.inverter_details).create_entities(