finished with core refactoring for Inverter, Battery next

givenergy_modbus/model/__init__.py

@@ -6,9 +6,7 @@
 from enum import IntEnum
 from typing import TYPE_CHECKING
 
-from pydantic import BaseModel, create_model
-
-from givenergy_modbus.model.inverter import InverterConfig, InverterRegisterGetter
+from pydantic import BaseModel
 
 if TYPE_CHECKING:
     from givenergy_modbus.model.register_cache import RegisterCache
@@ -69,6 +67,3 @@ def from_repr(cls, start: int | str, end: int | str):
 # Inverter = inverter.Inverter
 # Battery = battery.Battery
 # RegisterCache = register_cache.RegisterCache
-Inverter = create_model(
-    'Inverter', __config__=InverterConfig, **InverterRegisterGetter.to_fields(), **{'model': 'Foo'}
-)  # type: ignore[call-overload]

givenergy_modbus/model/inverter.py

@@ -1,24 +1,28 @@
-from enum import IntEnum
+from enum import IntEnum, Enum
 
-from pydantic import BaseConfig
+from pydantic import BaseConfig, create_model
 
-from givenergy_modbus.model.device import DataType as DT
-from givenergy_modbus.model.device import DeviceRegisterGetter
-from givenergy_modbus.model.device import RegisterDefinition as Def
-from givenergy_modbus.model.register import HR, IR, Register, RegisterEncoder
-from givenergy_modbus.model.register_cache import RegisterCache, RegisterCacheEncoder
+from givenergy_modbus.model.register import HR, IR
+from givenergy_modbus.model.register import DataType as DT
+from givenergy_modbus.model.register import RegisterDefinition as Def
+from givenergy_modbus.model.register import RegisterGetter
 
 
-class Model(IntEnum):
+class Model(Enum):
     """Known models of inverters."""
 
-    HYBRID = 2
-    AC = 3
-    HYBRID_3PH = 4
-    AC_3PH = 6
-    EMS = 5
-    GATEWAY = 7
-    ALL_IN_ONE = 8
+    HYBRID = '2'
+    AC = '3'
+    HYBRID_3PH = '4'
+    AC_3PH = '6'
+    EMS = '5'
+    GATEWAY = '7'
+    ALL_IN_ONE = '8'
+
+    @classmethod
+    def _missing_(cls, key: str):
+        """Pick model from the first digit of the device type code."""
+        return cls(key[0])
 
 
 class UsbDevice(IntEnum):
@@ -85,12 +89,13 @@ class InverterStatus(IntEnum):
     FLASHING_FIRMWARE = 4
 
 
