yau-plant-assistant/scripts/gen_historian_items.py

#!/usr/bin/env python3
"""Regenerate db/seed/historian_items.csv from the CI Server configuration.

    python scripts/gen_historian_items.py

WHY THIS EXISTS
---------------
The historian is keyed on CI Server ITEM names - `AID.WRPS.STN.LEVEL` - and
nothing else. Four namespaces describe the same measurement and only the last
one is what `imh` actually stores:

    LIT-101                instrument tag      the field device
    %QW0                   PLC symbol address  the register the PLC writes
    WRPS_PLC:STN_LEVEL     CI Server point     Modbus station : point name
    AID.WRPS.STN.LEVEL     CI Server item      what the historian is keyed on

db/002_fixtures.sql was once written against the point layer, which is why a
tag-level lookup matched zero history rows. This script pins the item layer
into the repository so the stand-in historian and the real one agree by
construction rather than by review.

`PS_*` IS NOT A NAMESPACE. An earlier delivery of the point list carried names
like PS_STN_WET_WELL_LEVEL where the item name belongs. They were a proposal
derived from the PLC register map that CI Server never adopted. Do not
reintroduce them. The only legitimate PS_ names are the four Modbus poll
groups - PS_STATUS_BITS, PS_PUBLISHED, PS_SETPOINTS, PS_SIM_CONTROL - which
are live configuration and appear only in the poll_group column.

WHAT IT READS - all of it under db/seed/scada-source/, delivered by hand and
not controlled. See the README there.

    ci-server-points.csv        49 points keyed on the item they feed
    wrps_item_df.qli            49 items, units, formats
    wrps_section_df.qli         the six AID.WRPS sections
    wrps_modbus_point_df.qli    point scaling, IO address, station
    item_his.qli                which items are historised
    export_his_group.qli        the LIVE historisation rates

Nothing downstream is hand-edited. If the CI Server configuration changes, ask
for a fresh delivery, replace the files there, and re-run this. The output CSV
is checked in so a deploy on lin001 needs nothing but this repository.

ON THE HISTORISATION GROUPS
    The source project's `his_group.qli` (the intended import) and the live
    system's `export_his_group.qli` DISAGREE: the file says WRPS_ONE_SEC is a
    1 second group and pairs it with a 60 second WRPS_ONE_MIN; the live server
    runs WRPS_ONE_SEC at 5 seconds and has WRPS_THIRTY_SEC at 30 seconds
    instead. The live server wins here - the stand-in exists to behave like
    the thing it stands in for. `--groups-from-repo` selects the other
    reading. Either way the chosen rates are written into the CSV, so
    db/002_fixtures.sql never hardcodes a sample interval.
"""

from __future__ import annotations

import argparse
import csv
import io
import re
import sys
from pathlib import Path

BS = chr(92)  # backslash; kept out of literals so shell heredocs cannot mangle it

# item_his.qli names the group each item belongs to. Those names are the
# repository's; map them onto whatever the live server actually calls the
# equivalent group. Identity for WRPS_EVENT - both agree it is on-change.
LIVE_GROUP_FOR = {
    "WRPS_ONE_SEC": "WRPS_ONE_SEC",
    "WRPS_ONE_MIN": "WRPS_THIRTY_SEC",
    "WRPS_EVENT": "WRPS_EVENT",
}

# The tag seed is keyed on the CI Server ITEM name for everything the historian
# carries, so the match is the item name itself and needs no lookup table. The
# old delivery forced one: it reused PS_STN_HIGH_LEVEL_ALARM for BOTH the coil
# 10 status bit and the holding register 1032 setpoint, so the name was not
# unique and the two had to be told apart by hand. The item layer distinguishes
# them on its own - STN.HIGH_LEVEL against SP.HIGH_ALARM - and all 49 item
# names are unique, so the special case is gone.

# Items deliberately NOT answerable by the assistant. They are historised, so
# they exist in the stand-in historian and in public.historian_items, but they
# carry no tag and no equipment and the agent cannot resolve a question onto
# them. Being listed here with a reason is the whole point: an item that is
# simply missing from the tag seed fails as "no records found", which an
# operator cannot tell apart from an absence of data. That was finding (a).
EXCLUDED_ITEMS = {
    "AID.WRPS.SIM.INFLOW":
        "simulation control, not a plant measurement - answering from it would "
        "report the scenario driver as though it were the real inflow",
    "AID.WRPS.SIM.SCENARIO":
        "simulation control, not a plant measurement",
    "AID.WRPS.SIM.RESET":
        "simulation control, not a plant measurement",
    "AID.WRPS.SIM.TIME_SCALE":
        "simulation control - a non-unity time scale means wall-clock durations "
        "in the history are compressed and must not be quoted as real durations",
}


