| 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 |
|
|
