feat(em511): Add the driver with some functions and unit test

Dexter9532 · Dexter9532 · commit 077df2307300 · 2025-10-27T15:22:08.000+01:00
Signed-off-by: Bobo Bäck Engström &lt;bobo@id8-engineering.io&gt;
diff --git 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
@@ -0,0 +1,212 @@
+# ruff: noqa: N802
+"""Driver class for EM511."""
+
+import time
+from decimal import Decimal, getcontext
+
+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_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
+
+    TIMEOUT = 0.5
+
+    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
+        getcontext().prec = 3
+
+    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:
+            ModbusException: 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)
+
+    def _timeout_check(self, start_timer: float, address: int) -> None:
+        """Init timeout counter.
+
+        Timeout function to check timeout error.
+
+        Args:
+            start_timer: Counter init.
+            address: address to the register.
+
+        Raises:
+            TimeoutError: If elapsed time is to long.
+        """
+        elapsed = time.perf_counter() - start_timer
+        if elapsed >= self.TIMEOUT:
+            msg = f"Timeout error, response failed, device_address={self.device_address} address={address}"
+            raise TimeoutError(msg)
+
+    @property
+    def V(self) -> Decimal:
+        """Voltage (V).
+
+        Returns:
+            Decimal: Current voltage value.
+
+        Raises:
+            TimeoutError: If elapsed time is to long.
+            ValueError: If input is at max value or above.
+            ModbusException: If failed to read input register.
+        """
+        start_timer = time.perf_counter()
+
+        regs = self._read_input_registers(self.EM511_REGISTER_V, self.INT32_REG_COUNT)
+        self._timeout_check(start_timer, self.EM511_REGISTER_V)
+        return Decimal(self._unpack(regs, self.EM511_REGISTER_V)) / self.SCALE_10
+
+    @property
+    def A(self) -> Decimal:
+        """Current (A).
+
+        Returns:
+            Decimal: Current ampere value.
+
+        Raises:
+            TimeoutError: If elapsed time is to long.
+            ValueError: If input is at max value or above.
+            ModbusException: If failed to read input register.
+        """
+        start_timer = time.perf_counter()
+        regs = self._read_input_registers(self.EM511_REGISTER_A, self.INT32_REG_COUNT)
+        self._timeout_check(start_timer, self.EM511_REGISTER_A)
+        return Decimal(self._unpack(regs, self.EM511_REGISTER_A)) / self.SCALE_1000
+
+    @property
+    def password(self) -> int:
+        """Password.
+
+        Returns:
+            int: Current password value.
+
+        Raises:
+            TimeoutError: If elapsed time is to long.
+            ValueError: If input is at max value or above.
+            ModbusException: If failed to read input register.
+        """
+        start_timer = time.perf_counter()
+        regs = self._read_input_registers(self.EM511_REGISTER_PASSWORD, self.INT16_REG_COUNT)
+        self._timeout_check(start_timer, self.EM511_REGISTER_PASSWORD)
+        return int(self._unpack(regs, self.EM511_REGISTER_PASSWORD))
+
+    @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.
+        """
+        if value not in range(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
@@ -0,0 +1,165 @@
+# ruff: noqa: S101,PLR2004, N802
+
+"""Test file for driver."""
+
+from contextlib import nullcontext
+from unittest.mock import MagicMock, patch
+
+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 4: Should raise exception due to taking more then 500 ms to respond."""
+    mock_result.registers = [0x1860, 0x0023]
+    meter.TIMEOUT = 0.01
+    client.read_input_registers.return_value = mock_result
+    with (
+        patch("em511.em511.time.perf_counter", side_effect=[0.0, 0.02]),
+        pytest.raises(TimeoutError, match="Timeout error, response failed"),
+    ):
+        _ = meter.V
+
+    """Test 5: Should NOT raise exception when within timeout limit."""
+    mock_result.registers = [0x1860, 0x0023]
+    meter.TIMEOUT = 0.02
+    client.read_input_registers.return_value = mock_result
+    with patch("em511.em511.time.perf_counter", side_effect=[0.0, 0.01]), nullcontext():
+        _ = 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 4: Should raise exception due to taking more then 500 ms to respond."""
+    mock_result.registers = [0x1860, 0x0023]
+    meter.TIMEOUT = 0.01
+    client.read_input_registers.return_value = mock_result
+    with (
+        patch("em511.em511.time.perf_counter", side_effect=[0.0, 0.02]),
+        pytest.raises(TimeoutError, match="Timeout error, response failed"),
+    ):
+        _ = meter.A
+
+    """Test 5: Should NOT raise exception when within timeout limit."""
+    mock_result.registers = [0x1860, 0x0023]
+    meter.TIMEOUT = 0.02
+    client.read_input_registers.return_value = mock_result
+
+    with patch("em511.em511.time.perf_counter", side_effect=[0.0, 0.01]), nullcontext():
+        _ = 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 due to taking more then 500 ms to respond."""
+    mock_result.registers = [0x1860, 0x0023]
+    meter.TIMEOUT = 0.01
+    client.read_input_registers.return_value = mock_result
+    with (
+        patch("em511.em511.time.perf_counter", side_effect=[0.0, 0.02]),
+        pytest.raises(TimeoutError, match="Timeout error, response failed"),
+    ):
+        _ = meter.password
+
+    """Test 3: Should NOT raise exception when within timeout limit."""
+    mock_result.registers = [0x1860, 0x0023]
+    meter.TIMEOUT = 0.02
+    client.read_input_registers.return_value = mock_result
+
+    with patch("em511.em511.time.perf_counter", side_effect=[0.0, 0.01]), nullcontext():
+        _ = meter.password
+
+    """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.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)
+
+    """Test 2: Try set password out of range."""
+    with pytest.raises(ValueError, match="Invalid password value:"):
+        meter.password = 12345
diff --git a/src/em511/test_hello.py b/src/em511/test_hello.py
