analyze_fit.py

#!/usr/bin/env python3
"""
FIT File Analyzer - Gear and Metrics Visualization
Analyzes FIT files to extract and visualize gear data and performance metrics
"""

import fitparse
import pandas as pd
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from datetime import datetime, timedelta
import numpy as np

def parse_fit_file(filename):
    """Parse FIT file and extract relevant data"""
    fitfile = fitparse.FitFile(filename)

    records = []
    gear_events = []

    # Extract record data (main metrics)
    for record in fitfile.get_messages('record'):
        record_data = {}

        for data in record:
            if data.name in ['timestamp', 'power', 'cadence', 'speed', 'altitude',
                           'heart_rate', 'position_lat', 'position_long',
                           'front_gear_num', 'rear_gear_num', 'front_gear', 'rear_gear']:
                record_data[data.name] = data.value

        if record_data:
            records.append(record_data)

    # Extract gear events
    for event in fitfile.get_messages('event'):
        gear_data = {}
        for data in event:
            if data.name in ['timestamp', 'front_gear', 'front_gear_num',
                           'rear_gear', 'rear_gear_num', 'gear_change_data']:
                gear_data[data.name] = data.value

        # Only keep events that have gear information
        if any(k in gear_data for k in ['front_gear', 'front_gear_num', 'rear_gear', 'rear_gear_num']):
            gear_events.append(gear_data)

    # Create main dataframe
    df = pd.DataFrame(records)

    # Create gear events dataframe
    if gear_events:
        gear_df = pd.DataFrame(gear_events)

        # Convert timestamps to datetime
        if 'timestamp' in df.columns:
            df['timestamp'] = pd.to_datetime(df['timestamp'])
        if 'timestamp' in gear_df.columns:
            gear_df['timestamp'] = pd.to_datetime(gear_df['timestamp'])

        # Merge gear data with records using forward fill
        # This propagates the last known gear state to each record
        df = pd.merge_asof(df.sort_values('timestamp'),
                          gear_df.sort_values('timestamp'),
                          on='timestamp',
                          direction='backward',
                          suffixes=('', '_gear'))

    return df

def calculate_gear_ratio(df):
    """Calculate gear ratio from front and rear gear data"""
    # Canyon Aeroad CF SLX 8 AXS SPEED specifications
    # Front chainrings: 48T (inner=1), 35T (outer=2)
    # Rear cassette: 10-30T with 12 sprockets
    # Cassette: 10, 11, 12, 13, 15, 17, 19, 21, 24, 27, 30T (SRAM 12-speed typical spacing)

    FRONT_CHAINRING = {
        1: 35,  # Small chainring
        2: 48   # Large chainring
    }

    # SRAM 10-30T 12-speed cassette (actual configuration)
    REAR_CASSETTE = {
        1: 10,
        2: 11,
        3: 12,
        4: 13,
        5: 14,
        6: 15,
        7: 17,
        8: 19,
        9: 21,
        10: 24,
        11: 27,
        12: 30
    }

    # Try different possible field names
    front_gear_num = None
    rear_gear_num = None

    if 'front_gear_num' in df.columns:
        front_gear_num = df['front_gear_num']
    elif 'front_gear' in df.columns:
        front_gear_num = df['front_gear']

    if 'rear_gear_num' in df.columns:
        rear_gear_num = df['rear_gear_num']
    elif 'rear_gear' in df.columns:
        rear_gear_num = df['rear_gear']

    # Map gear numbers to actual teeth count
    if front_gear_num is not None:
        df['front_teeth'] = front_gear_num.map(FRONT_CHAINRING)

    if rear_gear_num is not None:
        df['rear_teeth'] = rear_gear_num.map(REAR_CASSETTE)

    # Calculate gear ratio if both front and rear are available
    if 'front_teeth' in df.columns and 'rear_teeth' in df.columns:
        df['gear_ratio'] = df['front_teeth'] / df['rear_teeth'].replace(0, np.nan)

    return df

