Add files via upload

Author: rosenrot00

custom_components/solax_modbus/__init__.py

@@ -1584,22 +1584,23 @@ def treat_address(self, data: dict[str, Any], regs: list[int], idx: int, descr:
                 return_value = round(val * descr.scale * read_scale, descr.rounding)
             except Exception:
                 return_value = val  # probably a REGISTER_WORDS instance
-            if descr.native_unit_of_measurement == UnitOfFrequency.HERTZ:
+            native_unit = getattr(descr, "native_unit_of_measurement", None)
+            if native_unit == UnitOfFrequency.HERTZ:
                 min_val = getattr(descr, "min_value", 20)
                 max_val = getattr(descr, "max_value", 80)
-            if descr.native_unit_of_measurement == PERCENTAGE:
+            if native_unit == PERCENTAGE:
                 min_val = getattr(descr, "min_value", 0)
                 max_val = getattr(descr, "max_value", 100)
-            elif descr.native_unit_of_measurement == UnitOfTemperature.CELSIUS:
+            elif native_unit == UnitOfTemperature.CELSIUS:
                 min_val = getattr(descr, "min_value", -100)
                 max_val = getattr(descr, "max_value", 200)
-            elif descr.native_unit_of_measurement == UnitOfPower.KILO_WATT:
+            elif native_unit == UnitOfPower.KILO_WATT:
                 min_val = getattr(descr, "min_value", -self.inverterPowerKw * 2)
                 max_val = getattr(descr, "max_value", +self.inverterPowerKw * 2)
-            elif descr.native_unit_of_measurement == UnitOfElectricCurrent.AMPERE:
+            elif native_unit == UnitOfElectricCurrent.AMPERE:
                 min_val = getattr(descr, "min_value", -self.inverterPowerKw * 2)
                 max_val = getattr(descr, "max_value", +self.inverterPowerKw * 2)
-            elif descr.native_unit_of_measurement == UnitOfElectricPotential.VOLT:
+            elif native_unit == UnitOfElectricPotential.VOLT:
                 min_val = getattr(descr, "min_value", 0)
                 max_val = getattr(descr, "max_value", 2000)
             else:

custom_components/solax_modbus/const.py

@@ -307,14 +307,26 @@ class BaseModbusTimeEntityDescription(TimeEntityDescription):
     allowedtypes: int = 0  # overload with ALLDEFAULT from plugin
     modbus_min: int | None = None  # Minimum supported Modbus protocol document version, e.g. 102 for V001.02.
     modbus_max: int | None = None  # Maximum supported Modbus protocol document version.
-    register: int | None = None
+    scale: float | dict[Any, Any] | Callable[[Any, Any, dict[str, Any]], Any] = 1
+    read_scale_exceptions: list[Any] | None = None
+    read_scale: float = 1
+    register: int = -1
+    rounding: int = 1
+    register_type: int | None = None  # REG_HOLDING or REG_INPUT or REG_DATA
+    register_data_type: str | None = REGISTER_U16  # REGISTER_U16, REGISTER_S32, REGISTER_F32, etc.
+    scan_group: str | None = None  # SCAN_GROUP_MEDIUM, SCAN_GROUP_FAST, SCAN_GROUP_DEFAULT, etc.
+    newblock: bool = False  # set to True to start a new modbus read block operation
     option_dict: dict[int, str] | None = None
     reverse_option_dict: dict[str, int] | None = None  # autocomputed
     blacklist: list[str] | None = None  # none or list of serial number prefixes
     write_method: int = WRITE_SINGLE_MODBUS  # WRITE_SINGLE_MOBUS or WRITE_MULTI_MODBUS or WRITE_DATA_LOCAL
     initvalue: int | None = None  # initial default value for WRITE_DATA_LOCAL entities
-    register_data_type: str | None = None  # REGISTER_U16, REGISTER_S32, REGISTER_F32, etc.
     wordcount: int | None = None  # number of registers to write (for separate register format, e.g., hours and minutes in adjacent registers)
+    sleepmode: int | None = SLEEPMODE_LAST  # or SLEEPMODE_ZERO, SLEEPMODE_NONE or SLEEPMODE_LASTAWAKE
+    ignore_readerror: bool | Any = False
+    min_value: int | None = None
+    max_value: int | None = None
+    depends_on: list[str] | None = None  # list of modbus register keys that must be read
 
 
 @dataclass(kw_only=True, frozen=True)

custom_components/solax_modbus/plugin_solax.py

@@ -1,6 +1,6 @@
 import logging
 from collections.abc import Sequence
-from dataclasses import dataclass, replace
+from dataclasses import dataclass, fields, replace
 from time import time
 from typing import Any
 
@@ -11288,6 +11288,39 @@ def value_function_battery_voltage_cell_difference(initval: int, descr: Any, dat
     ),
 ]
 
