hao-ai-lab-FastVideo/fastvideo/performance_dashboard/service.py

# SPDX-License-Identifier: Apache-2.0
"""Pure data transforms for the local performance dashboard.

The functions in this module operate on normalized records from
``fastvideo/tests/performance/compare_baseline.py``. They intentionally avoid
network and FastAPI concerns so they can be tested with in-memory fixtures.
"""

from __future__ import annotations

import statistics
from collections import defaultdict
from datetime import datetime, timezone
from typing import Any

from fastvideo.performance.hf_store import is_baseline_eligible_record, safe_float
from fastvideo.performance.metric_policy import regression_delta, resolve_metric_policies

Record = dict[str, Any]
CohortKey = tuple[str, ...]
COMPARISON_COHORT_KEYS = (
    "workload_id",
    "variant_id",
    "benchmark_version",
    "recipe_fingerprint",
    "hardware_profile_id",
    "software_profile_id",
)


def parse_timestamp(value: Any) -> datetime | None:
    if not value:
        return None
    if isinstance(value, datetime):
        ts = value
    else:
        try:
            ts = datetime.fromisoformat(str(value))
        except ValueError:
            return None
    if ts.tzinfo is None:
        return ts.replace(tzinfo=timezone.utc)
    return ts.astimezone(timezone.utc)


def record_sort_key(record: Record) -> tuple[datetime, str]:
    ts = parse_timestamp(record.get("timestamp"))
    return (ts or datetime.min.replace(tzinfo=timezone.utc), str(record.get("commit_sha") or ""))


def filter_records(
    records: list[Record],
    *,
    model_id: str | None = None,
    gpu_type: str | None = None,
    run_source: str | None = None,
    success: bool | None = None,
) -> list[Record]:
    filtered = records
    if model_id:
        filtered = [record for record in filtered if record.get("model_id") == model_id]
    if gpu_type:
        filtered = [record for record in filtered if record.get("gpu_type") == gpu_type]
    if run_source:
        filtered = [record for record in filtered if record_run_source(record) == run_source]
    if success is not None:
        filtered = [record for record in filtered if bool(record.get("success", True)) == success]
    return sorted(filtered, key=record_sort_key)


def record_run_source(record: Record) -> str:
    value = str(record.get("run_source") or "unknown")
    return value if value in {"pr", "local", "scheduled_main", "unknown"} else "unknown"


def record_metadata(record: Record) -> Record:
    return {
        "run_source": record_run_source(record),
        "baseline_eligible": is_baseline_eligible_record(record),
        "branch": record.get("branch") or "",
        "pr_number": record.get("pr_number") or "",
        "test_scope": record.get("test_scope") or "",
        "build_url": record.get("build_url") or "",
        "build_id": record.get("build_id") or "",
        "job_id": record.get("job_id") or "",
    }


def _cohort_metadata_value(value: Any) -> Any:
    if value is None:
        return ""
    if isinstance(value, str) and not value.strip():
        return ""
    return value


def _cohort_key_value(value: Any) -> str:
    return str(_cohort_metadata_value(value))


def record_comparison_metadata(record: Record) -> Record:
    return {key: _cohort_metadata_value(record.get(key)) for key in COMPARISON_COHORT_KEYS}


def _record_has_complete_v2_identity(record: Record) -> bool:
    return all(_cohort_metadata_value(record.get(key)) != "" for key in COMPARISON_COHORT_KEYS)


def comparison_cohort_key(record: Record) -> CohortKey:
    if _record_has_complete_v2_identity(record):
        return ("v2", *(_cohort_key_value(record.get(key)) for key in COMPARISON_COHORT_KEYS))
    return (
        "legacy",
        str(record.get("model_id") or "unknown"),
        str(record.get("gpu_type") or "unknown"),
        *(_cohort_key_value(record.get(key)) for key in COMPARISON_COHORT_KEYS),
    )


def group_by_comparison_cohort(records: list[Record]) -> dict[CohortKey, list[Record]]:
    groups: dict[CohortKey, list[Record]] = defaultdict(list)
    for record in records:
        groups[comparison_cohort_key(record)].append(record)
    return {key: sorted(value, key=record_sort_key) for key, value in groups.items()}


def comparison_sort_key(record: Record) -> tuple[str, ...]:
    return (
        str(record.get("model_id") or "unknown"),
        str(record.get("gpu_type") or "unknown"),
        *(_cohort_key_value(record.get(key)) for key in COMPARISON_COHORT_KEYS),
    )


