time_series_gradio.py

import pandas as pd
import numpy as np
from datetime import datetime
from data import extract_model_data
import gradio as gr
import plotly.express as px
import plotly.graph_objects as go

COLORS = {
    'passed': '#4CAF50',
    'failed': '#E53E3E',
    'skipped': '#FFD54F',
    'error': '#8B0000',
    'amd': '#ED1C24',
    'nvidia': '#76B900'
}

def get_time_series_summary_dfs(historical_df: pd.DataFrame) -> dict:
    daily_stats = []
    dates = sorted(historical_df['date'].unique())
    for date in dates:
        date_data = historical_df[historical_df['date'] == date]
        amd_passed = date_data['success_amd'].sum() if 'success_amd' in date_data.columns else 0
        amd_failed = (date_data['failed_multi_no_amd'].sum() + date_data['failed_single_no_amd'].sum()) if 'failed_multi_no_amd' in date_data.columns else 0
        amd_skipped = date_data['skipped_amd'].sum() if 'skipped_amd' in date_data.columns else 0
        amd_total = amd_passed + amd_failed + amd_skipped
        amd_failure_rate = (amd_failed / amd_total * 100) if amd_total > 0 else 0

        nvidia_passed = date_data['success_nvidia'].sum() if 'success_nvidia' in date_data.columns else 0
        nvidia_failed = (date_data['failed_multi_no_nvidia'].sum() + date_data['failed_single_no_nvidia'].sum()) if 'failed_multi_no_nvidia' in date_data.columns else 0
        nvidia_skipped = date_data['skipped_nvidia'].sum() if 'skipped_nvidia' in date_data.columns else 0
        nvidia_total = nvidia_passed + nvidia_failed + nvidia_skipped
        nvidia_failure_rate = (nvidia_failed / nvidia_total * 100) if nvidia_total > 0 else 0

        daily_stats.append({
            'date': date,
            'amd_failure_rate': amd_failure_rate,
            'nvidia_failure_rate': nvidia_failure_rate,
            'amd_passed': amd_passed,
            'amd_failed': amd_failed,
            'amd_skipped': amd_skipped,
            'nvidia_passed': nvidia_passed,
            'nvidia_failed': nvidia_failed,
            'nvidia_skipped': nvidia_skipped
        })

    failure_rate_data = []
    for i, stat in enumerate(daily_stats):
        amd_change = stat['amd_failure_rate'] - daily_stats[i-1]['amd_failure_rate'] if i > 0 else 0
        nvidia_change = stat['nvidia_failure_rate'] - daily_stats[i-1]['nvidia_failure_rate'] if i > 0 else 0
        failure_rate_data.extend([
            {'date': stat['date'], 'failure_rate': stat['amd_failure_rate'], 'platform': 'AMD', 'change': amd_change},
            {'date': stat['date'], 'failure_rate': stat['nvidia_failure_rate'], 'platform': 'NVIDIA', 'change': nvidia_change}
        ])
    failure_rate_df = pd.DataFrame(failure_rate_data)

    amd_data = []
    for i, stat in enumerate(daily_stats):
        passed_change = stat['amd_passed'] - daily_stats[i-1]['amd_passed'] if i > 0 else 0
        failed_change = stat['amd_failed'] - daily_stats[i-1]['amd_failed'] if i > 0 else 0
        skipped_change = stat['amd_skipped'] - daily_stats[i-1]['amd_skipped'] if i > 0 else 0
        amd_data.extend([
            {'date': stat['date'], 'count': stat['amd_passed'], 'test_type': 'Passed', 'change': passed_change},
            {'date': stat['date'], 'count': stat['amd_failed'], 'test_type': 'Failed', 'change': failed_change},
            {'date': stat['date'], 'count': stat['amd_skipped'], 'test_type': 'Skipped', 'change': skipped_change}
        ])
    amd_df = pd.DataFrame(amd_data)

    nvidia_data = []
    for i, stat in enumerate(daily_stats):
        passed_change = stat['nvidia_passed'] - daily_stats[i-1]['nvidia_passed'] if i > 0 else 0
        failed_change = stat['nvidia_failed'] - daily_stats[i-1]['nvidia_failed'] if i > 0 else 0
        skipped_change = stat['nvidia_skipped'] - daily_stats[i-1]['nvidia_skipped'] if i > 0 else 0
        nvidia_data.extend([
            {'date': stat['date'], 'count': stat['nvidia_passed'], 'test_type': 'Passed', 'change': passed_change},
            {'date': stat['date'], 'count': stat['nvidia_failed'], 'test_type': 'Failed', 'change': failed_change},
            {'date': stat['date'], 'count': stat['nvidia_skipped'], 'test_type': 'Skipped', 'change': skipped_change}
        ])
    nvidia_df = pd.DataFrame(nvidia_data)

    return {
        'failure_rates_df': failure_rate_df,
        'amd_tests_df': amd_df,
        'nvidia_tests_df': nvidia_df,
    }