+
+_TIME_OPTION_DICTS = (TIME_OPTIONS, TIME_OPTIONS_GEN4, TIME_OPTIONS_SEPARATE_REGISTERS)
+_TIME_ENTITY_FIELD_NAMES = {field.name for field in fields(SolaXModbusTimeEntityDescription)}
+_TIME_ENTITY_KEYS = {(time_entity.key, time_entity.register, time_entity.allowedtypes) for time_entity in TIME_TYPES}
+
+
+def _is_time_select(select_entity: SolaxModbusSelectEntityDescription) -> bool:
+    """Return True for legacy time selects that should be exposed as TimeEntity."""
+    return any(select_entity.option_dict is option_dict for option_dict in _TIME_OPTION_DICTS)
+
+
+def _time_entity_from_select(select_entity: SolaxModbusSelectEntityDescription) -> SolaXModbusTimeEntityDescription:
+    """Convert a legacy time select descriptor into a native HA TimeEntity descriptor."""
+    return SolaXModbusTimeEntityDescription(
+        **{
+            field_name: getattr(select_entity, field_name)
+            for field_name in _TIME_ENTITY_FIELD_NAMES
+            if hasattr(select_entity, field_name)
+        }
+    )
+
+
+for _select_entity in SELECT_TYPES:
+    if not _is_time_select(_select_entity):
+        continue
+    _time_key = (_select_entity.key, _select_entity.register, _select_entity.allowedtypes)
+    if _time_key in _TIME_ENTITY_KEYS:
+        continue
+    TIME_TYPES.append(_time_entity_from_select(_select_entity))
+    _TIME_ENTITY_KEYS.add(_time_key)
+
+SELECT_TYPES = [select_entity for select_entity in SELECT_TYPES if not _is_time_select(select_entity)]
+
 # ============================ plugin declaration =================================================
 
 

custom_components/solax_modbus/time.py

@@ -115,6 +115,24 @@ def _handle_local_data_loaded(self, event: Any) -> None:
             return
         self.modbus_data_updated()
 
