sstent/Garmin_Analyser
1
0
Fork 0
mirror of https://github.com/sstent/Garmin_Analyser.git synced 2025-12-06 08:01:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

removing old cli.py

Browse Source
This commit is contained in:
sstent
2025-10-06 16:42:15 -07:00
parent 3fae31d8b2
commit 44776dea4b
3 changed files with 262 additions and 667 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
README.md
View File

@@ -1,6 +1,6 @@
# Garmin Analyser # Garmin Analyser
A comprehensive Python application for analyzing Garmin workout data from FIT, TCX, and GPX files, as well as direct integration with Garmin Connect. Provides detailed power, heart rate, and performance analysis with beautiful visualizations and comprehensive reports. A comprehensive Python application for analyzing Garmin workout data from FIT, TCX, and GPX files, as well as direct integration with Garmin Connect. Provides detailed power, heart rate, and performance analysis with beautiful visualizations and comprehensive reports via a modular command-line interface.
## Features ## Features
@@ -112,6 +112,10 @@ Output:
Charts saved to output/charts/ when --charts is used Charts saved to output/charts/ when --charts is used
``` ```
## Deprecation Notice
The Text User Interface (TUI) and legacy analyzer have been removed in favor of the more robust and maintainable modular command-line interface (CLI). The project now relies exclusively on the modular CLI (`main.py` and `cli.py`) for all operations. All functionality from the legacy components has been successfully migrated to the modular stack.
## Setup credentials ## Setup credentials
Canonical environment variables: Canonical environment variables:

382
cli.py
View File

