time

Author: kellerza

.github/workflows/lint.yml

@@ -7,8 +7,8 @@ on:
   pull_request:
     branches:
       - main
-  schedule:
-    - cron: "0 0 * * *"
+  # schedule:
+  #   - cron: "0 0 * * *"
 
 jobs:
   find:

pyproject.toml

@@ -26,7 +26,7 @@ dependencies = [
   "mqtt-entity[options]>=1.1",
   "prettytable>=3.17,<4",
   "pymodbus[serial]==3.11.4",
-  "whenever>=0.9.5",
+  "whenever>=0.10,<11",
 ]
 optional-dependencies.solarman = [ "pysolarmanv5==3.0.6" ]
 optional-dependencies.umodbus = [ "async-modbus==0.2.3", "connio==0.2", "umodbus==1.0.4" ]

src/sunsynk/helpers.py

@@ -4,6 +4,7 @@
 import math
 import struct
 from collections.abc import Iterable
+from dataclasses import InitVar, dataclass
 
 _LOG = logging.getLogger(__name__)
 
@@ -75,7 +76,7 @@ def ensure_tuple[T](val: T | tuple[T, ...]) -> tuple[T, ...]:
         return ()
     if isinstance(val, Iterable) and not isinstance(val, str):
         return tuple(val)
-    return (val,)
+    return (val,)  # type: ignore[return-value]
 
 
 def int_round(val: NumType) -> NumType:
@@ -123,28 +124,20 @@ def patch_bitmask(value: int, patch: int, bitmask: int) -> int:
     return (patch & bitmask) + (value & (0xFFFF - bitmask))
 
 
+@dataclass
 class SSTime:
     """Deals with inverter time format conversion complexities."""
 
     minutes: int = 0
+    strv: InitVar[str] = ""
+    regv: InitVar[int] = 0
 
-    def __init__(
-        self,
-        *,
-        minutes: int | None = None,
-        register: int | None = None,
-        string: str | None = None,
-    ) -> None:
-        """Init the time. All mutually exclusive."""
-        if minutes is not None:
-            assert register is None
-            assert string is None
-            self.minutes = minutes
-        elif register is not None:
-            assert string is None
-            self.reg_value = register
-        elif string is not None:
-            self.str_value = string
+    def __post_init__(self, strv: str, regv: int) -> None:
+        """Initialize from string or register value."""
+        if strv:
+            self.str_value = strv
+        elif regv:
+            self.reg_value = regv
 
     @property
     def reg_value(self) -> int:
@@ -162,6 +155,7 @@ def reg_value(self, reg_value: int) -> None:
     def str_value(self) -> str:
         """Get the value in hh:mm format."""
         hours, minutes = divmod(self.minutes, 60)
+        hours = hours % 24
         return f"{hours}:{minutes:02}"
 
     @str_value.setter

src/sunsynk/rwsensors.py

@@ -4,7 +4,6 @@
 
 import logging
 import re
-from collections.abc import Generator
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING
 
@@ -230,19 +229,17 @@ class TimeRWSensor(RWSensor):
 
     def available_values(self, step_minutes: int, state: InverterState) -> list[str]:
         """Get the available values for this sensor."""
-        full_day = 24 * 60
-
-        min_val = (
-            SSTime(string=str(state.get(self.min, "0:00"))).minutes if self.min else 0
-        )
-        max_val = (
-            SSTime(string=str(state.get(self.max, "24:00"))).minutes
-            if self.max
-            else full_day
-        )
-        val = SSTime(string=str(state.get(self, "0:00"))).minutes
-        time_range = self._range(min_val, max_val, val, step_minutes, full_day)
-        return list(map(lambda m: SSTime(minutes=m).str_value, time_range))
+        day = list(range(0, 24 * 60, step_minutes))
+        minv = SSTime(strv=str(state.get(self.min, "0:00"))).minutes if self.min else 0
+        maxv = SSTime(strv=str(state.get(self.max, "0:00"))).minutes if self.max else 0
+        if minv >= maxv:
+            maxv += 24 * 60
+        opt = [minv, *[v for v in day if (v > minv and v < maxv)], maxv]
+        val = SSTime(strv=str(state.get(self, "0:00"))).minutes
+        if val not in opt:
+            opt.append(val)
+        opt.sort()
+        return [SSTime(minutes=m).str_value for m in opt]
 
     @property
     def dependencies(self) -> list[Sensor]:
@@ -251,22 +248,10 @@ def dependencies(self) -> list[Sensor]:
 
     def reg_to_value(self, regs: RegType) -> ValType:
         """Decode the time from a register."""
-        return SSTime(register=regs[0]).str_value
+        return SSTime(regv=regs[0]).str_value
 
     def value_to_reg(self, value: ValType, state: InverterState) -> RegType:
         """Get the reg value from a display value."""
         if not self.address:
             raise NotImplementedError("Cannot write to a sensor with no address")
-        return self.reg(SSTime(string=str(value)).reg_value)
-
-    @staticmethod
-    def _range(
-        start: int, end: int, val: int, step: int, modulo: int
-    ) -> Generator[int]:
-        if val % step != 0:
-            yield val
-        stop = end if start <= end else end + modulo
-        for i in range(start, stop, step):
-            yield i % modulo
-        if start == end or start != end % modulo:
-            yield end
+        return self.reg(SSTime(strv=str(value)).reg_value)

src/tests/sunsynk/test_rwsensors.py

@@ -223,26 +223,30 @@ def test_time_rw(state: InverterState) -> None:
 
     state.update(
         {
-            50: 2330,
-            60: 2330,
-            70: 30,
+            50: 2300,  # min
+            60: 2330,  # value
+            70: 2400,  # max
         }
     )
     assert s.available_values(15, state) == [
+        "23:00",
+        "23:15",
         "23:30",
         "23:45",
         "0:00",
-        "0:15",
-        "0:30",
+        # "0:00",
+        # "0:15",
+        # "0:30",
     ]
 
     state.update(
         {
-            50: 200,
-            70: 200,
+            50: 203,
+            60: 210,
+            70: 222,
         }
     )
-    assert s.available_values(15, state) == ["2:00"]
+    assert s.available_values(15, state) == ["2:03", "2:10", "2:15", "2:22"]
 
 
 # def test_update_sensor(caplog) -> None: