wrps-demo-kit/04-scada/modbus_points/gen_scada_points.py

#!/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 ./scada-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 / "scada-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


def tag_for(row, description):
    """Tag name per CLAUDE.md section 8: PS_<EQUIP><NN>_<MEAS>.

    Built from the cleaned description, so the "SIM ONLY:" marker does
    not end up inside the tag name.
    """
    equip = row["tag"].replace("-", "")
    meas = description.split("(")[0].split(",")[0].strip()
    meas = "".join(ch if ch.isalnum() else "_" for ch in meas)
    meas = "_".join(p for p in meas.split("_") if p)[:28].upper()
    return f"PS_{equip}_{meas}".upper()


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(
            {
                "scada_tag": tag_for(r, description),
                "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()