home-automation/scripts/modbus_cli.py

"""Modbus CLI — read-only manual test tool for Modbus TCP devices.

Entry point::

    python -m scripts.modbus_cli <subcommand> [options]

Subcommands
-----------
read --host H --port P --unit U --profile NAME
    Read all metrics defined in the named profile from the device and print
    decoded engineering values as a formatted table.  Use this to verify the
    full decode chain end-to-end.

probe --host H --port P --unit U --fc {3,4} --address ADDR --count N
      [--decode float32]
    Send a raw FC03 (holding) or FC04 (input) read request for exactly N
    registers starting at ADDR, and print the raw 16-bit register values in
    hex.  Optionally decode consecutive register pairs as big-endian float32.
    Use this for first-contact verification of gateway reachability, framer
    mode, and unit_id before setting up a full profile.

Design
------
- **Read-only**: only FC03 and FC04 are exposed.  There is no write path.
  This prevents accidental modification of device configuration registers
  (Meter ID, baud rate, etc.) which could cause communication loss.
- **No DB dependency**: the tool connects directly to the gateway without
  reading from the application database.  It is safe to use before any
  device has been configured in the application.
- **Non-zero exit on connection failure** so shell scripts can detect errors.
"""

from __future__ import annotations

import argparse
import sys
from pathlib import Path

# Ensure the project root is on sys.path when invoked as ``python -m scripts.modbus_cli``
# or directly.
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from app.integrations.modbus.driver import (
    ModbusConnectionError,
    ModbusDriverError,
    ModbusResponseError,
    registers_to_float,
)
from app.integrations.modbus.profiles import ProfileNotFoundError, ProfileValidationError, decode, load_profile

# ---------------------------------------------------------------------------
# Helper — raw FC03/FC04 read (used by ``probe``)
# ---------------------------------------------------------------------------


def _raw_read(
    host: str,
    port: int,
    unit_id: int,
    fc: int,
    address: int,
    count: int,
    timeout: float = 5.0,
) -> list[int]:
    """Perform a raw FC03 or FC04 read and return the register list.

    Raises ``ModbusConnectionError`` or ``ModbusResponseError`` on failure.
    """
    from pymodbus.client import ModbusTcpClient
    from pymodbus.exceptions import ConnectionException

    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}"
            )

        if fc == 4:
            response = client.read_input_registers(address, count=count, device_id=unit_id)
        elif fc == 3:
            response = client.read_holding_registers(address, count=count, device_id=unit_id)
        else:
            raise ValueError(f"Unsupported function code FC{fc:02d} (only FC03/FC04 are allowed)")

        if response.isError():
            raise ModbusResponseError(
                f"Modbus exception reading FC{fc:02d} registers 0x{address:04X}+{count}: "
                f"{response}"
            )

        regs = response.registers
        if len(regs) != count:
            raise ModbusResponseError(
                f"Expected {count} registers, got {len(regs)}"
            )
        return list(regs)

    except ConnectionException as exc:
        raise ModbusConnectionError(
            f"Connection to {host}:{port} failed: {exc}"
        ) from exc
    finally:
        client.close()


# ---------------------------------------------------------------------------
# Sub-command: read
# ---------------------------------------------------------------------------


def cmd_read(args: argparse.Namespace) -> None:
    """Read all metrics from a named profile and print decoded values."""
    profile_name: str = args.profile
    host: str = args.host
    port: int = args.port
    unit_id: int = args.unit

    # Load and validate the profile.
    try:
        profile = load_profile(profile_name)
    except ProfileNotFoundError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        sys.exit(1)
    except ProfileValidationError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        sys.exit(1)

    print(f"Profile  : {profile.name} — {profile.description}")
    print(f"Gateway  : {host}:{port}  unit_id={unit_id}")
    print(f"Blocks   : {[(b.start, b.count) for b in profile.blocks]}")
    print()

    # Convert profile blocks to the dict format expected by read_blocks.
    blocks = [{"start": b.start, "count": b.count} for b in profile.blocks]

    # Perform the read.
    from app.integrations.modbus.driver import read_blocks

    try:
        registers = read_blocks(host, port, unit_id, blocks)
    except ModbusConnectionError as exc:
        print(f"Connection error: {exc}", file=sys.stderr)
        sys.exit(1)
    except ModbusResponseError as exc:
        print(f"Modbus response error: {exc}", file=sys.stderr)
        sys.exit(1)
    except ModbusDriverError as exc:
        print(f"Modbus driver error: {exc}", file=sys.stderr)
        sys.exit(1)

    # Decode.
    values = decode(profile, registers)

    # Print as table.
    print(f"{'Metric':<20} {'Value':>14}  Unit")
    print("-" * 45)
    for metric in profile.metrics:
        key = metric.key
        if key in values:
            unit = metric.unit if metric.unit else "—"
            print(f"{key:<20} {values[key]:>14.4f}  {unit}")
        else:
            print(f"{key:<20} {'(not available)':>14}")


