app.py

from __future__ import annotations

from pathlib import Path
from typing import Dict, Iterable, Tuple

import numpy as np
import pandas as pd
import streamlit as st

# -------------------- Streamlit page config -------------------- #
st.set_page_config(
    page_title="Running Optimizer · Model Results",
    page_icon="🏃",
    layout="wide",
)


ROOT = Path(__file__).resolve().parent
SAMPLE_RESULTS_PATH = ROOT / "assets" / "sample_results.csv"

CANONICAL_COLUMNS: Dict[str, Iterable[str]] = {
    "dataset": ["dataset", "dataset_name", "ds"],
    "table": ["table", "table_name", "target", "task"],
    "model_name": ["model_name", "model", "estimator", "algo", "algorithm"],
    "cv_mae": ["cv_mae", "cv_mae_mean", "cv_mae_score", "cv_mae_sec"],
    "test_mae": ["test_mae", "test_mae_score", "mae", "test_mae_sec"],
    "timestamp": ["timestamp", "time", "date", "created_at", "run_at"],
}

PREFERRED_COLUMNS = ["dataset", "table", "model_name", "cv_mae", "test_mae", "timestamp"]
NUMERIC_COLUMNS = ["cv_mae", "test_mae"]


# -------------------- Helpers -------------------- #
def normalize_columns(df: pd.DataFrame) -> pd.DataFrame:
    """Normalize known columns to a standard set of names."""
    if df.empty:
        return df

    rename_map: Dict[str, str] = {}
    normalized = {col: col.strip().lower().replace(" ", "_") for col in df.columns}

    for canonical, aliases in CANONICAL_COLUMNS.items():
        if canonical in df.columns:
            continue
        for alias in aliases:
            for original, normalized_name in normalized.items():
                if alias == normalized_name:
                    rename_map[original] = canonical
                    break
            if canonical in rename_map.values():
                break

    return df.rename(columns=rename_map)


def coerce_numeric(df: pd.DataFrame, columns: Iterable[str]) -> pd.DataFrame:
    for column in columns:
        if column in df.columns:
            df[column] = pd.to_numeric(df[column], errors="coerce")
    return df


def load_results() -> Tuple[pd.DataFrame, str]:
    if not SAMPLE_RESULTS_PATH.exists():
        raise FileNotFoundError(f"Sample results CSV not found at {SAMPLE_RESULTS_PATH.as_posix()}")
    df = pd.read_csv(SAMPLE_RESULTS_PATH)
    source = "Sample data (assets/sample_results.csv)"

    df = normalize_columns(df)
    df = coerce_numeric(df, NUMERIC_COLUMNS)
    return df, source


def apply_filters(df: pd.DataFrame) -> pd.DataFrame:
    filtered = df.copy()

    if "dataset" in df.columns:
        options = sorted(df["dataset"].dropna().unique().tolist())
        selected = st.sidebar.multiselect("Dataset", options, default=options)
        if selected:
            filtered = filtered[filtered["dataset"].isin(selected)]
        else:
            filtered = filtered.iloc[0:0]
    else:
        st.sidebar.caption("Dataset column not found in CSV")

    if "table" in df.columns:
        options = sorted(df["table"].dropna().unique().tolist())
        selected = st.sidebar.multiselect("Table", options, default=options)
        if selected:
            filtered = filtered[filtered["table"].isin(selected)]
        else:
            filtered = filtered.iloc[0:0]
    else:
        st.sidebar.caption("Table column not found in CSV")

    if "model_name" in df.columns:
        options = sorted(df["model_name"].dropna().unique().tolist())
        selected = st.sidebar.multiselect("Model", options, default=options)
        if selected:
            filtered = filtered[filtered["model_name"].isin(selected)]
        else:
            filtered = filtered.iloc[0:0]
    else:
        st.sidebar.caption("Model column not found in CSV")

    return filtered


def pick_best_model(df: pd.DataFrame) -> Tuple[str, float | None]:
    if df.empty:
        return "—", None

    score_col = (
        "test_mae" if "test_mae" in df.columns else "cv_mae" if "cv_mae" in df.columns else None
    )
    if score_col is None:
        return "—", None

    scores = pd.to_numeric(df[score_col], errors="coerce")
    if scores.dropna().empty:
        return "—", None

    best_idx = scores.idxmin()
    best_row = df.loc[best_idx]
    best_model = str(best_row.get("model_name", "—"))
    return best_model, float(scores.loc[best_idx])


def format_metric(value: float | None) -> str:
    if value is None or (isinstance(value, float) and np.isnan(value)):
        return "—"
    return f"{value:.3f}"


# -------------------- UI -------------------- #
st.title("🏃 Running Optimizer Results")
st.caption("Browse model evaluation results from the committed sample data.")

st.sidebar.header("Results Filters")

try:
    results_df, source_label = load_results()
except Exception as exc:  # pragma: no cover
    st.error(f"Failed to load results: {exc}")
    st.stop()

if results_df.empty:
    st.warning("No rows found in the results CSV.")
    st.stop()

st.sidebar.caption(f"Data source: {source_label}")

filtered_df = apply_filters(results_df)

st.info(f"Data source: {source_label}")

# -------------------- KPI Cards -------------------- #
rows_count = int(len(filtered_df))
best_model, _ = pick_best_model(filtered_df)
cv_best = (
    float(pd.to_numeric(filtered_df["cv_mae"], errors="coerce").min())
    if "cv_mae" in filtered_df.columns and not filtered_df.empty
    else None
)
test_best = (
    float(pd.to_numeric(filtered_df["test_mae"], errors="coerce").min())
    if "test_mae" in filtered_df.columns and not filtered_df.empty
    else None
)

kpi1, kpi2, kpi3, kpi4 = st.columns(4)
kpi1.metric("Rows", f"{rows_count}")
kpi2.metric("Best model", best_model)
kpi3.metric("Best CV MAE", format_metric(cv_best))
kpi4.metric("Best Test MAE", format_metric(test_best))

# -------------------- Trend Chart -------------------- #
if "timestamp" in filtered_df.columns:
    time_df = filtered_df.copy()
    time_df["timestamp"] = pd.to_datetime(time_df["timestamp"], errors="coerce")
    time_df = time_df.dropna(subset=["timestamp"]).sort_values("timestamp")

    metric_columns = [col for col in ["cv_mae", "test_mae"] if col in time_df.columns]

    if not time_df.empty and metric_columns:
        st.subheader("Metric trend over time")
        st.line_chart(time_df.set_index("timestamp")[metric_columns])
    elif metric_columns:
        st.info("No valid timestamps found to plot a trend.")
    else:
        st.info("Timestamp column found, but no metric columns to chart.")
else:
    st.caption("No timestamp column detected; skipping trend chart.")

# -------------------- Results Table -------------------- #
st.subheader("Results table")

if filtered_df.empty:
    st.warning("No rows match the current filters.")
else:
    display_cols = [col for col in PREFERRED_COLUMNS if col in filtered_df.columns]
    if not display_cols:
        display_cols = filtered_df.columns.tolist()

    st.dataframe(
        filtered_df[display_cols].sort_values(display_cols[0]),
        use_container_width=True,
        hide_index=True,
    )