Multi-Sensor Correlation to Reduce False Positives

"""Multi-sensor correlation engine for pharmaceutical cold chain excursion gating.

Consumes aligned telemetry, suppresses single-probe false positives via
deterministic hard gates plus probabilistic consensus, and emits a
tamper-evident, hash-chained decision record for every evaluation.
"""
from __future__ import annotations

import hashlib
import json
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from decimal import Decimal, ROUND_HALF_UP
from enum import Enum
from typing import Deque, Mapping, Sequence


class Decision(str, Enum):
    CONFIRMED = "CONFIRMED"
    SUPPRESSED = "SUPPRESSED"


class HardGate(str, Enum):
    NONE = "NONE"
    DOOR = "DOOR"
    VIBRATION = "VIBRATION"
    DEFROST = "DEFROST"


@dataclass(frozen=True)
class CorrelationConfig:
    """Versioned correlation parameters. The fingerprint of this object is
    bound to every record to satisfy change control over decision logic
    (21 CFR 11.10(k))."""
    window_seconds: int = 90
    confidence_threshold: Decimal = Decimal("0.70")
    min_corroborating_sensors: int = 2
    # Per-sensor vote weight: reliability x calibration x spatial proximity,
    # risk-assessed and documented under ICH Q9 quality risk management.
    sensor_weights: Mapping[str, Decimal] = field(
        default_factory=lambda: {
            "rtd_a": Decimal("1.00"),
            "rtd_b": Decimal("0.95"),
            "rtd_c": Decimal("0.90"),
        }
    )
    vibration_g_threshold: Decimal = Decimal("0.80")

    def fingerprint(self) -> str:
        # Canonical, sorted serialisation so the fingerprint is reproducible
        # on any host (11.10(a) validated, reproducible system behaviour).
        payload = {
            "window_seconds": self.window_seconds,
            "confidence_threshold": str(self.confidence_threshold),
            "min_corroborating_sensors": self.min_corroborating_sensors,
            "sensor_weights": {k: str(v) for k, v in sorted(self.sensor_weights.items())},
            "vibration_g_threshold": str(self.vibration_g_threshold),
        }
        canon = json.dumps(payload, separators=(",", ":"), sort_keys=True)
        return hashlib.sha256(canon.encode("utf-8")).hexdigest()


@dataclass(frozen=True)
class Reading:
    """A single aligned telemetry sample (timestamps already UTC-normalised
    upstream by the time-series alignment layer)."""
    sensor_id: str
    deviation_c: Decimal          # signed distance outside the in-band envelope
    ts: datetime                  # timezone-aware UTC
    door_open: bool = False
    compressor_active: bool = False
    vibration_g: Decimal = Decimal("0")
    humidity_drop: bool = False


def _q(value: Decimal, places: str) -> Decimal:
    # Half-up quantisation gives a deterministic, inspector-reproducible value
    # for the persisted score (11.10(b) accurate and complete records).
    return value.quantize(Decimal(places), rounding=ROUND_HALF_UP)


