realtime_pipeline.py

import argparse
import time
import numpy as np
import joblib
from collections import deque

import tensorflow as tf

from socket_client import run as socket_run
from detection import log_alert, severity_score

WINDOW_SIZE = 30
FEATURE_NAMES = ["cpu_usage_pct", "ram_usage_pct"]
SCALER_PATH = "sentinel_scaler.pkl"


class RealtimePipeline:
    def __init__(self, tflite_path, threshold):
        self.buffer = deque(maxlen=WINDOW_SIZE)
        self.threshold = threshold

        self.interpreter = tf.lite.Interpreter(model_path=tflite_path)
        self.interpreter.allocate_tensors()
        self.input_details = self.interpreter.get_input_details()
        self.output_details = self.interpreter.get_output_details()

        self.input_is_quantized = self.input_details[0]["dtype"] == np.int8
        if self.input_is_quantized:
            self.input_scale, self.input_zero_point = self.input_details[0][
                "quantization"
            ]

        self.scaler = self._load_scaler()

        self._n_inferences = 0
        self._latencies = []

    def _load_scaler(self):
        try:
            scaler = joblib.load(SCALER_PATH)
            print(f"[pipeline] scaler caricato da {SCALER_PATH}")
            return scaler
        except FileNotFoundError:
            print(
                f"[pipeline] WARNING: {SCALER_PATH} non trovato. "
                "Usando scaling identity (no normalizzazione)."
            )
            return None
        except Exception as e:
            print(f"[pipeline] WARNING: errore caricamento scaler: {e}")
            return None

    def _normalize(self, raw):
        if self.scaler is not None:
            return self.scaler.transform(raw)
        return raw

    def on_metric(self, data):
        self._last_data = data
        cpu = data.get("cpu_usage_pct", 0.0)
        ram = data.get("ram_usage_pct", 0.0)
        self.buffer.append([cpu, ram])

        if len(self.buffer) < WINDOW_SIZE:
            return

        self._run_inference()

    def _run_inference(self):
        start_time = time.perf_counter()

        raw_2d = np.array(self.buffer, dtype=np.float32)
        normalized_2d = self._normalize(raw_2d)
        sequence = normalized_2d.reshape(1, WINDOW_SIZE, 2).astype(np.float32)

        if self.input_is_quantized:
            sequence = (
                sequence / self.input_scale + self.input_zero_point
            ).astype(np.int8)

        self.interpreter.set_tensor(
            self.input_details[0]["index"], sequence
        )
        self.interpreter.invoke()
        output = self.interpreter.get_tensor(
            self.output_details[0]["index"]
        )

        if self.input_is_quantized:
            out_float = output.astype(np.float32)
            inp_float = sequence.astype(np.float32)
        else:
            out_float = output
            inp_float = sequence

        mse = float(np.mean(np.square(inp_float - out_float)))

        is_anomaly = mse > self.threshold
        severity = severity_score(np.array([mse]), self.threshold)[0]

        end_time = time.perf_counter()
        latency_ms = (end_time - start_time) * 1000
        self._latencies.append(latency_ms)
        self._n_inferences += 1

        last_data = getattr(self, "_last_data", {})
        ts = last_data.get("timestamp_ms", int(time.time() * 1000))

        status = "ANOMALIA" if is_anomaly else "OK"
        print(
            f"[pipeline] #{self._n_inferences:>4} | "
            f"MSE: {mse:.5f} | "
            f"soglia: {self.threshold:.5f} | "
            f"{status:>8} | "
            f"severità: {severity:>8} | "
            f"latenza: {latency_ms:.2f}ms"
        )

        if is_anomaly:
            event = {
                "timestamp_ms": ts,
                "reconstruction_error": mse,
                "threshold": self.threshold,
                "severity": severity,
                "ratio": mse / self.threshold if self.threshold > 0 else 0.0,
                "source": "realtime_pipeline",
                "inference_index": self._n_inferences,
            }
            log_alert(event)

    def on_disconnect(self):
        print("[pipeline] socket disconnesso — in attesa di riconnessione...")

    def summary(self):
        if self._latencies:
            print("\n" + "=" * 55)
            print("  RIEPILOGO PIPELINE")
            print("=" * 55)
            print(f"  Inferenze eseguite:    {self._n_inferences}")
            print(f"  Latenza media:         {np.mean(self._latencies):.2f} ms")
            print(f"  Latenza std:           {np.std(self._latencies):.2f} ms")
            print(f"  Latenza min:           {np.min(self._latencies):.2f} ms")
            print(f"  Latenza max:           {np.max(self._latencies):.2f} ms")
            print("=" * 55)


def compute_threshold():
    """Calcola la soglia di anomalia dal validation set.

    Il risultato viene stampato a schermo così l'utente può
    passarlo come argomento --threshold nelle esecuzioni successive,
    evitando di dover caricare il modello Keras ogni volta.
    """
    from detection import compute_threshold as det_threshold
    from evaluate import reconstruction_error
    from preprocessing import (
        load_and_segment,
        normalize_features,
        build_sequences,
        split_train_val,
    )

    print("[pipeline] calcolo soglia di anomalia dal validation set...")
    model = tf.keras.models.load_model("sentinel_autoencoder.keras")
    df = load_and_segment("sentinel_training.csv")
    df, scaler = normalize_features(df, ["cpu_usage_pct", "ram_usage_pct"])
    sequences = build_sequences(df, ["cpu_usage_pct", "ram_usage_pct"], WINDOW_SIZE)
    _, val_seqs = split_train_val(sequences)
    normal_errors = reconstruction_error(model, val_seqs)
    threshold = det_threshold(normal_errors)
    print(f"[pipeline] soglia calcolata: {threshold:.5f}")
    return threshold


def main():
    parser = argparse.ArgumentParser(
        description="Sentinel Realtime Pipeline — UDS → TFLite → Detection"
    )
    parser.add_argument(
        "--tflite",
        default="sentinel_fp16.tflite",
        help="Modello TFLite (default: sentinel_fp16.tflite)",
    )
    parser.add_argument(
        "--threshold",
        type=float,
        default=None,
        help="Soglia anomalia (default: calcolata automaticamente)",
    )
    args = parser.parse_args()

    threshold = args.threshold if args.threshold is not None else compute_threshold()

    pipeline = RealtimePipeline(args.tflite, threshold)

    print(f"[pipeline] pipeline avviata")
    print(f"[pipeline]   modello TFLite:      {args.tflite}")
    print(f"[pipeline]   soglia anomalia:     {threshold:.5f}")
    print(f"[pipeline]   WINDOW_SIZE:         {WINDOW_SIZE}")
    print(f"[pipeline]   finestra inferenza:  ogni nuovo campione")
    print(f"[pipeline] in attesa di {WINDOW_SIZE} campioni per la prima inferenza...")

    try:
        socket_run(
            on_metric=pipeline.on_metric,
            on_disconnect=pipeline.on_disconnect,
        )
    except KeyboardInterrupt:
        print("\n[pipeline] arresto richiesto (Ctrl+C)")
    finally:
        pipeline.summary()


if __name__ == "__main__":
    main()