BAT/tests/test_calc.py

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"]}})
    # Area2 = 5 baseline rows (i % 4 == 2) + 18 flood rows
    assert b["meta"]["activation_count"] == 23
    # dropdown options enumerate the UNfiltered data (never collapse)
    assert len(b["meta"]["areas"]) >= 4
    # event_count reflects only events of filtered instances
    b_all = calc.build_bundle(synth_rows(), [], None, start, end, end)
    assert 0 < b["meta"]["event_count"] < b_all["meta"]["event_count"]


def test_build_bundle_overview_stays_unfiltered():
    start, end = T0, T0 + 4 * H
    b_all = calc.build_bundle(synth_rows(), [], None, start, end, end)
    assert b_all["overview"] is None  # no filters -> Overview uses main bundle
    b = calc.build_bundle(synth_rows(), [], None, start, end, end,
                          {"filters": {"areas": ["Area2"]}})
    ov = b["overview"]
    assert ov is not None
    # main sections are filtered; overview mirrors the unfiltered plant view
    assert b["meta"]["activation_count"] == 23
    assert ov["meta"]["activation_count"] == b_all["meta"]["activation_count"]
    assert ov["meta"]["event_count"] == b_all["meta"]["event_count"]
    assert len(ov["pareto"]["rows"]) == len(b_all["pareto"]["rows"])
    assert ov.get("overview", None) is None  # no recursion


def test_build_bundle_filters_apply_to_status_kpis():
    start, end = T0, T0 + 4 * H
    status = [
        {"source": "prov:default:/tag:Plant/Area1/S1:/alm:S1", "display_path": "",
         "priority": 3, "priority_name": "High", "active_ms": T0, "unacked": True},
        {"source": "prov:default:/tag:Plant/Area2/S2:/alm:S2", "display_path": "",
         "priority": 1, "priority_name": "Low", "active_ms": T0, "unacked": False},
    ]
    b = calc.build_bundle(synth_rows(), status, 0, start, end, end,
                          {"filters": {"areas": ["Area2"]}})
    assert b["kpis"]["active_now"]["value"] == 1
    assert b["kpis"]["unacked_now"]["value"] == 0
    assert all(r["area"] == "Area2" for r in b["active_now"])


def test_apply_status_filters_semantics():
    rows = [
        {"source": "prov:default:/tag:Plant/Area1/Pump:/alm:Pump", "display_path": "",
         "priority": 4, "unacked": True},
        {"source": "prov:default:/tag:Plant/Area2/Valve:/alm:Valve", "display_path": "",
         "priority": 1, "unacked": False},
    ]
    assert calc.apply_status_filters(rows, None) == rows
    assert calc.apply_status_filters(rows, {"priorities": [4]}) == [rows[0]]
    assert calc.apply_status_filters(rows, {"states": ["unacked"]}) == [rows[0]]
    assert calc.apply_status_filters(rows, {"states": ["acked"]}) == [rows[1]]
    assert calc.apply_status_filters(rows, {"states": ["active"]}) == rows
    assert calc.apply_status_filters(rows, {"search": "valve"}) == [rows[1]]