#!/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())