app.py

"""
Predictive Equipment Maintenance
Detect sensor anomalies and estimate maintenance urgency from time-series signals.
"""

from pathlib import Path

import numpy as np
import pandas as pd
import plotly.express as px
import streamlit as st


st.set_page_config(page_title="Predictive Equipment Maintenance", page_icon="⚙️", layout="wide")


def load_shared_css() -> None:
    current_dir = Path(__file__).resolve().parent
    candidates = [
        current_dir / "shared" / "styles.css",
        current_dir.parent / "shared" / "styles.css",
    ]
    css_path = next(path for path in candidates if path.exists())
    st.markdown(f"<style>{css_path.read_text(encoding='utf-8')}</style>", unsafe_allow_html=True)


load_shared_css()


def make_sensor_data(hours: int, drift: float, vibration_spike: float, seed: int = 7) -> pd.DataFrame:
    rng = np.random.default_rng(seed)
    t = np.arange(hours)
    temperature = 68 + 0.035 * t * drift + 2.2 * np.sin(t / 13) + rng.normal(0, 0.7, hours)
    vibration = 1.8 + 0.009 * t * drift + 0.4 * np.sin(t / 7) + rng.normal(0, 0.12, hours)
    pressure = 38 - 0.015 * t * drift + rng.normal(0, 0.45, hours)
    spike_start = max(0, int(hours * 0.72))
    vibration[spike_start:] += vibration_spike
    return pd.DataFrame({
        "hour": t,
        "temperature_c": temperature,
        "vibration_mm_s": vibration,
        "pressure_bar": pressure,
    })


def add_anomaly_scores(df: pd.DataFrame) -> pd.DataFrame:
    scored = df.copy()
    for column in ["temperature_c", "vibration_mm_s", "pressure_bar"]:
        rolling_mean = scored[column].rolling(24, min_periods=6).mean()
        rolling_std = scored[column].rolling(24, min_periods=6).std().replace(0, np.nan)
        scored[f"{column}_z"] = ((scored[column] - rolling_mean) / rolling_std).fillna(0).abs()
    scored["anomaly_score"] = scored[["temperature_c_z", "vibration_mm_s_z", "pressure_bar_z"]].mean(axis=1)
    scored["risk_band"] = pd.cut(
        scored["anomaly_score"],
        bins=[-0.1, 1.2, 2.0, 99],
        labels=["Normal", "Watch", "Intervene"],
    )
    return scored


def maintenance_summary(scored: pd.DataFrame):
    latest = scored.iloc[-1]
    recent = scored.tail(24)
    high_risk_hours = int((recent["risk_band"] == "Intervene").sum())
    if latest["anomaly_score"] >= 2.0 or high_risk_hours >= 4:
        action = "Schedule inspection within 24 hours."
    elif latest["anomaly_score"] >= 1.2:
        action = "Monitor closely and check lubrication, cooling, and mounting."
    else:
        action = "Continue normal operation and keep collecting telemetry."

    remaining_useful_life = max(12, int(220 - latest["anomaly_score"] * 55 - high_risk_hours * 6))
    return latest, high_risk_hours, remaining_useful_life, action


st.markdown("""
<div class="hero">
  <div class="hf-badge">Time-Series ML</div>
  <h1>⚙️ Predictive Equipment Maintenance</h1>
  <p>Explore anomaly scoring, remaining useful life estimation, and maintenance triage on sensor streams.</p>
  <div class="pill-row">
    <span class="hf-chip">Rolling z-scores</span>
    <span class="hf-chip">RUL estimate</span>
    <span class="hf-chip">Telemetry dashboard</span>
  </div>
</div>
""", unsafe_allow_html=True)

with st.sidebar:
    st.markdown("### Sensor Scenario")
    hours = st.slider("Hours of telemetry", 96, 720, 240, 24)
    drift = st.slider("Wear drift", 0.0, 3.0, 1.2, 0.1)
    vibration_spike = st.slider("Late vibration spike", 0.0, 4.0, 1.4, 0.1)
    threshold = st.slider("Alert threshold", 0.8, 3.0, 2.0, 0.1)

df = make_sensor_data(hours, drift, vibration_spike)
scored = add_anomaly_scores(df)
latest, high_risk_hours, rul, action = maintenance_summary(scored)

metric_cols = st.columns(4)
metric_cols[0].metric("Current anomaly score", f"{latest['anomaly_score']:.2f}")
metric_cols[1].metric("Recent high-risk hours", high_risk_hours)
metric_cols[2].metric("Estimated RUL", f"{rul} hours")
metric_cols[3].metric("Risk band", str(latest["risk_band"]))

tab1, tab2, tab3 = st.tabs(["Telemetry", "Anomaly Score", "Decision Notes"])

with tab1:
    long_df = scored.melt(
        id_vars=["hour"],
        value_vars=["temperature_c", "vibration_mm_s", "pressure_bar"],
        var_name="sensor",
        value_name="value",
    )
    fig = px.line(
        long_df,
        x="hour",
        y="value",
        color="sensor",
        title="Synthetic equipment telemetry",
        color_discrete_sequence=["#ffad7a", "#b8a9d9", "#7accff"],
    )
    st.plotly_chart(fig, use_container_width=True)

with tab2:
    fig = px.line(scored, x="hour", y="anomaly_score", title="Rolling anomaly score")
    fig.add_hline(y=threshold, line_dash="dash", line_color="#e8935c", annotation_text="alert threshold")
    st.plotly_chart(fig, use_container_width=True)
    st.dataframe(scored.tail(36), use_container_width=True, hide_index=True)

with tab3:
    st.markdown(f"""
### Recommendation

**{action}**

This demo uses rolling z-score features so the math is inspectable. A production Hugging Face Space could replace the scorer with an Isolation Forest, Autoformer, TimesFM, or a fine-tuned time-series transformer, then publish benchmark telemetry as a Dataset.
""")