"""Modbus TCP driver — thin wrapper around pymodbus. This module provides a single public function ``read_blocks`` that performs one or more block reads against a Modbus TCP gateway — using either FC04 (Read Input Registers) or FC03 (Read Holding Registers), selected per call via the ``function_code`` argument — and returns a flat ``dict[register_address -> 16-bit_value]`` map. The function code comes from the device profile (e.g. SDM120 uses FC04, DDSU666 uses FC03). Design decisions ---------------- - **Only read function codes** (FC03/FC04) are exposed. There is no write path. - float32 decoding uses ``struct.unpack('>f', ...)`` directly rather than the pymodbus ``BinaryPayloadDecoder`` helper, which has had API churn across 3.x releases. Big-endian word order + big-endian byte order means the 4 raw bytes are already in standard network (big-endian) order. - ``ModbusTcpClient`` is used in synchronous mode (``connect()`` / ``close()``). The client is created fresh per call; this keeps the driver stateless and avoids threading concerns at the cost of one TCP handshake per read cycle. APScheduler jobs call this from a thread pool, so stateless is safer. - pymodbus 3.13.x uses ``device_id=`` as the slave-address keyword argument (renamed from ``slave=`` in earlier 3.x releases). Exceptions ---------- ``ModbusDriverError`` Base exception for all driver-level errors. ``ModbusConnectionError`` Raised when the TCP connection to the gateway cannot be established or is lost during the read. ``ModbusResponseError`` Raised when the gateway responds with a Modbus exception frame or when the returned register count does not match the requested count. """ from __future__ import annotations import logging import struct from typing import TYPE_CHECKING from pymodbus.client import ModbusTcpClient from pymodbus.exceptions import ConnectionException, ModbusException if TYPE_CHECKING: from collections.abc import Sequence logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Custom exceptions # --------------------------------------------------------------------------- class ModbusDriverError(RuntimeError): """Base class for all Modbus driver errors.""" class ModbusConnectionError(ModbusDriverError): """Cannot connect to (or lost connection with) the Modbus TCP gateway.""" class ModbusResponseError(ModbusDriverError): """Modbus gateway returned an exception frame or an unexpected response.""" # --------------------------------------------------------------------------- # Block definition type alias # --------------------------------------------------------------------------- #: A single read block: ``{"start": int, "count": int}``. Block = dict[str, int] # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def read_blocks( host: str, port: int, unit_id: int, blocks: Sequence[Block], *, function_code: int = 4, timeout: float = 3.0, ) -> dict[int, int]: """Read one or more contiguous register blocks via FC03 or FC04. Parameters ---------- host: Hostname or IP address of the Modbus TCP gateway. port: TCP port (typically 502). unit_id: Modbus slave / unit address (the device's Meter ID for SDM120). blocks: Sequence of ``{"start": int, "count": int}`` dicts describing the contiguous register ranges to read. ``count`` is the number of 16-bit registers (not bytes). function_code: Modbus read function code: ``4`` for FC04 (Read Input Registers, default — SDM120) or ``3`` for FC03 (Read Holding Registers — DDSU666 and other devices that expose measurements as holding registers). Comes from the device profile's ``function_code`` field. timeout: TCP connect/read timeout in seconds (default 3 s). Returns ------- dict[int, int] Flat mapping of ``{register_address: 16-bit_register_value}`` for every register returned across all blocks. Raises ------ ModbusDriverError If ``function_code`` is neither 3 nor 4 (validated before any connection is attempted). ModbusConnectionError If the TCP connection to the gateway fails. ModbusResponseError If the gateway returns a Modbus exception frame or an unexpected number of registers. """ if function_code not in (3, 4): raise ModbusDriverError( f"Unsupported read function code FC{function_code:02d} " f"(only FC03 holding-register and FC04 input-register reads are supported)" ) client = ModbusTcpClient(host, port=port, timeout=timeout) try: connected = client.connect() if not connected: raise ModbusConnectionError( f"Could not connect to Modbus gateway at {host}:{port}" ) registers: dict[int, int] = {} for block in blocks: start: int = block["start"] count: int = block["count"] _read_block(client, unit_id, start, count, registers, function_code=function_code) return registers except ConnectionException as exc: raise ModbusConnectionError( f"Connection to Modbus gateway {host}:{port} failed: {exc}" ) from exc except ModbusException as exc: raise ModbusResponseError(f"Modbus protocol error: {exc}") from exc finally: client.close() def _read_block( client: ModbusTcpClient, unit_id: int, start: int, count: int, result: dict[int, int], *, function_code: int, ) -> None: """Read one block and merge into *result*. Raises on any error. ``function_code`` is assumed already validated to be 3 or 4 by the caller (``read_blocks``); 3 dispatches FC03 (holding) and 4 dispatches FC04 (input). """ try: if function_code == 3: response = client.read_holding_registers(start, count=count, device_id=unit_id) else: # function_code == 4 (input registers) response = client.read_input_registers(start, count=count, device_id=unit_id) except ConnectionException as exc: raise ModbusConnectionError( f"Lost connection while reading registers 0x{start:04X}+{count}: {exc}" ) from exc if response.isError(): raise ModbusResponseError( f"Modbus exception reading registers 0x{start:04X}+{count}: {response}" ) regs = response.registers if len(regs) != count: raise ModbusResponseError( f"Expected {count} registers from 0x{start:04X}, got {len(regs)}" ) for offset, value in enumerate(regs): result[start + offset] = value # --------------------------------------------------------------------------- # Decoding helpers # --------------------------------------------------------------------------- def registers_to_float(hi_reg: int, lo_reg: int) -> float: """Decode two 16-bit registers to a big-endian IEEE-754 float32. The SDM120 (and most Modbus float devices) use big-endian word order: the high 16-bit register comes first, then the low register. Combined with big-endian byte order within each register, the raw 4-byte sequence is standard network-byte-order (big-endian) float32. Parameters ---------- hi_reg: The first (higher-address-range, most-significant) 16-bit register. lo_reg: The second (lower-address-range, least-significant) 16-bit register. Returns ------- float The decoded IEEE-754 single-precision floating-point value. Example ------- >>> registers_to_float(0x4366, 0x3334) 230.20001220703125 """ raw = struct.pack(">HH", hi_reg, lo_reg) return struct.unpack(">f", raw)[0]