def create_visualizations(df, filename):
    """Create comprehensive visualization of gear and metrics data"""

    # Convert timestamp to datetime if it's not already
    if 'timestamp' in df.columns:
        if not pd.api.types.is_datetime64_any_dtype(df['timestamp']):
            df['timestamp'] = pd.to_datetime(df['timestamp'])

        # Create relative time in minutes from start
        df['time_minutes'] = (df['timestamp'] - df['timestamp'].min()).dt.total_seconds() / 60

    # Convert speed from m/s to km/h
    if 'speed' in df.columns:
        df['speed_kmh'] = df['speed'] * 3.6

    # Determine which metrics are available
    available_metrics = []
    metric_configs = {
        'power': {'label': 'Power (W)', 'color': 'red'},
        'cadence': {'label': 'Cadence (rpm)', 'color': 'blue'},
        'speed_kmh': {'label': 'Speed (km/h)', 'color': 'green'},
        'altitude': {'label': 'Elevation (m)', 'color': 'brown'},
        'heart_rate': {'label': 'Heart Rate (bpm)', 'color': 'purple'}
    }

    for metric in metric_configs.keys():
        if metric in df.columns and df[metric].notna().any():
            available_metrics.append(metric)

    # Check for gear data
    has_front_gear = 'front_teeth' in df.columns
    has_rear_gear = 'rear_teeth' in df.columns
    has_gear_ratio = 'gear_ratio' in df.columns
    has_combined_gear = has_front_gear and has_rear_gear

    # Calculate number of subplots needed (combine front/rear into one plot)
    gear_plots = 1 if has_combined_gear else 0
    gear_plots += 1 if has_gear_ratio else 0
    total_plots = gear_plots + len(available_metrics)

    if total_plots == 0:
        print("No data to visualize")
        return

    # Create figure with subplots
    fig, axes = plt.subplots(total_plots, 1, figsize=(14, 3*total_plots), sharex=True)

    # Make axes always a list for consistent indexing
    if total_plots == 1:
        axes = [axes]

    plot_idx = 0

    # Plot combined front and rear gear data
    if has_combined_gear:
        ax1 = axes[plot_idx]
        ax2 = ax1.twinx()

        # Plot front gear on left axis
        line1 = ax1.plot(df['time_minutes'], df['front_teeth'], 'o-',
                        color='darkblue', markersize=3, linewidth=2,
                        label='Front Chainring', alpha=0.8)
        ax1.set_ylabel('Front Chainring (T)', fontsize=10, color='darkblue')
        ax1.set_yticks([35, 48])
        ax1.set_ylim([30, 53])
        ax1.tick_params(axis='y', labelcolor='darkblue')

        # Plot rear gear on right axis
        line2 = ax2.plot(df['time_minutes'], df['rear_teeth'], 'o-',
                        color='darkgreen', markersize=2, linewidth=1,
                        label='Rear Sprocket', alpha=0.8)
        ax2.set_ylabel('Rear Sprocket (T)', fontsize=10, color='darkgreen')
        ax2.set_yticks([10, 15, 20, 25, 30])
        ax2.set_ylim([8, 32])
        ax2.tick_params(axis='y', labelcolor='darkgreen')

        # Add legend
        lines = line1 + line2
        labels = [l.get_label() for l in lines]
        ax1.legend(lines, labels, loc='upper left', fontsize=8)

        ax1.grid(True, alpha=0.3)
        plot_idx += 1

    if has_gear_ratio:
        axes[plot_idx].plot(df['time_minutes'], df['gear_ratio'], '-', color='orange', linewidth=1.5)
        axes[plot_idx].set_ylabel('Gear Ratio', fontsize=10)
        axes[plot_idx].grid(True, alpha=0.3)
        # Add reference lines for min/max ratios
        axes[plot_idx].axhline(y=48/10, color='red', linestyle='--', alpha=0.3, label='Max (48/10)')
        axes[plot_idx].axhline(y=35/30, color='blue', linestyle='--', alpha=0.3, label='Min (35/30)')
        axes[plot_idx].legend(loc='upper right', fontsize=8)
        plot_idx += 1

    # Plot metrics
    for metric in available_metrics:
        config = metric_configs[metric]
        axes[plot_idx].plot(df['time_minutes'], df[metric], '-',
                          color=config['color'], linewidth=1.5, alpha=0.8)
        axes[plot_idx].set_ylabel(config['label'], fontsize=10)
        axes[plot_idx].grid(True, alpha=0.3)
        plot_idx += 1

    # Set x-axis label only on bottom plot
    axes[-1].set_xlabel('Time (minutes)', fontsize=10)

    # Add title
    fig.suptitle(f'FIT File Analysis: {filename}', fontsize=14, fontweight='bold')

    # Adjust layout
    plt.tight_layout()

    # Save figure
    output_filename = filename.replace('.fit', '_analysis.png')
    plt.savefig(output_filename, dpi=150, bbox_inches='tight')
    print(f"Visualization saved to: {output_filename}")

    # Show plot
    plt.show()

    return fig

def print_data_summary(df):
    """Print summary of available data"""
    print("\n" + "="*60)
    print("DATA SUMMARY")
    print("="*60)
    print(f"Total records: {len(df)}")

    if 'timestamp' in df.columns:
        duration = (df['timestamp'].max() - df['timestamp'].min()).total_seconds() / 60
        print(f"Duration: {duration:.1f} minutes")

    print("\nAvailable fields:")
    for col in df.columns:
        non_null = df[col].notna().sum()
        pct = (non_null / len(df)) * 100
        print(f"  - {col}: {non_null}/{len(df)} ({pct:.1f}%)")

    print("\nMetrics summary:")
    metrics = ['power', 'cadence', 'speed', 'altitude', 'heart_rate']
    for metric in metrics:
        if metric in df.columns and df[metric].notna().any():
            print(f"\n  {metric.upper()}:")
            print(f"    Mean: {df[metric].mean():.2f}")
            print(f"    Max: {df[metric].max():.2f}")
            print(f"    Min: {df[metric].min():.2f}")

    print("="*60 + "\n")

def main():
    filename = '20861120068_ACTIVITY.fit'

    print(f"Parsing FIT file: {filename}")
    df = parse_fit_file(filename)

    print(f"Extracted {len(df)} records")

    # Calculate gear ratio
    df = calculate_gear_ratio(df)

    # Print summary
    print_data_summary(df)

    # Create visualizations
    create_visualizations(df, filename)

    # Save data to CSV for inspection
    csv_filename = filename.replace('.fit', '_data.csv')
    df.to_csv(csv_filename, index=False)
    print(f"Data exported to: {csv_filename}")

if __name__ == '__main__':
    main()