From 235188a66cbdc20c6d09416f76bf67530737a254 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bobo=20B=C3=A4ck=20Engstr=C3=B6m?= Date: Tue, 21 Oct 2025 15:29:11 +0200 Subject: [PATCH] feat(em511): Add base driver MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Bobo Bäck Engström --- src/em511/__init__.py | 11 ++- src/em511/em511.py | 190 ++++++++++++++++++++++++++++++++++++++++ src/em511/test_em511.py | 133 ++++++++++++++++++++++++++++ src/em511/test_hello.py | 7 -- 4 files changed, 330 insertions(+), 11 deletions(-) create mode 100644 src/em511/em511.py create mode 100644 src/em511/test_em511.py delete mode 100644 src/em511/test_hello.py diff --git a/src/em511/__init__.py b/src/em511/__init__.py index 2d8f2cf..39f1c07 100644 --- a/src/em511/__init__.py +++ b/src/em511/__init__.py @@ -1,6 +1,9 @@ -"""Hello package.""" +"""Top-level package for the EM511 driver. +Provides the `Em511` class for reading and writing Modbus registers +using Carlo Gavazzi EM511 energy meters. +""" -def hello() -> str: - """Hello docstring.""" - return "Hello from python-em511!" +from .em511 import Em511 + +__all__ = ["Em511"] diff --git a/src/em511/em511.py b/src/em511/em511.py new file mode 100644 index 0000000..4da7c7e --- /dev/null +++ b/src/em511/em511.py @@ -0,0 +1,190 @@ +# ruff: noqa: N802 +"""Driver class for EM511.""" + +from decimal import Decimal + +from pymodbus.client import ModbusSerialClient +from pymodbus.exceptions import ModbusException + + +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. + + Attributes: + device_address (int): Modbus address of the target device. + client (ModbusSerialClient): Connected Modbus client. + """ + + 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 + + def __init__(self, device_address: int, client: ModbusSerialClient) -> None: + """Initialize an Em511 driver instance with an existing Modbus client. + + Args: + device_address: Modbus address for the EM511 meter. + client: An initialized ModbusSerialClient instance to use for communication. + """ + 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. + + Args: + address: Register address to read from. + count: Number of register to read from. + + Returns: + list of registers. + + Raises: + ModbusException: If read operation fails. + """ + 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} " + ) + raise ModbusException(msg) + return result.registers + + def _write_register(self, address: int, value: int) -> None: + """Write to register. + + Internal helper to write to single register. + + Args: + address: Register to write to. + value: Value to write the given register with. + + Raises: + ModbusException: If 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}" + ) + raise ModbusException(msg) + + def _unpack(self, regs: list[int], address: int) -> int: + """Unpack registers. + + Internal helper to unpack register list. + + Args: + regs: List of registers to unpack. + address: address to the register. + + Returns: + Unpacked integer value. + + Raises: + ValueError: If unexpected number of registers is given. + """ + if len(regs) == self.INT16_REG_COUNT: + value = regs[0] + if value == self.INPUT_MAX_VALUE_16: + msg = f"Input overflow EEE for 16-bit register: device_address={self.device_address} address={address}" + raise ValueError(msg) + return value + + if len(regs) == self.INT32_REG_COUNT: + value = (regs[1] << 16) + regs[0] + if value == self.INPUT_MAX_VALUE_32: + msg = f"Input overflow EEE for 32-bit register: device_address={self.device_address} address={address}" + raise ValueError(msg) + return value + + msg = f"Unexpected register count: {len(regs)}." + 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/test_em511.py b/src/em511/test_em511.py new file mode 100644 index 0000000..8c8bac5 --- /dev/null +++ b/src/em511/test_em511.py @@ -0,0 +1,133 @@ +# ruff: noqa: S101,PLR2004, N802 + +"""Test file for driver.""" + +from unittest.mock import MagicMock + +import pytest + +from em511 import Em511 + + +def test_V() -> None: + """Test Get v.""" + client = MagicMock() + mock_result = MagicMock() + mock_result.isError.return_value = False + meter = Em511(1, client) + + """Test 1: should pass""" + mock_result.registers = [0x08FC, 0x0000] + client.read_input_registers.return_value = mock_result + value = meter.V + assert value == 230 + + """Test 2: Should pass.""" + mock_result.registers = [0x9A28, 0x0001] + client.read_input_registers.return_value = mock_result + 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.""" + client = MagicMock() + mock_result = MagicMock() + mock_result.isError.return_value = False + meter = Em511(1, client) + + """Test 1: should pass""" + mock_result.registers = [0x2904, 0x0000] + client.read_input_registers.return_value = mock_result + value = meter.A + assert value == 10.5 + + """Test 2: Should pass.""" + mock_result.registers = [0x1860, 0x0023] + client.read_input_registers.return_value = mock_result + 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.""" + client = MagicMock() + mock_result = MagicMock() + mock_result.isError.return_value = False + meter = Em511(1, client) + + """Test 1: should pass""" + mock_result.registers = [1234] + client.read_input_registers.return_value = mock_result + 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.""" + mock_result.registers = [0x186A0, 0x0000] + client.read_input_registers.return_value = mock_result + with pytest.raises(ValueError, match="Invalid password value: "): + _ = meter.password + + +def test_set_password() -> None: + """Test Set Password.""" + client = MagicMock() + mock_result = MagicMock() + mock_result.isError.return_value = False + meter = Em511(1, client) + + """Test 1: Set password""" + mock_result.registers = [4096] + client.write_register.return_value = mock_result + meter.password = 1236 + client.write_register.assert_called_once_with(address=4096, value=1236, device_id=1) + + client.write_register.reset_mock() + + """Test 2: Try set password out of range.""" + with pytest.raises(ValueError, match="Invalid password value:"): + meter.password = 12345 + + """Test 3: Try set password at maximum value.""" + mock_result.registers = [4096] + client.write_register.return_value = mock_result + meter.password = 9999 + client.write_register.assert_called_once_with(address=4096, value=9999, device_id=1) + + client.write_register.reset_mock() + + """Test 4: Try set password at lowest value.""" + mock_result.registers = [4096] + client.write_register.return_value = mock_result + meter.password = 0 + client.write_register.assert_called_once_with(address=4096, value=0, device_id=1) diff --git a/src/em511/test_hello.py b/src/em511/test_hello.py deleted file mode 100644 index 4ade184..0000000 --- a/src/em511/test_hello.py +++ /dev/null @@ -1,7 +0,0 @@ -# noqa: D100 -import em511 - - -def test_hello() -> None: - """Test hello function.""" - assert em511.hello() == "Hello from python-em511!" # noqa: S101