class CorrelationEngine:
    def __init__(self, config: CorrelationConfig, genesis_hash: str = "0" * 64):
        self.config = config
        self._buffer: Deque[Reading] = deque()
        self._prev_hash = genesis_hash
        self._config_version = config.fingerprint()

    def _prune(self, now: datetime) -> None:
        horizon = now - timedelta(seconds=self.config.window_seconds)
        while self._buffer and self._buffer[0].ts < horizon:
            self._buffer.popleft()

    def _hard_gate(self, primary: Reading) -> HardGate:
        # Deterministic vetoes: an unambiguous physical cause invalidates the
        # excursion without a probabilistic vote (11.10(a) defensible logic).
        if primary.door_open and primary.compressor_active:
            return HardGate.DOOR
        if primary.vibration_g >= self.config.vibration_g_threshold:
            return HardGate.VIBRATION
        if primary.humidity_drop and primary.compressor_active:
            return HardGate.DEFROST
        return HardGate.NONE

    def _consensus(self, primary: Reading) -> tuple[Decimal, int]:
        """Weighted agreement that the deviation is real. A corroborating
        sensor is one whose own deviation has the same sign and at least half
        the primary magnitude within the window."""
        agree = Decimal("0")
        total = Decimal("0")
        corroborating = 0
        threshold_mag = abs(primary.deviation_c) / 2
        for r in self._buffer:
            if r.sensor_id == primary.sensor_id:
                continue
            weight = self.config.sensor_weights.get(r.sensor_id, Decimal("0.50"))
            total += weight
            same_sign = (r.deviation_c >= 0) == (primary.deviation_c >= 0)
            if same_sign and abs(r.deviation_c) >= threshold_mag:
                agree += weight
                corroborating += 1
        if total == 0:
            return Decimal("0"), 0
        return _q(agree / total, "0.0001"), corroborating

    def evaluate(self, primary: Reading) -> dict:
        self._buffer.append(primary)
        self._prune(primary.ts)

        gate = self._hard_gate(primary)
        score, corroborating = self._consensus(primary)

        if gate is not HardGate.NONE:
            decision = Decision.SUPPRESSED
        elif corroborating < self.config.min_corroborating_sensors:
            # Insufficient corroboration is NOT evidence of a false positive.
            # Fail safe to CONFIRMED so absent data can never mask a real
            # excursion (ICH Q9 bounded false-negative risk; WHO GDP continuous
            # control during sensor dropout).
            decision = Decision.CONFIRMED
        elif score >= self.config.confidence_threshold:
            decision = Decision.CONFIRMED
        else:
            decision = Decision.SUPPRESSED

        return self._seal(primary, gate, score, corroborating, decision)

    def _seal(self, primary: Reading, gate: HardGate, score: Decimal,
              corroborating: int, decision: Decision) -> dict:
        # Append-only hash chain: every decision, confirmed or suppressed, is a
        # tamper-evident record linked to its predecessor (11.10(e) audit trail).
        record = {
            "event_id": hashlib.sha256(
                f"{primary.sensor_id}{primary.ts.isoformat()}".encode()
            ).hexdigest(),
            "primary_sensor_id": primary.sensor_id,
            "peak_deviation_c": str(_q(primary.deviation_c, "0.0001")),
            "corroborating_sensors": corroborating,
            "consensus_score": str(score),
            "hard_gate": gate.value,
            "decision": decision.value,
            "config_version": self._config_version,
            "evaluated_at": primary.ts.astimezone(timezone.utc).isoformat(),
            "prev_hash": self._prev_hash,
        }
        canon = json.dumps(record, separators=(",", ":"), sort_keys=True)
        record["record_hash"] = hashlib.sha256(canon.encode("utf-8")).hexdigest()
        self._prev_hash = record["record_hash"]
        return record


def verify_chain(records: Sequence[Mapping], genesis_hash: str = "0" * 64) -> bool:
    # Re-derive each hash to prove no record was edited after the fact, the
    # tamper-evidence check an inspector applies under 11.10(e).
    prev = genesis_hash
    for rec in records:
        if rec["prev_hash"] != prev:
            return False
        body = {k: v for k, v in rec.items() if k != "record_hash"}
        canon = json.dumps(body, separators=(",", ":"), sort_keys=True)
        if hashlib.sha256(canon.encode("utf-8")).hexdigest() != rec["record_hash"]:
            return False
        prev = rec["record_hash"]
    return True


if __name__ == "__main__":
    cfg = CorrelationConfig()
    engine = CorrelationEngine(cfg)
    now = datetime.now(timezone.utc)

    # Door-open ingress: hard-gate suppression.
    out = engine.evaluate(Reading(
        sensor_id="rtd_a", deviation_c=Decimal("2.5"), ts=now,
        door_open=True, compressor_active=True,
    ))
    print(out["decision"], out["hard_gate"])  # SUPPRESSED DOOR

import copy
from decimal import Decimal
from datetime import datetime, timezone

def test_door_event_is_suppressed():
    cfg = CorrelationConfig()
    engine = CorrelationEngine(cfg)
    rec = engine.evaluate(Reading(
        sensor_id="rtd_a", deviation_c=Decimal("2.5"),
        ts=datetime.now(timezone.utc),
        door_open=True, compressor_active=True,
    ))
    # Transient ingress must never reach CAPA (ICH Q9 proportional action).
    assert rec["decision"] == "SUPPRESSED"
    assert rec["hard_gate"] == "DOOR"

def test_chain_detects_tampering():
    cfg = CorrelationConfig()
    engine = CorrelationEngine(cfg)
    recs = [engine.evaluate(Reading("rtd_a", Decimal("3.0"),
            datetime.now(timezone.utc)))]
    assert verify_chain(recs) is True
    forged = copy.deepcopy(recs)
    forged[0]["decision"] = "SUPPRESSED"   # post-hoc edit
    # 11.10(e) tamper-evidence: any edit must break verification.
    assert verify_chain(forged) is False