murphy_unified/diagnostics.py

"""Physics threshold diagnostics for simulation scalars.

Seven checks covering flow, temperature, pressure, efficiency, and valve
behaviour.  Each triggered finding carries a severity level (CRITICAL or
WARNING) so the dashboard can surface the most urgent issues first.
"""

from __future__ import annotations

import operator
from typing import Any, Dict, List

CHECKS: List[Dict[str, Any]] = [
    {"name": "low_flow",         "field": "mdot_kgpm",        "op": "<", "threshold": 0.5,   "severity": "CRITICAL"},
    {"name": "high_temp",        "field": "Tc_peak_K",        "op": ">", "threshold": 200,   "severity": "WARNING"},
    {"name": "over_pressure",    "field": "pc_peak_barg",     "op": ">", "threshold": 960,   "severity": "CRITICAL"},
    {"name": "low_efficiency",   "field": "mass_eff",         "op": "<", "threshold": 0.5,   "severity": "WARNING"},
    {"name": "icv_stuck",        "field": "ICVmax_open_frac", "op": "<", "threshold": 0.01,  "severity": "CRITICAL"},
    {"name": "dcv_slow",         "field": "DCV_ct_s",         "op": ">", "threshold": 0.02,  "severity": "WARNING"},
    {"name": "excessive_blowby", "field": "mass_eff",         "op": "<", "threshold": 0.3,   "severity": "CRITICAL"},
]

_OPS = {
    "<": operator.lt,
    ">": operator.gt,
}


def evaluate(scalars: Dict[str, Any]) -> List[Dict[str, Any]]:
    """Run all threshold checks against *scalars* and return triggered findings.

    Checks whose ``field`` is absent from *scalars* are silently skipped. If
    multiple checks on the same ``field`` trigger, only the highest-severity
    one is reported (CRITICAL > WARNING).
    """
    triggered: List[Dict[str, Any]] = []
    for check in CHECKS:
        field = check["field"]
        if field not in scalars:
            continue
        actual = scalars[field]
        compare = _OPS[check["op"]]
        if compare(actual, check["threshold"]):
            triggered.append(
                {
                    "name": check["name"],
                    "severity": check["severity"],
                    "field": field,
                    "threshold": check["threshold"],
                    "actual": actual,
                }
            )

    # Collapse per-field: prefer CRITICAL over WARNING on the same field.
    _rank = {"CRITICAL": 2, "WARNING": 1}
    by_field: Dict[str, Dict[str, Any]] = {}
    for f in triggered:
        key = f["field"]
        if key in by_field:
            if _rank[f["severity"]] > _rank[by_field[key]["severity"]]:
                by_field[key] = f
        else:
            by_field[key] = f
    return list(by_field.values())