def load_qli(path: Path, tag: str) -> list[dict[str, str]]:
    """Parse a CI Server .qli export into a list of records.

    The format is a @FIELDS block naming the columns, then a @<TAG> block of
    records. Records are comma-separated, quoted, and continued across lines
    with a trailing backslash.
    """
    text = path.read_text(encoding="utf-8", errors="replace").replace("\r\n", "\n")
    header = text.split("@FIELDS", 1)[1].split(tag, 1)[0].replace(BS, "")
    fields = [
        f.strip()
        for f in re.split(r"[,\n]", header)
        if f.strip() and not f.strip().startswith("!")
    ]
    body = text.split(tag, 1)[1].replace(BS + "\n", "")
    records = [line.strip() for line in body.split("\n") if line.strip().startswith('"')]
    return [
        dict(zip(fields, next(csv.reader(io.StringIO(r)))))
        for r in records
    ]


def io_address_parts(io_address: str) -> tuple[str, int]:
    """'RO:1025' -> ('holding', 1024).

    CI Server's IO_ADDRESS numbering is 1-based - confirmed on the WRPS system
    2026-08-14 by cross-checking STN.LEVEL against a direct pymodbus read - so
    holding register 0 is RO:01. DO is a coil, read with FC01.
    """
    kind, _, number = io_address.partition(":")
    zero_based = int(number) - 1
    return ("coil" if kind == "DO" else "holding"), zero_based


def scan_interval(groups: list[dict[str, str]], name: str) -> str:
    """Seconds between samples for a Scan/Time group; empty for event groups."""
    for g in groups:
        if g["NAME"] == name:
            return "" if g["COL_STOR_TYPE"] == "Event/Item" else g["SCAN_INTERVAL"]
    raise SystemExit(f"historisation group {name!r} is not defined on the server")


def life_time(groups: list[dict[str, str]], name: str) -> str:
    for g in groups:
        if g["NAME"] == name:
            return g["LIFE_TIME"]
    return ""


def build(source: Path, groups_from_repo: bool, tags_csv: Path) -> list[dict[str, str]]:
    modbus = source

    with tags_csv.open(encoding="utf-8") as fh:
        known_tags = {row["tag_id"] for row in csv.DictReader(fh)}

    items = load_qli(modbus / "wrps_item_df.qli", "@ITEM_DF")
    sections = {s["NAME"]: s for s in load_qli(modbus / "wrps_section_df.qli", "@SECTION_DF")}
    points = {p["NAME"]: p for p in load_qli(modbus / "wrps_modbus_point_df.qli", "@MODBUS_POINT_DF")}
    item_his = {h["ITEM_NAME"]: h for h in load_qli(modbus / "item_his.qli", "@ITEM_HIS_DF")}

    group_file = "his_group.qli" if groups_from_repo else "export_his_group.qli"
    groups = load_qli(modbus / group_file, "@HIS_GROUP_DF")

    # ci-server-points.csv carries the engineering gain, the poll group and the
    # IEC address - SCADA-side facts that live nowhere in the .qli definitions.
    # Key it the same way the point definitions are keyed, by resolved address.
    point_row_by_address: dict[tuple[str, int], dict[str, str]] = {}
    with (modbus / "ci-server-points.csv").open(encoding="utf-8") as fh:
        for row in csv.DictReader(fh):
            kind = "coil" if row["function_code"] == "FC01" else "holding"
            point_row_by_address[(kind, int(row["modbus_address"]))] = row

    out: list[dict[str, str]] = []
    for item in items:
        name = item["NAME"]
        point = points[item["POINT_NAME"]]
        kind, address = io_address_parts(point["IO_ADDRESS"])
        published = point_row_by_address.get((kind, address), {})

        his = item_his.get(name)
        repo_group = his["GROUP_NAME"] if his else ""
        group = "" if not repo_group else (
            repo_group if groups_from_repo else LIVE_GROUP_FOR.get(repo_group, repo_group)
        )

        # Resolve the item onto a tag, or onto a stated reason for having none.
        # Anything else is a build failure - see check_mapping below. The tag
        # seed is keyed on the item name, so this is an identity match.
        tag_id = name if name in known_tags else ""

        section_path = item["SECTION_PATH"]
        out.append(
            {
                "item_name": name,
                "tag_id": tag_id,
                "exclusion_reason": "" if tag_id else EXCLUDED_ITEMS.get(name, ""),
                "section_path": section_path,
                "section": item["UNIT"],
                "attribute": item["TAG"],
                "section_description": sections.get(section_path, {}).get("DESCRIPTION", ""),
                "description": item["DESCRIPTION"],
                "eng_unit": item["ENG_UNIT"],
                "value_format": item["VALUE_FORMAT"],
                "conv_type": point["CONV_TYPE"],
                "has_sign": point["HAS_SIGN"],
                "phys_low": point["PHYS_LOW"],
                "phys_high": point["PHYS_HIGH"],
                "eng_gain": published.get("eng_gain", ""),
                "raw_to_eng": published.get("raw_to_eng", ""),
                "his_group": group,
                # Empty for the on-change group. db/002_fixtures.sql reads this
                # rather than assuming a rate - the old fixtures hardcoded 60
                # seconds in two Cube measures and were wrong by 12x.
                "scan_interval_seconds": scan_interval(groups, group) if group else "",
                "life_time": life_time(groups, group) if group else "",
                "ci_station": point["STATION"],
                "ci_point": point["POINT"],
                "poll_group": published.get("poll_group", ""),
                "iec_address": published.get("iec_address", ""),
                "modbus_kind": kind,
                "modbus_address": str(address),
                "data_type": published.get("data_type", ""),
                # CI Server stamps every one of these points "Date+time GMT"
                # (TIME_ZONE in wrps_modbus_point_df.qli), and the WRPS history
                # groups all carry CORRECT_DAYLIGHT=0. Storage is UTC. Carried
                # per row so the claim is checkable at the point of use.
                "point_time_zone": point["TIME_ZONE"],
            }
        )
    return out


