#!/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 @ 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())