dashboard/data.py

from __future__ import annotations

from functools import lru_cache
from pathlib import Path

import numpy as np
import pandas as pd

TARGETS = ["mill_power_mw", "feed_rate_tph", "target_impurity_pct"]
TARGET_LABELS = {
    "mill_power_mw": "Mill Power (MW)",
    "feed_rate_tph": "Feed Rate (tph)",
    "target_impurity_pct": "Impurity (%)",
}
CONTEXT_WINDOWS = {
    "48 hours": 48,
    "96 hours (4 days)": 96,
    "168 hours (1 week)": 168,
}
HORIZONS = {
    "12 hours": 12,
    "24 hours": 24,
    "48 hours": 48,
}
RANDOM_SEED = 42
ASSETS_DIR = Path(__file__).resolve().parent.parent / "assets"
TELEMETRY_FILE = "telemetry_synthetic.parquet"


def generate_mineral_processing_telemetry(n_timestamps: int = 1000) -> pd.DataFrame:
    """Synthetic comminution/flotation plant telemetry."""
    rng = np.random.default_rng(RANDOM_SEED)
    timestamps = pd.date_range("2024-01-01", periods=n_timestamps, freq="h", tz="UTC")

    hour = np.arange(n_timestamps) % 24
    day = np.arange(n_timestamps) // 24

    shift_cycle = 0.6 * np.sin(2 * np.pi * hour / 24) + 0.25 * np.sin(2 * np.pi * hour / 12)
    hardness_drift = 0.15 * np.sin(2 * np.pi * day / 14)

    feed_rate_tph = 285 + 55 * shift_cycle + 20 * hardness_drift + rng.normal(0, 8, n_timestamps)
    feed_rate_tph = np.clip(feed_rate_tph, 180, 420)

    mill_power_mw = (
        7.5
        + 0.018 * feed_rate_tph
        + 2.2 * hardness_drift
        + 0.35 * shift_cycle
        + rng.normal(0, 0.25, n_timestamps)
    )
    mill_power_mw = np.clip(mill_power_mw, 5.0, 16.0)

    impurity_driver = (
        2.4
        - 0.004 * feed_rate_tph
        + 0.35 * hardness_drift
        + rng.normal(0, 0.08, n_timestamps)
    )
    target_impurity_pct = np.zeros(n_timestamps)
    target_impurity_pct[0] = impurity_driver[0]
    for t in range(1, n_timestamps):
        target_impurity_pct[t] = 0.82 * target_impurity_pct[t - 1] + 0.18 * impurity_driver[t]
    target_impurity_pct = np.clip(target_impurity_pct, 0.4, 4.5)

    return pd.DataFrame(
        {
            "timestamp": timestamps,
            "mill_power_mw": mill_power_mw.round(3),
            "feed_rate_tph": feed_rate_tph.round(2),
            "target_impurity_pct": target_impurity_pct.round(4),
        }
    )


def ensure_telemetry_asset() -> Path:
    """Persist synthetic telemetry to assets/ for fast Space startup."""
    ASSETS_DIR.mkdir(parents=True, exist_ok=True)
    path = ASSETS_DIR / TELEMETRY_FILE
    if not path.exists():
        generate_mineral_processing_telemetry().to_parquet(path, index=False)
    return path


@lru_cache(maxsize=1)
def load_telemetry() -> pd.DataFrame:
    path = ensure_telemetry_asset()
    frame = pd.read_parquet(path)
    frame["timestamp"] = pd.to_datetime(frame["timestamp"], utc=True)
    return frame.sort_values("timestamp").reset_index(drop=True)


def telemetry_summary(telemetry: pd.DataFrame) -> pd.DataFrame:
    rows = []
    for col in TARGETS:
        rows.append(
            {
                "Signal": TARGET_LABELS[col],
                "Rows": len(telemetry),
                "Start (UTC)": telemetry["timestamp"].min(),
                "End (UTC)": telemetry["timestamp"].max(),
                "Min": round(telemetry[col].min(), 4),
                "Max": round(telemetry[col].max(), 4),
                "Mean": round(telemetry[col].mean(), 4),
                "Std": round(telemetry[col].std(), 4),
            }
        )
    return pd.DataFrame(rows)


def correlation_matrix(telemetry: pd.DataFrame) -> pd.DataFrame:
    return telemetry[TARGETS].corr().round(3)


def build_context_frame(
    telemetry: pd.DataFrame,
    target: str,
    anchor_time: pd.Timestamp,
    context_steps: int,
) -> pd.DataFrame:
    history = telemetry.loc[telemetry["timestamp"] <= anchor_time].tail(context_steps)
    if len(history) < max(24, context_steps // 4):
        raise ValueError("Not enough history before the selected anchor time.")
    return pd.DataFrame(
        {
            "item_id": target,
            "timestamp": history["timestamp"],
            "target": history[target].astype(float),
        }
    )


def holdout_actuals(
    telemetry: pd.DataFrame,
    target: str,
    anchor_time: pd.Timestamp,
    horizon: int,
) -> pd.DataFrame:
    future = telemetry.loc[
        (telemetry["timestamp"] > anchor_time)
        & (telemetry["timestamp"] <= anchor_time + pd.Timedelta(hours=horizon))
    ]
    if future.empty:
        return pd.DataFrame(columns=["timestamp", "actual"])
    return future[["timestamp", target]].rename(columns={target: "actual"})


def compute_metrics(actual: pd.Series, predicted: pd.Series) -> dict[str, float]:
    diff = actual.values - predicted.values
    mae = float(np.mean(np.abs(diff)))
    rmse = float(np.sqrt(np.mean(diff**2)))
    mape = float(np.mean(np.abs(diff / np.clip(actual.values, 1e-6, None))) * 100)
    return {"mae": mae, "rmse": rmse, "mape_pct": mape}