feat(em511): Add the driver with some functions

Working on format

Signed-off-by: Bobo Bäck Engström <bobo@id8-engineering.io>

Author: Dexter9532

src/em511/__init__.py

@@ -1,6 +1,5 @@
 """Hello package."""
 
+from .em511 import Em511
 
-def hello() -> str:
-    """Hello docstring."""
-    return "Hello from python-em511!"
+__all__ = ["Em511"]

src/em511/em511.py

@@ -0,0 +1,282 @@
+"""Driver class for EM511."""
+
+from pymodbus.client import ModbusSerialClient
+from pymodbus.exceptions import ModbusException
+
+
+class Em511:
+    """Em511 driver class."""
+
+    INT16_REG_COUNT = 1
+    INT32_REG_COUNT = 2
+
+    def __init__(self, device_address: int, client: ModbusSerialClient) -> None:
+        """Constructor.
+
+        Constructor for the object
+
+        Args:
+            device_address: Modbus address for the meter in use.
+            client: Client for serial port.
+        """
+        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 adress to read from.
+            count: Number of register to read from
+
+        Returns:
+            value of registers.
+        """
+        result = self.client.read_input_registers(address=address, count=count, device_id=self.device_address)
+        if result.isError():
+            msg = f"Failed to read input register. device_address={self.device_address} address={address} count={count}"
+            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.
+
+        Returns:
+            True if register was written to.
+        """
+        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], count: int) -> int:
+        """Unpack registers.
+
+        Internal helper to unpuck register list.
+        """
+        if count == self.INT16_REG_COUNT:
+            return regs[0]
+
+        if count == self.INT32_REG_COUNT:
+            return (regs[1] << 16) + regs[0]
+        msg = f"Failed to unpack register, not speciefied. device_address={self.device_address} regs={regs}"
+        raise ModbusException(msg)
+
+    @property
+    def v(self) -> float:
+        """Return Voltage.
+
+        Returns:
+            Voltage
+        """
+        regs = self._read_input_registers(0, self.INT32_REG_COUNT)
+        return self._unpack(regs, self.INT32_REG_COUNT) / 10
+
+    @property
+    def a(self) -> float:
+        """Return A value."""
+        regs = self._read_input_registers(2, self.INT32_REG_COUNT)
+        return self._unpack(regs, self.INT32_REG_COUNT) / 1000
+
+    @property
+    def w(self) -> float:
+        """Return W value."""
+        regs = self._read_input_registers(4, self.INT32_REG_COUNT)
+        return self._unpack(regs, self.INT32_REG_COUNT) / 10
+
+    @property
+    def w_dmd(self) -> float:
+        """Return W Dmd value."""
+        regs = self._read_input_registers(10, self.INT32_REG_COUNT)
+        return self._unpack(regs, self.INT32_REG_COUNT) / 10
+
+    @property
+    def w_dmd_peak(self) -> float:
+        """Return W dmd peak value."""
+        regs = self._read_input_registers(12, self.INT32_REG_COUNT)
+        return self._unpack(regs, self.INT32_REG_COUNT) / 10
+
+    @property
+    def hz(self) -> float:
+        """Return Hz value."""
+        regs = self._read_input_registers(15, self.INT16_REG_COUNT)
+        return self._unpack(regs, self.INT16_REG_COUNT) / 10
+
+    @property
+    def kwh_tot(self) -> float:
+        """Return total kWh value."""
+        regs = self._read_input_registers(16, self.INT32_REG_COUNT)
+        return self._unpack(regs, self.INT32_REG_COUNT) / 10
+
+    @property
+    def kwh_partial(self) -> float:
+        """Return partial kwh value."""
+        regs = self._read_input_registers(20, self.INT32_REG_COUNT)
+        return self._unpack(regs, self.INT32_REG_COUNT) / 10
+
+    @property
+    def hour_counter(self) -> float:
+        """Return hour count."""
+        regs = self._read_input_registers(44, self.INT32_REG_COUNT)
+        return self._unpack(regs, self.INT32_REG_COUNT) / 100
+
+    @property
+    def lifetime_counter(self) -> float:
+        """Return partial hour count."""
+        regs = self._read_input_registers(54, self.INT32_REG_COUNT)
+        return self._unpack(regs, self.INT32_REG_COUNT) / 100
+
+    @property
+    def hour_counter_partial(self) -> float:
+        """Return partial hour count."""
+        regs = self._read_input_registers(54, self.INT32_REG_COUNT)
+        return self._unpack(regs, self.INT32_REG_COUNT) / 100
+
+    @property
+    def a_dmd(self) -> float:
+        """Return demand amps."""
+        regs = self._read_input_registers(58, self.INT32_REG_COUNT)
+        return self._unpack(regs, self.INT32_REG_COUNT) / 1000
+
+    @property
+    def a_dmd_peak(self) -> float:
+        """Return demand amps."""
+        regs = self._read_input_registers(60, self.INT32_REG_COUNT)
+        return self._unpack(regs, self.INT32_REG_COUNT) / 1000
+
+    @property
+    def password(self) -> int:
+        """Return password."""
+        regs = self._read_input_registers(4096, self.INT16_REG_COUNT)
+        return self._unpack(regs, self.INT16_REG_COUNT)
+
+    @property
+    def dmd_integr_time(self) -> int:
+        """Return Integration time for  DMD Power calculation unit. Default = 0."""
+        regs = self._read_input_registers(4112, self.INT32_REG_COUNT)
+        return self._unpack(regs, self.INT32_REG_COUNT)
+
+    @property
+    def alarm_state(self) -> int:
+        """Return alarm state."""
+        regs = self._read_input_registers(4116, self.INT16_REG_COUNT)
+        return self._unpack(regs, self.INT16_REG_COUNT)
+
+    @property
+    def alarm_mode(self) -> int:
+        """Read alarm linked state, default = 1."""
+        regs = self._read_input_registers(4117, self.INT16_REG_COUNT)
+        return self._unpack(regs, self.INT16_REG_COUNT)
+
+    @property
+    def alarm_delay(self) -> int:
+        """Return alarm delay, 0 - 3600 s. Default = 0."""
+        regs = self._read_input_registers(4122, self.INT16_REG_COUNT)
+        return self._unpack(regs, self.INT16_REG_COUNT)
+
+    @property
+    def device_id(self) -> int:
+        """Return device id."""
+        regs = self._read_input_registers(8192, self.INT16_REG_COUNT)
+        return self._unpack(regs, self.INT16_REG_COUNT)
+
+    @property
+    def baud_rate(self) -> int:
+        """Return baud rate."""
+        regs = self._read_input_registers(8193, self.INT16_REG_COUNT)
+        return self._unpack(regs, self.INT16_REG_COUNT)
+
+    @property
+    def parity(self) -> int:
+        """Return parity."""
+        regs = self._read_input_registers(8194, self.INT16_REG_COUNT)
+        return self._unpack(regs, self.INT16_REG_COUNT)
+
+    @property
+    def stop_bit(self) -> int:
+        """Return stop bits."""
+        regs = self._read_input_registers(8195, self.INT16_REG_COUNT)
+        return self._unpack(regs, self.INT16_REG_COUNT)
+
+    @property
+    def reply_delay(self) -> int:
+        """Return reply_delay."""
+        regs = self._read_input_registers(8196, self.INT16_REG_COUNT)
+        return self._unpack(regs, self.INT16_REG_COUNT)
+
+    @password.setter
+    def password(self, *, value: int) -> None:
+        """Set password."""
+        self._write_register(4096, value)
+
+    @dmd_integr_time.setter
+    def dmd_integr_time(self, *, value: int) -> None:
+        """Set IIntegration time for DMD power calculation, default = 3, self.INT16_REG_COUNT5 min."""
+        self._write_register(4112, value)
+
+    @alarm_state.setter
+    def alarm_state(self, *, value: bool) -> None:
+        """Set alarm state."""
+        self._write_register(4116, value)
+
+    @alarm_mode.setter
+    def alarm_mode(self, *, value: int) -> None:
+        """Set alarm linked state, default = 1."""
+        self._write_register(4117, value)
+
+    @alarm_delay.setter
+    def alarm_delay(self, *, value: int) -> None:
+        """Set alarm delay 0 - 3600 s, Default = 0."""
+        self._write_register(4122, value)
+
+    @device_id.setter
+    def device_id(self, *, value: int) -> None:
+        """Set device id."""
+        self._write_register(8192, value)
+
+    @baud_rate.setter
+    def baud_rate(self, *, value: int) -> None:
+        """Set baud rate."""
+        self._write_register(8193, value)
+
+    @parity.setter
+    def parity(self, *, value: int) -> None:
+        """Set parity."""
+        self._write_register(8194, value)
+
+    @stop_bit.setter
+    def stop_bit(self, *, value: int) -> None:
+        """Set stop bits."""
+        self._write_register(8195, value)
+
+    @reply_delay.setter
+    def reply_delay(self, *, value: int) -> None:
+        """Set reply_delay."""
+        self._write_register(8196, value)
+
+    def reset_tot_energy_and_run_hour_counter(self) -> None:
+        """Reset total energy + total run hour counters (excluding lifetime)."""
+        self._write_register(16387, self.INT16_REG_COUNT)
+
+    def reset_partial_energy_and_partial_hour_counter(self) -> None:
+        """Reset partial energy + partial run hour counters."""
+        self._write_register(16388, self.INT16_REG_COUNT)
+
+    def reset_dmd_and_dmd_max(self) -> None:
+        """Reset DMD and DMD max values."""
+        self._write_register(16389, self.INT16_REG_COUNT)
+
+    def reset_to_factory_settings(self) -> None:
+        """Factory Restore (Default settings)."""
+        self._write_register(16416, 2570)
+        self._write_register(16416, 49568)

src/em511/test_hello.py

@@ -1,7 +1,18 @@
-# noqa: D100
-import em511
+"""Test file for driver."""
 
+from pymodbus.client import ModbusSerialClient
 
-def test_hello() -> None:
-    """Test hello function."""
-    assert em511.hello() == "Hello from python-em511!"  # noqa: S101
+from em511 import Em511
+
+"""Init client."""
+client = ModbusSerialClient(
+    port="/dev/ttyUSB0",
+    baudrate=9600,
+    parity="E",
+    stopbits=1,
+    bytesize=8,
+    timeout=1,
+)
+
+"""Init constructor"""
+AC_meter = Em511(102, client)