From bed824a9a5fd982e576d972b6da5a2a81bee42b5 Mon Sep 17 00:00:00 2001 From: Clio Liu Date: Wed, 2 Sep 2026 17:27:30 +1000 Subject: [PATCH] test: Modbus verification harness The old 06-tests, with paths retargeted and both tools proven to run. Verified end to end during the move - fake_plc.py served on a spare port and verify_modbus.py read it: read : 69/69 points OK writable : 15/15 RW points OK PASS : the runtime matches the register map fake_plc.py --expect was BROKEN by the PS_* removal in the previous commit: it fell back to the scada_tag column, which no longer exists, and died with KeyError. It now uses ci_item, so the names it prints are the CI Server items an operator actually sees. Regenerated expected_readings.txt; fake_values.json came back byte-identical, which confirms the seed really is deterministic as the docstring claims. README corrections: - default host was dev-ubuntu; now 10.0.0.17, with the container fallback for when that address is not routable - said 11 RW points; there are 15 (11 control + 4 simulation) - the 'reading the output' section described the FIELD build - tripped pumps, alarm word 16500, seeded %MW defaults - as though it were what you would see. The deployed PLC runs the SIMULATION build. Both are now shown side by side, with the simulation column marked as the one that is live. - says plainly that none of the 20 acceptance tests are implemented here: this harness verifies the contract, not the behaviour - adds a caution against running fake_plc.py on port 502 of the live host, where the real PLC is serving and CI Server is polling --- 05-tests/README.md | 129 ++++++++++++++ 05-tests/expected_readings.txt | 53 ++++++ 05-tests/fake_plc.py | 304 +++++++++++++++++++++++++++++++++ 05-tests/fake_values.json | 61 +++++++ 05-tests/verify_modbus.py | 181 ++++++++++++++++++++ 5 files changed, 728 insertions(+) create mode 100644 05-tests/README.md create mode 100644 05-tests/expected_readings.txt create mode 100644 05-tests/fake_plc.py create mode 100644 05-tests/fake_values.json create mode 100644 05-tests/verify_modbus.py diff --git a/05-tests/README.md b/05-tests/README.md new file mode 100644 index 0000000..c84e138 --- /dev/null +++ b/05-tests/README.md @@ -0,0 +1,129 @@ +# 05-tests — Modbus verification harness + +Two tools. One reads the PLC and checks it against the register map; the other +*is* a PLC, serving deliberately distinctive values so the SCADA side can be +tested on its own. + +```bash +python verify_modbus.py --host 10.0.0.17 --port 502 --unit 1 +python verify_modbus.py --write-test # also prove the 15 RW points accept writes +python verify_modbus.py --quiet # failures and summary only +``` + +Requires `pymodbus` (`pip install pymodbus`). Exit code is 0 only if everything +passed, so it works as a gate. + +`verify_modbus.py` reads every row of `../03-plc/register-map.csv` from the live +runtime, using the function code each object type implies — FC1 coils, FC2 +discrete inputs, FC3 holding registers, FC4 input registers. + +## If you cannot reach 10.0.0.17 + +That address is only routable from inside the VNet or over the WireGuard VPN. From +anywhere else, poll from a throwaway container on the host's own Docker network: + +```bash +ssh lin001 "docker run --rm --network openplc-net python:3.12-alpine \ + sh -c 'pip install -q pymodbus && python - < [!CAUTION] +> **Do not run `fake_plc.py` on port 502 of the live host.** The real PLC is +> serving there and CI Server is polling it. Use a spare port, or stop the +> container first and remember that stopping it stops the demo. + +## Notes + +- **`%QW17` is decoded unsigned**; bit 15 does not fit a signed INT. Every other + register is treated as signed, per the map — including `%QW7`, which genuinely + goes negative. +- **`--write-test` writes each RW point back with the value it already holds.** No + value changes. See the caveat in the script docstring about `%MW1`, the command + word — a non-zero value there is a live command. +- **If everything fails to connect**, the cause is usually not this harness. The + runtime opens its Modbus slave only while a program is **running**, and only if + the Editor project defines a Modbus **Server**. See `03-plc/DEPLOY.md` §A3. A + refusal looks like a firewall drop and is not: Docker DNATs to the container, + which returns RST because nothing is bound inside. + +## Verified 2026-09-02 + +`fake_plc.py` served on a spare port and `verify_modbus.py` read it end to end: + +``` +read : 69/69 points OK +writable : 15/15 RW points OK +PASS : the runtime matches the register map +``` + +Both tools work against the current 69-point map. diff --git a/05-tests/expected_readings.txt b/05-tests/expected_readings.txt new file mode 100644 index 0000000..908c4ab --- /dev/null +++ b/05-tests/expected_readings.txt @@ -0,0 +1,53 @@ +Expected CI Server readings while fake_plc.py is serving (seed 20260814) + +ITEM ADDRESS RAW SHOULD READ NOTE +------------------------------------------------------------------------------ +PU301.RUN_CMD FC01:0 1 1 +PU302.RUN_CMD FC01:1 0 0 +PU303.RUN_CMD FC01:2 1 1 +PU301.RUNNING FC01:3 0 0 +PU302.RUNNING FC01:4 1 1 +PU303.RUNNING FC01:5 1 1 +PU301.AVAILABLE FC01:6 0 0 +PU302.AVAILABLE FC01:7 0 0 +PU303.AVAILABLE FC01:8 0 0 +STN.IN_AUTO FC01:9 1 1 +STN.HIGH_LEVEL FC01:10 1 1 +STN.SPILL_ACTIVE FC01:11 0 0 +PU301.TRIPPED FC01:12 1 1 +PU302.TRIPPED FC01:13 1 1 +PU303.TRIPPED FC01:14 0 0 +STN.LEVEL FC03:0 12124 202.1 % +STN.INFLOW FC03:1 20837 7501.3 m3/h +STN.DISCHARGE FC03:2 6350 2286.0 m3/h +STN.PUMPS_RUNNING FC03:3 3 3 count +STN.SPEED FC03:4 376 75.2 % +STN.TIME_TO_SPILL FC03:5 32767 32767 s 32767 - the 'drawing down' sentinel +STN.TIME_TO_LSHH FC03:6 26213 26213 s +STN.NET_ACCUM FC03:7 -981 -353.2 m3/h NEGATIVE - proves the point is signed +PU301.RUN_HOURS FC03:8 21439 21439 h +PU302.RUN_HOURS FC03:9 4852 4852 h +PU303.RUN_HOURS FC03:10 16665 16665 h +STN.VOL_TO_SPILL FC03:11 8878 8878 m3 +STN.STATE FC03:12 5 5 +PU301.STATE FC03:13 1 1 +PU302.STATE FC03:14 2 2 +PU303.STATE FC03:15 2 2 +STN.DUTY_PUMP FC03:16 1 1 +STN.ALARM_WORD FC03:17 42435 42435 bit15 set - proves the point is UNSIGNED +STN.CMD_ACK FC03:20 80 80 +SP.MODE FC03/FC06:1024 3 3 +SP.CMD_WORD FC03/FC06:1025 65 65 +SP.CMD_PARAM FC03/FC06:1026 5 5 +SP.LEVEL_SP FC03/FC06:1027 16343 272.4 % +SP.START_DUTY FC03/FC06:1028 28156 469.3 % +SP.START_P2 FC03/FC06:1029 11569 192.8 % +SP.START_P3 FC03/FC06:1030 23382 389.7 % +SP.STOP_ALL FC03/FC06:1031 6795 113.3 % +SP.HIGH_ALARM FC03/FC06:1032 18519 308.7 % +SP.MIN_SPEED FC03/FC06:1033 232 46.4 % +SP.SERVICE_HRS FC03/FC06:1034 13745 13745 h +SIM.INFLOW FC03/FC06:1044 22258 8012.9 m3/h +SIM.SCENARIO FC03/FC06:1045 5 5 +SIM.RESET FC03/FC06:1046 4 4 +SIM.TIME_SCALE FC03/FC06:1047 129 129 x diff --git a/05-tests/fake_plc.py b/05-tests/fake_plc.py new file mode 100644 index 0000000..97136d9 --- /dev/null +++ b/05-tests/fake_plc.py @@ -0,0 +1,304 @@ +#!/usr/bin/env python3 +"""A fake PLC: serves known values on Modbus TCP so the SCADA side can be +tested on its own. + + python3 fake_plc.py --serve # a spare port - NOT 502 on the live host + python fake_plc.py --expect # anywhere: what CI Server should show + +WHY THIS EXISTS +--------------- +Everything downstream of the register map - CI Server's Modbus config, the +conversions, the item bindings, the display masks - has only ever been +tested against the simulation, whose values are plausible. Plausible is +exactly what you cannot check: a level of 70% reads fine whether it came +from the right register or the one next to it. + +So: stop the PLC, serve *distinctive* values instead, and read the HMI. +Every register gets a different value, none of them round, none of them +equal to a neighbour, and the pattern is deterministic - the same seed +gives the same numbers, so the expected table can be printed anywhere, +including on a machine that cannot reach the server. + +Stdlib only, on purpose. The Ubuntu box has no pymodbus and a broken +ensurepip, and installing packages on someone's VM to run a test is a +worse trade than writing the 6 bytes of MBAP header out by hand. + +WHAT IT SPEAKS +-------------- +FC01 read coils, FC02 read discrete inputs, FC03 read holding registers, +FC04 read input registers, FC06 write single register, FC16 write multiple +registers. Writes are accepted and stored, so a setpoint written from the +HMI can be read back - that tests the write path in the same session. +Anything else gets exception 01 (illegal function). +""" + +import argparse +import socket +import socketserver +import struct +import sys +from pathlib import Path + +SEED = 20260814 # the date; change it for a fresh pattern +UNIT = 1 + +# ---------------------------------------------------------------- values + + +VALUES = Path(__file__).resolve().parent / "fake_values.json" +POINTS = Path(__file__).resolve().parents[1] / "04-scada" / "modbus_points" / "ci-server-points.csv" + +# A deliberately lopsided bit pattern. Not alternating: 1,0,1,0... is the +# one pattern that still looks like itself when the mapping slips, and an +# LCG's low bit produces exactly that. This one is distinguishable from +# every shift of itself. +COIL_BITS = 0b011011000110101 + + +def lcg(seed): + """A tiny deterministic generator - only used for the middle digits.""" + x = seed + while True: + x = (1103515245 * x + 12345) & 0x7FFFFFFF + yield x >> 8 # low bits of an LCG are not random; drop them + + +def make(): + """Build the test values from the point list, and write them out. + + Two constraints fight each other and both matter: + + - a value must be DISTINCTIVE, so a point reading its neighbour's + register is obvious. Plausible values are what makes a mis-mapped + point invisible, which is the whole reason for this exercise. + - a value must FIT THE DISPLAY MASK. The masks are sized for the + plant's real ranges, so a test value that overflows one renders as + nothing and looks like a fault that is not there. The first + attempt did exactly this - CMD_ACK got 2065 against a mask of 99. + + So the value is chosen per point, from its own mask and gain: as large + and as odd-looking as the mask can render, and no larger. + """ + import csv + import json + g = lcg(SEED) + coils, holding, notes = {}, {}, {} + + for i, r in enumerate(csv.DictReader(POINTS.open(newline="", encoding="utf-8"))): + addr, iec, mask = int(r["modbus_address"]), r["iec_address"], r["format_mask"] + if r["function_code"] == "FC01": + coils[addr] = bool(COIL_BITS >> addr & 1) + continue + + gain = float(r["eng_gain"]) + int_digits = len(mask.split(".")[0]) + max_eng = 10 ** int_digits - 1 # widest the mask renders + max_raw = min(int(max_eng / gain), 32000) # ...and a register holds + + if iec == "%QW17": # alarm bitmask, unsigned + holding[addr] = 0xA5C3 + notes[iec] = "bit15 set - proves the point is UNSIGNED" + elif iec == "%QW7": # net accumulation, signed + holding[addr] = -(next(g) % 900 + 100) + notes[iec] = "NEGATIVE - proves the point is signed" + elif iec == "%QW5": # time to spill + holding[addr] = 32767 + notes[iec] = "32767 - the 'drawing down' sentinel" + elif max_raw <= 9: # enums, counts, modes + holding[addr] = 1 + (next(g) % min(6, max_raw)) + else: + # SPREAD, not clustered. The first attempt put the address in + # the leading digits, which made the eleven setpoints land + # within 46-55% of each other - so a point reading its + # neighbour's register looked entirely reasonable, which is the + # one thing this test must not allow. Each point now takes a + # different fraction of its own renderable range, stepping by + # 37/89 so consecutive points land far apart, with an odd tail + # so no two are equal. + frac = ((i * 37 + 11) % 89 + 6) / 100.0 + v = int(max_raw * frac) + v = v - v % 100 + ((i * 13) % 89) + 7 if v > 200 else v + holding[addr] = max(1, min(v, max_raw)) + + data = {"seed": SEED, "coils": coils, "holding": holding, "notes": notes} + VALUES.write_text(json.dumps(data, indent=1, sort_keys=True) + "\n", + encoding="utf-8") + print("wrote %s (%d coils, %d holding registers)" + % (VALUES.name, len(coils), len(holding))) + return data + + +def pattern(): + import json + if not VALUES.is_file(): + sys.exit("missing %s - run: python fake_plc.py --make" % VALUES.name) + d = json.loads(VALUES.read_text(encoding="utf-8")) + return {"coils": {int(k): v for k, v in d["coils"].items()}, + "holding": {int(k): v for k, v in d["holding"].items()}, + "notes": d.get("notes", {})} + + +# ---------------------------------------------------------------- server +class Handler(socketserver.BaseRequestHandler): + def handle(self): + data = self.server.data + while True: + # MBAP is 7 bytes: transaction, protocol, length, unit. `length` + # counts the unit byte plus the PDU, so the PDU is length - 1. + head = self._recv(7) + if not head: + return + tid, pid, length, unit = struct.unpack(">HHHB", head) + body = self._recv(length - 1) + if not body: + return + fc = body[0] + try: + resp = self.dispatch(fc, body[1:], data) + except Exception: + resp = bytes([fc | 0x80, 0x02]) + self.request.sendall( + struct.pack(">HHHB", tid, pid, len(resp) + 1, unit) + resp) + + def _recv(self, n): + buf = b"" + while len(buf) < n: + chunk = self.request.recv(n - len(buf)) + if not chunk: + return b"" + buf += chunk + return buf + + def dispatch(self, fc, args, data): + if fc in (1, 2): # read bits + addr, count = struct.unpack(">HH", args[:4]) + bits = [data["coils"].get(addr + i, False) for i in range(count)] + packed = bytearray((count + 7) // 8) + for i, b in enumerate(bits): + if b: + packed[i // 8] |= 1 << (i % 8) + return bytes([fc, len(packed)]) + bytes(packed) + + if fc in (3, 4): # read registers + addr, count = struct.unpack(">HH", args[:4]) + out = b"" + for i in range(count): + v = data["holding"].get(addr + i, 0) if fc == 3 else 0 + out += struct.pack(">H", v & 0xFFFF) + return bytes([fc, len(out)]) + out + + if fc == 6: # write one + addr, value = struct.unpack(">HH", args[:4]) + data["holding"][addr] = value - 65536 if value > 32767 else value + data["writes"].append((addr, data["holding"][addr])) + print(" WRITE holding %d <- %d" % (addr, data["holding"][addr]), + flush=True) + return bytes([fc]) + args[:4] + + if fc == 16: # write many + addr, count = struct.unpack(">HH", args[:4]) + for i in range(count): + v, = struct.unpack(">H", args[5 + i * 2:7 + i * 2]) + data["holding"][addr + i] = v - 65536 if v > 32767 else v + data["writes"].append((addr + i, data["holding"][addr + i])) + print(" WRITE holding %d <- %d" % (addr + i, + data["holding"][addr + i]), + flush=True) + return bytes([fc]) + args[:4] + + return bytes([fc | 0x80, 0x01]) + + +class Server(socketserver.ThreadingTCPServer): + allow_reuse_address = True + daemon_threads = True + + +def serve(port): + data = pattern() + data["writes"] = [] + srv = Server(("0.0.0.0", port), Handler) + srv.data = data + print("fake PLC listening on 0.0.0.0:%d (seed %d)" % (port, SEED), flush=True) + print(" %d coils, %d holding registers served" + % (len(data["coils"]), len(data["holding"])), flush=True) + print(" writes from SCADA are stored and logged below", flush=True) + try: + srv.serve_forever() + except KeyboardInterrupt: + pass + + +# ---------------------------------------------------------------- expect +def half_up(value, decimals): + """Round the way CI Server does, not the way Python does. + + Python's %.1f rounds an exact half to even: 113.25 -> "113.2". + CI Server rounds it up: 113.3. Two of the 49 test values landed + exactly on a half, and both were reported as mismatches by the first + run of this table when the SCADA side was in fact correct. A test + that cries wolf twice in fifty is worse than no test. + """ + from decimal import Decimal, ROUND_HALF_UP + q = Decimal("1." + "0" * decimals) if decimals else Decimal("1") + return Decimal(repr(value)).quantize(q, rounding=ROUND_HALF_UP) + + +def expect(): + """What CI Server should display, in engineering units.""" + import csv + pts = Path(__file__).resolve().parents[1] / "04-scada" / "modbus_points" / "ci-server-points.csv" + rows = list(csv.DictReader(pts.open(newline="", encoding="utf-8"))) + data = pattern() + + # ci-server-points.csv carries the CI Server item name directly, so the + # name shown here is the one an operator sees on a display. (An earlier + # version fell back to a `scada_tag` column of PS_* placeholder names; + # those named nothing that exists and have been removed - see + # 04-scada/modbus_points/README.md.) + leaf = {r["iec_address"]: r["ci_item"].replace("AID.WRPS.", "") for r in rows} + + print("Expected CI Server readings while fake_plc.py is serving " + "(seed %d)\n" % SEED) + print("%-22s %-10s %8s %-14s %s" + % ("ITEM", "ADDRESS", "RAW", "SHOULD READ", "NOTE")) + print("-" * 78) + for r in rows: + a = int(r["modbus_address"]) + iec = r["iec_address"] + if r["function_code"] == "FC01": + raw = int(data["coils"].get(a, False)) + shown = str(raw) + else: + raw = data["holding"].get(a, 0) + gain = float(r["eng_gain"]) + dec = len(r["format_mask"].split(".")[1]) if "." in r["format_mask"] else 0 + eng = (raw & 0xFFFF) * gain if iec == "%QW17" else raw * gain + shown = "%s %s" % (half_up(eng, dec), r["eng_units"]) + print("%-22s %-10s %8d %-14s %s" + % (leaf.get(iec, r["ci_item"].replace("AID.WRPS.", "")), + "%s:%d" % (r["function_code"], a), raw, shown, + data["notes"].get(iec, ""))) + + +def main(): + ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + ap.add_argument("--make", action="store_true", + help="generate the test values from the point list") + ap.add_argument("--serve", action="store_true", help="run the server") + ap.add_argument("--expect", action="store_true", help="print expected readings") + ap.add_argument("--port", type=int, default=502) + a = ap.parse_args() + if a.make: + make() + elif a.serve: + serve(a.port) + elif a.expect: + expect() + else: + ap.print_help() + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/05-tests/fake_values.json b/05-tests/fake_values.json new file mode 100644 index 0000000..3dcb6bc --- /dev/null +++ b/05-tests/fake_values.json @@ -0,0 +1,61 @@ +{ + "coils": { + "0": true, + "1": false, + "2": true, + "3": false, + "4": true, + "5": true, + "6": false, + "7": false, + "8": false, + "9": true, + "10": true, + "11": false, + "12": true, + "13": true, + "14": false + }, + "holding": { + "0": 12124, + "1": 20837, + "2": 6350, + "3": 3, + "4": 376, + "5": 32767, + "6": 26213, + "7": -981, + "8": 21439, + "9": 4852, + "10": 16665, + "11": 8878, + "12": 5, + "13": 1, + "14": 2, + "15": 2, + "16": 1, + "17": 42435, + "20": 80, + "1024": 3, + "1025": 65, + "1026": 5, + "1027": 16343, + "1028": 28156, + "1029": 11569, + "1030": 23382, + "1031": 6795, + "1032": 18519, + "1033": 232, + "1034": 13745, + "1044": 22258, + "1045": 5, + "1046": 4, + "1047": 129 + }, + "notes": { + "%QW17": "bit15 set - proves the point is UNSIGNED", + "%QW5": "32767 - the 'drawing down' sentinel", + "%QW7": "NEGATIVE - proves the point is signed" + }, + "seed": 20260814 +} diff --git a/05-tests/verify_modbus.py b/05-tests/verify_modbus.py new file mode 100644 index 0000000..b52df1e --- /dev/null +++ b/05-tests/verify_modbus.py @@ -0,0 +1,181 @@ +#!/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())