def latest_row_sort_key(row: Record) -> tuple[Any, ...]:
    return (row["status"] != "fail", *comparison_sort_key(row))


def baseline_value(records: list[Record], metric_key: str) -> float | None:
    values = [safe_float(record.get(metric_key)) for record in records]
    values = [value for value in values if value is not None]
    if not values:
        return None
    return float(statistics.median(values))


def build_latest_summary(records: list[Record],
                         *,
                         baseline_window: int = 5,
                         run_source: str | None = None) -> list[Record]:
    rows: list[Record] = []
    for group in group_by_comparison_cohort(records).values():
        latest_candidates = group
        if run_source:
            latest_candidates = [record for record in group if record_run_source(record) == run_source]
        if not latest_candidates:
            continue

        latest = latest_candidates[-1]
        model_id = str(latest.get("model_id") or "unknown")
        gpu_type = str(latest.get("gpu_type") or "unknown")
        latest_index = next(index for index, record in enumerate(group) if record is latest)
        baseline_pool = [
            record for record in group[:latest_index]
            if record.get("success", True) and is_baseline_eligible_record(record)
        ]
        baseline_records = baseline_pool[-baseline_window:]
        metric_policies = resolve_metric_policies(latest.get("regression_thresholds"))

        metrics: dict[str, Record] = {}
        regressions: list[float] = []
        failing_metrics: list[str] = []
        threshold_exceeded_metrics: list[str] = []
        for policy in metric_policies:
            current = safe_float(latest.get(policy.key))
            baseline = baseline_value(baseline_records, policy.key)
            delta = None
            if current is not None and baseline is not None:
                delta = regression_delta(policy, current, baseline)
            regression = None if delta is None else delta.percent * 100.0
            metrics[policy.key] = {
                "current": current,
                "baseline": baseline,
                "regression_pct": regression,
                "absolute_delta": None if delta is None else delta.absolute,
                "threshold_percent": policy.threshold_percent * 100.0,
                "threshold_absolute": policy.threshold_absolute,
                "gated": policy.gated,
                "threshold_exceeded": False if delta is None else delta.threshold_exceeded,
                "regressed": False if delta is None else delta.regressed,
                "label": policy.label,
                "lower_is_better": policy.lower_is_better,
                "precision": policy.precision,
            }
            if regression is not None:
                regressions.append(regression)
            if delta is not None and delta.threshold_exceeded:
                threshold_exceeded_metrics.append(policy.key)
            if delta is not None and delta.regressed:
                failing_metrics.append(policy.key)

        worst_regression = max(regressions) if regressions else None
        success = bool(latest.get("success", True))
        status = "pass" if success else "fail"

        rows.append({
            "model_id": model_id,
            "gpu_type": gpu_type,
            "timestamp": latest.get("timestamp"),
            "commit_sha": latest.get("commit_sha"),
            **record_metadata(latest),
            **record_comparison_metadata(latest),
            "success": success,
            "baseline_n": len(baseline_records),
            "worst_regression_pct": worst_regression,
            "threshold_exceeded_metrics": threshold_exceeded_metrics,
            "failing_metrics": failing_metrics,
            "computed_regression_status": "fail" if failing_metrics else "pass",
            "status": status,
            "metrics": metrics,
        })

    return sorted(rows, key=latest_row_sort_key)


def build_trends(records: list[Record]) -> list[Record]:
    trends: list[Record] = []
    for group in group_by_comparison_cohort(records).values():
        latest = group[-1]
        model_id = str(latest.get("model_id") or "unknown")
        gpu_type = str(latest.get("gpu_type") or "unknown")
        points = []
        for record in group:
            metric_policies = resolve_metric_policies(record.get("regression_thresholds"))
            point = {
                "timestamp": record.get("timestamp"),
                "commit_sha": record.get("commit_sha"),
                **record_metadata(record),
                **record_comparison_metadata(record),
                "success": bool(record.get("success", True)),
                "metrics": {
                    policy.key: safe_float(record.get(policy.key))
                    for policy in metric_policies
                },
            }
            points.append(point)
        trends.append({
            "model_id": model_id,
            "gpu_type": gpu_type,
            **record_comparison_metadata(latest),
            "points": points,
        })
    return sorted(trends, key=comparison_sort_key)