-class InverterRegisterGetter(DeviceRegisterGetter):
+class InverterRegisterGetter(RegisterGetter):
     """Structured format for all inverter attributes."""
 
     REGISTER_LUT = {
         # Holding Registers, block 0-59
         'device_type_code': Def(DT.hex, None, HR(0)),
+        'model': Def(DT.hex, Model, HR(0)),
         'module': Def(DT.uint32, (DT.hex, 8), HR(1), HR(2)),
         'num_mppt': Def((DT.duint8, 0), None, HR(3)),
         'num_phases': Def((DT.duint8, 1), None, HR(3)),
@@ -103,7 +108,7 @@ class InverterRegisterGetter(DeviceRegisterGetter):
         'enable_charge_target': Def(DT.bool, None, HR(20)),
         'arm_firmware_version': Def(DT.uint16, None, HR(21)),
         'usb_device_inserted': Def(DT.uint16, UsbDevice, HR(22)),
-        'select_arm_chip': Def(DT.bool, UsbDevice, HR(23)),
+        'select_arm_chip': Def(DT.bool, None, HR(23)),
         # variable_address=rc[HR(24)],
         # variable_value=rc[HR(25)],
         'grid_port_max_power_output': Def(DT.uint16, None, HR(26)),
@@ -214,10 +219,6 @@ class InverterConfig(BaseConfig):
 
     orm_mode = True
     getter_dict = InverterRegisterGetter
-    json_encoders = {
-        RegisterCache: RegisterCacheEncoder.encode,
-        Register: RegisterEncoder.encode,
-    }
 
 
 # class Inverter(GivEnergyBaseModel):
@@ -568,3 +569,8 @@ class InverterConfig(BaseConfig):
 #     # def compute_e_pv_day(e_pv1_day: float, e_pv2_day: float, **kwargs) -> float:
 #     #     """Computes the discharge slot 2."""
 #     #     return e_pv1_day + e_pv2_day
+
+Inverter = create_model(
+    'Inverter', __config__=InverterConfig, **InverterRegisterGetter.to_fields()
+)  # type: ignore[call-overload]
+# , **{'model': 'Foo'}

givenergy_modbus/model/plant.py

@@ -1,8 +1,9 @@
 import logging
 from typing import Any
 
-from givenergy_modbus.model import GivEnergyBaseModel, Inverter
+from givenergy_modbus.model import GivEnergyBaseModel
 from givenergy_modbus.model.battery import Battery
+from givenergy_modbus.model.inverter import Inverter
 from givenergy_modbus.model.register import HR, IR
 from givenergy_modbus.model.register_cache import RegisterCache, RegisterCacheEncoder
 from givenergy_modbus.pdu import (

givenergy_modbus/model/register.py

@@ -1,5 +1,135 @@
+from dataclasses import dataclass
 from json import JSONEncoder
-from typing import Any
+from typing import Any, Callable, Union
+
+from pydantic.utils import GetterDict
+
+from givenergy_modbus.model import TimeSlot
+
+
+class DataType:
+    """Type of data register represents. Encoding is always big-endian."""
+
+    @staticmethod
+    def uint16(val: int) -> int:
+        """Simply return the raw unsigned 16-bit integer register value."""
+        if val is not None:
+            return int(val)
+
+    @staticmethod
+    def duint8(val: int, *idx: int) -> int:
+        """Split one register into two unsigned 8-bit ints and return the specified index."""
+        if val is not None:
+            vals = (val >> 8), (val & 0xFF)
+            return vals[idx[0]]
+
+    @staticmethod
+    def uint32(high_val: int, low_val: int) -> int:
+        """Combine two registers into an unsigned 32-bit int."""
+        if high_val is not None and low_val is not None:
+            return (high_val << 16) + low_val
+
+    @staticmethod
+    def timeslot(start_time: int, end_time: int) -> TimeSlot:
+        """Interpret register as a time slot."""
+        if start_time is not None and end_time is not None:
+            return TimeSlot.from_repr(start_time, end_time)
+
+    @staticmethod
+    def bool(val: int) -> bool:
+        """Interpret register as a bool."""
+        if val is not None:
+            return bool(val)
+
+    @staticmethod
+    def string(*vals: int) -> str:
+        """Represent one or more registers as a concatenated string."""
+        if vals is not None and None not in vals:
+            return b''.join((v or 0).to_bytes(2, byteorder='big') for v in vals).decode(encoding='latin1').upper()
+        return ''
+
+    @staticmethod
+    def hex(val: int, width: int = 4) -> str:
+        """Represent a register value as a 4-character hex string."""
+        if val is not None:
+            return f'{val:0{width}x}'
+
+
+@dataclass(init=False)
+class RegisterDefinition:
+    """Specifies how to convert raw register values into their actual representation."""
+
+    pre_conv: Union[Callable, tuple, None]
+    post_conv: Union[Callable, tuple[Callable, Any], None]
+    registers: tuple['Register']
+
+    def __init__(self, *args, **kwargs):
+        self.pre_conv = args[0]
+        self.post_conv = args[1]
+        self.registers = args[2:]  # type: ignore[assignment]
+
+    def __hash__(self):
+        return hash(self.registers)
+
+
+class RegisterGetter(GetterDict):
+    """Specifies how device attributes are derived from raw register values."""
+
+    REGISTER_LUT: dict[str, RegisterDefinition]
+
+    def get(self, key: str, default: Any = None) -> Any:
+        """Return a named register's value, after pre- and post-conversion."""
+        try:
+            r = self.REGISTER_LUT[key]
+        except KeyError:
+            return default
+
+        regs = [self._obj.get(r) for r in r.registers]
+
+        if None in regs:
+            return None
+
+        if r.pre_conv:
+            if isinstance(r.pre_conv, tuple):
+                args = regs + list(r.pre_conv[1:])
+                val = r.pre_conv[0](*args)
+            else:
+                val = r.pre_conv(*regs)
+        else:
+            val = regs
+
+        if r.post_conv:
+            if isinstance(r.post_conv, tuple):
+                return r.post_conv[0](val, *r.post_conv[1:])
+            else:
+                return r.post_conv(val)
+        return val
+
+    @classmethod
+    def to_fields(cls) -> dict[str, tuple[Any, None]]:
+        """Determine a pydantic fields definition for the class."""
+
+        def infer_return_type(obj: Any):
+            if hasattr(obj, '__annotations__') and (ret := obj.__annotations__.get('return', None)):
+                return ret
+            return obj  # assume it is a class/type already?
+
+        def return_type(v: RegisterDefinition):
+            if v.post_conv:
+                if isinstance(v.post_conv, tuple):
+                    return infer_return_type(v.post_conv[0])
+                else:
+                    return infer_return_type(v.post_conv)
+            elif v.pre_conv:
+                if isinstance(v.pre_conv, tuple):
+                    return infer_return_type(v.pre_conv[0])
+                else:
+                    return infer_return_type(v.pre_conv)
+            return Any
+
+        register_fields = {k: (return_type(v), None) for k, v in cls.REGISTER_LUT.items()}
+
+        return register_fields
 
 
 class RegisterEncoder(JSONEncoder):

givenergy_modbus/model/register_cache.py

@@ -92,4 +92,6 @@ def to_datetime(self, y: Register, m: Register, d: Register, h: Register, min: R
 
     def to_timeslot(self, start: Register, end: Register) -> 'TimeSlot':
         """Combine two registers into a time slot."""
+        from givenergy_modbus.model import TimeSlot
+
         return TimeSlot.from_repr(self[start], self[end])