#!/usr/bin/env python3 """Derive the SCADA-side point list from the PLC-side register map. python gen_scada_points.py Reads ../../03-plc/register-map.csv (PLC side, generated by build.py) Writes ./ci-server-points.csv (SCADA side, for CI Server config) CLAUDE.md section 3: the two sides are two views of the same points, and the PLC side leads. This script makes the SCADA view reproducible instead of hand-transcribed, so the two cannot drift apart silently. It is a *starting point* for CI Server configuration, not a CI Server import file. CI Server's .qli item exports carry no Modbus addressing (that lives in the front-end I/O configuration), so the mapping is applied by hand in CI Server from this list. The four judgements this script encodes --------------------------------------- 1. **Poll groups.** Four, not one per function code. The holding registers split by purpose and address range: read-only published data at 0-20, operator setpoints at 1024-1034, simulation control at 1044-1047. Each group is then contiguous and polls as one request; a single holding-register group would span 1048 mostly-empty addresses. 2. **%QW17 is UNSIGNED.** Bit 15 does not fit a signed INT. Configure it as a 16-bit unsigned register or the alarm word goes negative exactly when the most severe alarm is set. 3. **Engineering units are chosen here, not in the PLC.** The PLC publishes mm, L/s x10 and Hz x10 and keeps doing so; a unit on the SCADA side is a gain on the raw register (see UNITS below). One unit per register, though - CI Server rejects a second item on an I/O address it already has. 4. **%IW / %IX are excluded by default.** They are the *field* inputs. In the simulation build nothing writes them, so they read 0 forever and would show a dead plant on the HMI. Every live measurement is published in the %QW block. Pass --include-field to emit them anyway, for a real field deployment. """ import argparse import csv import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[2] SRC_MAP = ROOT / "03-plc" / "register-map.csv" OUT = Path(__file__).resolve().parent / "ci-server-points.csv" # Modbus object -> (function code, poll group name) # # Holding registers are split into three groups rather than one. They # share a function code but not a purpose or an address range: %QW is # read-only published data at 0-1023, %MW is writable at 1024+. One # group spanning both would be a single 1048-register request over a # mostly empty address space. GROUPS = { "Coil": ("FC01", "PS_STATUS_BITS"), "Discrete input": ("FC02", "PS_FIELD_BITS"), "Input register": ("FC04", "PS_FIELD_ANALOG"), } HR_PUBLISHED = ("FC03", "PS_PUBLISHED") # %QW, read-only HR_SETPOINTS = ("FC03/FC06", "PS_SETPOINTS") # %MW0-10, operator writable HR_SIM = ("FC03/FC06", "PS_SIM_CONTROL") # %MW20-23, simulation only FIELD_ONLY = {"Input register", "Discrete input"} UNSIGNED = {"%QW17"} # --------------------------------------------------------------------------- # Engineering units, SCADA side # --------------------------------------------------------------------------- # # The PLC publishes mm, L/s x10 and Hz x10 - that is the contract in # register-map.csv and it does not change. What the operator reads is a # presentation choice, and CI Server already applies a linear conversion # per Modbus point, so a unit here is nothing but a gain on the raw # register: # # engineering value = raw register * gain (offset is always 0) # # Zero offset matters: it keeps the full-scale ELEC -> PHYS mapping in # gen_ciserver_qli.py exact, and it means a second unit on the same # register is just a second point with a different gain. # # ONE ITEM PER I/O ADDRESS. A register carries exactly one engineering # unit, because CI Server will not accept two items on the same Modbus # address - an import that tries it fails with # # EQP-E-DUP_ITEM, I/O address of item already defined # DSSP-E-INSREC, Failed to insert a record in the dataset ITEM_DF # # (confirmed on R1.03, 2026-08-14, trying to publish level in m *and* %). # So showing one measurement in two units needs two *PLC* registers, not # two views of one. Level and speed are published in % alone. # # Gains chosen 2026-08-14: # level % 100% = the 6.000 m spill weir, so a full bar means spilling # and the reading reconciles with VOL_TO_SPILL. 1/60 per mm. # flow m3/h = L/s x 3.6, and the register is L/s x10: 0.36. # speed % 100% = 50 Hz, the drive's maximum. Register Hz x10: 0.2. # # iec address -> (units, gain, format mask) UNITS = { "%QW0": ("%", 1 / 60.0, "999.9"), # wet well level, was mm "%QW1": ("m3/h", 0.36, "9999.9"), # inflow, was L/s "%QW2": ("m3/h", 0.36, "9999.9"), # total discharge, was L/s "%QW4": ("%", 0.2, "999.9"), # common drive speed, was Hz "%QW7": ("m3/h", 0.36, "9999.9"), # net accumulation, signed "%MW3": ("%", 1 / 60.0, "999.9"), # level setpoint, was mm "%MW4": ("%", 1 / 60.0, "999.9"), # start duty "%MW5": ("%", 1 / 60.0, "999.9"), # start pump 2 "%MW6": ("%", 1 / 60.0, "999.9"), # start pump 3 "%MW7": ("%", 1 / 60.0, "999.9"), # stop all "%MW8": ("%", 1 / 60.0, "999.9"), # high level alarm "%MW9": ("%", 0.2, "999.9"), # minimum drive speed, was Hz "%MW20": ("m3/h", 0.36, "9999.9"), # SIM manual inflow, was L/s } # Points that keep their PLC unit but not the default mask. Enums, modes # and small counts are one or two digits wide; a five-digit mask on them # is not wrong, only wide, and it would disagree with the HMI's masks in # 04-scada/hmi/point_format.py. iec address -> mask MASK_ONLY = { "%QW3": "9", # pumps running, 0-3 "%QW11": "9999", # volume remaining to spill, m3 "%QW12": "9", # station state enum "%QW13": "9", "%QW14": "9", "%QW15": "9", # pump state enums "%QW16": "9", # duty pump, 0-3 "%QW20": "99", # command acknowledge "%MW0": "9", # station mode "%MW1": "99", # command word "%MW2": "9", # command parameter, pump number "%MW21": "9", # SIM scenario 0-3 "%MW22": "9", # SIM reset, write 1 "%MW23": "999", # SIM time scale 1-120 } def eng_view(iec, scaling): """(units override, gain, mask override) for a register. Falls back to the PLC's own unit and the raw scaling when the point is not in UNITS - most points are already in the unit the operator wants, and inventing a conversion for them would only add risk. """ if iec in UNITS: return UNITS[iec] gain = 0.1 if scaling == "x10" else 1.0 default = "9999.9" if scaling == "x10" else "99999" return None, gain, MASK_ONLY.get(iec, default) def gain_expr(gain): """The gain as the human-readable expression the CSV carries.""" if gain == 1.0: return "value" if gain == 0.1: return "value / 10" if abs(gain - 1 / 60.0) < 1e-12: return "value / 60" return "value * %g" % gain # The CI Server naming, imported from the .qli generator so the two files # cannot disagree. LEAF is keyed on IEC address and is the single place # section/leaf names are defined. # # An earlier version of this script emitted a `scada_tag` column holding # names like PS_STN_WET_WELL_LEVEL, derived from the PLC register map. # **CI Server never adopted them.** They named nothing that exists, and # they caused a real defect downstream when read as if they did. The only # legitimate PS_ names are the four Modbus POLL GROUPS below, which are # live configuration. sys.path.insert(0, str(Path(__file__).resolve().parent)) from gen_ciserver_qli import LEAF, ROOT, STATION def ci_names(iec): """(item, station, point) - the three CI Server namespaces. AID.WRPS.STN.LEVEL item what the historian is keyed on WRPS_PLC station the Modbus station STN_LEVEL point the Modbus point on that station """ if iec not in LEAF: sys.exit(f"no leaf name defined for {iec} - add it to LEAF in gen_ciserver_qli.py") section, leaf = LEAF[iec] return (f"{ROOT}.{section}.{leaf}", STATION, ("%s_%s" % (section, leaf))[:24]) def main(): ap = argparse.ArgumentParser(description="Derive the SCADA point list.") ap.add_argument( "--include-field", action="store_true", help="also emit %%IW/%%IX field inputs (they read 0 in the simulation build)", ) args = ap.parse_args() if not SRC_MAP.is_file(): sys.exit(f"missing {SRC_MAP} - run 04-plc/wrps-plc/build.py first") rows = list(csv.DictReader(open(SRC_MAP, newline="", encoding="utf-8"))) out_rows, skipped = [], 0 for r in rows: obj = r["modbus_object"] if obj in FIELD_ONLY and not args.include_field: skipped += 1 continue iec = r["iec_address"] sim_only = r["description"].startswith("SIM ONLY") writable = r["access"] == "RW" description = r["description"].replace("SIM ONLY: ", "") if obj == "Holding register": if sim_only: fc, group = HR_SIM elif writable: fc, group = HR_SETPOINTS else: fc, group = HR_PUBLISHED else: fc, group = GROUPS[obj] if obj in ("Coil", "Discrete input"): data_type = "Boolean" elif iec in UNSIGNED: data_type = "Unsigned 16-bit" else: data_type = "Signed 16-bit" units_override, gain, mask = eng_view(iec, r["scaling"]) out_rows.append( { "ci_item": ci_names(iec)[0], "ci_station": ci_names(iec)[1], "ci_point": ci_names(iec)[2], "description": description, "poll_group": group, "function_code": fc, "modbus_address": r["modbus_address"], "data_type": data_type, "access": "Read/Write" if writable else "Read", "eng_units": units_override if units_override is not None else r["units"], "raw_to_eng": gain_expr(gain), "eng_gain": repr(gain), "format_mask": mask, "iec_address": iec, "plc_tag": r["tag"], "notes": "SIMULATION CONTROL - simulation build only" if sim_only else "", } ) with open(OUT, "w", newline="", encoding="utf-8") as fh: w = csv.DictWriter(fh, fieldnames=list(out_rows[0].keys())) w.writeheader() w.writerows(out_rows) groups = {} for r in out_rows: groups.setdefault(r["poll_group"], []).append(int(r["modbus_address"])) print(f"OK {OUT} ({len(out_rows)} points)") for g, addrs in groups.items(): print(f" {g:<18} {len(addrs):>3} points, addresses {min(addrs)}-{max(addrs)}") if skipped: print( f" skipped {skipped} %IW/%IX field points - they read 0 in the\n" f" simulation build; pass --include-field for a real field deployment" ) if __name__ == "__main__": main()