localgenai/opencode/scripts/fetch_garmin_data.py

import os
import sqlite3
from datetime import datetime
from garminconnect import Garmin
from garmin_fit_sdk import Decoder

def get_db_path():
    """Return path to SQLite database."""
    return os.path.join('garmin_data', 'garmin.db')

def init_database(db_path):
    """Initialize SQLite database with required tables."""
    os.makedirs(os.path.dirname(db_path), exist_ok=True)
    
    with sqlite3.connect(db_path) as conn:
        cursor = conn.cursor()
        
        # Create activities table
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS activities (
                id INTEGER PRIMARY KEY,
                start_time TEXT,
                end_time TEXT,
                distance REAL,
                duration INTEGER,
                activity_type TEXT,
                avg_heart_rate INTEGER,
                max_heart_rate INTEGER,
                avg_speed REAL,
                max_speed REAL,
                calories INTEGER,
                climb INTEGER,
                UNIQUE(id)
            )
        ''')
        
        # Create records table
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS records (
                id INTEGER PRIMARY KEY,
                activity_id INTEGER,
                timestamp TEXT,
                heart_rate INTEGER,
                cadence INTEGER,
                speed REAL,
                altitude REAL,
                latitude REAL,
                longitude REAL,
                power INTEGER,
                distance REAL,
                FOREIGN KEY (activity_id) REFERENCES activities (id)
            )
        ''')
        
        # Create indexes for better query performance
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_records_activity ON records(activity_id)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_records_time ON records(timestamp)')
        
        conn.commit()

def get_garmin_client(email, password):
    """Authenticate with Garmin Connect."""
    try:
        client = Garmin(email, password)
        client.login()
        return client
    except Exception as e:
        print(f"Error authenticating: {e}")
        return None

def download_activities(client):
    """Download activity list."""
    try:
        return client.get_activities(0, 1)  # Get most recent activity
    except Exception as e:
        print(f"Error downloading activity list: {e}")
        return None

def download_fit_file(client, activity_id, output_dir):
    """Download FIT file for activity."""
    try:
        fit_data = client.download_activity(activity_id, dl_fmt=client.ActivityDownloadFormat.ORIGINAL)
        fit_path = os.path.join(output_dir, f"{activity_id}.fit")
        os.makedirs(output_dir, exist_ok=True)
        with open(fit_path, 'wb') as f:
            f.write(fit_data)
        return fit_path
    except Exception as e:
        print(f"Error downloading FIT file: {e}")
        return None

def extract_fit_data(fit_file_path):
    """Extract data from FIT file."""
    try:
        decoder = Decoder()
        messages, errors = decoder.read_fit_file(fit_file_path)
        return messages, errors
    except Exception as e:
        print(f"Error decoding FIT file: {e}")
        return None, None

def save_to_database(db_path, messages):
    """Save extracted data to SQLite database."""
    with sqlite3.connect(db_path) as conn:
        cursor = conn.cursor()
        
        # Extract activity metadata
        activity_message = next((m for m in messages if m['type'] == 'session'), None)
        if not activity_message:
            print("No session message found")
            return
        
        # Insert activity
        cursor.execute('''
            INSERT OR IGNORE INTO activities 
            (id, start_time, end_time, distance, duration, activity_type, 
             avg_heart_rate, max_heart_rate, avg_speed, max_speed, 
             calories, climb) 
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        ''', (
            activity_message['message']['start_time'],
            activity_message['message']['end_time'],
            activity_message['message']['total_distance'],
            activity_message['message']['total_elapsed_time'],
            activity_message['message']['sport'],
            activity_message['message']['avg_heart_rate'],
            activity_message['message']['max_heart_rate'],
            activity_message['message']['avg_speed'],
            activity_message['message']['max_speed'],
            activity_message['message']['total_calories'],
            activity_message['message']['total_ascent']
        ))
        
        # Insert records
        record_messages = [m for m in messages if m['type'] == 'record']
        for message in record_messages:
            record = message['message']
            cursor.execute('''
                INSERT INTO records 
                (activity_id, timestamp, heart_rate, cadence, speed, 
                 altitude, latitude, longitude, power, distance) 
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            ''', (
                activity_message['message']['start_time'],
                record.get('timestamp'),
                record.get('heart_rate'),
                record.get('cadence'),
                record.get('speed'),
                record.get('altitude'),
                record.get('position_lat'),
                record.get('position_long'),
                record.get('power'),
                record.get('distance')
            ))
        
        conn.commit()
        print(f"Saved {len(record_messages)} records to database")

def main():
    # Authentication credentials
    email = os.getenv('GARMIN_EMAIL')
    password = os.getenv('GARMIN_PASSWORD')
    
    if not email or not password:
        print("Please set GARMIN_EMAIL and GARMIN_PASSWORD environment variables")
        return
    
    # Initialize database
    db_path = get_db_path()
    init_database(db_path)
    
    # Initialize client
    client = get_garmin_client(email, password)
    if not client:
        return
    
    # Get activities
    activities = download_activities(client)
    if not activities:
        return
    
    activity_id = activities[0]['activityId']
    
    # Download FIT file
    fit_path = download_fit_file(client, activity_id, 'garmin_data')
    if not fit_path:
        return
    
    # Extract data
    messages, errors = extract_fit_data(fit_path)
    if errors:
        print(f"FIT file errors: {errors}")
    
    # Save to database
    save_to_database(db_path, messages)

if __name__ == "__main__":
    main()