src/anomaly_model.py

"""
anomaly_model.py
-----------------
Isolation Forest based anomaly detector for conveyor / crane motor sensor
streams (motor temperature, vibration, current draw, belt speed). This
powers the "Predictive Maintenance" tab -- flags abnormal equipment
behaviour before it causes an unplanned stoppage, which is exactly the kind
of workload Daifuku's intralogistics platforms (e.g. AS/RS, sorters, AGVs)
generate continuously in production.
"""

from dataclasses import dataclass

import joblib
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler

FEATURES = ["motor_temp_c", "vibration_mm_s", "current_amps", "belt_speed_mps"]


@dataclass
class AnomalyResult:
    is_anomaly: bool
    anomaly_score: float  # higher = more anomalous, roughly in [0, 1]
    raw_score: float


def build_model(contamination: float = 0.1, seed: int = 42) -> IsolationForest:
    return IsolationForest(
        n_estimators=200,
        contamination=contamination,
        random_state=seed,
    )


def score_reading(model: IsolationForest, scaler: StandardScaler, reading: dict) -> AnomalyResult:
    x = np.array([[reading[f] for f in FEATURES]])
    x_scaled = scaler.transform(x)
    raw = model.decision_function(x_scaled)[0]  # higher = more normal
    pred = model.predict(x_scaled)[0]  # 1 = normal, -1 = anomaly
    # squash raw decision_function (~[-0.5, 0.5]) into a 0-1 "anomaly score"
    anomaly_score = float(np.clip(0.5 - raw, 0, 1))
    return AnomalyResult(is_anomaly=(pred == -1), anomaly_score=anomaly_score, raw_score=float(raw))


def save_artifacts(model, scaler, model_path: str, scaler_path: str):
    joblib.dump(model, model_path)
    joblib.dump(scaler, scaler_path)


def load_artifacts(model_path: str, scaler_path: str):
    return joblib.load(model_path), joblib.load(scaler_path)