home-automation/app/integrations/modbus/driver.py

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