def get_model_time_series_dfs(historical_df: pd.DataFrame, model_name: str) -> dict:
    model_data = historical_df[historical_df.index.str.lower() == model_name.lower()]
    
    if model_data.empty:
        empty_df = pd.DataFrame({'date': [], 'count': [], 'test_type': [], 'change': []})
        return {'amd_df': empty_df.copy(), 'nvidia_df': empty_df.copy()}

    dates = sorted(model_data['date'].unique())
    amd_data = []
    nvidia_data = []
    for i, date in enumerate(dates):
        date_data = model_data[model_data['date'] == date]
        row = date_data.iloc[0]

        amd_passed = row.get('success_amd', 0)
        amd_failed = row.get('failed_multi_no_amd', 0) + row.get('failed_single_no_amd', 0)
        amd_skipped = row.get('skipped_amd', 0)
        prev_row = model_data[model_data['date'] == dates[i-1]].iloc[0] if i > 0 and not model_data[model_data['date'] == dates[i-1]].empty else None
        amd_passed_change = amd_passed - (prev_row.get('success_amd', 0) if prev_row is not None else 0)
        amd_failed_change = amd_failed - (prev_row.get('failed_multi_no_amd', 0) + prev_row.get('failed_single_no_amd', 0) if prev_row is not None else 0)
        amd_skipped_change = amd_skipped - (prev_row.get('skipped_amd', 0) if prev_row is not None else 0)
        amd_data.extend([
            {'date': date, 'count': amd_passed, 'test_type': 'Passed', 'change': amd_passed_change},
            {'date': date, 'count': amd_failed, 'test_type': 'Failed', 'change': amd_failed_change},
            {'date': date, 'count': amd_skipped, 'test_type': 'Skipped', 'change': amd_skipped_change}
        ])

        nvidia_passed = row.get('success_nvidia', 0)
        nvidia_failed = row.get('failed_multi_no_nvidia', 0) + row.get('failed_single_no_nvidia', 0)
        nvidia_skipped = row.get('skipped_nvidia', 0)
        if prev_row is not None:
            prev_nvidia_passed = prev_row.get('success_nvidia', 0)
            prev_nvidia_failed = prev_row.get('failed_multi_no_nvidia', 0) + prev_row.get('failed_single_no_nvidia', 0)
            prev_nvidia_skipped = prev_row.get('skipped_nvidia', 0)
        else:
            prev_nvidia_passed = prev_nvidia_failed = prev_nvidia_skipped = 0
        nvidia_data.extend([
            {'date': date, 'count': nvidia_passed, 'test_type': 'Passed', 'change': nvidia_passed - prev_nvidia_passed},
            {'date': date, 'count': nvidia_failed, 'test_type': 'Failed', 'change': nvidia_failed - prev_nvidia_failed},
            {'date': date, 'count': nvidia_skipped, 'test_type': 'Skipped', 'change': nvidia_skipped - prev_nvidia_skipped}
        ])

    return {'amd_df': pd.DataFrame(amd_data), 'nvidia_df': pd.DataFrame(nvidia_data)}