def check_mapping(rows: list[dict[str, str]]) -> list[str]:
    """Every historised item must resolve to a tag or to a stated reason.

    This is the check that stops finding (a) recurring. The old failure was an
    item the history was keyed on with no matching row in the tag seed: the
    join silently matched nothing and the assistant reported "no records
    found". Here that is a build error instead, and the only way past it is to
    add the tag or to write down why the item is deliberately unanswerable.
    """
    problems = []
    for r in rows:
        if not r["his_group"]:
            continue
        if not r["tag_id"] and not r["exclusion_reason"]:
            problems.append(
                f"  {r['item_name']} (point {r['ci_point'] or '?'}) has no row in "
                f"db/seed/tags.csv and no entry in EXCLUDED_ITEMS"
            )
    return problems


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--source", type=Path,
                        default=Path(__file__).parent.parent / "db" / "seed" / "scada-source",
                        help="folder holding the delivered CI Server configuration")
    parser.add_argument("--out", type=Path,
                        default=Path(__file__).parent.parent / "db" / "seed" / "historian_items.csv")
    parser.add_argument("--groups-from-repo", action="store_true",
                        help="take historisation rates from his_group.qli (the intended "
                             "import) rather than export_his_group.qli (what the server runs)")
    args = parser.parse_args()

    if not (args.source / "ci-server-points.csv").is_file():
        print(f"no CI Server configuration in {args.source}", file=sys.stderr)
        return 1

    tags_csv = args.out.parent / "tags.csv"
    rows = build(args.source, args.groups_from_repo, tags_csv)

    problems = check_mapping(rows)
    if problems:
        print("historised items that resolve to nothing:", file=sys.stderr)
        print("\n".join(problems), file=sys.stderr)
        print("\nAdd the tag to db/seed/tags.csv, or add the item to "
              "EXCLUDED_ITEMS with a reason. Nothing is written.", file=sys.stderr)
        return 1

    args.out.parent.mkdir(parents=True, exist_ok=True)
    with args.out.open("w", encoding="utf-8", newline="") as fh:
        writer = csv.DictWriter(fh, fieldnames=list(rows[0].keys()), lineterminator="\n")
        writer.writeheader()
        writer.writerows(rows)

    historised = [r for r in rows if r["his_group"]]
    print(f"wrote {len(rows)} items to {args.out}")
    print(f"  historised: {len(historised)}"
          f"   answerable: {sum(1 for r in historised if r['tag_id'])}"
          f"   deliberately excluded: {sum(1 for r in historised if r['exclusion_reason'])}")
    for group in sorted({r['his_group'] for r in historised}):
        members = [r for r in historised if r["his_group"] == group]
        interval = members[0]["scan_interval_seconds"] or "on change"
        print(f"    {group:16} {len(members):2} items   every {interval}"
              f"   life {members[0]['life_time']}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())