transforms.py

from __future__ import annotations

from datetime import datetime, timezone
import re
from typing import Iterable

import numpy as np
import pandas as pd


_CANONICAL_METRIC_MAP = {
    "precision": "Precision",
    "recall": "Recall",
    "f1": "F1",
    "meaniou": "Mean_IoU",
    "meantpiou": "Mean_TP_IoU",
}


def normalize_metrics_dataframe(dataframe: pd.DataFrame) -> pd.DataFrame:
    if dataframe.empty:
        return pd.DataFrame()

    rename_map: dict[str, str] = {}
    class_column: str | None = None

    for column in dataframe.columns:
        normalized = _normalize_key(column)
        if normalized == "class":
            class_column = str(column)
            rename_map[str(column)] = "Class"
            continue

        if normalized.endswith("std"):
            metric_key = normalized[: -len("std")]
            metric_name = _CANONICAL_METRIC_MAP.get(metric_key)
            if metric_name:
                rename_map[str(column)] = f"{metric_name}_std"
            continue

        metric_name = _CANONICAL_METRIC_MAP.get(normalized)
        if metric_name:
            rename_map[str(column)] = metric_name

    if class_column is None:
        return pd.DataFrame()

    cleaned = dataframe.rename(columns=rename_map).copy()
    cleaned["Class"] = cleaned["Class"].astype(str).str.strip()
    cleaned = cleaned[cleaned["Class"] != ""]

    for column in cleaned.columns:
        if column == "Class":
            continue
        cleaned[column] = cleaned[column].apply(_parse_float)

    return cleaned.reset_index(drop=True)


def table_to_long_dataframe(
    dataframe: pd.DataFrame,
    repo_id: str,
    revision: str,
    committed_at: datetime | None,
    metrics: Iterable[str],
) -> pd.DataFrame:
    if dataframe.empty:
        return pd.DataFrame()

    timestamp = pd.to_datetime(committed_at or datetime.now(timezone.utc), utc=True)

    frames: list[pd.DataFrame] = []
    for metric in metrics:
        if metric not in dataframe.columns:
            continue

        metric_frame = pd.DataFrame(
            {
                "repo_id": repo_id,
                "revision": revision,
                "timestamp": timestamp,
                "class": dataframe["Class"].astype(str),
                "metric": metric,
                "value": dataframe[metric],
                "std": dataframe.get(f"{metric}_std", np.nan),
            }
        )
        frames.append(metric_frame)

    if not frames:
        return pd.DataFrame()

    long_df = pd.concat(frames, ignore_index=True)
    long_df["value"] = pd.to_numeric(long_df["value"], errors="coerce")
    long_df["std"] = pd.to_numeric(long_df["std"], errors="coerce")
    long_df["timestamp"] = pd.to_datetime(long_df["timestamp"], utc=True, errors="coerce")
    long_df = long_df.dropna(subset=["timestamp"])
    return long_df


def add_latest_flags(history_df: pd.DataFrame) -> pd.DataFrame:
    if history_df.empty:
        result = history_df.copy()
        result["is_latest"] = False
        return result

    unique_runs = history_df[["repo_id", "revision", "timestamp"]].drop_duplicates()
    latest_runs = (
        unique_runs.sort_values(["repo_id", "timestamp", "revision"])
        .groupby("repo_id", as_index=False)
        .tail(1)
        .assign(is_latest=True)
    )

    merged = history_df.merge(
        latest_runs,
        on=["repo_id", "revision", "timestamp"],
        how="left",
    )
    merged["is_latest"] = merged["is_latest"].fillna(False)
    return merged


def filter_history(
    history_df: pd.DataFrame,
    repos: Iterable[str],
    classes: Iterable[str],
    metrics: Iterable[str],
) -> pd.DataFrame:
    if history_df.empty:
        return history_df

    filtered = history_df.copy()
    repos = list(repos)
    classes = list(classes)
    metrics = list(metrics)

    if repos:
        filtered = filtered[filtered["repo_id"].isin(repos)]
    if classes:
        filtered = filtered[filtered["class"].isin(classes)]
    if metrics:
        filtered = filtered[filtered["metric"].isin(metrics)]

    return filtered


def build_latest_snapshot_table(history_df: pd.DataFrame) -> pd.DataFrame:
    if history_df.empty:
        return pd.DataFrame()

    latest_df = history_df[history_df["is_latest"]].copy()
    if latest_df.empty:
        return pd.DataFrame()

    value_pivot = latest_df.pivot_table(
        index=["repo_id", "class", "revision", "timestamp"],
        columns="metric",
        values="value",
        aggfunc="first",
    )
    std_pivot = latest_df.pivot_table(
        index=["repo_id", "class", "revision", "timestamp"],
        columns="metric",
        values="std",
        aggfunc="first",
    )

    if not std_pivot.empty:
        std_pivot.columns = [f"{column}_std" for column in std_pivot.columns]

    snapshot = value_pivot.join(std_pivot, how="left")
    snapshot = snapshot.reset_index().sort_values(["repo_id", "class"])
    return snapshot


def _normalize_key(value: str) -> str:
    return re.sub(r"[^a-z0-9]", "", str(value).strip().lower())


def _parse_float(value: object) -> float:
    if value is None:
        return np.nan

    text = str(value).strip()
    if text == "":
        return np.nan

    lowered = text.lower()
    if lowered in {"nan", "na", "n/a", "none", "null", "-"}:
        return np.nan

    cleaned = text.replace(",", "")
    try:
        return float(cleaned)
    except ValueError:
        return np.nan