def create_time_series_summary_gradio(historical_df: pd.DataFrame) -> dict:
    if historical_df.empty or 'date' not in historical_df.columns:
        # Create empty Plotly figure
        empty_fig = go.Figure()
        empty_fig.update_layout(
            title="No historical data available",
            height=400,
            font=dict(size=16, color='#CCCCCC'),
            paper_bgcolor='#000000',
            plot_bgcolor='#1a1a1a',
        )
        return {
            'failure_rates': gr.Plot(value=empty_fig),
            'amd_tests': gr.Plot(value=empty_fig),
            'nvidia_tests': gr.Plot(value=empty_fig)
        }
    
    daily_stats = []
    dates = sorted(historical_df['date'].unique())
    
    for date in dates:
        date_data = historical_df[historical_df['date'] == date]
        
        amd_passed = date_data['success_amd'].sum() if 'success_amd' in date_data.columns else 0
        amd_failed = (date_data['failed_multi_no_amd'].sum() + date_data['failed_single_no_amd'].sum()) if 'failed_multi_no_amd' in date_data.columns else 0
        amd_skipped = date_data['skipped_amd'].sum() if 'skipped_amd' in date_data.columns else 0
        amd_total = amd_passed + amd_failed + amd_skipped
        amd_failure_rate = (amd_failed / amd_total * 100) if amd_total > 0 else 0
        
        nvidia_passed = date_data['success_nvidia'].sum() if 'success_nvidia' in date_data.columns else 0
        nvidia_failed = (date_data['failed_multi_no_nvidia'].sum() + date_data['failed_single_no_nvidia'].sum()) if 'failed_multi_no_nvidia' in date_data.columns else 0
        nvidia_skipped = date_data['skipped_nvidia'].sum() if 'skipped_nvidia' in date_data.columns else 0
        nvidia_total = nvidia_passed + nvidia_failed + nvidia_skipped
        nvidia_failure_rate = (nvidia_failed / nvidia_total * 100) if nvidia_total > 0 else 0
        
        daily_stats.append({
            'date': date,
            'amd_failure_rate': amd_failure_rate,
            'nvidia_failure_rate': nvidia_failure_rate,
            'amd_passed': amd_passed,
            'amd_failed': amd_failed,
            'amd_skipped': amd_skipped,
            'nvidia_passed': nvidia_passed,
            'nvidia_failed': nvidia_failed,
            'nvidia_skipped': nvidia_skipped
        })
    
    failure_rate_data = []
    for i, stat in enumerate(daily_stats):
        amd_change = nvidia_change = 0
        if i > 0:
            amd_change = stat['amd_failure_rate'] - daily_stats[i-1]['amd_failure_rate']
            nvidia_change = stat['nvidia_failure_rate'] - daily_stats[i-1]['nvidia_failure_rate']
        
        failure_rate_data.extend([
            {'date': stat['date'], 'failure_rate': stat['amd_failure_rate'], 'platform': 'AMD', 'change': amd_change},
            {'date': stat['date'], 'failure_rate': stat['nvidia_failure_rate'], 'platform': 'NVIDIA', 'change': nvidia_change}
        ])
    
    failure_rate_df = pd.DataFrame(failure_rate_data)
    
    amd_data = []
    for i, stat in enumerate(daily_stats):
        passed_change = failed_change = skipped_change = 0
        if i > 0:
            passed_change = stat['amd_passed'] - daily_stats[i-1]['amd_passed']
            failed_change = stat['amd_failed'] - daily_stats[i-1]['amd_failed']
            skipped_change = stat['amd_skipped'] - daily_stats[i-1]['amd_skipped']
        
        amd_data.extend([
            {'date': stat['date'], 'count': stat['amd_passed'], 'test_type': 'Passed', 'change': passed_change},
            {'date': stat['date'], 'count': stat['amd_failed'], 'test_type': 'Failed', 'change': failed_change},
            {'date': stat['date'], 'count': stat['amd_skipped'], 'test_type': 'Skipped', 'change': skipped_change}
        ])
    
    amd_df = pd.DataFrame(amd_data)
    
    nvidia_data = []
    for i, stat in enumerate(daily_stats):
        passed_change = failed_change = skipped_change = 0
        if i > 0:
            passed_change = stat['nvidia_passed'] - daily_stats[i-1]['nvidia_passed']
            failed_change = stat['nvidia_failed'] - daily_stats[i-1]['nvidia_failed']
            skipped_change = stat['nvidia_skipped'] - daily_stats[i-1]['nvidia_skipped']
        
        nvidia_data.extend([
            {'date': stat['date'], 'count': stat['nvidia_passed'], 'test_type': 'Passed', 'change': passed_change},
            {'date': stat['date'], 'count': stat['nvidia_failed'], 'test_type': 'Failed', 'change': failed_change},
            {'date': stat['date'], 'count': stat['nvidia_skipped'], 'test_type': 'Skipped', 'change': skipped_change}
        ])
    
    nvidia_df = pd.DataFrame(nvidia_data)
    
    # Create Plotly figure for failure rates
    fig_failure_rates = px.line(
        failure_rate_df,
        x='date',
        y='failure_rate',
        color='platform',
        color_discrete_map={"AMD": COLORS['amd'], "NVIDIA": COLORS['nvidia']},
        title="Overall Failure Rates Over Time",
        labels={'failure_rate': 'Failure Rate (%)', 'date': 'Date', 'platform': 'Platform'}
    )
    fig_failure_rates.update_traces(mode='lines+markers', hovertemplate='%{y:.2f}%<br>%{x}<extra></extra>')
    fig_failure_rates.update_layout(
        height=400,
        font=dict(size=16, color='#CCCCCC'),
        paper_bgcolor='#000000',
        plot_bgcolor='#1a1a1a',
        title_font_size=20,
        legend=dict(font=dict(size=16), bgcolor='rgba(0,0,0,0.5)'),
        xaxis=dict(
            title_font_size=16,
            tickfont_size=14,
            gridcolor='#333333',
            showgrid=True
        ),
        yaxis=dict(
            title_font_size=16,
            tickfont_size=14,
            gridcolor='#333333',
            showgrid=True
        ),
        hovermode='x unified'
    )
    
    # Create Plotly figure for AMD tests
    fig_amd = px.line(
        amd_df,
        x='date',
        y='count',
        color='test_type',
        color_discrete_map={"Passed": COLORS['passed'], "Failed": COLORS['failed'], "Skipped": COLORS['skipped']},
        title="AMD Test Results Over Time",
        labels={'count': 'Number of Tests', 'date': 'Date', 'test_type': 'Test Type'}
    )
    fig_amd.update_traces(mode='lines+markers', hovertemplate='%{y}<br>%{x}<extra></extra>')
    fig_amd.update_layout(
        height=400,
        font=dict(size=16, color='#CCCCCC'),
        paper_bgcolor='#000000',
        plot_bgcolor='#1a1a1a',
        title_font_size=20,
        legend=dict(font=dict(size=16), bgcolor='rgba(0,0,0,0.5)'),
        xaxis=dict(
            title_font_size=16,
            tickfont_size=14,
            gridcolor='#333333',
            showgrid=True
        ),
        yaxis=dict(
            title_font_size=16,
            tickfont_size=14,
            gridcolor='#333333',
            showgrid=True
        ),
        hovermode='x unified'
    )
    
    # Create Plotly figure for NVIDIA tests
    fig_nvidia = px.line(
        nvidia_df,
        x='date',
        y='count',
        color='test_type',
        color_discrete_map={"Passed": COLORS['passed'], "Failed": COLORS['failed'], "Skipped": COLORS['skipped']},
        title="NVIDIA Test Results Over Time",
        labels={'count': 'Number of Tests', 'date': 'Date', 'test_type': 'Test Type'}
    )
    fig_nvidia.update_traces(mode='lines+markers', hovertemplate='%{y}<br>%{x}<extra></extra>')
    fig_nvidia.update_layout(
        height=400,
        font=dict(size=16, color='#CCCCCC'),
        paper_bgcolor='#000000',
        plot_bgcolor='#1a1a1a',
        title_font_size=20,
        legend=dict(font=dict(size=16), bgcolor='rgba(0,0,0,0.5)'),
        xaxis=dict(
            title_font_size=16,
            tickfont_size=14,
            gridcolor='#333333',
            showgrid=True
        ),
        yaxis=dict(
            title_font_size=16,
            tickfont_size=14,
            gridcolor='#333333',
            showgrid=True
        ),
        hovermode='x unified'
    )
    
    return {
        'failure_rates': gr.Plot(value=fig_failure_rates),
        'amd_tests': gr.Plot(value=fig_amd),
        'nvidia_tests': gr.Plot(value=fig_nvidia)
    }


