garmin-health-analysis/scripts/garmin_activity_files.py

#!/usr/bin/env python3
"""
Download and analyze Garmin activity FIT/GPX files.
Extract GPS, elevation, pace, heart rate, power, cadence, etc.
"""

import json
import sys
import os
from datetime import datetime
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))
from garmin_auth import get_client

# Check for optional dependencies
try:
    import fitparse
    HAS_FITPARSE = True
except ImportError:
    HAS_FITPARSE = False

try:
    import gpxpy
    import gpxpy.gpx
    HAS_GPXPY = True
except ImportError:
    HAS_GPXPY = False


def download_activity_file(client, activity_id, file_format="fit", output_dir="/tmp"):
    """Download activity FIT or GPX file."""
    try:
        output_path = f"{output_dir}/activity_{activity_id}.{file_format.lower()}"
        
        if file_format.lower() == "fit":
            data = client.download_activity(activity_id, dl_fmt=client.ActivityDownloadFormat.ORIGINAL)
        elif file_format.lower() == "gpx":
            data = client.download_activity(activity_id, dl_fmt=client.ActivityDownloadFormat.GPX)
        elif file_format.lower() == "tcx":
            data = client.download_activity(activity_id, dl_fmt=client.ActivityDownloadFormat.TCX)
        else:
            return {"error": f"Unsupported format: {file_format}"}
        
        with open(output_path, 'wb') as f:
            f.write(data)
        
        return {"file": output_path, "activity_id": activity_id, "format": file_format}
    
    except Exception as e:
        return {"error": str(e), "activity_id": activity_id}


def parse_fit_file(file_path):
    """Parse FIT file and extract all data points."""
    if not HAS_FITPARSE:
        return {"error": "fitparse library not installed. Run: pip install fitparse"}
    
    try:
        fitfile = fitparse.FitFile(file_path)
        
        # Extract different record types
        records = []
        laps = []
        sessions = []
        
        for record in fitfile.get_messages('record'):
            data_point = {}
            for field in record:
                if field.value is not None:
                    data_point[field.name] = field.value
            if data_point:
                records.append(data_point)
        
        for record in fitfile.get_messages('lap'):
            lap_data = {}
            for field in record:
                if field.value is not None:
                    lap_data[field.name] = field.value
            if lap_data:
                laps.append(lap_data)
        
        for record in fitfile.get_messages('session'):
            session_data = {}
            for field in record:
                if field.value is not None:
                    session_data[field.name] = field.value
            if session_data:
                sessions.append(session_data)
        
        return {
            "records": records,
            "laps": laps,
            "sessions": sessions,
            "total_records": len(records)
        }
    
    except Exception as e:
        return {"error": str(e)}


def parse_gpx_file(file_path):
    """Parse GPX file and extract track points."""
    if not HAS_GPXPY:
        return {"error": "gpxpy library not installed. Run: pip install gpxpy"}
    
    try:
        with open(file_path, 'r') as f:
            gpx = gpxpy.parse(f)
        
        points = []
        for track in gpx.tracks:
            for segment in track.segments:
                for point in segment.points:
                    points.append({
                        "latitude": point.latitude,
                        "longitude": point.longitude,
                        "elevation": point.elevation,
                        "time": point.time.isoformat() if point.time else None,
                        "speed": point.speed,
                        "hr": point.extensions.get("hr") if point.extensions else None
                    })
        
        return {
            "points": points,
            "total_points": len(points),
            "bounds": {
                "min_lat": gpx.get_bounds().min_latitude,
                "max_lat": gpx.get_bounds().max_latitude,
                "min_lon": gpx.get_bounds().min_longitude,
                "max_lon": gpx.get_bounds().max_longitude
            } if gpx.get_bounds() else None
        }
    
    except Exception as e:
        return {"error": str(e)}


def query_data_at_distance(data, distance_meters):
    """Find data point closest to a specific distance."""
    if "records" in data:
        records = data["records"]
    elif "points" in data:
        records = data["points"]
    else:
        return {"error": "No data records found"}
    
    # Find closest by distance
    closest = None
    min_diff = float('inf')
    
    for record in records:
        if "distance" in record:
            diff = abs(record["distance"] - distance_meters)
            if diff < min_diff:
                min_diff = diff
                closest = record
    
    return closest


def query_data_at_time(data, target_time):
    """Find data point at a specific time."""
    if "records" in data:
        records = data["records"]
    elif "points" in data:
        records = data["points"]
    else:
        return {"error": "No data records found"}
    
    # Parse target time
    if isinstance(target_time, str):
        try:
            target_dt = datetime.fromisoformat(target_time.replace("Z", "+00:00"))
        except:
            return {"error": f"Invalid time format: {target_time}"}
    else:
        target_dt = target_time
    
    target_ts = target_dt.timestamp()
    
    closest = None
    min_diff = float('inf')
    
    for record in records:
        if "timestamp" in record:
            if isinstance(record["timestamp"], datetime):
                rec_ts = record["timestamp"].timestamp()
            else:
                continue
            
            diff = abs(rec_ts - target_ts)
            if diff < min_diff:
                min_diff = diff
                closest = record
    
    return closest