# ---------------------------------------------------------------------------
# Sub-command: probe
# ---------------------------------------------------------------------------


def cmd_probe(args: argparse.Namespace) -> None:
    """Perform a raw FC03/FC04 read and print raw register values."""
    host: str = args.host
    port: int = args.port
    unit_id: int = args.unit
    fc: int = args.fc
    address: int = args.address
    count: int = args.count
    do_decode: bool = args.decode == "float32"

    if fc not in (3, 4):
        print(f"Error: --fc must be 3 or 4 (got {fc})", file=sys.stderr)
        sys.exit(1)

    print(f"Gateway  : {host}:{port}  unit_id={unit_id}")
    print(f"Request  : FC{fc:02d}  start=0x{address:04X}  count={count}")
    print()

    try:
        regs = _raw_read(host, port, unit_id, fc, address, count)
    except ModbusConnectionError as exc:
        print(f"Connection error: {exc}", file=sys.stderr)
        sys.exit(1)
    except ModbusResponseError as exc:
        print(f"Modbus response error: {exc}", file=sys.stderr)
        sys.exit(1)
    except ModbusDriverError as exc:
        print(f"Modbus driver error: {exc}", file=sys.stderr)
        sys.exit(1)

    # Print raw register values.
    print(f"{'Addr':>8}  {'Hex':>6}  {'Dec':>7}")
    print("-" * 28)
    for i, val in enumerate(regs):
        print(f"0x{address + i:04X}    0x{val:04X}  {val:>7d}")

    # Optional float32 decoding — consecutive register pairs.
    if do_decode and len(regs) >= 2:
        print()
        print("float32 decode (big-endian, high register first):")
        print(f"{'Addr pair':>14}  {'Float value':>16}")
        print("-" * 35)
        for i in range(0, len(regs) - 1, 2):
            hi = regs[i]
            lo = regs[i + 1]
            fval = registers_to_float(hi, lo)
            print(f"0x{address + i:04X}+0x{address + i + 1:04X}  {fval:>16.6f}")


# ---------------------------------------------------------------------------
# Argument parser
# ---------------------------------------------------------------------------


def _auto_int(value: str) -> int:
    """Parse an integer that may be hex (``0x...``) or decimal."""
    return int(value, 0)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="python -m scripts.modbus_cli",
        description=(
            "Modbus CLI — read-only manual test tool for Modbus TCP devices. "
            "Use 'read' to decode all profile metrics; use 'probe' for raw "
            "first-contact register inspection."
        ),
    )
    subparsers = parser.add_subparsers(dest="command", metavar="<command>")
    subparsers.required = True

    # -- read --
    rp = subparsers.add_parser(
        "read",
        help=(
            "Read and decode all metrics defined in a profile. "
            "Prints a table of engineering values."
        ),
    )
    rp.add_argument("--host", required=True, metavar="HOST", help="Gateway IP/hostname.")
    rp.add_argument(
        "--port", type=int, default=502, metavar="PORT", help="Gateway TCP port (default 502)."
    )
    rp.add_argument(
        "--unit",
        type=int,
        required=True,
        metavar="UNIT_ID",
        help="Modbus slave / unit ID (Meter ID, typically 1).",
    )
    rp.add_argument(
        "--profile",
        required=True,
        metavar="NAME",
        help="Profile name (e.g. 'sdm120').",
    )
    rp.set_defaults(func=cmd_read)

    # -- probe --
    pp = subparsers.add_parser(
        "probe",
        help=(
            "Raw read of N registers starting at ADDR via FC03 or FC04. "
            "Prints raw hex values; optionally decodes as big-endian float32."
        ),
    )
    pp.add_argument("--host", required=True, metavar="HOST", help="Gateway IP/hostname.")
    pp.add_argument(
        "--port", type=int, default=502, metavar="PORT", help="Gateway TCP port (default 502)."
    )
    pp.add_argument(
        "--unit",
        type=int,
        required=True,
        metavar="UNIT_ID",
        help="Modbus slave / unit ID.",
    )
    pp.add_argument(
        "--fc",
        type=int,
        choices=[3, 4],
        default=4,
        metavar="FC",
        help="Function code: 3 (holding) or 4 (input registers, default).",
    )
    pp.add_argument(
        "--address",
        type=_auto_int,
        required=True,
        metavar="ADDR",
        help="Start register address (hex e.g. 0x0000 or decimal).",
    )
    pp.add_argument(
        "--count",
        type=int,
        required=True,
        metavar="N",
        help="Number of registers to read.",
    )
    pp.add_argument(
        "--decode",
        choices=["float32"],
        default=None,
        metavar="TYPE",
        help="Optional: decode consecutive register pairs as 'float32'.",
    )
    pp.set_defaults(func=cmd_probe)

    return parser


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------


def main(argv: list[str] | None = None) -> None:
    parser = build_parser()
    args = parser.parse_args(argv)
    args.func(args)


if __name__ == "__main__":
    main()