def create_model_time_series_gradio(historical_df: pd.DataFrame, model_name: str) -> dict:
    if historical_df.empty or 'date' not in historical_df.columns:
        # Create empty Plotly figures
        empty_fig_amd = go.Figure()
        empty_fig_amd.update_layout(
            title=f"{model_name.upper()} - AMD Results Over Time",
            height=400,
            font=dict(size=16, color='#CCCCCC'),
            paper_bgcolor='#000000',
            plot_bgcolor='#1a1a1a',
        )
        empty_fig_nvidia = go.Figure()
        empty_fig_nvidia.update_layout(
            title=f"{model_name.upper()} - NVIDIA Results Over Time",
            height=400,
            font=dict(size=16, color='#CCCCCC'),
            paper_bgcolor='#000000',
            plot_bgcolor='#1a1a1a',
        )
        return {
            'amd_plot': gr.Plot(value=empty_fig_amd),
            'nvidia_plot': gr.Plot(value=empty_fig_nvidia)
        }
    
    model_data = historical_df[historical_df.index.str.lower() == model_name.lower()]
    
    if model_data.empty:
        # Create empty Plotly figures
        empty_fig_amd = go.Figure()
        empty_fig_amd.update_layout(
            title=f"{model_name.upper()} - AMD Results Over Time",
            height=400,
            font=dict(size=16, color='#CCCCCC'),
            paper_bgcolor='#000000',
            plot_bgcolor='#1a1a1a',
        )
        empty_fig_nvidia = go.Figure()
        empty_fig_nvidia.update_layout(
            title=f"{model_name.upper()} - NVIDIA Results Over Time",
            height=400,
            font=dict(size=16, color='#CCCCCC'),
            paper_bgcolor='#000000',
            plot_bgcolor='#1a1a1a',
        )
        return {
            'amd_plot': gr.Plot(value=empty_fig_amd),
            'nvidia_plot': gr.Plot(value=empty_fig_nvidia)
        }
    
    dates = sorted(model_data['date'].unique())
    
    amd_data = []
    nvidia_data = []
    
    for i, date in enumerate(dates):
        date_data = model_data[model_data['date'] == date]
        
        if not date_data.empty:
            row = date_data.iloc[0]
            
            amd_passed = row.get('success_amd', 0)
            amd_failed = row.get('failed_multi_no_amd', 0) + row.get('failed_single_no_amd', 0)
            amd_skipped = row.get('skipped_amd', 0)
            
            passed_change = failed_change = skipped_change = 0
            if i > 0:
                prev_date_data = model_data[model_data['date'] == dates[i-1]]
                if not prev_date_data.empty:
                    prev_row = prev_date_data.iloc[0]
                    prev_amd_passed = prev_row.get('success_amd', 0)
                    prev_amd_failed = prev_row.get('failed_multi_no_amd', 0) + prev_row.get('failed_single_no_amd', 0)
                    prev_amd_skipped = prev_row.get('skipped_amd', 0)
                    
                    passed_change = amd_passed - prev_amd_passed
                    failed_change = amd_failed - prev_amd_failed
                    skipped_change = amd_skipped - prev_amd_skipped
            
            amd_data.extend([
                {'date': date, 'count': amd_passed, 'test_type': 'Passed', 'change': passed_change},
                {'date': date, 'count': amd_failed, 'test_type': 'Failed', 'change': failed_change},
                {'date': date, 'count': amd_skipped, 'test_type': 'Skipped', 'change': skipped_change}
            ])
            
            nvidia_passed = row.get('success_nvidia', 0)
            nvidia_failed = row.get('failed_multi_no_nvidia', 0) + row.get('failed_single_no_nvidia', 0)
            nvidia_skipped = row.get('skipped_nvidia', 0)
            
            nvidia_passed_change = nvidia_failed_change = nvidia_skipped_change = 0
            if i > 0:
                prev_date_data = model_data[model_data['date'] == dates[i-1]]
                if not prev_date_data.empty:
                    prev_row = prev_date_data.iloc[0]
                    prev_nvidia_passed = prev_row.get('success_nvidia', 0)
                    prev_nvidia_failed = prev_row.get('failed_multi_no_nvidia', 0) + prev_row.get('failed_single_no_nvidia', 0)
                    prev_nvidia_skipped = prev_row.get('skipped_nvidia', 0)
                    
                    nvidia_passed_change = nvidia_passed - prev_nvidia_passed
                    nvidia_failed_change = nvidia_failed - prev_nvidia_failed
                    nvidia_skipped_change = nvidia_skipped - prev_nvidia_skipped
            
            nvidia_data.extend([
                {'date': date, 'count': nvidia_passed, 'test_type': 'Passed', 'change': nvidia_passed_change},
                {'date': date, 'count': nvidia_failed, 'test_type': 'Failed', 'change': nvidia_failed_change},
                {'date': date, 'count': nvidia_skipped, 'test_type': 'Skipped', 'change': nvidia_skipped_change}
            ])
    
    amd_df = pd.DataFrame(amd_data)
    nvidia_df = pd.DataFrame(nvidia_data)
    
    # Create Plotly figure for AMD
    fig_amd = px.line(
        amd_df,
        x='date',
        y='count',
        color='test_type',
        color_discrete_map={"Passed": COLORS['passed'], "Failed": COLORS['failed'], "Skipped": COLORS['skipped']},
        title=f"{model_name.upper()} - AMD Results Over Time",
        labels={'count': 'Number of Tests', 'date': 'Date', 'test_type': 'Test Type'}
    )
    fig_amd.update_traces(mode='lines+markers', hovertemplate='%{y}<br>%{x}<extra></extra>')
    fig_amd.update_layout(
        height=400,
        font=dict(size=16, color='#CCCCCC'),
        paper_bgcolor='#000000',
        plot_bgcolor='#1a1a1a',
        title_font_size=20,
        legend=dict(font=dict(size=16), bgcolor='rgba(0,0,0,0.5)'),
        xaxis=dict(
            title_font_size=16,
            tickfont_size=14,
            gridcolor='#333333',
            showgrid=True
        ),
        yaxis=dict(
            title_font_size=16,
            tickfont_size=14,
            gridcolor='#333333',
            showgrid=True
        ),
        hovermode='x unified'
    )
    
    # Create Plotly figure for NVIDIA
    fig_nvidia = px.line(
        nvidia_df,
        x='date',
        y='count',
        color='test_type',
        color_discrete_map={"Passed": COLORS['passed'], "Failed": COLORS['failed'], "Skipped": COLORS['skipped']},
        title=f"{model_name.upper()} - NVIDIA Results Over Time",
        labels={'count': 'Number of Tests', 'date': 'Date', 'test_type': 'Test Type'}
    )
    fig_nvidia.update_traces(mode='lines+markers', hovertemplate='%{y}<br>%{x}<extra></extra>')
    fig_nvidia.update_layout(
        height=400,
        font=dict(size=16, color='#CCCCCC'),
        paper_bgcolor='#000000',
        plot_bgcolor='#1a1a1a',
        title_font_size=20,
        legend=dict(font=dict(size=16), bgcolor='rgba(0,0,0,0.5)'),
        xaxis=dict(
            title_font_size=16,
            tickfont_size=14,
            gridcolor='#333333',
            showgrid=True
        ),
        yaxis=dict(
            title_font_size=16,
            tickfont_size=14,
            gridcolor='#333333',
            showgrid=True
        ),
        hovermode='x unified'
    )
    
    return {
        'amd_plot': gr.Plot(value=fig_amd),
        'nvidia_plot': gr.Plot(value=fig_nvidia)
    }