| from __future__ import annotations |
|
|
| import logging |
| import os |
| import time |
| from typing import Any |
|
|
| import pandas as pd |
| import streamlit as st |
|
|
| from charts import ( |
| build_class_small_multiples, |
| build_latest_comparison_chart, |
| build_trend_chart, |
| ) |
| from config import AppConfig, load_config |
| from hf_client import HubClient, RepoRevision |
| from parsing import extract_performance_metrics_table |
| from transforms import ( |
| add_latest_flags, |
| build_latest_snapshot_table, |
| filter_history, |
| normalize_metrics_dataframe, |
| table_to_long_dataframe, |
| ) |
|
|
| try: |
| from streamlit_autorefresh import st_autorefresh |
| except Exception: |
| def st_autorefresh(*args: Any, **kwargs: Any) -> None: |
| return None |
|
|
|
|
| CACHE_TTL_SECONDS = int(os.getenv("EVAL_CACHE_TTL_SECONDS", "600")) |
| LOG_LEVEL = os.getenv("EVAL_LOG_LEVEL", "INFO").upper() |
| logging.basicConfig(level=getattr(logging, LOG_LEVEL, logging.INFO)) |
| logger = logging.getLogger("segmentation_eval_dashboard") |
|
|
|
|
| st.set_page_config( |
| page_title="Segmentation Evaluation Dashboard", |
| layout="wide", |
| ) |
|
|
|
|
| @st.cache_data(ttl=CACHE_TTL_SECONDS, show_spinner="Loading metrics from Hugging Face Hub...") |
| def collect_dashboard_data(config: AppConfig) -> tuple[pd.DataFrame, pd.DataFrame, list[str]]: |
| client = HubClient(config) |
| repos = client.discover_dataset_repos() |
| logger.info("Discovered %d candidate repos", len(repos)) |
| start_time = time.monotonic() |
|
|
| history_frames: list[pd.DataFrame] = [] |
| status_rows: list[dict[str, Any]] = [] |
|
|
| for repo_id in repos: |
| elapsed = time.monotonic() - start_time |
| if elapsed >= config.max_load_seconds: |
| status_rows.append( |
| { |
| "repo_id": "__load_guard__", |
| "status": "partial", |
| "revisions_checked": 0, |
| "revisions_parsed": 0, |
| "points": 0, |
| "latest_timestamp": pd.NaT, |
| "message": f"Stopped early after {config.max_load_seconds}s to keep app responsive", |
| } |
| ) |
| logger.warning("Load guard triggered after %.1fs", elapsed) |
| break |
|
|
| revisions_checked = 0 |
| revisions_parsed = 0 |
| points = 0 |
| message = "" |
| commit_error_message = "" |
|
|
| try: |
| commit_limit = config.max_commits_per_repo if config.include_history else 1 |
| revisions = client.list_dataset_commits(repo_id=repo_id, max_commits=max(1, commit_limit)) |
| except Exception as exc: |
| revisions = [RepoRevision(repo_id=repo_id, revision="main", committed_at=None)] |
| commit_error_message = f"Commit listing failed ({type(exc).__name__}); fallback to main" |
| logger.warning("Repo %s: %s", repo_id, commit_error_message) |
|
|
| repo_frames: list[pd.DataFrame] = [] |
| seen_revisions: set[str] = set() |
| for revision_info in revisions: |
| revision = revision_info.revision |
| if revision in seen_revisions: |
| continue |
|
|
| revisions_checked += 1 |
| seen_revisions.add(revision) |
|
|
| readme = client.fetch_dataset_readme(repo_id=repo_id, revision=revision) |
| if not readme: |
| continue |
|
|
| metrics_table = extract_performance_metrics_table( |
| markdown=readme, |
| heading=config.metrics_section_heading, |
| ) |
| if metrics_table is None or metrics_table.empty: |
| continue |
|
|
| normalized = normalize_metrics_dataframe(metrics_table) |
| if normalized.empty: |
| continue |
|
|
| long_df = table_to_long_dataframe( |
| dataframe=normalized, |
| repo_id=repo_id, |
| revision=revision, |
| committed_at=revision_info.committed_at, |
| metrics=config.metrics, |
| ) |
| if long_df.empty: |
| continue |
|
|
| revisions_parsed += 1 |
| points += len(long_df) |
| repo_frames.append(long_df) |
|
|
| if repo_frames: |
| repo_history = pd.concat(repo_frames, ignore_index=True) |
| history_frames.append(repo_history) |
| latest_timestamp = repo_history["timestamp"].max() |
| status = "ok" |
| message = f"Parsed {revisions_parsed}/{revisions_checked} revisions" |
| if commit_error_message: |
| message = f"{message}. {commit_error_message}" |
| else: |
| latest_timestamp = pd.NaT |
| status = "error" if commit_error_message else "no_metrics" |
| message = commit_error_message or "No valid Performance Metrics table found" |
|
|
| status_rows.append( |
| { |
| "repo_id": repo_id, |
| "status": status, |
| "revisions_checked": revisions_checked, |
| "revisions_parsed": revisions_parsed, |
| "points": points, |
| "latest_timestamp": latest_timestamp, |
| "message": message, |
| } |
| ) |
| logger.info( |
| "Repo %s: status=%s revisions_checked=%d revisions_parsed=%d points=%d", |
| repo_id, |
| status, |
| revisions_checked, |
| revisions_parsed, |
| points, |
| ) |
|
|
| if history_frames: |
| history_df = pd.concat(history_frames, ignore_index=True) |
| history_df = history_df.sort_values(["timestamp", "repo_id", "revision", "class", "metric"]) |
| history_df = add_latest_flags(history_df) |
| else: |
| history_df = pd.DataFrame( |
| columns=["repo_id", "revision", "timestamp", "class", "metric", "value", "std", "is_latest"] |
| ) |
|
|
| status_df = pd.DataFrame(status_rows) |
| if not status_df.empty and "latest_timestamp" in status_df.columns: |
| status_df["latest_timestamp"] = pd.to_datetime(status_df["latest_timestamp"], utc=True, errors="coerce") |
|
|
| return history_df, status_df, repos |
|
|
|
|
| def main() -> None: |
| config = load_config() |
|
|
| if config.auto_refresh_seconds > 0: |
| st_autorefresh(interval=config.auto_refresh_seconds * 1000, key="dashboard-refresh") |
|
|
| st.title("Segmentation Evaluation Metrics Dashboard") |
| st.caption("Source: Hugging Face dataset cards (`### Performance Metrics`).") |
|
|
| with st.sidebar: |
| st.header("Controls") |
| if st.button("Refresh now", use_container_width=True): |
| st.cache_data.clear() |
| st.rerun() |
|
|
| st.caption( |
| "Filtering/discovery is configured by env vars: " |
| "`HF_OWNER`, `EVAL_REPO_ALLOWLIST`, `EVAL_REPO_PREFIXES`, `EVAL_REPO_ID_CONTAINS`." |
| ) |
| show_error_bands = st.checkbox("Show std error bands", value=True) |
| show_small_multiples = st.checkbox("Show class small multiples", value=False) |
|
|
| history_df, status_df, discovered_repos = collect_dashboard_data(config) |
| logger.info( |
| "UI load complete: discovered=%d parsed_rows=%d status_rows=%d", |
| len(discovered_repos), |
| len(history_df), |
| len(status_df), |
| ) |
|
|
| if not discovered_repos: |
| st.warning("No dataset repositories were discovered. Check owner/prefix/allowlist environment variables.") |
| st.stop() |
|
|
| available_repos = sorted(history_df["repo_id"].dropna().unique().tolist()) if not history_df.empty else discovered_repos |
| available_classes = sorted(history_df["class"].dropna().unique().tolist()) if not history_df.empty else [] |
| available_metrics = sorted(history_df["metric"].dropna().unique().tolist()) if not history_df.empty else list(config.metrics) |
|
|
| with st.sidebar: |
| selected_repos = st.multiselect("Repos", options=available_repos, default=available_repos) |
| selected_classes = st.multiselect("Classes", options=available_classes, default=available_classes) |
| default_metrics = [metric for metric in config.metrics if metric in available_metrics] or available_metrics |
| selected_metrics = st.multiselect("Metrics", options=available_metrics, default=default_metrics) |
|
|
| filtered_history = filter_history( |
| history_df=history_df, |
| repos=selected_repos, |
| classes=selected_classes, |
| metrics=selected_metrics, |
| ) |
|
|
| latest_filtered = filtered_history[filtered_history["is_latest"]].copy() |
|
|
| success_repos = int((status_df["status"] == "ok").sum()) if not status_df.empty else 0 |
| latest_runs = latest_filtered[["repo_id", "revision"]].drop_duplicates().shape[0] if not latest_filtered.empty else 0 |
| latest_f1 = latest_filtered[latest_filtered["metric"] == "F1"]["value"].mean() if not latest_filtered.empty else float("nan") |
| latest_timestamp = filtered_history["timestamp"].max() if not filtered_history.empty else pd.NaT |
|
|
| col1, col2, col3, col4 = st.columns(4) |
| col1.metric("Discovered repos", len(discovered_repos)) |
| col2.metric("Repos parsed", success_repos) |
| col3.metric("Latest runs in view", latest_runs) |
| col4.metric("Latest mean F1", f"{latest_f1:.3f}" if pd.notna(latest_f1) else "n/a") |
|
|
| if pd.notna(latest_timestamp): |
| st.caption(f"Most recent datapoint: {pd.to_datetime(latest_timestamp).strftime('%Y-%m-%d %H:%M:%S %Z')}") |
|
|
| st.subheader("Trend Over Time") |
| trend_chart = build_trend_chart(filtered_history, show_error_bands=show_error_bands) |
| if trend_chart is None: |
| st.info("No trend data available for the active filters.") |
| else: |
| st.altair_chart(trend_chart, use_container_width=True) |
|
|
| st.subheader("Latest Snapshot") |
| snapshot_table = build_latest_snapshot_table(filtered_history) |
| if snapshot_table.empty: |
| st.info("No latest snapshot rows for current filters.") |
| else: |
| st.dataframe(snapshot_table, use_container_width=True, hide_index=True) |
|
|
| st.subheader("Repo Comparison (Latest)") |
| comparison_metric_options = selected_metrics or available_metrics |
| comparison_metric = None |
| if comparison_metric_options: |
| comparison_metric = st.selectbox( |
| "Metric for repo comparison", |
| options=comparison_metric_options, |
| index=0, |
| ) |
|
|
| if comparison_metric is not None: |
| comparison_chart = build_latest_comparison_chart(filtered_history, metric=comparison_metric) |
| if comparison_chart is None: |
| st.info("No latest comparison data available for this metric.") |
| else: |
| st.altair_chart(comparison_chart, use_container_width=True) |
|
|
| if show_small_multiples and comparison_metric: |
| st.subheader("Per-Class Small Multiples") |
| small_multiples_chart = build_class_small_multiples(filtered_history, metric=comparison_metric) |
| if small_multiples_chart is None: |
| st.info("No small-multiple data available for this metric.") |
| else: |
| st.altair_chart(small_multiples_chart, use_container_width=True) |
|
|
| st.subheader("Export") |
| csv_bytes = filtered_history.to_csv(index=False).encode("utf-8") |
| st.download_button( |
| "Export filtered history as CSV", |
| data=csv_bytes, |
| file_name="segmentation_metrics_history.csv", |
| mime="text/csv", |
| use_container_width=False, |
| ) |
|
|
| with st.expander("Health / Status", expanded=False): |
| if status_df.empty: |
| st.info("No status rows to display.") |
| else: |
| st.dataframe(status_df.sort_values(["status", "repo_id"]), use_container_width=True, hide_index=True) |
|
|
| with st.expander("Raw Parsed Data", expanded=False): |
| if filtered_history.empty: |
| st.info("No parsed rows for current filters.") |
| else: |
| st.dataframe( |
| filtered_history.sort_values(["timestamp", "repo_id", "class", "metric"], ascending=[False, True, True, True]), |
| use_container_width=True, |
| hide_index=True, |
| ) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
