diff --git a/src/em511/em511.py b/src/em511/em511.py index 4da7c7e..22e4901 100644 --- a/src/em511/em511.py +++ b/src/em511/em511.py @@ -1,108 +1,232 @@ -# ruff: noqa: N802 -"""Driver class for EM511.""" +"""Driver class for Carlo Gavazzi EM511 Modbus energy meters. +This module provides a generic, configurable driver for the Carlo Gavazzi EM511 +series of energy meters, using pymodbus for serial communication. + +The design uses dataclasses to define register specifications and a class +decorator to automatically generate @property accessors for all defined +registers. This minimizes boilerplate and ensures consistency across multiple +registers. +""" + +from dataclasses import dataclass from decimal import Decimal +from typing import Final, TypeVar from pymodbus.client import ModbusSerialClient from pymodbus.exceptions import ModbusException +T = TypeVar("T", bound=type) -class Em511: - """Driver for Carlo Gavazzi EM511 series energy meters. - This class provides read and write access to Modbus registers - via a connected `pymodbus.client.ModbusSerialClient` instance. +@dataclass(frozen=True) +class RegisterSpec: + """Specification for a Modbus register mapping. + + Defines the register address, scaling, precision, and read/write behavior. Attributes: - device_address (int): Modbus address of the target device. - client (ModbusSerialClient): Connected Modbus client. + address (int): Modbus register address. + count (int): Number of consecutive registers to read. + scale (int): Scaling factor to apply to the raw integer value. + decimals (int): Number of decimal places to round the scaled value to. + writable (bool): Whether this register can be written to. + range (bool): Whether range validation should be performed. + min (int): Minimum allowed value for range validation. + max (int): Maximum allowed value for range validation. + """ + + address: int + count: int + decimals: int = 0 + scale: int = 1 + range: bool = False + min: int = 0 + max: int = 0x7FFFFFFF + writable: bool = False + return_type: type[int] | type[Decimal] = Decimal + + +def register_properties(cls: T) -> T: + """Class decorator that auto-generates @property accessors for Modbus registers. + + For each entry in `cls._register_specs`, this decorator dynamically creates + a corresponding @property getter, and optionally a setter if `writable=True`. + + The generated getter automatically calls `_read_register(register_name)` + and the setter calls `_write_register(address, value)` with range validation + if enabled in the `RegisterSpec`. + + Args: + cls: The target class to which properties will be added. + + Returns: + The same class with dynamically added properties. + """ + for name, spec in cls._register_specs.items(): + + def getter(self: "Em511", _name: str = name, _spec: "RegisterSpec" = spec) -> Decimal | int: + """Auto-generated register reader. + + Returns the current value of the register. If range validation is + enabled, ensures that the returned value is within the expected range. + + Raises: + ValueError: If the register value is outside its defined range. + """ + _value = self._read_register(_name) + if _spec.range and not (_spec.min <= _value <= _spec.max): + msg = f"Invalid value for '{_name}': {_value}. Must be between {_spec.min} and {_spec.max}." + raise ValueError(msg) + return _value + + def setter(self: "Em511", value: int, _name: str = name, _spec: "RegisterSpec" = spec) -> None: + """Auto-generated register writer. + + Writes a new value to the register and performs range validation + if defined in the corresponding `RegisterSpec`. + + Raises: + AttributeError: If the register is read-only. + ValueError: If the written value is outside its defined range. + """ + if not _spec.writable: + msg = f"Register '{_name}' is read-only." + raise AttributeError(msg) + + if _spec.range and not (_spec.min <= value <= _spec.max): + msg = f"Invalid value for '{_name}': {value}. Must be between {_spec.min} and {_spec.max}." + raise ValueError(msg) + self._write_register(_spec.address, int(value)) + + prop = property(getter, setter) if spec.writable else property(getter) + + prop.__doc__ = f"{name} ({'read/write' if spec.writable else 'read-only'})" + ( + f" range=[{spec.min}, {spec.max}]" if spec.range else "" + ) + + setattr(cls, name, prop) + + return cls + + +@register_properties +class Em511: + """Driver for Carlo Gavazzi EM511 series energy meters. + + Provides read and write access to Modbus registers via an existing + `pymodbus.client.ModbusSerialClient` instance. Register definitions are + dynamically mapped to @property accessors based on `_register_specs`. """ INT16_REG_COUNT = 1 INT32_REG_COUNT = 2 - PASSWORD_MIN_VALUE = 0 - PASSWORD_MAX_VALUE = 9999 INPUT_MAX_VALUE_32 = 0x7FFFFFFF INPUT_MAX_VALUE_16 = 0x7FFF - EM511_REGISTER_V = 0x0000 - EM511_REGISTER_A = 0x0002 - EM511_REGISTER_PASSWORD = 0x1000 - - SCALE_10 = 10 - SCALE_100 = 100 - SCALE_1000 = 1000 + _register_specs: Final[dict[str, RegisterSpec]] = { + "V": RegisterSpec(address=0x0000, count=2, decimals=1, scale=10), + "A": RegisterSpec(address=0x0002, count=2, decimals=3, scale=1000), + "W": RegisterSpec(address=0x0004, count=2, decimals=1, scale=10), + "password": RegisterSpec( + address=0x1000, + count=1, + range=True, + min=0, + max=9999, + return_type=int, + writable=True, + ), + "alarm_status": RegisterSpec(address=0x0307, count=1, return_type=int), + } def __init__(self, device_address: int, client: ModbusSerialClient) -> None: - """Initialize an Em511 driver instance with an existing Modbus client. + """Initialize an Em511 driver instance. Args: device_address: Modbus address for the EM511 meter. - client: An initialized ModbusSerialClient instance to use for communication. + client: A connected `ModbusSerialClient` instance. """ self.device_address = device_address self.client = client def _read_input_registers(self, address: int, count: int) -> list[int]: - """Read input registers. - - Internal helper to read Modbus registers safely. + """Safely read input registers from the Modbus device. Args: - address: Register address to read from. - count: Number of register to read from. + address: Starting register address to read. + count: Number of registers to read. Returns: - list of registers. + A list of integer register values. Raises: - ModbusException: If read operation fails. + ModbusException: If the read operation fails or returns an error. """ result = self.client.read_input_registers(address=address, count=count, device_id=self.device_address) if result.isError(): msg = ( "Failed to read input register. " - f"device_address={self.device_address} address={address} count={count} result={result} " + f"device_address={self.device_address} address={address} count={count} result={result}" ) raise ModbusException(msg) return result.registers - def _write_register(self, address: int, value: int) -> None: - """Write to register. + def _read_register(self, register_name: str) -> Decimal | int: + """Read and scale the specified register. - Internal helper to write to single register. + Args: + register_name: Name of the register as defined in `_register_specs`. + + Returns: + A Decimal value representing the scaled register reading. + + Raises: + ValueError: If register unpacking fails or returns overflow values. + ModbusException: If Modbus read operation fails. + """ + spec = self._register_specs[register_name] + regs = self._read_input_registers(spec.address, spec.count) + if spec.return_type is Decimal: + value = Decimal(self._unpack(regs, spec.address)) / spec.scale + return round(value, spec.decimals) + return self._unpack(regs, spec.address) + + def _write_register(self, address: int, value: int) -> None: + """Write a single Modbus register. Args: - address: Register to write to. - value: Value to write the given register with. + address: Register address to write. + value: Integer value to write to the register. Raises: - ModbusException: If write operation fails. + ModbusException: If the write operation fails. """ result = self.client.write_register(address=address, value=value, device_id=self.device_address) - if result.isError(): msg = ( - "Failed to write to single register." - f"device_address={self.device_address} address={address} count={value}" + "Failed to write to single register. " + f"device_address={self.device_address} address={address} value={value}" ) raise ModbusException(msg) def _unpack(self, regs: list[int], address: int) -> int: - """Unpack registers. + """Unpack raw Modbus register data into an integer value. - Internal helper to unpack register list. + Supports both 16-bit and 32-bit register combinations and performs + overflow detection for "EEE" values reported by the meter. Args: - regs: List of registers to unpack. - address: address to the register. + regs: The list of register values to unpack. + address: The base register address (used for error reporting). Returns: - Unpacked integer value. + The unpacked integer representation of the registers. Raises: - ValueError: If unexpected number of registers is given. + ValueError: If an invalid number of registers is provided or an + overflow marker is detected. """ if len(regs) == self.INT16_REG_COUNT: value = regs[0] @@ -118,73 +242,5 @@ def _unpack(self, regs: list[int], address: int) -> int: raise ValueError(msg) return value - msg = f"Unexpected register count: {len(regs)}." + msg = f"Unexpected register count: {len(regs)} for address={address}" raise ValueError(msg) - - @property - def V(self) -> Decimal: - """Voltage (V). - - Returns: - Decimal: Current voltage value. - - Raises: - ValueError: If input is at max value or above. - ModbusException: If failed to read input register. - """ - regs = self._read_input_registers(self.EM511_REGISTER_V, self.INT32_REG_COUNT) - value = Decimal(self._unpack(regs, self.EM511_REGISTER_V)) / self.SCALE_10 - return round(value, 1) - - @property - def A(self) -> Decimal: - """Current (A). - - Returns: - Decimal: Current ampere value. - - Raises: - ValueError: If input is at max value or above. - ModbusException: If failed to read input register. - """ - regs = self._read_input_registers(self.EM511_REGISTER_A, self.INT32_REG_COUNT) - value = Decimal(self._unpack(regs, self.EM511_REGISTER_A)) / self.SCALE_1000 - return round(value, 3) - - @property - def password(self) -> int: - """Password. - - Returns: - int: Current password value. - - Raises: - ValueError: If input is at max value or above. - ValueError: If password is out of range. - ModbusException: If failed to read input register. - """ - regs = self._read_input_registers(self.EM511_REGISTER_PASSWORD, self.INT16_REG_COUNT) - value = self._unpack(regs, self.EM511_REGISTER_PASSWORD) - if not (self.PASSWORD_MIN_VALUE <= value <= self.PASSWORD_MAX_VALUE): - msg = f"Invalid password value: {value}. Must be between 0 and 9999." - raise ValueError(msg) - return value - - @password.setter - def password(self, value: int) -> None: - """Password. - - Min value: 0 (no password). - Max value: 9999. - - Args: - value (int): Set Password. - - Raises: - ModbusException: If failed to write to single register. - ValueError: If password value is out of range. - """ - if not (self.PASSWORD_MIN_VALUE <= value <= self.PASSWORD_MAX_VALUE): - msg = f"Invalid password value: {value}. Must be between 0 and 9999." - raise ValueError(msg) - self._write_register(self.EM511_REGISTER_PASSWORD, value) diff --git a/src/em511/em511.pyi b/src/em511/em511.pyi new file mode 100644 index 0000000..bc83fae --- /dev/null +++ b/src/em511/em511.pyi @@ -0,0 +1,14 @@ +from decimal import Decimal + +from pymodbus.client import ModbusSerialClient + +class Em511: + def __init__(self, device_address: int, client: ModbusSerialClient) -> None: ... + V: Decimal + A: Decimal + password: int + + def _unpack(self, registers: list[int], address: int) -> int: ... + def _write_register(self, address: int, value: int) -> None: ... + def _read_register(self, register_name: str) -> Decimal | int: ... + def _read_input_registers(self, address: int, count: int) -> list[int]: ... diff --git a/src/em511/test_em511.py b/src/em511/test_em511.py index 8c8bac5..0b9daa7 100644 --- a/src/em511/test_em511.py +++ b/src/em511/test_em511.py @@ -9,6 +9,27 @@ from em511 import Em511 +def test_unpack() -> None: + """Test unpack.""" + client = MagicMock() + meter = Em511(1, client) + + """Test 1: Should raise exception due to more registers in use than allowed.""" + registers = [0x1860, 0x0023, 0x4244] + with pytest.raises(ValueError, match="Unexpected register count:"): + _ = meter._unpack(registers, 0x0001) # noqa: SLF001 + + """Test 2: Should raise exception due to 16-bit register overflow""" + registers = [0x7FFF] + with pytest.raises(ValueError, match="Input overflow EEE for 16-bit register: "): + _ = meter._unpack(registers, 0x0001) # noqa: SLF001 + + """Test 3: Should raise exception due to 32-bit register overflow""" + registers = [0xFFFF, 0x7FFF] + with pytest.raises(ValueError, match="Input overflow EEE for 32-bit register: "): + _ = meter._unpack(registers, 0x0001) # noqa: SLF001 + + def test_V() -> None: """Test Get v.""" client = MagicMock() @@ -28,18 +49,6 @@ def test_V() -> None: value = meter.V assert value == 10500 - """Test 3: Should raise exception due to more registers in use than allowed.""" - mock_result.registers = [0x1860, 0x0023, 0x4244] - client.read_input_registers.return_value = mock_result - with pytest.raises(ValueError, match="Unexpected register count:"): - _ = meter.V - - """Test 6: Should raise exception if input value exceeds maximum value, display shows 'EEE', 32-bit register.""" - mock_result.registers = [0xFFFF, 0x7FFF] - client.read_input_registers.return_value = mock_result - with pytest.raises(ValueError, match="Input overflow EEE for 32-bit register: "): - _ = meter.V - def test_get_A() -> None: """Test Get a.""" @@ -60,18 +69,6 @@ def test_get_A() -> None: value = meter.A assert value == 2300 - """Test 3: Should raise exception due to more registers in use than allowed.""" - mock_result.registers = [0x1860, 0x0023, 0x4244] - client.read_input_registers.return_value = mock_result - with pytest.raises(ValueError, match="Unexpected register count:"): - _ = meter.A - - """Test 6: Should raise exception if input value exceeds maximum value, display shows 'EEE', 32-bit register.""" - mock_result.registers = [0xFFFF, 0x7FFF] - client.read_input_registers.return_value = mock_result - with pytest.raises(ValueError, match="Input overflow EEE for 32-bit register: "): - _ = meter.A - def test_get_password() -> None: """Test Get password.""" @@ -86,16 +83,10 @@ def test_get_password() -> None: value = meter.password assert value == 1234 - """Test 2: Should raise exception if input value exceeds maximum value, display shows 'EEE', 32-bit register.""" - mock_result.registers = [0xFFFF, 0x7FFF] - client.read_input_registers.return_value = mock_result - with pytest.raises(ValueError, match="Input overflow EEE for 32-bit register: "): - _ = meter.password - - """Test 3: Should raise exception if password return a value out of its range of 0-9999.""" + """Test 2: Should raise exception if password return a value out of its range of 0-9999.""" mock_result.registers = [0x186A0, 0x0000] client.read_input_registers.return_value = mock_result - with pytest.raises(ValueError, match="Invalid password value: "): + with pytest.raises(ValueError, match="Invalid value for"): _ = meter.password @@ -115,7 +106,7 @@ def test_set_password() -> None: client.write_register.reset_mock() """Test 2: Try set password out of range.""" - with pytest.raises(ValueError, match="Invalid password value:"): + with pytest.raises(ValueError, match="Invalid value for"): meter.password = 12345 """Test 3: Try set password at maximum value."""