"Updated Simulator"
This commit is contained in:
477
tools/seed_journal.py
Normal file
477
tools/seed_journal.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user