def analyze_activity(data):
    """Analyze activity data and provide insights."""
    if "error" in data:
        return data
    
    records = data.get("records", [])
    if not records:
        return {"error": "No data records to analyze"}
    
    # Calculate statistics
    hr_values = [r.get("heart_rate") for r in records if r.get("heart_rate")]
    elevation_values = [r.get("altitude") or r.get("elevation") for r in records if r.get("altitude") or r.get("elevation")]
    speed_values = [r.get("speed") for r in records if r.get("speed")]
    cadence_values = [r.get("cadence") for r in records if r.get("cadence")]
    power_values = [r.get("power") for r in records if r.get("power")]
    
    analysis = {
        "total_points": len(records),
        "duration_seconds": None,
        "distance_meters": None,
        "heart_rate": {
            "avg": sum(hr_values) / len(hr_values) if hr_values else None,
            "max": max(hr_values) if hr_values else None,
            "min": min(hr_values) if hr_values else None
        },
        "elevation": {
            "max": max(elevation_values) if elevation_values else None,
            "min": min(elevation_values) if elevation_values else None,
            "gain": None  # Would need to calculate from sequential points
        },
        "speed": {
            "avg": sum(speed_values) / len(speed_values) if speed_values else None,
            "max": max(speed_values) if speed_values else None
        },
        "cadence": {
            "avg": sum(cadence_values) / len(cadence_values) if cadence_values else None
        } if cadence_values else None,
        "power": {
            "avg": sum(power_values) / len(power_values) if power_values else None,
            "max": max(power_values) if power_values else None
        } if power_values else None
    }
    
    # Get duration and distance from first/last records
    if records:
        if "timestamp" in records[0] and "timestamp" in records[-1]:
            if isinstance(records[0]["timestamp"], datetime):
                duration = (records[-1]["timestamp"] - records[0]["timestamp"]).total_seconds()
                analysis["duration_seconds"] = duration
        
        if "distance" in records[-1]:
            analysis["distance_meters"] = records[-1]["distance"]
    
    return analysis


def main():
    import argparse
    
    parser = argparse.ArgumentParser(description="Analyze Garmin activity files")
    parser.add_argument("action", choices=["download", "parse", "query", "analyze"],
                       help="Action to perform")
    parser.add_argument("--activity-id", type=int, help="Activity ID")
    parser.add_argument("--format", choices=["fit", "gpx", "tcx"], default="fit",
                       help="File format for download")
    parser.add_argument("--file", help="Path to local FIT/GPX file")
    parser.add_argument("--distance", type=float, help="Query data at distance (meters)")
    parser.add_argument("--time", help="Query data at time (ISO format)")
    parser.add_argument("--output-dir", default="/tmp", help="Output directory")
    
    args = parser.parse_args()
    
    if args.action == "download":
        if not args.activity_id:
            print('{"error": "activity-id required for download"}')
            sys.exit(1)
        
        client = get_client()
        if not client:
            print('{"error": "Not authenticated"}')
            sys.exit(1)
        
        result = download_activity_file(client, args.activity_id, args.format, args.output_dir)
        print(json.dumps(result, indent=2))
    
    elif args.action == "parse":
        if not args.file:
            print('{"error": "file path required for parse"}')
            sys.exit(1)
        
        if args.file.endswith('.fit'):
            result = parse_fit_file(args.file)
        elif args.file.endswith('.gpx'):
            result = parse_gpx_file(args.file)
        else:
            result = {"error": "Unsupported file type. Use .fit or .gpx"}
        
        print(json.dumps(result, indent=2, default=str))
    
    elif args.action == "query":
        if not args.file:
            print('{"error": "file path required for query"}')
            sys.exit(1)
        
        # First parse the file
        if args.file.endswith('.fit'):
            data = parse_fit_file(args.file)
        elif args.file.endswith('.gpx'):
            data = parse_gpx_file(args.file)
        else:
            print('{"error": "Unsupported file type"}')
            sys.exit(1)
        
        if "error" in data:
            print(json.dumps(data, indent=2))
            sys.exit(1)
        
        # Query
        if args.distance is not None:
            result = query_data_at_distance(data, args.distance)
        elif args.time:
            result = query_data_at_time(data, args.time)
        else:
            result = {"error": "Specify --distance or --time for query"}
        
        print(json.dumps(result, indent=2, default=str))
    
    elif args.action == "analyze":
        if not args.file:
            print('{"error": "file path required for analyze"}')
            sys.exit(1)
        
        # Parse and analyze
        if args.file.endswith('.fit'):
            data = parse_fit_file(args.file)
        elif args.file.endswith('.gpx'):
            data = parse_gpx_file(args.file)
        else:
            print('{"error": "Unsupported file type"}')
            sys.exit(1)
        
        result = analyze_activity(data)
        print(json.dumps(result, indent=2, default=str))


if __name__ == "__main__":
    main()