feat(em511): Add the driver with some functions

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"]

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)

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)