@@ -1,382 +0,0 @@
#!/usr/bin/env python3
"""
Command-line interface for Garmin Cycling Analyzer.
This module provides CLI tools for analyzing cycling workouts from Garmin devices.
"""
import argparse
import json
import os
import sys
from pathlib import Path
from typing import List, Optional
# Import from the new structure
from parsers.file_parser import FileParser
from analyzers.workout_analyzer import WorkoutAnalyzer
from config import settings
# Import for Garmin Connect functionality
try:
from clients.garmin_client import GarminClient
GARMIN_CLIENT_AVAILABLE = True
except ImportError:
GARMIN_CLIENT_AVAILABLE = False
print("Warning: Garmin Connect client not available. Install garminconnect package for download functionality.")
def create_parser() -> argparse.ArgumentParser:
"""Create the argument parser for CLI commands."""
parser = argparse.ArgumentParser(
description='Analyze cycling workouts from Garmin devices',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
%(prog)s analyze file.fit --output results.json
%(prog)s batch --input-dir ./workouts --output-dir ./results
%(prog)s config --show
"""
)
subparsers = parser.add_subparsers(dest='command', help='Available commands')
# Analyze command
analyze_parser = subparsers.add_parser('analyze', help='Analyze a single workout file')
analyze_parser.add_argument('file', help='Path to the workout file (.fit, .tcx, or .gpx)')
analyze_parser.add_argument('--output', '-o', help='Output file for results (JSON format)')
analyze_parser.add_argument('--cog-size', type=int, help='Chainring cog size (teeth)')
analyze_parser.add_argument('--format', choices=['json', 'summary'], default='json',
help='Output format (default: json)')
analyze_parser.add_argument('--ftp', type=int, help='Functional Threshold Power (W)')
analyze_parser.add_argument('--max-hr', type=int, help='Maximum heart rate (bpm)')
# Batch command
batch_parser = subparsers.add_parser('batch', help='Analyze multiple workout files')
batch_parser.add_argument('--input-dir', '-i', required=True, help='Directory containing workout files')
batch_parser.add_argument('--output-dir', '-o', required=True, help='Directory for output files')
batch_parser.add_argument('--cog-size', type=int, help='Chainring cog size (teeth)')
batch_parser.add_argument('--pattern', default='*.fit', help='File pattern to match (default: *.fit)')
batch_parser.add_argument('--ftp', type=int, help='Functional Threshold Power (W)')
batch_parser.add_argument('--max-hr', type=int, help='Maximum heart rate (bpm)')
# Config command
config_parser = subparsers.add_parser('config', help='Manage configuration')
config_parser.add_argument('--show', action='store_true', help='Show current configuration')
# Download command (from original garmin_cycling_analyzer.py)
download_parser = subparsers.add_parser('download', help='Download workouts from Garmin Connect')
download_parser.add_argument('--workout-id', '-w', type=int, help='Download specific workout by ID')
download_parser.add_argument('--download-all', action='store_true', help='Download all cycling activities')
download_parser.add_argument('--limit', type=int, default=50, help='Maximum number of activities to download')
# Reanalyze command (from original garmin_cycling_analyzer.py)
reanalyze_parser = subparsers.add_parser('reanalyze', help='Re-analyze downloaded workouts')
reanalyze_parser.add_argument('--input-dir', '-i', default='data', help='Directory containing downloaded workouts (default: data)')
reanalyze_parser.add_argument('--output-dir', '-o', default='reports', help='Directory for analysis reports (default: reports)')
return parser
def analyze_file(file_path: str, cog_size: Optional[int] = None,
ftp: Optional[int] = None, max_hr: Optional[int] = None,
output_format: str = 'json') -> dict:
"""
Analyze a single workout file.
Args:
file_path: Path to the workout file
cog_size: Chainring cog size for power estimation
ftp: Functional Threshold Power
max_hr: Maximum heart rate
output_format: Output format ('json' or 'summary')
Returns:
Analysis results as dictionary
"""
if not os.path.exists(file_path):
raise FileNotFoundError(f"File not found: {file_path}")
# Override settings with provided parameters
if ftp:
settings.FTP = ftp
if max_hr:
settings.MAX_HEART_RATE = max_hr
if cog_size:
settings.COG_SIZE = cog_size
# Parse the file
parser = FileParser()
workout = parser.parse_file(Path(file_path))
if not workout:
raise ValueError(f"Failed to parse file: {file_path}")
# Analyze the workout
analyzer = WorkoutAnalyzer()
results = analyzer.analyze_workout(workout)
return results
def batch_analyze(input_dir: str, output_dir: str, cog_size: Optional[int] = None,
ftp: Optional[int] = None, max_hr: Optional[int] = None,
pattern: str = '*.fit') -> List[str]:
"""
Analyze multiple workout files in a directory.
Args:
input_dir: Directory containing workout files
output_dir: Directory for output files
cog_size: Chainring cog size for power estimation
ftp: Functional Threshold Power
max_hr: Maximum heart rate
pattern: File pattern to match
Returns:
List of processed file paths
"""
input_path = Path(input_dir)
output_path = Path(output_dir)
if not input_path.exists():
raise FileNotFoundError(f"Input directory not found: {input_dir}")
# Create output directory if it doesn't exist
output_path.mkdir(parents=True, exist_ok=True)
# Override settings with provided parameters
if ftp:
settings.FTP = ftp
if max_hr:
settings.MAX_HEART_RATE = max_hr
if cog_size:
settings.COG_SIZE = cog_size
# Find matching files
files = list(input_path.glob(pattern))
processed_files = []
for file_path in files:
try:
print(f"Analyzing {file_path.name}...")
results = analyze_file(str(file_path))
# Save results
output_file = output_path / f"{file_path.stem}_analysis.json"
with open(output_file, 'w') as f:
json.dump(results, f, indent=2, default=str)
processed_files.append(str(file_path))
print(f" ✓ Results saved to {output_file.name}")
except Exception as e:
print(f" ✗ Error analyzing {file_path.name}: {e}")
return processed_files
def show_config():
"""Display current configuration."""
print("Current Configuration:")
print("-" * 30)
config_dict = {
'FTP': settings.FTP,
'MAX_HEART_RATE': settings.MAX_HEART_RATE,
'COG_SIZE': getattr(settings, 'COG_SIZE', None),
'ZONES_FILE': getattr(settings, 'ZONES_FILE', None),
'REPORTS_DIR': settings.REPORTS_DIR,
'DATA_DIR': settings.DATA_DIR,
}
for key, value in config_dict.items():
print(f"{key}: {value}")
def main():
"""Main CLI entry point."""
parser = create_parser()
args = parser.parse_args()
if not args.command:
parser.print_help()
return
try:
if args.command == 'analyze':
results = analyze_file(
args.file,
cog_size=getattr(args, 'cog_size', None),
ftp=getattr(args, 'ftp', None),
max_hr=getattr(args, 'max_hr', None),
output_format=args.format
)
if args.format == 'json':
if args.output:
with open(args.output, 'w') as f:
json.dump(results, f, indent=2, default=str)
print(f"Analysis complete. Results saved to {args.output}")
else:
print(json.dumps(results, indent=2, default=str))
elif args.format == 'summary':
print_summary(results)
elif args.command == 'batch':
processed = batch_analyze(
args.input_dir,
args.output_dir,
cog_size=getattr(args, 'cog_size', None),
ftp=getattr(args, 'ftp', None),
max_hr=getattr(args, 'max_hr', None),
pattern=args.pattern
)
print(f"\nBatch analysis complete. Processed {len(processed)} files.")
elif args.command == 'config':
if args.show:
show_config()
else:
show_config()
elif args.command == 'download':
download_workouts(
workout_id=getattr(args, 'workout_id', None),
download_all=args.download_all,
limit=getattr(args, 'limit', 50)
)
elif args.command == 'reanalyze':
reanalyze_workouts(
input_dir=getattr(args, 'input_dir', 'data'),
output_dir=getattr(args, 'output_dir', 'reports')
)
except Exception as e:
print(f"Error: {e}", file=sys.stderr)
sys.exit(1)
def print_summary(results: dict):
"""Print a human-readable summary of the analysis."""
metadata = results.get('metadata', {})
summary = results.get('summary', {})
print("\n" + "="*50)
print("WORKOUT SUMMARY")
print("="*50)
if metadata:
print(f"Activity: {metadata.get('activity_type', 'Unknown')}")
print(f"Date: {metadata.get('start_time', 'Unknown')}")
print(f"Duration: {summary.get('duration_minutes', 0):.1f} minutes")
if summary:
print(f"\nDistance: {summary.get('distance_km', 0):.1f} km")
print(f"Average Speed: {summary.get('avg_speed_kmh', 0):.1f} km/h")
if 'avg_power' in summary:
print(f"Average Power: {summary['avg_power']:.0f} W")
if 'max_power' in summary:
print(f"Max Power: {summary['max_power']:.0f} W")
print(f"Average Heart Rate: {summary.get('avg_heart_rate', 0):.0f} bpm")
print(f"Max Heart Rate: {summary.get('max_heart_rate', 0):.0f} bpm")
elevation = results.get('elevation_analysis', {})
if elevation:
print(f"Elevation Gain: {elevation.get('total_elevation_gain', 0):.0f} m")
zones = results.get('zones', {})
if zones and 'power' in zones:
print("\nPower Zone Distribution:")
for zone, data in zones['power'].items():
print(f" {zone}: {data['percentage']:.1f}% ({data['time_minutes']:.1f} min)")
print("="*50)
def download_workouts(workout_id: Optional[int] = None, download_all: bool = False, limit: int = 50):
"""
Download workouts from Garmin Connect.
Args:
workout_id: Specific workout ID to download
download_all: Download all cycling activities
limit: Maximum number of activities to download
"""
if not GARMIN_CLIENT_AVAILABLE:
print("Error: Garmin Connect client not available. Install garminconnect package:")
print(" pip install garminconnect")
return
try:
client = GarminClient()
if workout_id:
print(f"Downloading workout {workout_id}...")
success = client.download_workout(workout_id)
if success:
print(f"✓ Workout {workout_id} downloaded successfully")
else:
print(f"✗ Failed to download workout {workout_id}")
elif download_all:
print(f"Downloading up to {limit} cycling activities...")
downloaded = client.download_all_workouts(limit=limit)
print(f"✓ Downloaded {len(downloaded)} activities")
else:
print("Downloading latest cycling activity...")
success = client.download_latest_workout()
if success:
print("✓ Latest activity downloaded successfully")
else:
print("✗ Failed to download latest activity")
except Exception as e:
print(f"Error downloading workouts: {e}")
def reanalyze_workouts(input_dir: str = 'data', output_dir: str = 'reports'):
"""
Re-analyze all downloaded workouts.
Args:
input_dir: Directory containing downloaded workouts
output_dir: Directory for analysis reports
"""
input_path = Path(input_dir)
output_path = Path(output_dir)
if not input_path.exists():
print(f"Input directory not found: {input_dir}")
return
# Create output directory if it doesn't exist
output_path.mkdir(parents=True, exist_ok=True)
# Find all workout files
patterns = ['*.fit', '*.tcx', '*.gpx']
files = []
for pattern in patterns:
files.extend(input_path.glob(pattern))
if not files:
print(f"No workout files found in {input_dir}")
return
print(f"Found {len(files)} workout files to re-analyze")
processed = batch_analyze(
str(input_path),
str(output_path),
pattern='*.*' # Process all files
)
print(f"\nRe-analysis complete. Processed {len(processed)} files.")
if __name__ == '__main__':
main()

541
main.py
View File

@@ -33,136 +33,157 @@ def setup_logging(verbose: bool = False):
def parse_args() -> argparse.Namespace: def parse_args() -> argparse.Namespace:
"""Parse command line arguments. """Parse command line arguments."""
Returns:
Parsed arguments
"""
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description='Analyze Garmin workout data from files or Garmin Connect', description='Analyze Garmin workout data from files or Garmin Connect',
formatter_class=argparse.RawTextHelpFormatter,
epilog=( epilog=(
'Examples:\n' 'Examples:\n'
' Analyze latest workout from Garmin Connect: python main.py --garmin-connect\n' ' %(prog)s analyze --file path/to/workout.fit\n'
' Analyze specific workout by ID: python main.py --workout-id 123456789\n' ' %(prog)s batch --directory data/ --output-dir reports/\n'
' Download all cycling workouts: python main.py --download-all\n' ' %(prog)s download --all\n'
' Re-analyze all downloaded workouts: python main.py --reanalyze-all\n' ' %(prog)s reanalyze --input-dir data/\n'
' Analyze local FIT file: python main.py --file path/to/workout.fit\n' ' %(prog)s config --show'
' Analyze directory of workouts: python main.py --directory data/\n\n' )
'Configuration:\n'
' Set Garmin credentials in .env file: GARMIN_EMAIL and GARMIN_PASSWORD\n'
' Configure zones in config/config.yaml or use --zones flag\n'
' Override FTP with --ftp flag, max HR with --max-hr flag\n\n'
'Output:\n'
' Reports saved to output/ directory by default\n'
' Charts saved to output/charts/ when --charts is used'
),
formatter_class=argparse.RawTextHelpFormatter
) )
parser.add_argument(
'--config', '-c',
type=str,
default='config/config.yaml',
help='Configuration file path'
)
parser.add_argument( parser.add_argument(
'--verbose', '-v', '--verbose', '-v',
action='store_true', action='store_true',
help='Enable verbose logging' help='Enable verbose logging'
) )
# Input options subparsers = parser.add_subparsers(dest='command', help='Available commands')
input_group = parser.add_mutually_exclusive_group(required=True)
input_group.add_argument( # Analyze command
analyze_parser = subparsers.add_parser('analyze', help='Analyze a single workout or download from Garmin Connect')
analyze_parser.add_argument(
'--file', '-f', '--file', '-f',
type=str, type=str,
help='Path to workout file (FIT, TCX, or GPX)' help='Path to workout file (FIT, TCX, or GPX)'
) )
input_group.add_argument( analyze_parser.add_argument(
'--directory', '-d',
type=str,
help='Directory containing workout files'
)
input_group.add_argument(
'--garmin-connect', '--garmin-connect',
action='store_true', action='store_true',
help='Download from Garmin Connect' help='Download and analyze latest workout from Garmin Connect'
) )
input_group.add_argument( analyze_parser.add_argument(
'--workout-id', '--workout-id',
type=int, type=int,
help='Analyze specific workout by ID from Garmin Connect' help='Analyze specific workout by ID from Garmin Connect'
) )
input_group.add_argument( analyze_parser.add_argument(
'--download-all', '--ftp', type=int, help='Functional Threshold Power (W)'
action='store_true',
help='Download all cycling activities from Garmin Connect (no analysis)'
) )
input_group.add_argument( analyze_parser.add_argument(
'--reanalyze-all', '--max-hr', type=int, help='Maximum heart rate (bpm)'
action='store_true', )
help='Re-analyze all downloaded activities and generate reports' analyze_parser.add_argument(
'--zones', type=str, help='Path to zones configuration file'
)
analyze_parser.add_argument(
'--cog', type=int, help='Cog size (teeth) for power calculations. Auto-detected if not provided'
)
analyze_parser.add_argument(
'--output-dir', type=str, default='output', help='Output directory for reports and charts'
)
analyze_parser.add_argument(
'--format', choices=['html', 'pdf', 'markdown'], default='html', help='Report format'
)
analyze_parser.add_argument(
'--charts', action='store_true', help='Generate charts'
)
analyze_parser.add_argument(
'--report', action='store_true', help='Generate comprehensive report'
)
# Batch command
batch_parser = subparsers.add_parser('batch', help='Analyze multiple workout files in a directory')
batch_parser.add_argument(
'--directory', '-d', required=True, type=str, help='Directory containing workout files'
)
batch_parser.add_argument(
'--output-dir', type=str, default='output', help='Output directory for reports and charts'
)
batch_parser.add_argument(
'--format', choices=['html', 'pdf', 'markdown'], default='html', help='Report format'
)
batch_parser.add_argument(
'--charts', action='store_true', help='Generate charts'
)
batch_parser.add_argument(
'--report', action='store_true', help='Generate comprehensive report'
)
batch_parser.add_argument(
'--summary', action='store_true', help='Generate summary report for multiple workouts'
)
batch_parser.add_argument(
'--ftp', type=int, help='Functional Threshold Power (W)'
)
batch_parser.add_argument(
'--max-hr', type=int, help='Maximum heart rate (bpm)'
)
batch_parser.add_argument(
'--zones', type=str, help='Path to zones configuration file'
)
batch_parser.add_argument(
'--cog', type=int, help='Cog size (teeth) for power calculations. Auto-detected if not provided'
)
# Download command
download_parser = subparsers.add_parser('download', help='Download activities from Garmin Connect')
download_parser.add_argument(
'--all', action='store_true', help='Download all cycling activities'
)
download_parser.add_argument(
'--workout-id', type=int, help='Download specific workout by ID'
)
download_parser.add_argument(
'--limit', type=int, default=50, help='Maximum number of activities to download (with --all)'
)
download_parser.add_argument(
'--output-dir', type=str, default='data', help='Directory to save downloaded files'
) )
# Analysis options # Reanalyze command
parser.add_argument( reanalyze_parser = subparsers.add_parser('reanalyze', help='Re-analyze all downloaded activities')
'--ftp', reanalyze_parser.add_argument(
type=int, '--input-dir', type=str, default='data', help='Directory containing downloaded workouts'
help='Functional Threshold Power (W)'
) )
reanalyze_parser.add_argument(
parser.add_argument( '--output-dir', type=str, default='output', help='Output directory for reports and charts'
'--max-hr',
type=int,
help='Maximum heart rate (bpm)'
) )
reanalyze_parser.add_argument(
parser.add_argument( '--format', choices=['html', 'pdf', 'markdown'], default='html', help='Report format'
'--zones',
type=str,
help='Path to zones configuration file'
) )
reanalyze_parser.add_argument(
parser.add_argument( '--charts', action='store_true', help='Generate charts'
'--cog',
type=int,
help='Cog size (teeth) for power calculations. Auto-detected if not provided'
) )
reanalyze_parser.add_argument(
# Output options '--report', action='store_true', help='Generate comprehensive report'
parser.add_argument(
'--output-dir',
type=str,
default='output',
help='Output directory for reports and charts'
) )
reanalyze_parser.add_argument(
parser.add_argument( '--summary', action='store_true', help='Generate summary report for multiple workouts'
'--format',
choices=['html', 'pdf', 'markdown'],
default='html',
help='Report format'
) )
reanalyze_parser.add_argument(
parser.add_argument( '--ftp', type=int, help='Functional Threshold Power (W)'
'--charts',
action='store_true',
help='Generate charts'
) )
reanalyze_parser.add_argument(
parser.add_argument( '--max-hr', type=int, help='Maximum heart rate (bpm)'
'--report',
action='store_true',
help='Generate comprehensive report'
) )
reanalyze_parser.add_argument(
parser.add_argument( '--zones', type=str, help='Path to zones configuration file'
'--summary',
action='store_true',
help='Generate summary report for multiple workouts'
) )
reanalyze_parser.add_argument(
'--cog', type=int, help='Cog size (teeth) for power calculations. Auto-detected if not provided'
)
# Config command
config_parser = subparsers.add_parser('config', help='Manage configuration')
config_parser.add_argument(
'--show', action='store_true', help='Show current configuration'
)
return parser.parse_args() return parser.parse_args()
@@ -180,43 +201,57 @@ class GarminAnalyser:
# Create report templates # Create report templates
self.report_generator.create_report_templates() self.report_generator.create_report_templates()
def analyze_file(self, file_path: Path, cog_size: Optional[int] = None) -> dict: def _apply_analysis_overrides(self, args: argparse.Namespace):
"""Apply FTP, Max HR, and zones overrides from arguments."""
if hasattr(args, 'ftp') and args.ftp:
self.settings.FTP = args.ftp
if hasattr(args, 'max_hr') and args.max_hr:
self.settings.MAX_HEART_RATE = args.max_hr
if hasattr(args, 'zones') and args.zones:
self.settings.ZONES_FILE = args.zones
# Reload zones if the file path is updated
self.settings.load_zones(Path(args.zones))
def analyze_file(self, file_path: Path, args: argparse.Namespace) -> dict:
"""Analyze a single workout file. """Analyze a single workout file.
Args: Args:
file_path: Path to workout file file_path: Path to workout file
cog_size: Chainring teeth size for power calculations args: Command line arguments including analysis overrides
Returns: Returns:
Analysis results Analysis results
""" """
logging.info(f"Analyzing file: {file_path}") logging.info(f"Analyzing file: {file_path}")
self._apply_analysis_overrides(args)
# Parse workout file
workout = self.file_parser.parse_file(file_path) workout = self.file_parser.parse_file(file_path)
if not workout: if not workout:
raise ValueError(f"Failed to parse file: {file_path}") raise ValueError(f"Failed to parse file: {file_path}")
# Analyze workout # Determine cog size from args or auto-detect
analysis = self.workout_analyzer.analyze_workout(workout, cog_size=cog_size) cog_size = None
if hasattr(args, 'cog') and args.cog:
cog_size = args.cog
elif hasattr(args, 'auto_detect_cog') and args.auto_detect_cog:
# Implement auto-detection logic if needed, or rely on analyzer's default
pass
return { analysis = self.workout_analyzer.analyze_workout(workout, cog_size=cog_size)
'workout': workout, return {'workout': workout, 'analysis': analysis, 'file_path': file_path}
'analysis': analysis,
'file_path': file_path def batch_analyze_directory(self, directory: Path, args: argparse.Namespace) -> List[dict]:
} """Analyze multiple workout files in a directory.
def analyze_directory(self, directory: Path, cog_size: Optional[int] = None) -> List[dict]:
"""Analyze all workout files in a directory.
Args: Args:
directory: Directory containing workout files directory: Directory containing workout files
cog_size: Chainring teeth size for power calculations args: Command line arguments including analysis overrides
Returns: Returns:
List of analysis results List of analysis results
""" """
logging.info(f"Analyzing directory: {directory}") logging.info(f"Analyzing directory: {directory}")
self._apply_analysis_overrides(args)
results = [] results = []
supported_extensions = {'.fit', '.tcx', '.gpx'} supported_extensions = {'.fit', '.tcx', '.gpx'}
@@ -224,168 +259,105 @@ class GarminAnalyser:
for file_path in directory.rglob('*'): for file_path in directory.rglob('*'):
if file_path.suffix.lower() in supported_extensions: if file_path.suffix.lower() in supported_extensions:
try: try:
result = self.analyze_file(file_path, cog_size=cog_size) result = self.analyze_file(file_path, args)
results.append(result) results.append(result)
except Exception as e: except Exception as e:
logging.error(f"Error analyzing {file_path}: {e}") logging.error(f"Error analyzing {file_path}: {e}")
return results return results
def download_from_garmin(self, days: int = 30, cog_size: Optional[int] = None) -> List[dict]: def download_workouts(self, args: argparse.Namespace) -> List[dict]:
"""Download and analyze workouts from Garmin Connect. """Download workouts from Garmin Connect.
Args: Args:
days: Number of days to download args: Command line arguments for download options
cog_size: Chainring teeth size for power calculations
Returns: Returns:
List of analysis results List of downloaded workout data or analysis results
""" """
logging.info(f"Downloading workouts from Garmin Connect (last {days} days)") email, password = self.settings.get_garmin_credentials()
client = GarminClient(email=email, password=password)
download_output_dir = Path(getattr(args, 'output_dir', 'data'))
download_output_dir.mkdir(parents=True, exist_ok=True)
email, password = settings.get_garmin_credentials() downloaded_activities = []
client = GarminClient( if getattr(args, 'all', False):
email=email, logging.info(f"Downloading up to {getattr(args, 'limit', 50)} cycling activities...")
password=password downloaded_activities = client.download_all_workouts(limit=getattr(args, 'limit', 50), output_dir=download_output_dir)
) elif getattr(args, 'workout_id', None):
logging.info(f"Downloading workout {args.workout_id}...")
activity_path = client.download_activity_original(str(args.workout_id), output_dir=download_output_dir)
if activity_path:
downloaded_activities.append({'file_path': activity_path})
else:
logging.info("Downloading latest cycling activity...")
activity_path = client.download_latest_workout(output_dir=download_output_dir)
if activity_path:
downloaded_activities.append({'file_path': activity_path})
# Download workouts
workouts = client.get_all_cycling_workouts()
# Analyze each workout
results = [] results = []
for workout_summary in workouts: # Check if any analysis-related flags are set
try: if (getattr(args, 'charts', False)) or \
activity_id = workout_summary.get('activityId') (getattr(args, 'report', False)) or \
if not activity_id: (getattr(args, 'summary', False)) or \
logging.warning("Skipping workout with no activity ID.") (getattr(args, 'ftp', None)) or \
continue (getattr(args, 'max_hr', None)) or \
(getattr(args, 'zones', None)) or \
(getattr(args, 'cog', None)):
logging.info("Analyzing downloaded workouts...")
for activity_data in downloaded_activities:
file_path = activity_data['file_path']
try:
result = self.analyze_file(file_path, args)
results.append(result)
except Exception as e:
logging.error(f"Error analyzing downloaded file {file_path}: {e}")
return results if results else downloaded_activities # Return analysis results if analysis was requested, else just downloaded file paths
logging.info(f"Downloading workout file for activity ID: {activity_id}") def reanalyze_workouts(self, args: argparse.Namespace) -> List[dict]:
workout_file_path = client.download_activity_original(str(activity_id))
if workout_file_path and workout_file_path.exists():
workout = self.file_parser.parse_file(workout_file_path)
if workout:
analysis = self.workout_analyzer.analyze_workout(workout, cog_size=cog_size)
results.append({
'workout': workout,
'analysis': analysis,
'file_path': workout_file_path
})
else:
logging.error(f"Failed to download workout file for activity ID: {activity_id}")
except Exception as e:
logging.error(f"Error analyzing workout: {e}")
return results
def download_all_workouts(self) -> List[dict]:
"""Download all cycling activities from Garmin Connect without analysis.
Returns:
List of downloaded workouts
"""
email, password = settings.get_garmin_credentials()
client = GarminClient(
email=email,
password=password
)
# Download all cycling workouts
workouts = client.get_all_cycling_workouts()
# Save workouts to files
data_dir = Path('data')
data_dir.mkdir(exist_ok=True)
downloaded_workouts = []
for workout in workouts:
try:
# Generate filename
date_str = workout.metadata.start_time.strftime('%Y-%m-%d')
filename = f"{date_str}_{workout.metadata.activity_name.replace(' ', '_')}.fit"
file_path = data_dir / filename
# Save workout data
client.download_workout_file(workout.id, file_path)
downloaded_workouts.append({
'workout': workout,
'file_path': file_path
})
logging.info(f"Downloaded: {filename}")
except Exception as e:
logging.error(f"Error downloading workout {workout.id}: {e}")
logging.info(f"Downloaded {len(downloaded_workouts)} workouts")
return downloaded_workouts
def reanalyze_all_workouts(self, cog_size: Optional[int] = None) -> List[dict]:
"""Re-analyze all downloaded workout files. """Re-analyze all downloaded workout files.
Args: Args:
cog_size: Chainring teeth size for power calculations args: Command line arguments including input/output directories and analysis overrides
Returns: Returns:
List of analysis results List of analysis results
""" """
logging.info("Re-analyzing all downloaded workouts") logging.info("Re-analyzing all downloaded workouts")
self._apply_analysis_overrides(args)
data_dir = Path('data') input_dir = Path(getattr(args, 'input_dir', 'data'))
if not data_dir.exists(): if not input_dir.exists():
logging.error("No data directory found. Use --download-all first.") logging.error(f"Input directory not found: {input_dir}. Please download workouts first.")
return [] return []
results = [] results = []
supported_extensions = {'.fit', '.tcx', '.gpx'} supported_extensions = {'.fit', '.tcx', '.gpx'}
for file_path in data_dir.rglob('*'): for file_path in input_dir.rglob('*'):
if file_path.suffix.lower() in supported_extensions: if file_path.suffix.lower() in supported_extensions:
try: try:
result = self.analyze_file(file_path, cog_size=cog_size) result = self.analyze_file(file_path, args)
results.append(result) results.append(result)
except Exception as e: except Exception as e:
logging.error(f"Error re-analyzing {file_path}: {e}") logging.error(f"Error re-analyzing {file_path}: {e}")
logging.info(f"Re-analyzed {len(results)} workouts") logging.info(f"Re-analyzed {len(results)} workouts")
return results return results
def analyze_workout_by_id(self, workout_id: int, cog_size: Optional[int] = None) -> dict: def show_config(self):
"""Analyze a specific workout by ID from Garmin Connect. """Display current configuration."""
logging.info("Current Configuration:")
Args: logging.info("-" * 30)
workout_id: Garmin Connect workout ID config_dict = {
cog_size: Chainring teeth size for power calculations 'FTP': self.settings.FTP,
'MAX_HEART_RATE': self.settings.MAX_HEART_RATE,
Returns: 'ZONES_FILE': getattr(self.settings, 'ZONES_FILE', 'N/A'),
Analysis result 'REPORTS_DIR': self.settings.REPORTS_DIR,
""" 'DATA_DIR': self.settings.DATA_DIR,
logging.info(f"Analyzing workout ID: {workout_id}")
email, password = settings.get_garmin_credentials()
client = GarminClient(
email=email,
password=password
)
# Download specific workout
workout = client.get_workout_by_id(workout_id)
if not workout:
raise ValueError(f"Workout not found: {workout_id}")
# Analyze workout
analysis = self.workout_analyzer.analyze_workout(workout, cog_size=cog_size)
return {
'workout': workout,
'analysis': analysis,
'file_path': None
} }
for key, value in config_dict.items():
logging.info(f"{key}: {value}")
def generate_outputs(self, results: List[dict], args: argparse.Namespace): def generate_outputs(self, results: List[dict], args: argparse.Namespace):
"""Generate charts and reports based on results. """Generate charts and reports based on results.
@@ -393,26 +365,26 @@ class GarminAnalyser:
results: Analysis results results: Analysis results
args: Command line arguments args: Command line arguments
""" """
output_dir = Path(args.output_dir) output_dir = Path(getattr(args, 'output_dir', 'output'))
output_dir.mkdir(exist_ok=True) output_dir.mkdir(exist_ok=True)
if args.charts: if getattr(args, 'charts', False):
logging.info("Generating charts...") logging.info("Generating charts...")
for result in results: for result in results:
charts = self.chart_generator.generate_workout_charts( self.chart_generator.generate_workout_charts(
result['workout'], result['analysis'] result['workout'], result['analysis']
) )
logging.info(f"Charts saved to: {output_dir / 'charts'}") logging.info(f"Charts saved to: {output_dir / 'charts'}")
if args.report: if getattr(args, 'report', False):
logging.info("Generating reports...") logging.info("Generating reports...")
for result in results: for result in results:
report_path = self.report_generator.generate_workout_report( report_path = self.report_generator.generate_workout_report(
result['workout'], result['analysis'], args.format result['workout'], result['analysis'], getattr(args, 'format', 'html')
) )
logging.info(f"Report saved to: {report_path}") logging.info(f"Report saved to: {report_path}")
if args.summary and len(results) > 1: if getattr(args, 'summary', False) and len(results) > 1:
logging.info("Generating summary report...") logging.info("Generating summary report...")
workouts = [r['workout'] for r in results] workouts = [r['workout'] for r in results]
analyses = [r['analysis'] for r in results] analyses = [r['analysis'] for r in results]
@@ -428,56 +400,57 @@ def main():
setup_logging(args.verbose) setup_logging(args.verbose)
try: try:
# Override settings with command line arguments
if args.ftp:
settings.FTP = args.ftp
if args.max_hr:
settings.MAX_HEART_RATE = args.max_hr
# Initialize analyser
analyser = GarminAnalyser() analyser = GarminAnalyser()
# Analyze workouts
results = [] results = []
if args.file:
file_path = Path(args.file)
if not file_path.exists():
logging.error(f"File not found: {file_path}")
sys.exit(1)
results = [analyser.analyze_file(file_path, cog_size=args.cog)]
elif args.directory: if args.command == 'analyze':
if args.file:
file_path = Path(args.file)
if not file_path.exists():
logging.error(f"File not found: {file_path}")
sys.exit(1)
results = [analyser.analyze_file(file_path, args)]
elif args.garmin_connect or args.workout_id:
results = analyser.download_workouts(args)
else:
logging.error("Please specify a file, --garmin-connect, or --workout-id for the analyze command.")
sys.exit(1)
if results: # Only generate outputs if there are results
analyser.generate_outputs(results, args)
elif args.command == 'batch':
directory = Path(args.directory) directory = Path(args.directory)
if not directory.exists(): if not directory.exists():
logging.error(f"Directory not found: {directory}") logging.error(f"Directory not found: {directory}")
sys.exit(1) sys.exit(1)
results = analyser.analyze_directory(directory, cog_size=args.cog) results = analyser.batch_analyze_directory(directory, args)
if results: # Only generate outputs if there are results
analyser.generate_outputs(results, args)
elif args.garmin_connect: elif args.command == 'download':
results = analyser.download_from_garmin(cog_size=args.cog) # Download workouts and potentially analyze them if analysis flags are present
results = analyser.download_workouts(args)
if results:
# If analysis was part of download, generate outputs
if (getattr(args, 'charts', False) or getattr(args, 'report', False) or getattr(args, 'summary', False)):
analyser.generate_outputs(results, args)
else:
logging.info(f"Downloaded {len(results)} activities to {getattr(args, 'output_dir', 'data')}")
logging.info("Download command complete!")
elif args.workout_id: elif args.command == 'reanalyze':
try: results = analyser.reanalyze_workouts(args)
results = [analyser.analyze_workout_by_id(args.workout_id, cog_size=args.cog)] if results: # Only generate outputs if there are results
except ValueError as e: analyser.generate_outputs(results, args)
logging.error(f"Error: {e}")
sys.exit(1)
elif args.download_all: elif args.command == 'config':
analyser.download_all_workouts() if getattr(args, 'show', False):
logging.info("Download complete! Use --reanalyze-all to analyze downloaded workouts.") analyser.show_config()
return
elif args.reanalyze_all:
results = analyser.reanalyze_all_workouts(cog_size=args.cog)
# Generate outputs # Print summary for analyze, batch, reanalyze commands if results are available
if results: if args.command in ['analyze', 'batch', 'reanalyze'] and results:
analyser.generate_outputs(results, args)
# Print summary
if results:
logging.info(f"\nAnalysis complete! Processed {len(results)} workout(s)") logging.info(f"\nAnalysis complete! Processed {len(results)} workout(s)")
for result in results: for result in results:
workout = result['workout'] workout = result['workout']
@@ -490,7 +463,7 @@ def main():
) )
except Exception as e: except Exception as e:
logging.error(f"Error: {e}") logging.error(f"Error: {e}", file=sys.stderr)
if args.verbose: if args.verbose:
logging.exception("Full traceback:") logging.exception("Full traceback:")
sys.exit(1) sys.exit(1)