GPUMODE/kernelbot-data

1#!/usr/bin/env python32"""3Helper functions for analyzing kernelbot submissions.4 5Usage:6    from analyze_submissions import load_submissions, author_progression, top_contestants7"""8 9import pandas as pd10from pathlib import Path11 12 13def format_score(score, unit='us'):14    """15    Format score with appropriate units.16 17    Args:18        score: Score in seconds19        unit: 'us' for microseconds, 'ms' for milliseconds, 'auto' for automatic20 21    Returns:22        Formatted string with units23    """24    if pd.isna(score):25        return 'N/A'26 27    if unit == 'auto':28        if score < 0.001:  # Less than 1ms, show in microseconds29            return f"{score * 1_000_000:.2f} µs"30        elif score < 1:  # Less than 1s, show in milliseconds31            return f"{score * 1_000:.3f} ms"32        else:33            return f"{score:.4f} s"34    elif unit == 'us':35        return f"{score * 1_000_000:.2f} µs"36    elif unit == 'ms':37        return f"{score * 1_000:.3f} ms"38    else:39        return f"{score:.6f} s"40 41 42def load_submissions(parquet_path: str = None) -> pd.DataFrame:43    """Load deduplicated submissions from parquet file."""44    if parquet_path is None:45        parquet_path = Path(__file__).parent.parent.parent / "nvidia_nvfp4_submissions.parquet"46    return pd.read_parquet(parquet_path)47 48 49def author_progression(df: pd.DataFrame, user_id: str = None, user_name: str = None,50                       problem_name: str = None) -> pd.DataFrame:51    """52    Get submissions from an author sorted by time to see their progression.53 54    Args:55        df: DataFrame of submissions56        user_id: Filter by user ID (Discord ID)57        user_name: Filter by username (partial match, case-insensitive)58        problem_name: Filter by problem name59 60    Returns:61        DataFrame sorted by submission_time showing the author's journey62    """63    result = df.copy()64 65    if user_id:66        result = result[result['user_id'] == user_id]67 68    if user_name:69        result = result[result['user_name'].str.contains(user_name, case=False, na=False)]70 71    if problem_name:72        result = result[result['problem_name'] == problem_name]73 74    return result.sort_values('submission_time')75 76 77def top_contestants(df: pd.DataFrame, problem_name: str = None, n: int = 20,78                    passing_only: bool = True) -> pd.DataFrame:79    """80    Get top contestants sorted by their best score (fastest time).81 82    Args:83        df: DataFrame of submissions84        problem_name: Filter by problem name (required for meaningful results)85        n: Number of top contestants to return86        passing_only: Only include passing submissions87 88    Returns:89        DataFrame with top contestants and their best scores90    """91    result = df.copy()92 93    if problem_name:94        result = result[result['problem_name'] == problem_name]95 96    if passing_only:97        result = result[result['passed'] == True]98 99    # Filter out rows with NA scores100    result = result.dropna(subset=['score'])101 102    if result.empty:103        return pd.DataFrame(columns=['user_name', 'user_id', 'score', 'submission_time', 'problem_name'])104 105    # Get best score per user106    best_scores = result.loc[result.groupby('user_id')['score'].idxmin()]107 108    return best_scores.sort_values('score').head(n)[109        ['user_name', 'user_id', 'score', 'submission_time', 'problem_name']110    ]111 112 113def leaderboard_summary(df: pd.DataFrame, score_unit='us') -> pd.DataFrame:114    """115    Get summary statistics for each problem.116 117    Args:118        df: DataFrame of submissions119        score_unit: 'us' for microseconds, 'ms' for milliseconds, 's' for seconds120 121    Returns:122        DataFrame with submission counts, unique users, score ranges123    """124    summary = df.groupby('problem_name').agg({125        'submission_id': 'count',126        'user_id': 'nunique',127        'score': ['min', 'median', 'max'],128        'passed': 'sum'129    })130 131    summary.columns = ['submissions', 'unique_users', 'best_score', 'median_score',132                       'worst_score', 'passing_count']133 134    # Convert scores to specified unit135    if score_unit == 'us':136        multiplier = 1_000_000137        summary['best_score'] = (summary['best_score'] * multiplier).round(2)138        summary['median_score'] = (summary['median_score'] * multiplier).round(2)139        summary['worst_score'] = (summary['worst_score'] * multiplier).round(2)140    elif score_unit == 'ms':141        multiplier = 1_000142        summary['best_score'] = (summary['best_score'] * multiplier).round(3)143        summary['median_score'] = (summary['median_score'] * multiplier).round(3)144        summary['worst_score'] = (summary['worst_score'] * multiplier).round(3)145 146    return summary147 148 149def user_stats(df: pd.DataFrame, user_id: str = None, user_name: str = None) -> pd.DataFrame:150    """151    Get statistics for a specific user across all problems.152    """153    result = df.copy()154 155    if user_id:156        result = result[result['user_id'] == user_id]157    elif user_name:158        result = result[result['user_name'].str.contains(user_name, case=False, na=False)]159 160    return result.groupby('problem_name').agg({161        'submission_id': 'count',162        'score': 'min',163        'passed': 'sum'164    }).rename(columns={165        'submission_id': 'num_submissions',166        'score': 'best_score',167        'passed': 'passing_count'168    })169