wrps-demo-kit/05-tests/verify_modbus.py

#!/usr/bin/env python3
"""Verify the running PLC against the register map.

    python verify_modbus.py --host 10.0.0.17 --port 502 --unit 1

Reads every row of ../03-plc/register-map.csv from the live runtime and
reports what came back.  The map is the PLC-side half of the Modbus
contract (CLAUDE.md section 3); this harness is what proves the runtime
actually honours it.

What it checks
--------------
  * every mapped address is readable with the function code its object
    type implies (FC1 coils, FC2 discrete inputs, FC3 holding, FC4 input)
  * reads at the top of each mapped range succeed, i.e. the runtime's
    Modbus buffers are at least as large as the map claims
  * %QW17 is decoded as **unsigned**, since bit 15 does not fit a signed
    INT (see 03-plc/README.md)
  * with --write-test, every RW row is writable: each is written back
    with the value it already holds, then re-read.  This exercises FC6
    without changing any value.

    Caveat: "no value changed" is not the same as "no effect".  %MW1 is
    the command word, and re-writing a *pending* non-zero command could
    re-trigger it.  Harmless on the demo rig, where the command word is
    0 and acknowledged, but do not run --write-test against anything
    that matters without reading section 3.3 of the build brief first.

Exit code is 0 only if every check passed.

Nothing here asserts *correct control behaviour* — that is the 14 pass-1
acceptance tests in the build brief, which are a separate job.
"""

import argparse
import csv
import sys
from pathlib import Path

try:
    from pymodbus.client import ModbusTcpClient
except ImportError:
    sys.exit("pymodbus is not installed.  pip install pymodbus")

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_MAP = ROOT / "03-plc" / "register-map.csv"

# Registers that must be interpreted as unsigned 16-bit.
UNSIGNED = {"%QW17"}

# modbus_object -> (reader name, is_bit)
READERS = {
    "Coil": ("read_coils", True),
    "Discrete input": ("read_discrete_inputs", True),
    "Input register": ("read_input_registers", False),
    "Holding register": ("read_holding_registers", False),
}


def load_map(path):
    with open(path, newline="", encoding="utf-8") as fh:
        rows = list(csv.DictReader(fh))
    if not rows:
        sys.exit(f"register map is empty: {path}")
    unknown = {r["modbus_object"] for r in rows} - set(READERS)
    if unknown:
        sys.exit(f"register map has unknown modbus_object value(s): {sorted(unknown)}")
    for r in rows:
        r["modbus_address"] = int(r["modbus_address"])
    return rows


def read_one(client, row, unit):
    """Read a single mapped point.  Returns (value, error_or_None)."""
    reader_name, is_bit = READERS[row["modbus_object"]]
    reader = getattr(client, reader_name)
    try:
        rr = reader(address=row["modbus_address"], count=1, device_id=unit)
    except Exception as exc:  # noqa: BLE001 - report, don't crash the run
        return None, f"{type(exc).__name__}: {exc}"
    if rr.isError():
        return None, str(rr)
    value = rr.bits[0] if is_bit else rr.registers[0]
    if not is_bit and row["iec_address"] not in UNSIGNED and value > 32767:
        value -= 65536  # map's INTs are signed unless listed in UNSIGNED
    return value, None


def write_back(client, row, value, unit):
    """No-op write: put back the value already there.  Proves writability."""
    try:
        wr = client.write_register(address=row["modbus_address"], value=value, device_id=unit)
    except Exception as exc:  # noqa: BLE001
        return f"{type(exc).__name__}: {exc}"
    if wr.isError():
        return str(wr)
    readback, err = read_one(client, row, unit)
    if err:
        return f"re-read failed: {err}"
    if readback != value:
        return f"wrote {value}, read back {readback}"
    return None


def scaled(value, scaling):
    """Render a raw register with its documented scaling."""
    if value is None or scaling != "x10":
        return ""
    return f"({value / 10:g})"


def main():
    ap = argparse.ArgumentParser(description="Verify the PLC against the register map.")
    ap.add_argument("--host", default="10.0.0.17")
    ap.add_argument("--port", type=int, default=502)
    ap.add_argument("--unit", type=int, default=1, help="Modbus slave/unit id")
    ap.add_argument("--map", default=str(DEFAULT_MAP), help="path to register-map.csv")
    ap.add_argument(
        "--write-test",
        action="store_true",
        help="also prove RW rows are writable, by writing back the value already there",
    )
    ap.add_argument("--quiet", action="store_true", help="only print failures and the summary")
    args = ap.parse_args()

    rows = load_map(args.map)
    print(f"map    : {args.map} ({len(rows)} points)")
    print(f"target : {args.host}:{args.port} unit {args.unit}")

    client = ModbusTcpClient(args.host, port=args.port, timeout=3)
    if not client.connect():
        sys.exit(f"FAIL: cannot connect to {args.host}:{args.port}")

    read_failures, write_failures, write_ok = [], [], 0
    try:
        for obj in ("Input register", "Discrete input", "Holding register", "Coil"):
            group = [r for r in rows if r["modbus_object"] == obj]
            if not group:
                continue
            if not args.quiet:
                print(f"\n--- {obj} " + "-" * (58 - len(obj)))
            for row in group:
                value, err = read_one(client, row, args.unit)
                if err:
                    read_failures.append((row, err))
                    print(f"  FAIL {row['iec_address']:<8} @{row['modbus_address']:<5} "
                          f"{row['tag']:<10} {err}")
                    continue
                if not args.quiet:
                    note = "  [unsigned]" if row["iec_address"] in UNSIGNED else ""
                    print(f"  {row['iec_address']:<8} @{row['modbus_address']:<5} "
                          f"{row['tag']:<10} {str(value):<8}{scaled(value, row['scaling']):<8}"
                          f"{row['description'][:38]}{note}")

                if args.write_test and row["access"] == "RW" and not err:
                    werr = write_back(client, row, value, args.unit)
                    if werr:
                        write_failures.append((row, werr))
                        print(f"  FAIL {row['iec_address']:<8} not writable: {werr}")
                    else:
                        write_ok += 1
    finally:
        client.close()

    rw_total = sum(1 for r in rows if r["access"] == "RW")
    print("\n" + "=" * 66)
    print(f"read     : {len(rows) - len(read_failures)}/{len(rows)} points OK")
    if args.write_test:
        print(f"writable : {write_ok}/{rw_total} RW points OK")
    else:
        print(f"writable : not tested ({rw_total} RW points; pass --write-test)")

    if read_failures or write_failures:
        print(f"FAILED   : {len(read_failures)} read, {len(write_failures)} write")
        return 1
    print("PASS     : the runtime matches the register map")
    return 0


if __name__ == "__main__":
    sys.exit(main())