+    def _parse_time_string(self, time_val: str) -> datetime_time | None:
+        """Parse common time string formats into a datetime.time object."""
+        time_val = time_val.strip()
+        if not time_val:
+            _LOGGER.debug(f"{self._platform_name}: empty time string for {self._key}")
+            return None
+
+        for fmt in ["%H:%M", "%H:%M:%S", "%H:%M:%S.%f"]:
+            try:
+                parsed = datetime.strptime(time_val, fmt)
+                _LOGGER.debug(f"{self._platform_name}: parsed {self._key} as {fmt}: {parsed.time()}")
+                return parsed.time()
+            except ValueError:
+                continue
+
+        _LOGGER.debug(f"{self._platform_name}: unrecognized time format for {self._key}: {time_val}")
+        return None
+
     def _parse_time_value(self) -> datetime_time | None:
         """Parse the time value from hub.data and return a datetime.time object.
 
@@ -136,57 +154,17 @@ def _parse_time_value(self) -> datetime_time | None:
 
         # Handle string time values in hh:mm format
         if isinstance(time_val, str):
-            # Strip whitespace and handle empty strings
-            time_val = time_val.strip()
-            if not time_val:
-                _LOGGER.debug(f"{self._platform_name}: empty time string for {self._key}")
-                return None
-            # Common time formats
-            for fmt in ["%H:%M", "%H:%M:%S", "%H:%M:%S.%f"]:
-                try:
-                    parsed = datetime.strptime(time_val, fmt)
-                    _LOGGER.debug(f"{self._platform_name}: parsed {self._key} as {fmt}: {parsed.time()}")
-                    return parsed.time()
-                except ValueError:
-                    continue
-            # Try parsing as HH:MM:SS with seconds (8 chars like 05:25:30)
-            if len(time_val) == 8 and time_val[2] == ":" and time_val[5] == ":":
-                try:
-                    parsed = datetime.strptime(time_val, "%H:%M:%S")
-                    _LOGGER.debug(f"{self._platform_name}: parsed {self._key} as HH:MM:SS: {parsed.time()}")
-                    return parsed.time()
-                except ValueError:
-                    pass
-            # Try parsing as HH:MM (5 chars like 05:25)
-            if len(time_val) == 5 and time_val[2] == ":":
-                try:
-                    parsed = datetime.strptime(time_val, "%H:%M")
-                    _LOGGER.debug(f"{self._platform_name}: parsed {self._key} as HH:MM: {parsed.time()}")
-                    return parsed.time()
-                except ValueError:
-                    pass
-            # If we get here, the string format was not recognized
-            _LOGGER.debug(f"{self._platform_name}: unrecognized time format for {self._key}: {time_val}")
-            return None
+            return self._parse_time_string(time_val)
 
-        # Handle numeric values (e.g., from value_function_gen4time or value_function_gen23time)
+        # Handle raw Modbus payloads by translating through the descriptor's option table.
         if isinstance(time_val, (int, float)):
-            # Try to convert to string and parse
-            time_str = str(time_val)
-            if len(time_str) == 5 and time_str[2] == ":":
-                try:
-                    parsed = datetime.strptime(time_str, "%H:%M")
-                    _LOGGER.debug(f"{self._platform_name}: parsed numeric {self._key} as HH:MM: {parsed.time()}")
-                    return parsed.time()
-                except ValueError:
-                    pass
-            if len(time_str) == 8 and time_str[2] == ":" and time_str[5] == ":":
-                try:
-                    parsed = datetime.strptime(time_str, "%H:%M:%S")
-                    _LOGGER.debug(f"{self._platform_name}: parsed numeric {self._key} as HH:MM:SS: {parsed.time()}")
-                    return parsed.time()
-                except ValueError:
-                    pass
+            payload = int(time_val)
+            if self._option_dict is not None:
+                mapped_value = self._option_dict.get(payload)
+                if mapped_value is not None:
+                    return self._parse_time_string(mapped_value)
+            _LOGGER.debug(f"{self._platform_name}: no time option mapping for {self._key} payload {payload}")
+            return None
 
         _LOGGER.debug(f"{self._platform_name}: time value for {self._key} is not a string or datetime: {type(time_val)}")
         return None
@@ -225,7 +203,7 @@ async def async_set_value(self, value: datetime_time | None) -> None:
 
         # Find the corresponding payload from option_dict
         payload = None
-        for key, time_val in self._option_dict.items():
+        for key, time_val in (self._option_dict or {}).items():
             if time_val == time_str:
                 payload = key
                 break
@@ -259,7 +237,9 @@ async def async_set_value(self, value: datetime_time | None) -> None:
         elif self._write_method == WRITE_DATA_LOCAL:
             _LOGGER.info(f"*** local data written {self._key}: {time_str}")
             self._hub.localsUpdated = True  # mark to save permanently
-            self._hub.data[self._key] = time_str
+
+        self._hub.data[self._key] = time_str
+        self._attr_native_value = value
 
         self.async_write_ha_state()