YAU.POC/wrps-demo-kit
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

test: Modbus verification harness

Browse source 
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
This commit is contained in:
Clio Liu 2026-09-02 17:27:30 +10:00
parent 2b8f88d01f
commit bed824a9a5
5 changed files with 728 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

129
05-tests/README.md Normal file
View file

@ -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 - <<EOF
from pymodbus.client import ModbusTcpClient
c = ModbusTcpClient(\"openplc-runtime\", port=502, timeout=5); c.connect()
print(c.read_holding_registers(0, count=21, slave=1).registers)
EOF'"
```
The container resolves `openplc-runtime` by service name, so no IP is needed.
## What a pass proves, and what it does not
**Proves:** the runtime honours the register map — every mapped address exists and
responds, the `%MW` block really does start at holding register 1024, and the RW
points accept writes.
**Does not prove:** that the control logic is correct. **None of the 20 acceptance
tests from `01-design/WRPS-CTL-003` (§7 and §8.4) are implemented here.** A green
run against a completely wrong control program looks identical to a green run
against a right one.
That gap is the largest open item in the project. This harness verifies the
*contract*, not the *behaviour*.
## Reading the output
What you see depends on which build is running. **The deployed PLC runs the
simulation build**, so expect the second column.
| | Field build, nothing driving the inputs | **Simulation build (deployed)** |
|---|---|---|
| `%IW` / `%IX` | all 0 / False | all 0 / False — the mux takes simulated values instead |
| Pumps | **tripped** (`%QX1.4-1.6` True) — thermal-healthy reads FALSE | running normally |
| `%QW17` alarm word | **16500** — thermals, mains and `LSLL` all reading unhealthy | 0 |
| `%QW5` / `%QW6` | 32767, the `CFG_NO_TIME` sentinel | 32767 while drawing down, a real countdown while filling |
| `%QW11` | 720 — 120 m² × 6.0 m, the plant geometry | falls as the well fills |
| `%MW` block | the seeded defaults | **whatever was last written** — operators retune these live |
In the field build, tripped pumps and a non-zero alarm word are the program
responding **correctly** to an all-zero field. It is not broken.
`%MW` is worth stressing: `IO_MUX` seeds the defaults once on first scan, and
nothing overwrites them afterwards, so a live system shows the tuning someone
last wrote — not the values in `03-plc/src/10_globals.st`. See `03-plc/README.md`.
## fake_plc.py — testing the SCADA side without a PLC
```bash
python fake_plc.py --serve --port 502 # serve known values
python fake_plc.py --expect # print what CI Server should show
```
**Why it exists.** Everything downstream of the register map — CI Server's Modbus
configuration, the conversions, the item bindings, the display masks — had only
ever been checked against the simulation, whose values are *plausible*. Plausible
is exactly what you cannot verify: a level of 70% looks right whether it came from
the correct register or the one beside it.
So it serves **distinctive** values instead. Every register gets a different one;
none round, none equal to a neighbour, and the pattern is deterministic — the same
seed gives the same numbers, so `expected_readings.txt` can be printed on a
machine that cannot reach the server.
Stdlib only, deliberately: it writes the MBAP header by hand rather than requiring
`pymodbus` on a host you may not want to install packages on.
Speaks FC01, FC02, FC03, FC04, FC06 and FC16. Writes are accepted and stored, so a
setpoint written from CI Server reads back.
| File | What |
|---|---|
| `fake_values.json` | the served values, so they can be regenerated or inspected |
| `expected_readings.txt` | what CI Server should display, item by item, seed `20260814` |
> [!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.

53
05-tests/expected_readings.txt Normal file
View file

@ -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

304
05-tests/fake_plc.py Normal file
View file

@ -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()

61
05-tests/fake_values.json Normal file
View file

@ -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
}

181
05-tests/verify_modbus.py Normal file
View file

@ -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())