From ecb10b9b7b025a49c941f4417719c16cbd9f2dbe Mon Sep 17 00:00:00 2001 From: "b.peck" Date: Thu, 16 Jul 2026 13:08:02 -0500 Subject: [PATCH] Phase 1 data layer: calc/fmt/alarms + 31 CPython tests + live gateway P1 smoke pytest 31/31. In-gateway end-to-end: 559 activations correlated (uuid mode), 3 chatterers, 34 fleeting, 1 flood, health F(22) on the deliberately-bad sim plant. Gate G1 green. Co-Authored-By: Claude Fable 5 --- .../config/ignition/tags/valueStore.idb | Bin 20480 -> 20480 bytes .../PrimeControls/alarms/code.py | 396 ++++++++++ .../PrimeControls/alarms/resource.json | 12 + .../script-python/PrimeControls/calc/code.py | 737 ++++++++++++++++++ .../PrimeControls/calc/resource.json | 12 + .../script-python/PrimeControls/fmt/code.py | 325 ++++++++ .../PrimeControls/fmt/resource.json | 12 + .../ignition/script-python/probe/code.py | 57 +- .../ignition/timer/ProbeTick/resource.json | 2 +- tests/conftest.py | 21 + tests/test_calc.py | 306 ++++++++ tests/test_fmt.py | 94 +++ 12 files changed, 1971 insertions(+), 3 deletions(-) create mode 100644 ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/alarms/code.py create mode 100644 ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/alarms/resource.json create mode 100644 ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/calc/code.py create mode 100644 ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/calc/resource.json create mode 100644 ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/fmt/code.py create mode 100644 ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/fmt/resource.json create mode 100644 tests/conftest.py create mode 100644 tests/test_calc.py create mode 100644 tests/test_fmt.py diff --git a/ignition/gateway/config/ignition/tags/valueStore.idb b/ignition/gateway/config/ignition/tags/valueStore.idb index e02c8f0a06674d21c48aa4c72246b36398c5fb2f..366ba4a6a303d6676bf8eb85db0afb0992b6b920 100644 GIT binary patch delta 1142 zcmYjROHUI~6lOqPwW-*>)q&UbqwF5HL< zBSlV;S>*y@2*0Pb9k5vMfOwQZi_=n^Y+Wx5ok_OdhNI4?$Y|>rmeIDF@OJ2IbU;O2 zAR!L}t&s zi=jR5N4N@ZdL7W;Xo87Qp`0RG5xxb5*TQnB1isYg6K1{e3Srg(GtAffMOHTRh&P=1 z7pa7VhqH)qN;9Te3~#?<)-W3F*aS(J1Gtu#4XVV{s(YAFs}@5W2|be5X#=ARI~cY; zMo; z1lH#YAA%)>9@e=k>WaYDY({rwz~Qy}&iTxJ?Q$9^cP}DRKGYQuGt^wgNV5^!{``eG z`~1On1ED9=$_Y%ho73Q|af2k&ma`k9B~r?}%EQ0NpujPRwV=_n2GWZ3!|^wmXj!*v zz}66mF7hX7xl|d>*N2OHDaQg3Er$EP)e`myg!Uq6GtBz%??Dx=B0v>ppZbme0ecmGvH$=8 delta 1116 zcmYjROH31C5Ei9Ce^$0t3c|Lwz7dd+QcEec1+mXa#aE0Xkse4%euCmN;o$3FVxmZx zi(I79_(&uH6OogNiP4Lw^`Z&!QB!DIwgo+?7X|0vF2&0xGvCa7-+nW@mqJ;WLRn*Z zMvg{aXJ-xLv&&Nr-%WZ5`3cyJMM_aSiFr|14abarj*9oLv+&*zZx!sT)q|lX7d{zF z2u+s0LLe!W!=ll`)1ifN42KpsH&+tj00fL$sdUj*P?)pgIYNnNTcqc;T@^&w2OV`= z*Y_Atp6Wb(y8FcWbcxvf1xv);&CYTn9F}H;Jp_K%tUYE3Pe5(5aWwwuK8Ep; zd~X@bs)2XzT(vx8Nxyd+B9*Y}>voT-tg@m|6v^!E6+5FvnMWzgkO z$wM-=t(#%dtOq7zbf=D|sT;qrE7gD0y@?2IvcwWlnUNs&fjJWUSOI#oh0xiz4=|j4 zmk$AR15d@lRTzqc!Gi`OoQ1GO3nyzi=<${U@8IFD*@G?48P?+R!QR>?jz*(u45QI= zUR}m}ov%!GT*rWm!w8c%hFGUHAe!EIw4+*NtJS@4cs#2<=kY zh($waB0P%u#QU}I+R}jj-u#1Ma-aj$b&cHeI$Mh+1Lmy+uGZ-|nyU1$ag`&kA}~1g z@;fMtGfRw(KE{IBn8vM1J5&Z6t+>F9rC_(#A>V|G@lBk8%hqO|&IruA8Eu7cO?n$z zn73}=X`<~M<`ZpGPW94Bg!~-1>s5r#ymYXhmv@8GXpKM;}hP2y$)*mA`;b-{|z4+pP8NDhSwYJdF$lF^mjMdh*gM zN`E_H+d5p|cphS@@$TKZL^vd|g#1Oaym= max_events: + truncated = True + break + return rows, dropped, truncated, None + except: + return rows, dropped, truncated, _err_text() + + +def _fetch_status(): + out = [] + try: + result = system.alarm.queryStatus(state=["ActiveUnacked", "ActiveAcked"]) + for evt in result: + try: + source = str(evt.getSource() or "") + if source.startswith("evt:") or ":/alm:" not in source: + continue + active_ms = None + try: + data = evt.getActiveData() + if data is not None: + active_ms = _ms_of(data.get("eventTime")) + except: + active_ms = None + if active_ms is None: + try: + active_ms = _ms_of(evt.get("eventTime")) + except: + pass + prio_raw = None + try: + p = evt.getPriority() + try: + prio_raw = int(p.ordinal()) + except: + prio_raw = str(p) + except: + pass + lvl, name = calc.normalize_priority(prio_raw) + unacked = False + try: + unacked = "Unacknowledged" in str(evt.getState() or "") + except: + pass + display_path = "" + try: + display_path = str(evt.getDisplayPath() or "") + except: + pass + out.append({"source": source, "display_path": display_path, + "priority": lvl, "priority_name": name, + "active_ms": active_ms, "unacked": unacked}) + except: + continue + except: + pass + return out + + +def _fetch_shelved(): + try: + paths = system.alarm.getShelvedPaths() + return len(list(paths)) if paths is not None else 0 + except: + return None + + +def _clean_filters(options): + f = {"priorities": [], "areas": [], "states": [], "search": ""} + if not options: + return f + try: + for k in ("priorities", "areas", "states"): + v = options.get(k) + if v: + f[k] = [x for x in list(v)] + f["priorities"] = [int(x) for x in f["priorities"]] + f["areas"] = [str(x) for x in f["areas"]] + f["states"] = [str(x) for x in f["states"]] + f["search"] = str(options.get("search") or "") + except: + pass + return f + + +# ---------------- Perspective entry points ---------------- + + +def getDashboardBundle(startMs, endMs, options=None): + try: + s = int(startMs) + e = int(endMs) + opts = calc.opts_merged(None) + filters = _clean_filters(options) + span = max(e - s, 1) + rows, dropped, truncated, err = _fetch_journal( + s - span, e, int(opts["max_events"])) + active_now = _fetch_status() + shelved = _fetch_shelved() + bundle = calc.build_bundle(rows, active_now, shelved, s, e, _now_ms(), + {"filters": filters}) + bundle["meta"]["dropped_rows"] = dropped + bundle["meta"]["truncated"] = truncated + if err: + bundle["meta"]["error"] = err + bundle["insights"] = fmt.insights(bundle) + return bundle + except: + try: + bundle = calc.build_bundle([], [], None, int(startMs or 0), + int(endMs or 0), _now_ms(), None) + except: + bundle = calc.build_bundle([], [], None, 0, 0, 0, None) + bundle["meta"]["error"] = _err_text() + bundle["insights"] = [] + return bundle + + +STATE_LABELS = {"active": "Active", "clear": "Cleared", "ack": "Acked"} + + +def _journal_rows(start_ms, end_ms, filters, max_rows): + rows, dropped, truncated, err = _fetch_journal(start_ms, end_ms, max_rows) + out = [] + prios = set(filters.get("priorities") or []) + areas = set(filters.get("areas") or []) + states = set(filters.get("states") or []) + search = (filters.get("search") or "").strip().lower() + for r in rows: + if r["is_system"]: + continue + parsed = calc.parse_source(r["source"], r["display_path"]) + if prios and r["priority"] not in prios: + continue + if areas and parsed["area"] not in areas: + continue + if states: + tag = {"active": "active", "clear": "cleared", "ack": "acked"}[r["state"]] + if tag not in states and not ("unacked" in states and r["state"] == "active"): + continue + if search and search not in (r["source"] + " " + parsed["label"]).lower(): + continue + out.append({"id": r["event_id"], "time_ms": r["ts"], + "time_label": fmt.day_clock(r["ts"]), + "source": r["source"], "label": parsed["label"], + "area": parsed["area"], "state": r["state"], + "state_label": STATE_LABELS.get(r["state"], r["state"]), + "priority": r["priority"], "priority_name": r["priority_name"], + "ack_user": r["ack_user"] or ""}) + out.sort(key=lambda r: -(r["time_ms"] or 0)) + return out, truncated, err + + +def getJournalPage(startMs, endMs, filters=None, page=0, pageSize=50): + try: + f = _clean_filters(filters) + rows, truncated, err = _journal_rows(int(startMs), int(endMs), f, + int(calc.DEFAULTS["max_events"])) + page = max(0, int(page or 0)) + page_size = max(1, int(pageSize or 50)) + start = page * page_size + return {"rows": rows[start:start + page_size], "total": len(rows), + "page": page, "page_size": page_size, "truncated": truncated, + "error": err} + except: + return {"rows": [], "total": 0, "page": 0, "page_size": int(pageSize or 50), + "truncated": False, "error": _err_text()} + + +def journalCsv(startMs, endMs, filters=None, maxRows=10000): + try: + f = _clean_filters(filters) + rows, truncated, err = _journal_rows(int(startMs), int(endMs), f, int(maxRows)) + headers = ["time", "state", "priority", "label", "area", "source", "ack_user"] + data = [[r["time_label"], r["state_label"], r["priority_name"], r["label"], + r["area"], r["source"], r["ack_user"]] for r in rows] + ds = system.dataset.toDataSet(headers, data) + return system.dataset.toCSV(ds) + except: + return "error,%s" % _err_text().replace(",", ";") + + +def getSourceDetail(source, startMs, endMs): + try: + s = int(startMs) + e = int(endMs) + src = str(source or "") + rows, dropped, truncated, err = _fetch_journal(s, e, int(calc.DEFAULTS["max_events"])) + mine = [r for r in rows if r["source"] == src] + instances, _mode = calc.correlate_instances(mine) + parsed = calc.parse_source(src, mine[0]["display_path"] if mine else "") + day_bins = calc.rate_bins(instances, s, e, {"bin_ms": 86400000, "max_bins": 400}) + daily = [{"t0": b["t0"], "count": b["count"]} for b in day_bins["bins"]] + ttas = [i["tta_ms"] for i in instances if i.get("tta_ms") is not None] + durs = [i["duration_ms"] for i in instances if i.get("duration_ms") is not None] + fleet = len([i for i in instances + if i.get("duration_ms") is not None + and i["duration_ms"] < calc.DEFAULTS["fleeting_s"] * 1000]) + ack_users = {} + for r in mine: + if r["state"] == "ack" and r.get("ack_user"): + ack_users[r["ack_user"]] = ack_users.get(r["ack_user"], 0) + 1 + top_ack = [{"user": u, "count": c} for u, c in + sorted(ack_users.items(), key=lambda kv: -kv[1])[:5]] + events = [] + for r in sorted(mine, key=lambda r: -r["ts"])[:100]: + events.append({"id": r["event_id"], "time_ms": r["ts"], + "time_label": fmt.day_clock(r["ts"]), + "state_label": STATE_LABELS.get(r["state"], r["state"]), + "priority_name": r["priority_name"], + "ack_user": r["ack_user"] or ""}) + return {"label": parsed["label"], "area": parsed["area"], "events": events, + "daily": daily, + "stats": {"count": len([i for i in instances if i["active_ms"] is not None]), + "avg_tta_ms": (sum(ttas) / len(ttas)) if ttas else None, + "avg_active_ms": (sum(durs) / len(durs)) if durs else None, + "fleeting_count": fleet, "top_ack_users": top_ack}, + "error": err} + except: + return {"label": str(source or ""), "area": "", "events": [], "daily": [], + "stats": {"count": 0, "avg_tta_ms": None, "avg_active_ms": None, + "fleeting_count": 0, "top_ack_users": []}, + "error": _err_text()} + + +def getEventData(eventId, aroundMs, source): + try: + eid = str(eventId or "") + mid = int(aroundMs) + props = [] + seen = {} + try: + result = system.alarm.queryJournal(startDate=mid - 3600000, + endDate=mid + 3600000, + includeData=True) + except: + result = system.alarm.queryJournal(startDate=mid - 3600000, + endDate=mid + 3600000) + for evt in result: + try: + if str(evt.getId()) != eid: + continue + try: + names = evt.getProperties() + except: + names = [] + for p in names or []: + try: + n = str(p) + v = evt.get(p) + if n not in seen: + seen[n] = True + props.append({"name": n, "value": str(v)}) + except: + continue + except: + continue + props.sort(key=lambda kv: kv["name"]) + return {"props": props, "error": None} + except: + return {"props": [], "error": _err_text()} + + +def getShiftReportText(shiftHours=8, anchorHour=0, options=None): + try: + bounds = fmt.shift_bounds(_now_ms(), int(shiftHours), int(anchorHour)) + s, e = bounds["previous"] + bundle = getDashboardBundle(s, e, options) + report = fmt.shift_report(bundle, bounds["label"]) + return {"text": report["text"], "start_ms": s, "end_ms": e, + "label": bounds["label"]} + except: + return {"text": "Shift report unavailable: %s" % _err_text(), + "start_ms": 0, "end_ms": 0, "label": ""} diff --git a/ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/alarms/resource.json b/ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/alarms/resource.json new file mode 100644 index 0000000..e160ebe --- /dev/null +++ b/ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/alarms/resource.json @@ -0,0 +1,12 @@ +{ + "scope": "A", + "version": 1, + "restricted": false, + "overridable": true, + "files": [ + "code.py" + ], + "attributes": { + "hintScope": 2 + } +} diff --git a/ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/calc/code.py b/ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/calc/code.py new file mode 100644 index 0000000..e8af1f0 --- /dev/null +++ b/ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/calc/code.py @@ -0,0 +1,737 @@ +# PrimeControls.calc - PURE alarm analytics computation (no system.*, no I/O, no clock). +# Runs under Jython 2.7 (Ignition) AND CPython 3 (pytest). No f-strings. +from __future__ import division, print_function + +DEFAULTS = { + "bin_ms": 600000, "flood_per_10min": 10, "flood_end_per_10min": 5, + "target_per_hr": 6.0, + "chatter_per_hr": 10.0, "chatter_gap_s": 120, "chatter_min_count": 5, + "fleeting_s": 10, "fleeting_min_count": 3, + "standing_hours": 24, "trend_buckets": 24, + "max_events": 50000, "max_bins": 5000, + "delta_flat_pct": 5.0, "insights_max": 5, + "priority_target": {"low": 80.0, "medium": 15.0, "high": 5.0}, + "health_weights": {"rate": 0.30, "flood": 0.25, "chatter": 0.20, + "standing": 0.15, "priority": 0.10}, + "grade_cuts": [(90, "A"), (80, "B"), (70, "C"), (60, "D"), (0, "F")], + "list_cap": 50, "active_now_cap": 200, +} + +PRIORITY_NAMES = {0: "Diagnostic", 1: "Low", 2: "Medium", 3: "High", 4: "Critical"} + + +def opts_merged(opts): + out = dict(DEFAULTS) + if opts: + for k in opts: + out[k] = opts[k] + return out + + +def clamp(v, lo, hi): + return max(lo, min(hi, v)) + + +def _median(values): + vs = sorted(v for v in values if v is not None) + n = len(vs) + if n == 0: + return None + if n % 2: + return float(vs[n // 2]) + return (vs[n // 2 - 1] + vs[n // 2]) / 2.0 + + +def _mean(values): + vs = [v for v in values if v is not None] + if not vs: + return None + return sum(vs) / float(len(vs)) + + +def _safe_div(a, b): + try: + if not b: + return None + return a / float(b) + except Exception: + return None + + +# ---------------- path parsing ---------------- + + +def parse_source(source, display_path): + source = source or "" + display_path = display_path or "" + tag_segments = [] + alarm_name = "" + if ":/tag:" in source: + tag_part = source.split(":/tag:", 1)[1] + tag_path = tag_part.split(":", 1)[0] + tag_segments = [s for s in tag_path.split("/") if s] + if ":/alm:" in source: + alarm_name = source.split(":/alm:", 1)[1].split(":", 1)[0] + if display_path: + segments = [s for s in display_path.split("/") if s] + elif tag_segments: + segments = list(tag_segments) + if alarm_name and (not segments or alarm_name.lower() != segments[-1].lower()): + segments.append(alarm_name) + else: + segments = [source] if source else ["unknown"] + if len(segments) >= 2: + label = "/".join(segments[-2:]) + else: + label = segments[0] + base = tag_segments if tag_segments else segments + if len(base) >= 2: + area = base[-2] + else: + area = base[0] if base else "unknown" + return {"label": label, "area": area, "segments": segments} + + +def normalize_priority(value): + try: + iv = int(value) + if 0 <= iv <= 4: + return iv, PRIORITY_NAMES[iv] + except Exception: + pass + s = str(value).strip().lower() if value is not None else "" + for lvl in PRIORITY_NAMES: + if PRIORITY_NAMES[lvl].lower() in s and s: + return lvl, PRIORITY_NAMES[lvl] + return -1, (str(value) if value is not None else "Unknown") + + +# ---------------- correlation ---------------- + + +def _new_instance(row): + lvl, name = row.get("priority", -1), row.get("priority_name", "Unknown") + parsed = parse_source(row.get("source"), row.get("display_path")) + return {"event_id": row.get("event_id"), "source": row.get("source") or "", + "label": parsed["label"], "area": parsed["area"], + "priority": lvl, "priority_name": name, + "active_ms": None, "clear_ms": None, "ack_ms": None, + "duration_ms": None, "tta_ms": None, "open": True, + "ack_user": None} + + +def _finalize(inst): + a, c, k = inst["active_ms"], inst["clear_ms"], inst["ack_ms"] + inst["duration_ms"] = (c - a) if (a is not None and c is not None and c >= a) else None + inst["tta_ms"] = (k - a) if (a is not None and k is not None and k >= a) else None + inst["open"] = c is None + return inst + + +def correlate_instances(rows): + rows = [r for r in (rows or []) if r and not r.get("is_system") and r.get("ts") is not None] + rows.sort(key=lambda r: r["ts"]) + if not rows: + return [], "uuid" + with_id = [r for r in rows if r.get("event_id")] + mode = "uuid" if len(with_id) >= 0.9 * len(rows) else "source-sequence" + instances = [] + if mode == "uuid": + groups, order = {}, [] + for i, r in enumerate(rows): + k = r.get("event_id") or ("_noid_%d" % i) + if k not in groups: + groups[k] = [] + order.append(k) + groups[k].append(r) + for k in order: + grp = groups[k] + ident = None + for r in grp: + if r["state"] == "active": + ident = r + break + inst = _new_instance(ident or grp[0]) + acts = [r["ts"] for r in grp if r["state"] == "active"] + clears = [r["ts"] for r in grp if r["state"] == "clear"] + acks = [r["ts"] for r in grp if r["state"] == "ack"] + inst["active_ms"] = min(acts) if acts else None + inst["clear_ms"] = min(clears) if clears else None + inst["ack_ms"] = min(acks) if acks else None + for r in grp: + if r.get("ack_user"): + inst["ack_user"] = r["ack_user"] + break + instances.append(_finalize(inst)) + else: + open_by_src = {} + for r in rows: + src = r.get("source") or "" + st = r.get("state") + if st == "active": + if src in open_by_src: + instances.append(_finalize(open_by_src.pop(src))) + inst = _new_instance(r) + inst["active_ms"] = r["ts"] + open_by_src[src] = inst + else: + inst = open_by_src.get(src) + if inst is None: + inst = _new_instance(r) + open_by_src[src] = inst + if st == "clear" and inst["clear_ms"] is None: + inst["clear_ms"] = r["ts"] + elif st == "ack" and inst["ack_ms"] is None: + inst["ack_ms"] = r["ts"] + if r.get("ack_user"): + inst["ack_user"] = r["ack_user"] + for src in open_by_src: + instances.append(_finalize(open_by_src[src])) + instances.sort(key=lambda i: i["active_ms"] if i["active_ms"] is not None else 0) + return instances, mode + + +def partition_instances(instances, start_ms, end_ms): + prior_start = start_ms - (end_ms - start_ms) + current, prior, orphans = [], [], [] + for inst in instances or []: + a = inst.get("active_ms") + if a is None: + orphans.append(inst) + elif start_ms <= a <= end_ms: + current.append(inst) + elif prior_start <= a < start_ms: + prior.append(inst) + else: + orphans.append(inst) + return current, prior, orphans + + +def apply_filters(instances, filters): + if not filters: + return instances + prios = set(filters.get("priorities") or []) + areas = set(filters.get("areas") or []) + states = set(filters.get("states") or []) + search = (filters.get("search") or "").strip().lower() + if not prios and not areas and not states and not search: + return instances + out = [] + for inst in instances: + if prios and inst["priority"] not in prios: + continue + if areas and inst["area"] not in areas: + continue + if states: + tags = set() + if inst["open"]: + tags.add("active") + if inst["clear_ms"] is not None: + tags.add("cleared") + if inst["ack_ms"] is not None: + tags.add("acked") + else: + tags.add("unacked") + if not (states & tags): + continue + if search: + hay = (inst["source"] + " " + inst["label"]).lower() + if search not in hay: + continue + out.append(inst) + return out + + +# ---------------- rate / floods ---------------- + + +def rate_bins(instances, start_ms, end_ms, opts=None): + o = opts_merged(opts) + bin_ms = int(o["bin_ms"]) + empty = {"bins": [], "bin_ms": bin_ms, + "target_per_bin": o["target_per_hr"] * bin_ms / 3600000.0, + "flood_threshold": 1, "max_count": 0} + if end_ms is None or start_ms is None or end_ms <= start_ms: + return empty + while (end_ms - start_ms) // bin_ms + 1 > o["max_bins"]: + bin_ms *= 2 + origin = (start_ms // bin_ms) * bin_ms + n = int((end_ms - origin) // bin_ms) + 1 + counts = {} + for inst in instances or []: + a = inst.get("active_ms") + if a is None or a < origin or a > end_ms: + continue + idx = int((a - origin) // bin_ms) + counts[idx] = counts.get(idx, 0) + 1 + flood_threshold = max(1, int(round(o["flood_per_10min"] * bin_ms / 600000.0))) + bins = [] + max_count = 0 + for i in range(n): + c = counts.get(i, 0) + max_count = max(max_count, c) + bins.append({"t0": origin + i * bin_ms, "t1": origin + (i + 1) * bin_ms, + "count": c, "flood": c > flood_threshold}) + return {"bins": bins, "bin_ms": bin_ms, + "target_per_bin": o["target_per_hr"] * bin_ms / 3600000.0, + "flood_threshold": flood_threshold, "max_count": max_count} + + +def flood_episodes(bins_result, instances, start_ms, end_ms, opts=None): + o = opts_merged(opts) + bins = (bins_result or {}).get("bins") or [] + bin_ms = (bins_result or {}).get("bin_ms") or o["bin_ms"] + thr = (bins_result or {}).get("flood_threshold") or 1 + end_thr = max(1, int(round(o["flood_end_per_10min"] * bin_ms / 600000.0))) + episodes = [] + cur = None + for b in bins: + if cur is None: + if b["count"] > thr: + cur = {"start_ms": b["t0"], "end_ms": b["t1"], "peak_bin_count": b["count"]} + else: + if b["count"] >= end_thr: + cur["end_ms"] = b["t1"] + cur["peak_bin_count"] = max(cur["peak_bin_count"], b["count"]) + else: + episodes.append(cur) + cur = None + if cur is not None: + episodes.append(cur) + for ep in episodes: + ep["duration_ms"] = ep["end_ms"] - ep["start_ms"] + src_counts = {} + area_counts = {} + total = 0 + for inst in instances or []: + a = inst.get("active_ms") + if a is not None and ep["start_ms"] <= a < ep["end_ms"]: + total += 1 + key = (inst["source"], inst["label"]) + src_counts[key] = src_counts.get(key, 0) + 1 + area_counts[inst["area"]] = area_counts.get(inst["area"], 0) + 1 + ep["event_count"] = total + if src_counts: + best = sorted(src_counts.items(), key=lambda kv: (-kv[1], kv[0][1]))[0] + ep["top_source_label"] = best[0][1] + ep["top_source_count"] = best[1] + else: + ep["top_source_label"] = "" + ep["top_source_count"] = 0 + if area_counts: + ep["top_area"] = sorted(area_counts.items(), key=lambda kv: (-kv[1], kv[0]))[0][0] + else: + ep["top_area"] = "" + span = (end_ms - start_ms) if (end_ms and start_ms and end_ms > start_ms) else 0 + pct = 0.0 + if span: + pct = 100.0 * sum(ep["duration_ms"] for ep in episodes) / span + return {"episodes": episodes, "pct_time_in_flood": pct} + + +# ---------------- detections ---------------- + + +def chattering(instances, start_ms, end_ms, opts=None): + o = opts_merged(opts) + window_h = _safe_div(end_ms - start_ms, 3600000.0) if (end_ms and start_ms) else None + if not window_h or window_h <= 0: + return [] + groups = {} + for inst in instances or []: + a = inst.get("active_ms") + if a is None: + continue + groups.setdefault(inst["source"], []).append(inst) + out = [] + for src in groups: + insts = groups[src] + count = len(insts) + if count < o["chatter_min_count"]: + continue + per_hour = count / window_h + times = sorted(i["active_ms"] for i in insts) + gaps = [times[i + 1] - times[i] for i in range(len(times) - 1)] + med_gap = _median(gaps) + if per_hour > o["chatter_per_hr"] and med_gap is not None and med_gap <= o["chatter_gap_s"] * 1000: + ref = insts[0] + out.append({"source": src, "label": ref["label"], "area": ref["area"], + "priority_name": ref["priority_name"], "count": count, + "per_hour": per_hour, "median_gap_s": med_gap / 1000.0}) + out.sort(key=lambda r: (-r["per_hour"], r["label"])) + return out[:int(o["list_cap"])] + + +def fleeting(instances, opts=None): + o = opts_merged(opts) + thresh = o["fleeting_s"] * 1000 + groups = {} + total = 0 + for inst in instances or []: + d = inst.get("duration_ms") + if d is not None and d < thresh: + total += 1 + groups.setdefault(inst["source"], []).append(inst) + sources = [] + for src in groups: + insts = groups[src] + if len(insts) < o["fleeting_min_count"]: + continue + ref = insts[0] + sources.append({"source": src, "label": ref["label"], "area": ref["area"], + "priority_name": ref["priority_name"], "count": len(insts), + "median_s": (_median([i["duration_ms"] for i in insts]) or 0) / 1000.0}) + sources.sort(key=lambda r: (-r["count"], r["label"])) + return {"total": total, "sources": sources[:int(o["list_cap"])]} + + +def standing(active_now, open_instances, now_ms, opts=None): + o = opts_merged(opts) + thresh = o["standing_hours"] * 3600000.0 + rows = [] + if active_now: + mode = "status" + for s in active_now: + a = s.get("active_ms") + age = (now_ms - a) if (a is not None and now_ms) else None + parsed = parse_source(s.get("source"), s.get("display_path")) + rows.append({"source": s.get("source") or "", "label": parsed["label"], + "area": parsed["area"], + "priority_name": s.get("priority_name", "Unknown"), + "active_ms": a, "age_ms": age, + "age_h": (age / 3600000.0) if age is not None else None, + "unacked": bool(s.get("unacked"))}) + else: + mode = "journal" + for inst in open_instances or []: + a = inst.get("active_ms") + age = (now_ms - a) if (a is not None and now_ms) else None + rows.append({"source": inst["source"], "label": inst["label"], + "area": inst["area"], "priority_name": inst["priority_name"], + "active_ms": a, "age_ms": age, + "age_h": (age / 3600000.0) if age is not None else None, + "unacked": inst.get("ack_ms") is None}) + rows.sort(key=lambda r: -(r["age_ms"] or -1)) + count = len([r for r in rows if r["age_ms"] is not None and r["age_ms"] > thresh]) + return {"mode": mode, "count": count, "rows": rows[:int(o["list_cap"])]} + + +# ---------------- MTTA / MTTR ---------------- + + +def mtta_mttr(instances, start_ms, end_ms, opts=None): + o = opts_merged(opts) + ttas = [i["tta_ms"] for i in instances or [] if i.get("tta_ms") is not None and i["tta_ms"] >= 0] + durs = [i["duration_ms"] for i in instances or [] if i.get("duration_ms") is not None and i["duration_ms"] >= 0] + result = {"mtta": {"mean_ms": _mean(ttas), "median_ms": _median(ttas), "count": len(ttas)}, + "mttr": {"mean_ms": _mean(durs), "median_ms": _median(durs), "count": len(durs)}, + "trend": []} + if not end_ms or not start_ms or end_ms <= start_ms: + return result + span = end_ms - start_ms + bin_ms = int(o["bin_ms"]) + width = max(bin_ms, ((span // int(o["trend_buckets"])) // bin_ms + 1) * bin_ms) + origin = (start_ms // width) * width + n = int((end_ms - origin) // width) + 1 + buckets = {} + for inst in instances or []: + a = inst.get("active_ms") + if a is None: + continue + idx = int((a - origin) // width) + b = buckets.setdefault(idx, {"tta": [], "dur": []}) + if inst.get("tta_ms") is not None and inst["tta_ms"] >= 0: + b["tta"].append(inst["tta_ms"]) + if inst.get("duration_ms") is not None and inst["duration_ms"] >= 0: + b["dur"].append(inst["duration_ms"]) + for i in range(n): + b = buckets.get(i, {"tta": [], "dur": []}) + result["trend"].append({"t0": origin + i * width, + "mtta_ms": _median(b["tta"]), + "mttr_ms": _median(b["dur"]), + "ack_n": len(b["tta"]), "clear_n": len(b["dur"])}) + return result + + +# ---------------- pareto / heatmap / priority ---------------- + + +def pareto(instances, opts=None): + counts = {} + refs = {} + total = 0 + for inst in instances or []: + if inst.get("active_ms") is None: + continue + total += 1 + src = inst["source"] + counts[src] = counts.get(src, 0) + 1 + if src not in refs: + refs[src] = inst + ordered = sorted(counts.items(), key=lambda kv: (-kv[1], refs[kv[0]]["label"]))[:10] + rows = [] + cum = 0 + for i, (src, c) in enumerate(ordered): + cum += c + ref = refs[src] + rows.append({"rank": i + 1, "source": src, "label": ref["label"], + "area": ref["area"], "priority_name": ref["priority_name"], + "count": c, "pct": 100.0 * c / total if total else 0.0, + "cum_pct": 100.0 * cum / total if total else 0.0}) + return {"total_activations": total, "rows": rows} + + +def top_sources(pareto_result, n=5): + return list((pareto_result or {}).get("rows") or [])[:n] + + +def heatmap(instances, opts=None): + import time as _time + rows = [[0] * 24 for _ in range(7)] + total = 0 + mx = 0 + for inst in instances or []: + a = inst.get("active_ms") + if a is None: + continue + lt = _time.localtime(a / 1000.0) + rows[lt.tm_wday][lt.tm_hour] += 1 + total += 1 + if rows[lt.tm_wday][lt.tm_hour] > mx: + mx = rows[lt.tm_wday][lt.tm_hour] + return {"rows": rows, "row_labels": ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"], + "max_count": mx, "total": total} + + +def priority_distribution(instances, opts=None): + o = opts_merged(opts) + raw_counts = {} + buckets = {"low": 0, "medium": 0, "high": 0, "other": 0} + for inst in instances or []: + lvl = inst.get("priority", -1) + name = inst.get("priority_name", "Unknown") + key = (lvl, name) + raw_counts[key] = raw_counts.get(key, 0) + 1 + if lvl in (0, 1): + buckets["low"] += 1 + elif lvl == 2: + buckets["medium"] += 1 + elif lvl in (3, 4): + buckets["high"] += 1 + else: + buckets["other"] += 1 + raw = [{"name": k[1], "level": k[0], "count": raw_counts[k]} + for k in sorted(raw_counts.keys())] + bucketed = buckets["low"] + buckets["medium"] + buckets["high"] + pct = {"low": 0.0, "medium": 0.0, "high": 0.0} + sum_abs_dev = None + target = o["priority_target"] + if bucketed: + for k in pct: + pct[k] = 100.0 * buckets[k] / bucketed + sum_abs_dev = (abs(pct["low"] - target["low"]) + + abs(pct["medium"] - target["medium"]) + + abs(pct["high"] - target["high"])) + return {"raw": raw, "buckets": buckets, "pct": pct, + "target": dict(target), "sum_abs_dev": sum_abs_dev} + + +# ---------------- health / deltas ---------------- + + +def _piecewise(x, points): + if x is None: + return None + pts = sorted(points) + if x <= pts[0][0]: + return float(pts[0][1]) + if x >= pts[-1][0]: + return float(pts[-1][1]) + for i in range(len(pts) - 1): + x0, y0 = pts[i] + x1, y1 = pts[i + 1] + if x0 <= x <= x1: + if x1 == x0: + return float(y1) + return y0 + (y1 - y0) * (x - x0) / float(x1 - x0) + return float(pts[-1][1]) + + +def health_score(rate_per_hr, pct_flood, chatter_count, standing_count, + sum_abs_dev, has_activity, opts=None): + o = opts_merged(opts) + if not has_activity: + rate_s, flood_s, chatter_s = 100.0, 100.0, 100.0 + else: + rate_s = _piecewise(rate_per_hr, [(6, 100), (12, 60), (60, 0)]) + flood_s = _piecewise(pct_flood, [(0, 100), (1, 80), (5, 0)]) + chatter_s = clamp(100.0 - 25.0 * chatter_count, 0, 100) if chatter_count is not None else None + standing_s = clamp(100.0 - 20.0 * standing_count, 0, 100) if standing_count is not None else None + priority_s = clamp(100.0 - sum_abs_dev, 0, 100) if sum_abs_dev is not None else None + weights = o["health_weights"] + subs = [ + {"key": "rate", "label": "Alarm Rate", "score": rate_s, "weight": weights["rate"], + "detail": ("%.1f alarms/hr vs %.0f/hr ISA target" % (rate_per_hr, o["target_per_hr"])) + if rate_per_hr is not None else "no data"}, + {"key": "flood", "label": "Time in Flood", "score": flood_s, "weight": weights["flood"], + "detail": ("%.1f%% of window in flood (target <1%%)" % pct_flood) + if pct_flood is not None else "no data"}, + {"key": "chatter", "label": "Chattering Alarms", "score": chatter_s, "weight": weights["chatter"], + "detail": ("%d chattering alarm(s) (target 0)" % chatter_count) + if chatter_count is not None else "no data"}, + {"key": "standing", "label": "Standing Alarms", "score": standing_s, "weight": weights["standing"], + "detail": ("%d standing alarm(s) >24h (target 0)" % standing_count) + if standing_count is not None else "no data"}, + {"key": "priority", "label": "Priority Distribution", "score": priority_s, "weight": weights["priority"], + "detail": ("%.0f pts deviation from ISA 80/15/5" % sum_abs_dev) + if sum_abs_dev is not None else "no data"}, + ] + num = 0.0 + den = 0.0 + for s in subs: + if s["score"] is not None: + num += s["score"] * s["weight"] + den += s["weight"] + score = (num / den) if den else None + grade = None + if score is not None: + for cut, g in o["grade_cuts"]: + if score >= cut: + grade = g + break + if grade is None: + grade = "F" + return {"grade": grade, "score": score, "subs": subs} + + +def delta(current, prior, higher_is_worse=True, flat_pct=5.0): + d = {"value": current, "prior": prior, "delta_pct": None, "dir": "none", "good": None} + cur0 = current or 0 + pri0 = prior or 0 + if prior is None: + d["dir"] = "new" if cur0 > 0 else "none" + return d + if pri0 == 0: + d["dir"] = "new" if cur0 > 0 else "none" + return d + if current is None: + return d + dp = 100.0 * (current - prior) / float(prior) + d["delta_pct"] = dp + if abs(dp) < flat_pct: + d["dir"] = "flat" + return d + d["dir"] = "up" if dp > 0 else "down" + d["good"] = (d["dir"] == "down") == bool(higher_is_worse) + return d + + +# ---------------- the bundle ---------------- + + +def build_bundle(rows, active_now, shelved_count, start_ms, end_ms, now_ms, opts=None): + o = opts_merged(opts) + try: + start_ms = int(start_ms or 0) + end_ms = int(end_ms or 0) + now_ms = int(now_ms or 0) + except Exception: + start_ms, end_ms, now_ms = 0, 0, 0 + filters = (opts or {}).get("filters") if opts else None + instances, mode = correlate_instances(rows) + instances = apply_filters(instances, filters) + current, prior, orphans = partition_instances(instances, start_ms, end_ms) + window_h = _safe_div(end_ms - start_ms, 3600000.0) or 0.0 + + bins = rate_bins(current, start_ms, end_ms, o) + floods = flood_episodes(bins, current, start_ms, end_ms, o) + chat = chattering(current, start_ms, end_ms, o) + fleet = fleeting(current, o) + open_current = [i for i in current if i["open"]] + stand = standing(active_now, open_current, now_ms, o) + mm = mtta_mttr(current, start_ms, end_ms, o) + par = pareto(current, o) + prio = priority_distribution(current, o) + heat = heatmap(current, o) + + # prior mirrors (cheap subset for deltas) + p_start = start_ms - (end_ms - start_ms) + p_bins = rate_bins(prior, p_start, start_ms, o) + p_floods = flood_episodes(p_bins, prior, p_start, start_ms, o) + p_chat = chattering(prior, p_start, start_ms, o) + p_fleet = fleeting(prior, o) + p_mm = mtta_mttr(prior, p_start, start_ms, o) + + n_act = len(current) + p_act = len(prior) + rate = _safe_div(n_act, window_h) + p_rate = _safe_div(p_act, window_h) + unacked_now = len([s for s in active_now or [] if s.get("unacked")]) + flat = o["delta_flat_pct"] + + kpis = { + "activations": delta(n_act, p_act, True, flat), + "rate_per_hr": delta(rate, p_rate, True, flat), + "active_now": delta(len(active_now or []), None, True, flat), + "unacked_now": delta(unacked_now, None, True, flat), + "shelved_now": delta(shelved_count, None, True, flat), + "mtta_ms": delta(mm["mtta"]["median_ms"], p_mm["mtta"]["median_ms"], True, flat), + "mttr_ms": delta(mm["mttr"]["median_ms"], p_mm["mttr"]["median_ms"], True, flat), + "flood_pct": delta(floods["pct_time_in_flood"], p_floods["pct_time_in_flood"], True, flat), + "flood_count": delta(len(floods["episodes"]), len(p_floods["episodes"]), True, flat), + "chatter_count": delta(len(chat), len(p_chat), True, flat), + "standing_count": delta(stand["count"], None, True, flat), + "fleeting_count": delta(fleet["total"], p_fleet["total"], True, flat), + } + + has_activity = n_act > 0 + health = health_score(rate if has_activity else None, + floods["pct_time_in_flood"] if has_activity else None, + len(chat), stand["count"] if (active_now or open_current) or has_activity else None, + prio["sum_abs_dev"], has_activity, o) + if not has_activity and not (active_now or []): + health = {"grade": None, "score": None, "subs": health["subs"]} + + areas = sorted(set([i["area"] for i in current] + + [r["area"] for r in stand["rows"]])) + prios_seen = sorted(set([(i["priority"], i["priority_name"]) for i in current])) + active_now_rows = [] + for s in (active_now or [])[:int(o["active_now_cap"])]: + parsed = parse_source(s.get("source"), s.get("display_path")) + a = s.get("active_ms") + active_now_rows.append({"source": s.get("source") or "", "label": parsed["label"], + "area": parsed["area"], + "priority_name": s.get("priority_name", "Unknown"), + "active_ms": a, + "age_ms": (now_ms - a) if a is not None else None, + "unacked": bool(s.get("unacked"))}) + + return { + "meta": {"start_ms": start_ms, "end_ms": end_ms, "prior_start_ms": p_start, + "now_ms": now_ms, "window_hours": window_h, + "event_count": len(rows or []), "activation_count": n_act, + "dropped_rows": 0, "truncated": False, + "correlation_mode": mode, "standing_mode": stand["mode"], + "areas": areas, + "priorities_seen": [{"value": p[0], "label": p[1]} for p in prios_seen], + "source_count": len(set(i["source"] for i in current)), + "error": None, "opts": { + "bin_ms": bins["bin_ms"], "target_per_hr": o["target_per_hr"], + "flood_per_10min": o["flood_per_10min"], + "chatter_per_hr": o["chatter_per_hr"], + "fleeting_s": o["fleeting_s"], "standing_hours": o["standing_hours"]}}, + "kpis": kpis, + "health": health, + "insights": [], + "rate": bins, + "floods": floods, + "priority": prio, + "heatmap": heat, + "mtta_mttr": mm, + "pareto": par, + "top_sources": top_sources(par, 5), + "chattering": chat, + "fleeting": fleet, + "standing": stand, + "active_now": active_now_rows, + } diff --git a/ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/calc/resource.json b/ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/calc/resource.json new file mode 100644 index 0000000..e160ebe --- /dev/null +++ b/ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/calc/resource.json @@ -0,0 +1,12 @@ +{ + "scope": "A", + "version": 1, + "restricted": false, + "overridable": true, + "files": [ + "code.py" + ], + "attributes": { + "hintScope": 2 + } +} diff --git a/ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/fmt/code.py b/ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/fmt/code.py new file mode 100644 index 0000000..c9c2f00 --- /dev/null +++ b/ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/fmt/code.py @@ -0,0 +1,325 @@ +# PrimeControls.fmt - PURE formatting, insights, shift report (py2/3, no f-strings). +from __future__ import division, print_function + +import time as _time + +from PrimeControls import calc + +ARROW_UP = u"▲" +ARROW_DOWN = u"▼" +ARROW_FLAT = u"—" + + +def dur(ms): + if ms is None: + return "--" + try: + s = int(round(ms / 1000.0)) + except Exception: + return "--" + if s < 60: + return "%ds" % s + m = s // 60 + if m < 60: + return "%dm %ds" % (m, s % 60) if s % 60 else "%dm" % m + h = m // 60 + if h < 24: + return "%dh %dm" % (h, m % 60) if m % 60 else "%dh" % h + d = h // 24 + return "%dd %dh" % (d, h % 24) if h % 24 else "%dd" % d + + +def num(v, dec=0): + if v is None: + return "--" + try: + if dec: + s = ("%%.%df" % dec) % v + whole, frac = s.split(".") + return _group(whole) + "." + frac + return _group("%d" % round(v)) + except Exception: + return str(v) + + +def _group(digits): + neg = digits.startswith("-") + if neg: + digits = digits[1:] + out = "" + while len(digits) > 3: + out = "," + digits[-3:] + out + digits = digits[:-3] + return ("-" if neg else "") + digits + out + + +def pct(v, dec=0, signed=False): + if v is None: + return "--" + try: + fmtstr = "%%+.%df%%%%" % dec if signed else "%%.%df%%%%" % dec + return fmtstr % v + except Exception: + return str(v) + + +def clock(ms): + if ms is None: + return "--" + try: + return _time.strftime("%I:%M %p", _time.localtime(ms / 1000.0)).lstrip("0") + except Exception: + return "--" + + +def day_clock(ms): + if ms is None: + return "--" + try: + return _time.strftime("%a ", _time.localtime(ms / 1000.0)) + clock(ms) + except Exception: + return "--" + + +def delta_chip(d): + if not d: + return {"arrow": ARROW_FLAT, "text": "--", "good": None} + direction = d.get("dir") + if direction == "up": + arrow = ARROW_UP + elif direction == "down": + arrow = ARROW_DOWN + else: + arrow = ARROW_FLAT + if direction == "new": + text = "new" + elif direction == "none": + text = "--" + elif d.get("delta_pct") is not None: + text = pct(d["delta_pct"], 0, True) + else: + text = "--" + return {"arrow": arrow, "text": text, "good": d.get("good")} + + +def grade_color_key(grade): + return grade if grade in ("A", "B", "C", "D", "F") else "NA" + + +# ---------------- insights ---------------- + + +def insights(bundle, opts=None): + o = calc.opts_merged(opts) + if not bundle: + return [] + out = [] + for rule in (_i_rate, _i_floods, _i_chatter, _i_standing, + _i_priority, _i_pareto, _i_fleeting): + try: + r = rule(bundle) + if r: + out.append(r) + except Exception: + pass + out.sort(key=lambda i: (i["severity"], -i.get("magnitude", 0))) + out = out[:int(o["insights_max"])] + if not out: + k = bundle.get("kpis", {}).get("rate_per_hr", {}) + rate = k.get("value") + if bundle.get("meta", {}).get("activation_count", 0) == 0: + text = "No alarm events in the selected range." + else: + text = ("No significant changes vs the prior period; rate %.1f/hr is within " + "the ISA target of %.0f/hr." % (rate or 0.0, + bundle["meta"]["opts"]["target_per_hr"])) + out = [{"severity": 3, "icon": "check_circle", "text": text, + "tab": "overview", "magnitude": 0}] + for i in out: + i.pop("magnitude", None) + return out + + +def _i_rate(b): + k = b["kpis"]["activations"] + dp = k.get("delta_pct") + cur = k.get("value") or 0 + if dp is None or cur < 10: + return None + if dp >= 25: + sev = 0 if dp >= 100 else 1 + text = ("Alarm rate up %d%% vs prior period (%s -> %s activations)" + % (round(dp), num(k.get("prior")), num(cur))) + attr = _rate_attribution(b) + if attr: + text += " - largest increase from %s" % attr + return {"severity": sev, "icon": "trending_up", "text": text + ".", + "tab": "badactors", "magnitude": dp} + if dp <= -25: + return {"severity": 3, "icon": "trending_down", + "text": "Alarm rate down %d%% vs prior period - improving." % round(-dp), + "tab": "overview", "magnitude": -dp} + return None + + +def _rate_attribution(b): + rows = b.get("pareto", {}).get("rows") or [] + if rows: + return rows[0]["label"] + return None + + +def _i_floods(b): + eps = b.get("floods", {}).get("episodes") or [] + if not eps: + return None + worst = sorted(eps, key=lambda e: -e["event_count"])[0] + total_min = sum(e["duration_ms"] for e in eps) / 60000.0 + text = ("%d flood episode(s) totaling %s; worst %s-%s (%d events, mostly %s)." + % (len(eps), dur(total_min * 60000), day_clock(worst["start_ms"]), + clock(worst["end_ms"]), worst["event_count"], + worst["top_source_label"] or worst["top_area"] or "n/a")) + return {"severity": 0, "icon": "flood", "text": text, "tab": "analysis", + "magnitude": len(eps)} + + +def _i_chatter(b): + rows = b.get("chattering") or [] + if not rows: + return None + top = rows[0] + if len(rows) == 1: + text = ("%s is chattering (%.0f re-triggers/hr, median gap %.0fs)." + % (top["label"], top["per_hour"], top["median_gap_s"])) + else: + text = ("%d alarms are chattering (>%.0f re-triggers/hr); worst is %s at %.0f/hr." + % (len(rows), b["meta"]["opts"]["chatter_per_hr"], top["label"], + top["per_hour"])) + return {"severity": 1, "icon": "repeat", "text": text, "tab": "badactors", + "magnitude": len(rows)} + + +def _i_standing(b): + st = b.get("standing") or {} + if not st.get("count"): + return None + rows = st.get("rows") or [] + oldest = rows[0] if rows else None + text = "%d alarm(s) standing >%dh" % (st["count"], b["meta"]["opts"]["standing_hours"]) + if oldest and oldest.get("age_h"): + text += "; oldest is %s at %.1f days" % (oldest["label"], oldest["age_h"] / 24.0) + return {"severity": 1, "icon": "schedule", "text": text + ".", "tab": "badactors", + "magnitude": st["count"]} + + +def _i_priority(b): + p = b.get("priority") or {} + if p.get("sum_abs_dev") is None: + return None + high = p["pct"]["high"] + target = p["target"]["high"] + if high - target >= 10: + return {"severity": 2, "icon": "priority_high", + "text": ("High/Critical alarms are %d%% of events vs %d%% ISA target - " + "priorities may be over-assigned." % (round(high), round(target))), + "tab": "analysis", "magnitude": high - target} + return None + + +def _i_pareto(b): + rows = b.get("pareto", {}).get("rows") or [] + for k, row in enumerate(rows[:3]): + if row["cum_pct"] >= 50: + n = k + 1 + return {"severity": 2, "icon": "insights", + "text": ("%d source(s) account for %d%% of all activations - start " + "with %s." % (n, round(row["cum_pct"]), rows[0]["label"])), + "tab": "badactors", "magnitude": row["cum_pct"]} + return None + + +def _i_fleeting(b): + fl = b.get("fleeting") or {} + if (fl.get("total") or 0) < 10: + return None + return {"severity": 2, "icon": "bolt", + "text": ("%d fleeting alarms (<%ds active) in range - likely deadband or " + "nuisance issues." % (fl["total"], b["meta"]["opts"]["fleeting_s"])), + "tab": "badactors", "magnitude": fl["total"]} + + +# ---------------- shift report ---------------- + + +def shift_bounds(now_ms, shift_hours=8, anchor_hour=0): + lt = _time.localtime(now_ms / 1000.0) + midnight = _time.mktime((lt.tm_year, lt.tm_mon, lt.tm_mday, 0, 0, 0, + lt.tm_wday, lt.tm_yday, -1)) + anchor = midnight + anchor_hour * 3600 + now_s = now_ms / 1000.0 + if now_s < anchor: + anchor -= 86400 + shift_s = shift_hours * 3600 + k = int((now_s - anchor) // shift_s) + cur_start = anchor + k * shift_s + cur = (int(cur_start * 1000), int((cur_start + shift_s) * 1000)) + prev = (int((cur_start - shift_s) * 1000), int(cur_start * 1000)) + label = "%s - %s" % (day_clock(prev[0]), day_clock(prev[1])) + return {"current": cur, "previous": prev, "label": label} + + +def _kpi_line(name, k, formatter): + chip = delta_chip(k) + val = formatter(k.get("value")) + extra = "" if chip["text"] == "--" else " (%s %s)" % (chip["arrow"], chip["text"]) + return "%-14s %s%s" % (name + ":", val, extra) + + +def shift_report(bundle, label): + if not bundle: + return {"text": "No data.", "sections": []} + k = bundle["kpis"] + h = bundle["health"] + subs = " | ".join("%s %s" % (s["key"], "%d" % s["score"] if s["score"] is not None else "n/a") + for s in h.get("subs", [])) + sections = [] + head = ["Shift: %s" % label, + "Generated: %s" % day_clock(bundle["meta"]["now_ms"]), + "Health: %s (%s) [%s]" % (h.get("grade") or "N/A", + "%d" % h["score"] if h.get("score") is not None else "-", + subs)] + sections.append({"heading": "PRIME CONTROLS - ALARM SHIFT REPORT", "lines": head}) + kpis = [ + _kpi_line("Activations", k["activations"], num), + _kpi_line("Rate/hr", k["rate_per_hr"], lambda v: num(v, 1)), + _kpi_line("MTTA", k["mtta_ms"], dur), + _kpi_line("MTTR", k["mttr_ms"], dur), + _kpi_line("Flood time", k["flood_pct"], lambda v: pct(v, 1)), + _kpi_line("Chattering", k["chatter_count"], num), + _kpi_line("Standing", k["standing_count"], num), + _kpi_line("Fleeting", k["fleeting_count"], num), + ] + sections.append({"heading": "KPIs (vs prior shift)", "lines": kpis}) + actors = ["#%d %-40s %6s %s" % (r["rank"], r["label"][:40], num(r["count"]), + pct(r["cum_pct"])) + for r in bundle["pareto"]["rows"][:5]] or ["none"] + sections.append({"heading": "Top actors (count, cumulative %)", "lines": actors}) + floods = ["%s - %s %s, %d events, top: %s" + % (day_clock(e["start_ms"]), clock(e["end_ms"]), dur(e["duration_ms"]), + e["event_count"], e["top_source_label"] or e["top_area"]) + for e in bundle["floods"]["episodes"]] or ["none"] + sections.append({"heading": "Flood episodes", "lines": floods}) + standing_rows = bundle["standing"]["rows"][:8] + stand = ["%-40s active %s%s" % (r["label"][:40], dur(r["age_ms"]), + " (UNACKED)" if r["unacked"] else "") + for r in standing_rows] or ["none"] + sections.append({"heading": "Standing / handover attention", "lines": stand}) + sections.append({"heading": "Insights", + "lines": [i["text"] for i in bundle.get("insights") or []] or ["none"]}) + text_parts = [] + for s in sections: + text_parts.append(s["heading"]) + text_parts.append("-" * len(s["heading"])) + text_parts.extend(s["lines"]) + text_parts.append("") + return {"text": "\n".join(text_parts), "sections": sections} diff --git a/ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/fmt/resource.json b/ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/fmt/resource.json new file mode 100644 index 0000000..e160ebe --- /dev/null +++ b/ignition/gateway/projects/PrimeBAT/ignition/script-python/PrimeControls/fmt/resource.json @@ -0,0 +1,12 @@ +{ + "scope": "A", + "version": 1, + "restricted": false, + "overridable": true, + "files": [ + "code.py" + ], + "attributes": { + "hintScope": 2 + } +} diff --git a/ignition/gateway/projects/SimHarness/ignition/script-python/probe/code.py b/ignition/gateway/projects/SimHarness/ignition/script-python/probe/code.py index 56d2618..b6b88cb 100644 --- a/ignition/gateway/projects/SimHarness/ignition/script-python/probe/code.py +++ b/ignition/gateway/projects/SimHarness/ignition/script-python/probe/code.py @@ -11,7 +11,60 @@ import json OUT_DIR = "/usr/local/bin/ignition/data/projects/.probe" -PROBE_NAME = "P0" +PROBE_NAME = "P1" + + +def _p1(): + """Exec PrimeBAT's real PrimeControls modules (calc/fmt/alarms) in this scope + and run getDashboardBundle against the live journal - end-to-end data-layer + smoke with zero dev footprint inside PrimeBAT.""" + import imp + base = "/usr/local/bin/ignition/data/projects/PrimeBAT/ignition/script-python/PrimeControls/" + pkg = imp.new_module("PrimeControls") + pkg.__path__ = [base] + sys_modules = __import__("sys").modules + sys_modules["PrimeControls"] = pkg + for name in ("calc", "fmt", "alarms"): + m = imp.new_module("PrimeControls." + name) + m.__dict__["system"] = system + code = open(base + name + "/code.py").read() + compiled = compile(code, name, "exec") + # py2 qualified-exec statement form: required because _p1 contains + # closures, which forbid the unqualified exec(...) call form in Jython. + exec compiled in m.__dict__ + sys_modules["PrimeControls." + name] = m + setattr(pkg, name, m) + al = pkg.alarms + end = system.date.toMillis(system.date.now()) + start = end - 4 * 3600 * 1000 + b = al.getDashboardBundle(start, end, None) + page = al.getJournalPage(start, end, None, 0, 5) + shift = al.getShiftReportText() + return { + "meta": {"event_count": b["meta"]["event_count"], + "activation_count": b["meta"]["activation_count"], + "areas": b["meta"]["areas"], + "correlation_mode": b["meta"]["correlation_mode"], + "standing_mode": b["meta"]["standing_mode"], + "error": b["meta"]["error"], + "dropped": b["meta"]["dropped_rows"]}, + "health": {"grade": b["health"]["grade"], "score": b["health"]["score"], + "subs": [(s["key"], s["score"]) for s in b["health"]["subs"]]}, + "kpis": dict((k, b["kpis"][k]["value"]) for k in b["kpis"]), + "counts": {"chattering": len(b["chattering"]), + "standing": b["standing"]["count"], + "fleeting": b["fleeting"]["total"], + "floods": len(b["floods"]["episodes"]), + "pareto_rows": len(b["pareto"]["rows"]), + "bins": len(b["rate"]["bins"]), + "active_now": len(b["active_now"])}, + "insights": [i["text"] for i in b["insights"]], + "journal_page": {"total": page["total"], "rows": len(page["rows"]), + "error": page["error"], + "first": page["rows"][0] if page["rows"] else None}, + "shift_label": shift["label"], + "shift_text_head": shift["text"][:400], + } def _safe(fn): @@ -174,6 +227,6 @@ def _p0(): def run(): result = {"probe": PROBE_NAME, "ranAt": str(system.date.now())} - result["payload"] = _safe(_p0) + result["payload"] = _safe(_p1) _write(result) system.util.getLogger("probe").info("probe %s written to %s" % (PROBE_NAME, OUT_DIR)) diff --git a/ignition/gateway/projects/SimHarness/ignition/timer/ProbeTick/resource.json b/ignition/gateway/projects/SimHarness/ignition/timer/ProbeTick/resource.json index 1b34bba..943ccb0 100644 --- a/ignition/gateway/projects/SimHarness/ignition/timer/ProbeTick/resource.json +++ b/ignition/gateway/projects/SimHarness/ignition/timer/ProbeTick/resource.json @@ -12,4 +12,4 @@ "fixedDelay": true, "enabled": false } -} +} \ No newline at end of file diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..272677a --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,21 @@ +"""Register PrimeControls.{calc,fmt} from the Ignition project's code.py files so +pytest imports them exactly as the gateway does (single source of truth).""" +import importlib.util +import os +import sys +import types + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +SP = os.path.join(ROOT, "ignition", "gateway", "projects", "PrimeBAT", + "ignition", "script-python", "PrimeControls") + +pkg = types.ModuleType("PrimeControls") +pkg.__path__ = [SP] +sys.modules["PrimeControls"] = pkg +for mod in ("calc", "fmt"): + spec = importlib.util.spec_from_file_location( + "PrimeControls." + mod, os.path.join(SP, mod, "code.py")) + m = importlib.util.module_from_spec(spec) + sys.modules["PrimeControls." + mod] = m + spec.loader.exec_module(m) + setattr(pkg, mod, m) diff --git a/tests/test_calc.py b/tests/test_calc.py new file mode 100644 index 0000000..60e9329 --- /dev/null +++ b/tests/test_calc.py @@ -0,0 +1,306 @@ +import os +import time + +import pytest + +from PrimeControls import calc + +H = 3600000 +M = 60000 +T0 = 1784200000000 # fixed epoch anchor + + +def row(state, ts, src="prov:default:/tag:Sim/AreaA/T1:/alm:Alm1", eid=None, + prio=3, dp="", system=False, ack_user=None): + lvl, name = calc.normalize_priority(prio) + return {"event_id": eid, "source": src, "display_path": dp, + "priority": lvl, "priority_name": name, "state": state, "ts": ts, + "is_system": system, "ack_user": ack_user} + + +def lifecycle(eid, src, t_active, dur_ms=None, ack_ms=None, prio=3): + rows = [row("active", t_active, src, eid, prio)] + if ack_ms is not None: + rows.append(row("ack", t_active + ack_ms, src, eid, prio, ack_user="op1")) + if dur_ms is not None: + rows.append(row("clear", t_active + dur_ms, src, eid, prio)) + return rows + + +# ---------------- parse_source / normalize_priority ---------------- + +def test_parse_source_from_source_path(): + p = calc.parse_source("prov:default:/tag:Plant/TankFarm/T101_Hi:/alm:T101_Hi", "") + assert p["area"] == "TankFarm" + assert p["label"] == "TankFarm/T101_Hi" + + +def test_parse_source_display_path_wins_label_but_area_from_tag(): + p = calc.parse_source("prov:default:/tag:Plant/Boiler/DrumLo:/alm:DrumLo", "North/Drum Level") + assert p["label"] == "North/Drum Level" + assert p["area"] == "Boiler" + + +def test_parse_source_system_row(): + p = calc.parse_source("evt:System Startup", "") + assert p["label"] == "evt:System Startup" + + +def test_normalize_priority(): + assert calc.normalize_priority(3) == (3, "High") + assert calc.normalize_priority("Critical") == (4, "Critical") + assert calc.normalize_priority("AlarmPriority.Low") == (1, "Low") + lvl, _ = calc.normalize_priority("weird") + assert lvl == -1 + + +# ---------------- correlation ---------------- + +def test_correlate_full_lifecycle_uuid(): + rows = lifecycle("e1", "s1", T0, dur_ms=5 * M, ack_ms=2 * M) + insts, mode = calc.correlate_instances(rows) + assert mode == "uuid" + assert len(insts) == 1 + i = insts[0] + assert i["active_ms"] == T0 and i["duration_ms"] == 5 * M and i["tta_ms"] == 2 * M + assert not i["open"] and i["ack_user"] == "op1" + + +def test_correlate_ack_after_clear(): + rows = [row("active", T0, eid="e1"), row("clear", T0 + M, eid="e1"), + row("ack", T0 + 2 * M, eid="e1")] + i = calc.correlate_instances(rows)[0][0] + assert i["duration_ms"] == M and i["tta_ms"] == 2 * M + + +def test_correlate_straddler_and_open(): + rows = [row("clear", T0, eid="old"), row("active", T0 + M, eid="new")] + insts, _ = calc.correlate_instances(rows) + by_id = dict((i["event_id"], i) for i in insts) + assert by_id["old"]["active_ms"] is None and by_id["old"]["duration_ms"] is None + assert by_id["new"]["open"] and by_id["new"]["clear_ms"] is None + + +def test_correlate_negative_skew_none(): + rows = [row("active", T0, eid="e1"), row("ack", T0 - M, eid="e1")] + i = calc.correlate_instances(rows)[0][0] + assert i["tta_ms"] is None + + +def test_correlate_system_dropped_and_empty(): + assert calc.correlate_instances([row("active", T0, system=True)])[0] == [] + assert calc.correlate_instances([])[0] == [] + + +def test_correlate_source_sequence_fallback(): + rows = [row("active", T0, src="sA"), row("clear", T0 + M, src="sA"), + row("active", T0 + 2 * M, src="sA"), row("active", T0 + 3 * M, src="sB")] + insts, mode = calc.correlate_instances(rows) + assert mode == "source-sequence" + assert len(insts) == 3 + closed = [i for i in insts if i["source"] == "sA" and not i["open"]] + assert len(closed) == 1 and closed[0]["duration_ms"] == M + + +def test_partition_boundary(): + insts = [dict(active_ms=T0, source="a"), dict(active_ms=T0 - 1, source="b"), + dict(active_ms=None, source="c")] + cur, prior, orphans = calc.partition_instances(insts, T0, T0 + H) + assert len(cur) == 1 and len(prior) == 1 and len(orphans) == 1 + + +# ---------------- bins / floods ---------------- + +def make_insts(times, src="s1", prio=3): + out = [] + for i, t in enumerate(times): + rows = lifecycle("e%d_%s" % (i, src), src, t, dur_ms=30000, prio=prio) + out.extend(rows) + insts, _ = calc.correlate_instances(out) + return insts + + +def test_rate_bins_placement_and_edges(): + start = (T0 // 600000) * 600000 + insts = make_insts([start, start + 600000 - 1, start + 600000]) + r = calc.rate_bins(insts, start, start + 2 * 600000) + assert r["bins"][0]["count"] == 2 and r["bins"][1]["count"] == 1 + + +def test_rate_bins_empty_and_reversed(): + assert calc.rate_bins([], T0, T0)["bins"] == [] + assert calc.rate_bins([], T0 + 1, T0)["bins"] == [] + + +def test_flood_hysteresis_merge_and_split(): + start = (T0 // 600000) * 600000 + times = [] + times += [start + i for i in range(12)] # bin0: 12 (flood) + times += [start + 600000 + i for i in range(7)] # bin1: 7 (>=5 continues) + times += [start + 2 * 600000 + i for i in range(12)] # bin2: 12 + insts = make_insts(times) + r = calc.rate_bins(insts, start, start + 3 * 600000) + f = calc.flood_episodes(r, insts, start, start + 3 * 600000) + assert len(f["episodes"]) == 1 + times2 = [start + i for i in range(12)] + [start + 2 * 600000 + i for i in range(12)] + insts2 = make_insts(times2) + r2 = calc.rate_bins(insts2, start, start + 3 * 600000) + f2 = calc.flood_episodes(r2, insts2, start, start + 3 * 600000) + assert len(f2["episodes"]) == 2 + assert f2["episodes"][0]["event_count"] == 12 + assert f2["pct_time_in_flood"] > 0 + + +# ---------------- detections ---------------- + +def test_chattering_detects_sim_shape_not_slow(): + start = T0 + end = T0 + 2 * H + chatter_times = [start + i * 45000 for i in range(160)] # every 45s + slow_times = [start + i * 5 * M for i in range(24)] # every 5min -> 12/hr slow + insts = make_insts(chatter_times, src="chat") + make_insts(slow_times, src="slow") + out = calc.chattering(insts, start, end) + srcs = [r["source"] for r in out] + assert "chat" in srcs and "slow" not in srcs + + +def test_fleeting_and_standing(): + rows = [] + for i in range(4): + rows += lifecycle("f%d" % i, "fleety", T0 + i * M, dur_ms=3000) + rows += lifecycle("n1", "normal", T0, dur_ms=5 * M) + insts, _ = calc.correlate_instances(rows) + fl = calc.fleeting(insts) + assert fl["total"] == 4 and fl["sources"][0]["source"] == "fleety" + now = T0 + 30 * H + status = [{"source": "prov:default:/tag:A/B/C:/alm:C", "display_path": "", + "priority": 3, "priority_name": "High", + "active_ms": T0, "unacked": True}] + st = calc.standing(status, [], now) + assert st["mode"] == "status" and st["count"] == 1 + st2 = calc.standing([], [i for i in insts if i["open"]], now) + assert st2["mode"] == "journal" + + +def test_mtta_mttr_exclusions_and_median(): + rows = (lifecycle("a", "s", T0, dur_ms=2 * M, ack_ms=M) + + lifecycle("b", "s", T0 + M, dur_ms=4 * M, ack_ms=3 * M) + + lifecycle("c", "s", T0 + 2 * M, dur_ms=100 * M) + # no ack + lifecycle("d", "s", T0 + 3 * M)) # open, no clear + insts, _ = calc.correlate_instances(rows) + mm = calc.mtta_mttr(insts, T0, T0 + H) + assert mm["mtta"]["count"] == 2 and mm["mtta"]["median_ms"] == 2 * M + assert mm["mttr"]["count"] == 3 and mm["mttr"]["median_ms"] == 4 * M + + +def test_pareto_cum_pct_vs_total(): + insts = make_insts([T0 + i * M for i in range(6)], src="hot") + insts += make_insts([T0 + i * M for i in range(4)], src="warm") + p = calc.pareto(insts) + assert p["total_activations"] == 10 + assert p["rows"][0]["source"] == "hot" and abs(p["rows"][0]["cum_pct"] - 60.0) < 0.01 + assert abs(p["rows"][1]["cum_pct"] - 100.0) < 0.01 + assert calc.top_sources(p, 5) == p["rows"][:5] + + +def test_heatmap_placement(): + os.environ["TZ"] = "UTC" + time.tzset() + # 2026-07-16 is a Thursday; 10:00 UTC + ts = 1784196000000 # Thu Jul 16 2026 10:00:00 UTC + insts = make_insts([ts, ts, ts + H]) + hm = calc.heatmap(insts) + assert hm["rows"][3][10] == 2 and hm["rows"][3][11] == 1 + assert hm["max_count"] == 2 and hm["total"] == 3 + + +def test_priority_distribution_extremes(): + insts = make_insts([T0 + i * M for i in range(8)], src="lo", prio=1) + insts += make_insts([T0 + i * M for i in range(2)], src="hi", prio=4) + d = calc.priority_distribution(insts) + assert d["buckets"]["low"] == 8 and d["buckets"]["high"] == 2 + assert d["sum_abs_dev"] is not None + assert calc.priority_distribution([])["sum_abs_dev"] is None + + +# ---------------- health / delta ---------------- + +def test_health_anchors_and_grades(): + h = calc.health_score(6.0, 0.0, 0, 0, 0.0, True) + assert h["score"] == 100 and h["grade"] == "A" + assert calc.health_score(60.0, None, 0, 0, None, True)["subs"][0]["score"] == 0 + h2 = calc.health_score(None, 1.0, None, None, None, True) + flood_sub = [s for s in h2["subs"] if s["key"] == "flood"][0] + assert flood_sub["score"] == 80 + h3 = calc.health_score(None, None, None, None, None, True) + assert h3["score"] is None and h3["grade"] is None + + +def test_health_quiet_plant(): + h = calc.health_score(None, None, 0, 0, None, False) + assert h["score"] is not None and h["grade"] in ("A", "B") + + +def test_delta_semantics(): + assert calc.delta(10, 0)["dir"] == "new" + assert calc.delta(0, None)["dir"] == "none" + assert calc.delta(102, 100)["dir"] == "flat" + d = calc.delta(200, 100) + assert d["dir"] == "up" and d["good"] is False + d2 = calc.delta(50, 100, higher_is_worse=True) + assert d2["dir"] == "down" and d2["good"] is True + + +# ---------------- bundle ---------------- + +def synth_rows(): + rows = [] + base = T0 + for i in range(20): # baseline across areas/priorities + src = "prov:default:/tag:Plant/Area%d/Tag%d:/alm:A%d" % (i % 4, i, i) + rows += lifecycle("b%d" % i, src, base + i * 8 * M, dur_ms=3 * M, + ack_ms=M, prio=(i % 3) + 1) + chat_src = "prov:default:/tag:Plant/Area1/Chatty:/alm:Chatty" + for i in range(100): # every 50s for ~83 min -> ~25/hr over the 4h window + rows += lifecycle("c%d" % i, chat_src, base + i * 50000, dur_ms=8000, prio=3) + flood_t = base + 3 * H + for i in range(18): + rows += lifecycle("fl%d" % i, "prov:default:/tag:Plant/Area2/F%d:/alm:F%d" % (i, i), + flood_t + i * 20000, dur_ms=M, prio=2) + for i in range(5): + rows += lifecycle("fle%d" % i, "prov:default:/tag:Plant/Area3/Blip:/alm:Blip", + base + i * 12 * M, dur_ms=4000, prio=1) + return rows + + +def test_build_bundle_golden_and_empty(): + start, end = T0, T0 + 4 * H + status = [{"source": "prov:default:/tag:Plant/Area0/Stand:/alm:Stand", + "display_path": "", "priority": 3, "priority_name": "High", + "active_ms": T0 - 30 * H, "unacked": True}] + b = calc.build_bundle(synth_rows(), status, 0, start, end, end + M) + for key in ("meta", "kpis", "health", "rate", "floods", "priority", "heatmap", + "mtta_mttr", "pareto", "top_sources", "chattering", "fleeting", + "standing", "active_now", "insights"): + assert key in b, key + assert b["meta"]["activation_count"] > 0 + assert len(b["floods"]["episodes"]) >= 1 + assert any(r["source"].endswith("Chatty") for r in b["chattering"]) + assert b["standing"]["count"] == 1 + assert b["fleeting"]["total"] >= 5 + assert b["health"]["grade"] is not None + assert b["kpis"]["activations"]["value"] > 0 + assert len(b["meta"]["areas"]) >= 4 + empty = calc.build_bundle([], [], None, start, end, end) + assert empty["meta"]["activation_count"] == 0 + assert empty["health"]["grade"] is None + assert empty["pareto"]["rows"] == [] and empty["rate"]["bins"] != [] + + +def test_build_bundle_filters(): + start, end = T0, T0 + 4 * H + b = calc.build_bundle(synth_rows(), [], None, start, end, end, + {"filters": {"areas": ["Area2"]}}) + assert all(a == "Area2" for a in b["meta"]["areas"]) + # Area2 = 5 baseline rows (i % 4 == 2) + 18 flood rows + assert b["meta"]["activation_count"] == 23 diff --git a/tests/test_fmt.py b/tests/test_fmt.py new file mode 100644 index 0000000..377e985 --- /dev/null +++ b/tests/test_fmt.py @@ -0,0 +1,94 @@ +import os +import time + +from PrimeControls import calc, fmt + +H = 3600000 +M = 60000 +T0 = 1784200000000 + + +def test_dur_num_pct(): + assert fmt.dur(None) == "--" + assert fmt.dur(42000) == "42s" + assert fmt.dur(90 * M) == "1h 30m" + assert fmt.dur(26 * H) == "1d 2h" + assert fmt.num(None) == "--" + assert fmt.num(1234567) == "1,234,567" + assert fmt.num(12.345, 1) == "12.3" + assert fmt.pct(34.2) == "34%" + assert fmt.pct(34.2, 0, True) == "+34%" + + +def test_delta_chip(): + assert fmt.delta_chip(calc.delta(200, 100))["arrow"] == fmt.ARROW_UP + assert fmt.delta_chip(calc.delta(50, 100))["arrow"] == fmt.ARROW_DOWN + assert fmt.delta_chip(calc.delta(10, 0))["text"] == "new" + assert fmt.delta_chip(None)["text"] == "--" + + +def bundle_with(rows=None, status=None): + return calc.build_bundle(rows or [], status or [], 0, T0, T0 + 4 * H, T0 + 4 * H) + + +def lifecycle(eid, src, t, dur_ms=None, ack_ms=None, prio=3): + lvl, name = calc.normalize_priority(prio) + rows = [{"event_id": eid, "source": src, "display_path": "", "priority": lvl, + "priority_name": name, "state": "active", "ts": t, "is_system": False, + "ack_user": None}] + if dur_ms: + rows.append(dict(rows[0], state="clear", ts=t + dur_ms)) + return rows + + +def test_insights_fallback_and_cap(): + b = bundle_with() + ins = fmt.insights(b) + assert len(ins) == 1 and ins[0]["severity"] == 3 + assert "No alarm events" in ins[0]["text"] + + +def test_insights_chatter_and_flood_fire(): + rows = [] + for i in range(100): # ~25/hr over the 4h window, median gap 50s + rows += lifecycle("c%d" % i, "prov:default:/tag:P/A/Chat:/alm:Chat", + T0 + i * 50000, dur_ms=8000) + for i in range(15): + rows += lifecycle("f%d" % i, "prov:default:/tag:P/B/F%d:/alm:F%d" % (i, i), + T0 + 2 * H + i * 30000, dur_ms=M) + b = bundle_with(rows) + b["insights"] = fmt.insights(b) + texts = " | ".join(i["text"] for i in b["insights"]) + assert "chatter" in texts.lower() + assert "flood" in texts.lower() + assert len(b["insights"]) <= 5 + + +def test_shift_bounds_grid(): + os.environ["TZ"] = "UTC" + time.tzset() + # Thu Jul 16 2026 14:20 UTC + now = 1784211600000 + 20 * M + sb = fmt.shift_bounds(now, 8, 0) + cs, ce = sb["current"] + ps, pe = sb["previous"] + assert ce - cs == 8 * H and pe == cs + lt = time.localtime(cs / 1000.0) + assert lt.tm_hour in (0, 8, 16) + sb2 = fmt.shift_bounds(now, 12, 6) + lt2 = time.localtime(sb2["current"][0] / 1000.0) + assert lt2.tm_hour in (6, 18) + + +def test_shift_report_sections(): + rows = [] + for i in range(10): + rows += lifecycle("b%d" % i, "prov:default:/tag:P/A/T%d:/alm:T%d" % (i, i), + T0 + i * 10 * M, dur_ms=M) + b = bundle_with(rows) + b["insights"] = fmt.insights(b) + rep = fmt.shift_report(b, "Thu 6:00 AM - Thu 2:00 PM") + heads = [s["heading"] for s in rep["sections"]] + assert any("SHIFT REPORT" in h for h in heads) + assert any("KPIs" in h for h in heads) + assert "Activations" in rep["text"]