From 8a3a0acb7534ee42d4d20dc119994e7fb724361c Mon Sep 17 00:00:00 2001 From: "a.williamson" Date: Fri, 31 Jul 2026 15:34:31 -0500 Subject: [PATCH] "Updated Simulator" --- tools/seed_journal.py | 477 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 477 insertions(+) create mode 100644 tools/seed_journal.py diff --git a/tools/seed_journal.py b/tools/seed_journal.py new file mode 100644 index 0000000..8a71f4e --- /dev/null +++ b/tools/seed_journal.py @@ -0,0 +1,477 @@ +#!/usr/bin/env python3 +"""seed_journal.py - backdate synthetic alarm activity into the alarm journal. + +The alarm simulator (SimHarness/alarmsim) generates activity in real time, so +"give me 8 hours of alarms" would take 8 hours. This writes the same kind of +activity straight into the journal tables with backdated timestamps, shaped to +land on a target Alarm Health grade. + + python3 tools/seed_journal.py --hours 8 --grade B --dry-run # plan only + python3 tools/seed_journal.py --hours 8 --grade B # insert + +How the grade is hit (weights from PrimeControls.calc.DEFAULTS): + + rate 0.30 activations/hr vs the 6/hr ISA target + flood 0.25 % of window inside a flood episode (>10 activations/10 min) + chatter 0.20 sources re-activating >10x/hr with a median gap <=120 s + standing 0.15 alarms active >24 h (LIVE queryStatus - not seeded, see below) + priority 0.10 deviation from the ISA 80/15/5 low/medium/high mix + +Only `rate`, `flood` and `priority` are seedable. `standing` reads live gateway +alarm state via system.alarm.queryStatus, so whatever is actually active now is +measured as-is and the profile budgets for it (--standing-now). + +The plan is verified offline against the real PrimeControls.calc before anything +is written - the same module the gateway runs - and the run aborts if the +predicted grade misses the target. + +Journal conventions this mirrors (harvested from existing rows): + eventtype 0=active 1=clear 2=ack eventflags 0 for tag alarms + eventtime DATETIME in GATEWAY LOCAL time (America/Chicago), not UTC + event_data active/clear -> eventValue (dtype 0, intvalue 1/0) + ack -> ackUser + ackUserName (dtype 2, strvalue) + +Flood episodes are stable under window drift: calc bins on `origin = +(start_ms // bin_ms) * bin_ms`, always a multiple of bin_ms, so bin edges sit on +absolute 10-minute epoch marks no matter when the dashboard re-anchors "now". A +burst placed inside one such cell stays inside one bin. +""" +import argparse +import collections +import importlib.util +import json +import os +import random +import subprocess +import sys +import types +import uuid +from datetime import datetime + +try: + from zoneinfo import ZoneInfo +except ImportError: + ZoneInfo = None + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +PROJECT = os.path.join(ROOT, "ignition", "gateway", "projects", "PrimeBAT", + "ignition", "script-python", "PrimeControls") +TAGS_FILE = os.path.join(ROOT, "test-data", "simulation_tags.json") + +DB_CONTAINER = "buildathon-db" +DB_NAME = "ignition" +DB_USER = "ignition" +DB_PASS = "ignition" +EVENTS_TABLE = "PrimeControls_alarm_events" +DATA_TABLE = "PrimeControls_alarm_event_data" + +GATEWAY_TZ = "America/Chicago" +TAG_PROVIDER = "default" +TAG_ROOT = "BuildathonSim" + +PRIORITY_LEVEL = {"Diagnostic": 0, "Low": 1, "Medium": 2, "High": 3, "Critical": 4} +BUCKET_OF = {0: "low", 1: "low", 2: "medium", 3: "high", 4: "high"} +ACK_USERS = [("Prime", 953), ("operator", 284), ("admin", 122)] + +# Sources whose names advertise a pathology - excluded from a healthy seed. +BAD_ACTOR_MARKERS = ("_Chatter", "_Standing", "_Fleeting") + +BIN_MS = 600000 # calc DEFAULTS bin_ms (10 min) +HOUR_MS = 3600000 + +# Per-window profiles. `counts` is the exact low/medium/high activation split; +# `burst` is (hours_before_end, activation_count) placed inside one 10-min cell. +PROFILES = { + "B": { + "current": {"counts": {"low": 38, "medium": 9, "high": 3}, + "burst": (2.2, 11), "burst_area": "Packaging"}, + # Prior period only feeds the period-over-period delta chips; its grade + # is never scored. Made deliberately worse so deltas read as improving. + # Kept within the low-bucket capacity (13 sources x SOURCE_CAP). + "prior": {"counts": {"low": 46, "medium": 14, "high": 6}, + "burst": (2.6, 13), "burst_area": "BoilerHouse"}, + }, +} + +BASELINE_BIN_CAP = 4 # keeps non-burst cells under the flood-start (>10) and + # flood-continue (>=5) thresholds +SOURCE_CAP = 4 # < calc chatter_min_count (5) => structurally no chatter +MIN_DURATION_S = 45 # > calc fleeting_s (10) => structurally no fleeting +MAX_DURATION_S = 420 +ACK_RATE = 0.85 +CLEAR_RATE = 0.94 + + +# ---------------- the gateway's own calc module ---------------- + + +def load_calc(): + """Import PrimeControls.calc from the project exactly as tests/conftest.py + does, so the prediction uses the code the gateway actually runs.""" + pkg = types.ModuleType("PrimeControls") + pkg.__path__ = [PROJECT] + sys.modules["PrimeControls"] = pkg + out = {} + for mod in ("calc", "fmt"): + spec = importlib.util.spec_from_file_location( + "PrimeControls." + mod, os.path.join(PROJECT, mod, "code.py")) + m = importlib.util.module_from_spec(spec) + sys.modules["PrimeControls." + mod] = m + spec.loader.exec_module(m) + setattr(pkg, mod, m) + out[mod] = m + return out["calc"] + + +# ---------------- sources ---------------- + + +def load_sources(): + """[(rel_path, priority_level)] for every real (non-bad-actor) sim alarm.""" + with open(TAGS_FILE) as fh: + doc = json.load(fh) + rows = [] + + def walk(node, path): + name = node.get("name") + p = path + [name] if name else path + for alarm in node.get("alarms") or []: + rel = "/".join(p[1:]) + if any(m in rel for m in BAD_ACTOR_MARKERS): + continue + rows.append((rel, PRIORITY_LEVEL[alarm.get("priority")])) + for child in node.get("tags") or []: + walk(child, p) + + walk(doc, []) + return sorted(set(rows)) + + +def source_path(rel): + name = rel.split("/")[-1] + return "prov:%s:/tag:%s/%s:/alm:%s" % (TAG_PROVIDER, TAG_ROOT, rel, name) + + +# ---------------- planning ---------------- + + +def cell_start(ms): + """Absolute 10-minute epoch cell containing ms.""" + return (ms // BIN_MS) * BIN_MS + + +def allocate(sources, counts, burst_count, burst_area, rng): + """Pick (rel, level) for every activation, honouring the exact bucket split + and SOURCE_CAP. Returns (burst_picks, baseline_picks).""" + by_bucket = collections.defaultdict(list) + for rel, lvl in sources: + by_bucket[BUCKET_OF[lvl]].append((rel, lvl)) + + used = collections.Counter() + + def take(bucket, area=None): + pool = [s for s in by_bucket[bucket] if used[s[0]] < SOURCE_CAP + and (area is None or s[0].split("/")[0] == area)] + if not pool: + if area is not None: + return take(bucket, None) # area can't serve it; go global + raise SystemExit("exhausted %s sources (raise SOURCE_CAP)" % bucket) + pool.sort(key=lambda s: (used[s[0]], s[0])) + least = [s for s in pool if used[s[0]] == used[pool[0][0]]] + pick = rng.choice(least) + used[pick[0]] += 1 + return pick + + slots = [] + for bucket, n in counts.items(): + slots += [bucket] * n + rng.shuffle(slots) + + # The burst is an area cascade: draw its slots from one area where possible. + burst_slots, baseline_slots = slots[:burst_count], slots[burst_count:] + burst = [take(b, burst_area) for b in burst_slots] + baseline = [take(b) for b in baseline_slots] + return burst, baseline + + +def place_times(start_ms, end_ms, burst, baseline, burst_at_ms, rng): + """Assign an activation timestamp to every pick. + + Burst picks land inside the single 10-minute cell containing burst_at_ms + (with an edge margin so they cannot spill into the neighbouring cell). + Baseline picks spread over the remaining cells, capped at BASELINE_BIN_CAP. + """ + burst_cell = cell_start(burst_at_ms) + events = [] + + margin = 45000 + for pick in burst: + t = rng.randint(burst_cell + margin, burst_cell + BIN_MS - margin - 1) + events.append((t, pick)) + + cells = [] + c = cell_start(start_ms) + if c < start_ms: + c += BIN_MS # first whole cell inside the window + while c + BIN_MS <= end_ms: + if c != burst_cell: + cells.append(c) + c += BIN_MS + if not cells: + raise SystemExit("window too short to place baseline activations") + + capacity = len(cells) * BASELINE_BIN_CAP + if len(baseline) > capacity: + raise SystemExit("baseline %d exceeds capacity %d (cap %d/cell over %d cells)" + % (len(baseline), capacity, BASELINE_BIN_CAP, len(cells))) + + per_cell = collections.Counter() + for pick in baseline: + # Free choice among under-cap cells, so activity clumps the way real + # activity does. BASELINE_BIN_CAP (4) is below both the flood-start + # (>10) and flood-continue (>=5) thresholds, so no clump can start or + # extend a flood episode however the draws fall. + choices = [c for c in cells if per_cell[c] < BASELINE_BIN_CAP] + cell = rng.choice(choices) + per_cell[cell] += 1 + t = rng.randint(cell + 5000, cell + BIN_MS - 5000) + events.append((t, pick)) + + events.sort(key=lambda e: e[0]) + return events, burst_cell, per_cell + + +def build_instances(events, end_ms, rng): + """Expand each activation into its active / ack / clear lifecycle.""" + instances = [] + for active_ms, (rel, lvl) in events: + dur_s = rng.randint(MIN_DURATION_S, MAX_DURATION_S) + clear_ms = active_ms + dur_s * 1000 + # A clear past the window end would not be fetched; leave it open, which + # is what a still-active alarm looks like anyway. + if rng.random() > CLEAR_RATE or clear_ms > end_ms - 30000: + clear_ms = None + ack_ms = None + ack_user = None + if rng.random() < ACK_RATE: + ceiling = (clear_ms if clear_ms else end_ms) - 5000 + floor = active_ms + 15000 + if ceiling > floor: + ack_ms = rng.randint(floor, ceiling) + ack_user = rng.choices([u for u, _ in ACK_USERS], + weights=[w for _, w in ACK_USERS])[0] + instances.append({ + "event_id": str(uuid.uuid4()), "rel": rel, "source": source_path(rel), + "priority": lvl, "active_ms": active_ms, "ack_ms": ack_ms, + "clear_ms": clear_ms, "ack_user": ack_user, + }) + return instances + + +def plan_window(sources, spec, start_ms, end_ms, rng): + counts = spec["counts"] + burst_hours, burst_count = spec["burst"] + burst, baseline = allocate(sources, counts, burst_count, + spec.get("burst_area"), rng) + burst_at = end_ms - int(burst_hours * HOUR_MS) + events, burst_cell, per_cell = place_times(start_ms, end_ms, burst, baseline, + burst_at, rng) + instances = build_instances(events, end_ms, rng) + return instances, burst_cell, per_cell + + +# ---------------- verification against the real calc ---------------- + + +def to_journal_rows(instances, calc): + """Flatten instances into the row dicts alarms._norm_event would produce.""" + rows = [] + for inst in instances: + lvl, name = calc.normalize_priority(inst["priority"]) + base = {"event_id": inst["event_id"], "source": inst["source"], + "display_path": "", "priority": lvl, "priority_name": name, + "is_system": False, "ack_user": None} + rows.append(dict(base, state="active", ts=inst["active_ms"])) + if inst["ack_ms"]: + rows.append(dict(base, state="ack", ts=inst["ack_ms"], + ack_user=inst["ack_user"])) + if inst["clear_ms"]: + rows.append(dict(base, state="clear", ts=inst["clear_ms"])) + return rows + + +def predict(calc, rows, active_now, start_ms, end_ms, now_ms): + bundle = calc.build_bundle(rows, active_now, 0, start_ms, end_ms, now_ms, + {"filters": {"priorities": [], "areas": [], + "states": [], "search": ""}}) + return bundle + + +def synthetic_standing(calc, count, now_ms): + """Stand-ins for the live queryStatus rows, so the offline prediction sees + the same standing-alarm penalty the gateway will apply.""" + rows = [] + for i in range(count): + lvl, name = calc.normalize_priority(1) + rows.append({"source": source_path("TankFarm/ManifoldLeakDetect"), + "display_path": "", "priority": lvl, "priority_name": name, + "active_ms": now_ms - (48 + i) * HOUR_MS, "unacked": True}) + return rows + + +# ---------------- SQL ---------------- + + +def local_dt(ms): + if ZoneInfo is not None: + return datetime.fromtimestamp(ms / 1000.0, ZoneInfo(GATEWAY_TZ)).strftime( + "%Y-%m-%d %H:%M:%S") + return datetime.fromtimestamp(ms / 1000.0).strftime("%Y-%m-%d %H:%M:%S") + + +def sql_rows(instances, first_id): + """(event_values, data_values) as SQL literal tuples.""" + events, data = [], [] + next_id = first_id + for inst in instances: + stages = [(0, inst["active_ms"])] + if inst["ack_ms"]: + stages.append((2, inst["ack_ms"])) + if inst["clear_ms"]: + stages.append((1, inst["clear_ms"])) + for etype, ts in stages: + rid = next_id + next_id += 1 + events.append("(%d,'%s','%s','',%d,%d,0,'%s')" % ( + rid, inst["event_id"], inst["source"], inst["priority"], + etype, local_dt(ts))) + if etype == 0: + data.append("(%d,'eventValue',0,1,NULL,NULL)" % rid) + elif etype == 1: + data.append("(%d,'eventValue',0,0,NULL,NULL)" % rid) + else: + user = inst["ack_user"] + data.append("(%d,'ackUser',2,NULL,NULL,'usr-prov:%s:/usr:%s')" + % (rid, TAG_PROVIDER, user)) + data.append("(%d,'ackUserName',2,NULL,NULL,'%s')" % (rid, user)) + return events, data, next_id + + +def mariadb(sql, capture=True): + cmd = ["docker", "exec", "-i", DB_CONTAINER, "mariadb", + "-u" + DB_USER, "-p" + DB_PASS, DB_NAME, "-N", "-B", "-e", sql] + res = subprocess.run(cmd, capture_output=capture, text=True) + if res.returncode != 0: + raise SystemExit("mariadb failed: %s" % (res.stderr or "").strip()) + return (res.stdout or "").strip() + + +def max_id(): + out = mariadb("SELECT COALESCE(MAX(id),0) FROM %s;" % EVENTS_TABLE) + return int(out.split("\n")[0]) + + +# ---------------- reporting ---------------- + + +def show_health(label, bundle): + h = bundle["health"] + print(" %s: grade %s score %.2f" % (label, h["grade"], h["score"])) + for sub in h["subs"]: + score = "none" if sub["score"] is None else "%6.2f" % sub["score"] + print(" %-9s w=%.2f score=%s %s" + % (sub["key"], sub["weight"], score, sub["detail"])) + + +def main(): + ap = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--hours", type=float, default=8.0, + help="window length; must match the dashboard preset (default 8)") + ap.add_argument("--grade", default="B", choices=sorted(PROFILES), + help="target Alarm Health grade (default B)") + ap.add_argument("--standing-now", type=int, default=1, + help="live alarms active >24h, which cost 20 pts each on the " + "standing sub-score. NOT seedable - standing() reads " + "system.alarm.queryStatus. Confirm the real number with " + "the SimHarness ProbeTick probe (default 1)") + ap.add_argument("--no-prior", action="store_true", + help="skip the preceding comparison window (delta chips go 'new')") + ap.add_argument("--seed", type=int, default=20260731) + ap.add_argument("--dry-run", action="store_true", + help="plan and predict only; write nothing") + args = ap.parse_args() + + calc = load_calc() + sources = load_sources() + rng = random.Random(args.seed) + + now_ms = int(datetime.now().timestamp() * 1000) + span = int(args.hours * HOUR_MS) + end_ms, start_ms = now_ms, now_ms - span + prior_start = start_ms - span + + profile = PROFILES[args.grade] + + print("window %s -> %s (%.1f h)" + % (local_dt(start_ms), local_dt(end_ms), args.hours)) + print("sources %d real sim alarms (bad-actor tags excluded)" % len(sources)) + + cur, burst_cell, per_cell = plan_window(sources, profile["current"], + start_ms, end_ms, rng) + instances = list(cur) + if not args.no_prior: + pri, _, _ = plan_window(sources, profile["prior"], prior_start, start_ms, rng) + instances = pri + instances + + standing_n = args.standing_now + print("standing %d live alarm(s) >24h (measured, not seeded)" % standing_n) + + rows = to_journal_rows(instances, calc) + active_now = synthetic_standing(calc, standing_n, now_ms) + bundle = predict(calc, rows, active_now, start_ms, end_ms, now_ms) + + m = bundle["meta"] + print() + print("predicted (offline, via the gateway's own PrimeControls.calc):") + print(" activations %d over %.1f h = %.2f/hr" + % (m["activation_count"], m["window_hours"], bundle["kpis"]["rate_per_hr"]["value"])) + print(" floods %d episode(s), %.3f%% of window" + % (len(bundle["floods"]["episodes"]), bundle["floods"]["pct_time_in_flood"])) + print(" chattering %d fleeting %d standing %d" + % (len(bundle["chattering"]), bundle["fleeting"]["total"], + bundle["standing"]["count"])) + pct = bundle["priority"]["pct"] + print(" priority %.1f/%.1f/%.1f vs 80/15/5 (dev %.1f)" + % (pct["low"], pct["medium"], pct["high"], bundle["priority"]["sum_abs_dev"])) + print(" busiest 10-min cell (non-burst): %d burst cell: %d" + % (max(per_cell.values()) if per_cell else 0, + profile["current"]["burst"][1])) + show_health("health", bundle) + + grade = bundle["health"]["grade"] + if grade != args.grade: + raise SystemExit("\nABORT: predicted grade %s != target %s; nothing written." + % (grade, args.grade)) + print("\nprediction matches target grade %s" % args.grade) + + event_rows = sum(1 for _ in rows) + if args.dry_run: + print("dry run: would insert %d event rows for %d activations " + "(%d in the scored window)" + % (event_rows, len(instances), len(cur))) + return + + first = max_id() + 1 + events, data, last = sql_rows(instances, first) + stmt = ("INSERT INTO %s (id,eventid,source,displaypath,priority,eventtype," + "eventflags,eventtime) VALUES %s;\nINSERT INTO %s " + "(id,propname,dtype,intvalue,floatvalue,strvalue) VALUES %s;" + % (EVENTS_TABLE, ",".join(events), DATA_TABLE, ",".join(data))) + mariadb(stmt) + print("inserted %d event rows + %d data rows (ids %d..%d)" + % (len(events), len(data), first, last - 1)) + + +if __name__ == "__main__": + main()