sstent/AICyclingCoach
1
0
Fork 0
mirror of https://github.com/sstent/AICyclingCoach.git synced 2026-01-25 00:21:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

migrate to garmin connect library

Browse Source
This commit is contained in:
sstent
2025-10-02 13:08:11 -07:00
parent c2dc64f322
commit 7d4ffcd902
10 changed files with 31445 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

92
backend/app/services/garmin.py
View File

@@ -1,17 +1,22 @@
import os
from pathlib import Path
import garth
from garth.exc import GarthException
import asyncio
import logging
from typing import List, Dict, Any, Optional
from datetime import datetime, timedelta
from garminconnect import (
Garmin,
GarminConnectAuthenticationError,
GarminConnectConnectionError,
GarminConnectTooManyRequestsError,
)
from sqlalchemy.ext.asyncio import AsyncSession
import logging
logger = logging.getLogger(__name__)
class GarminService:
class GarminConnectService:
"""Service for interacting with Garmin Connect API."""
def __init__(self, db: Optional[AsyncSession] = None):
@@ -20,58 +25,93 @@ class GarminService:
self.password = os.getenv("GARMIN_PASSWORD")
self.session_dir = Path("data/sessions")
self.session_dir.mkdir(parents=True, exist_ok=True)
self.client: Optional[Garmin] = None
async def _get_garmin_client(self) -> Garmin:
"""Get or create a Garmin client instance."""
if self.client:
return self.client
self.client = Garmin()
return self.client
async def authenticate(self) -> bool:
"""Authenticate with Garmin Connect and persist session."""
client = await self._get_garmin_client()
try:
await asyncio.to_thread(garth.resume, self.session_dir)
logger.info("Loaded existing Garmin session")
except (FileNotFoundError, GarthException):
logger.warning("No existing session found. Attempting fresh authentication.")
logger.debug("Attempting to resume existing Garmin session.")
await asyncio.to_thread(client.login, str(self.session_dir))
logger.info("Successfully loaded existing Garmin session.")
except (FileNotFoundError, GarminConnectAuthenticationError, GarminConnectConnectionError):
logger.debug("No existing Garmin session found or session invalid.")
logger.info("Attempting fresh authentication with Garmin Connect.")
if not self.username or not self.password:
logger.error("Garmin username or password not set in environment variables.")
raise GarminAuthError("Garmin username or password not configured.")
try:
await asyncio.to_thread(garth.login, self.username, self.password)
await asyncio.to_thread(garth.save, self.session_dir)
logger.info("Successfully authenticated with Garmin Connect")
logger.debug(f"Attempting to log in with username: {self.username}")
# The login method of python-garminconnect returns (token1, token2) on successful login
# and handles MFA internally if prompt_mfa is provided.
await asyncio.to_thread(client.login, self.username, self.password)
await asyncio.to_thread(client.garth.dump, str(self.session_dir)) # Save tokens using garth.dump
logger.info("Successfully authenticated and saved new Garmin session.")
except Exception as e:
logger.error(f"Garmin authentication failed: {str(e)}")
raise GarminAuthError(f"Authentication failed: {str(e)}")
logger.error(f"Garmin fresh authentication failed: {e}", exc_info=True)
raise GarminAuthError(f"Authentication failed: {e}")
return True
async def get_activities(self, limit: int = 10, start_date: datetime = None) -> List[Dict[str, Any]]:
async def get_activities(self, limit: int = 10, start_date: Optional[datetime] = None) -> List[Dict[str, Any]]:
"""Fetch recent activities from Garmin Connect."""
await self.authenticate()
client = await self._get_garmin_client()
if not start_date:
start_date = datetime.now() - timedelta(days=7)
# Convert start_date to YYYY-MM-DD string as required by garminconnect.get_activities_by_date
start_date_str = start_date.strftime("%Y-%m-%d") if start_date else (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d")
end_date_str = datetime.now().strftime("%Y-%m-%d")
try:
logger.debug(f"Fetching Garmin activities with limit={limit}, start_date={start_date_str}.")
activities = await asyncio.to_thread(
garth.connectapi,
"/activity-service/activity/activities",
params={"limit": limit, "start": start_date.strftime("%Y-%m-%d")},
client.get_activities_by_date,
start_date_str,
end_date_str,
limit=limit
)
logger.info(f"Fetched {len(activities)} activities from Garmin")
logger.info(f"Successfully fetched {len(activities)} activities from Garmin.")
logger.debug(f"Garmin activities data: {activities}")
return activities or []
except (GarminConnectConnectionError, GarminConnectTooManyRequestsError) as e:
logger.error(f"Failed to fetch activities from Garmin: {e}", exc_info=True)
raise GarminAPIError(f"Failed to fetch activities: {e}")
except GarminConnectAuthenticationError as e:
logger.error(f"Garmin authentication failed while fetching activities: {e}", exc_info=True)
raise GarminAuthError(f"Authentication failed: {e}")
except Exception as e:
logger.error(f"Failed to fetch activities: {str(e)}")
raise GarminAPIError(f"Failed to fetch activities: {str(e)}")
logger.error(f"An unexpected error occurred while fetching activities from Garmin: {e}", exc_info=True)
raise GarminAPIError(f"Unexpected error: {e}")
async def get_activity_details(self, activity_id: str) -> Dict[str, Any]:
"""Get detailed activity data including metrics."""
await self.authenticate()
client = await self._get_garmin_client()
try:
logger.debug(f"Fetching detailed data for activity ID: {activity_id}.")
details = await asyncio.to_thread(
garth.connectapi, f"/activity-service/activity/{activity_id}"
client.get_activity_details, activity_id
)
logger.info(f"Fetched details for activity {activity_id}")
logger.info(f"Successfully fetched details for activity ID: {activity_id}.")
logger.debug(f"Garmin activity {activity_id} details: {details}")
return details
except (GarminConnectConnectionError, GarminConnectTooManyRequestsError) as e:
logger.error(f"Failed to fetch activity details for {activity_id}: {e}", exc_info=True)
raise GarminAPIError(f"Failed to fetch activity details: {e}")
except GarminConnectAuthenticationError as e:
logger.error(f"Garmin authentication failed while fetching activity details: {e}", exc_info=True)
raise GarminAuthError(f"Authentication failed: {e}")
except Exception as e:
logger.error(f"Failed to fetch activity details for {activity_id}: {str(e)}")
raise GarminAPIError(f"Failed to fetch activity details: {str(e)}")
logger.error(f"An unexpected error occurred while fetching activity details for {activity_id}: {e}", exc_info=True)
raise GarminAPIError(f"Unexpected error: {e}")
class GarminAuthError(Exception):

73
backend/app/services/workout_sync.py
View File

@@ -1,6 +1,6 @@
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, desc
from backend.app.services.garmin import GarminService, GarminAPIError, GarminAuthError
from backend.app.services.garmin import GarminConnectService as GarminService, GarminAPIError, GarminAuthError
from backend.app.models.workout import Workout
from backend.app.models.garmin_sync_log import GarminSyncLog, GarminSyncStatus
from datetime import datetime, timedelta
@@ -20,46 +20,65 @@ class WorkoutSyncService:
async def sync_recent_activities(self, days_back: int = 7) -> int:
"""Sync recent Garmin activities to database."""
logger.info(f"Starting Garmin activity sync for the last {days_back} days.")
sync_log = None # Initialize sync_log
try:
# Create sync log entry
sync_log = GarminSyncLog(status="in_progress")
sync_log = GarminSyncLog(status=GarminSyncStatus.IN_PROGRESS)
self.db.add(sync_log)
await self.db.commit()
await self.db.refresh(sync_log) # Refresh to get the generated ID
logger.debug(f"Created new GarminSyncLog with ID: {sync_log.id}")
# Calculate start date
start_date = datetime.now() - timedelta(days=days_back)
logger.debug(f"Fetching activities from Garmin starting from: {start_date}")
# Fetch activities from Garmin
activities = await self.garmin_service.get_activities(
limit=50, start_date=start_date
limit=50, start_date=start_date, end_date=datetime.now()
)
logger.debug(f"Found {len(activities)} activities from Garmin.")
synced_count = 0
for activity in activities:
activity_id = activity['activityId']
activity_id = str(activity['activityId'])
logger.debug(f"Processing activity ID: {activity_id}")
if await self.activity_exists(activity_id):
logger.debug(f"Activity {activity_id} already exists in DB, skipping.")
continue
# Get full activity details with retry logic
max_retries = 3
details = None
for attempt in range(max_retries):
try:
logger.debug(f"Attempt {attempt + 1} to fetch details for activity {activity_id}")
details = await self.garmin_service.get_activity_details(activity_id)
logger.debug(f"Successfully fetched details for activity {activity_id}.")
break
except (GarminAPIError, GarminAuthError) as e:
logger.warning(f"Failed to fetch details for {activity_id} (attempt {attempt + 1}/{max_retries}): {e}")
if attempt == max_retries - 1:
logger.error(f"Max retries reached for activity {activity_id}. Skipping details fetch.", exc_info=True)
raise
await asyncio.sleep(2 ** attempt)
logger.warning(f"Retrying activity details fetch for {activity_id}, attempt {attempt + 1}")
if details is None:
logger.warning(f"Skipping activity {activity_id} due to failure in fetching details.")
continue
# Merge basic activity data with detailed metrics
full_activity = {**activity, **details}
logger.debug(f"Merged activity data for {activity_id}.")
# Parse and create workout
workout_data = await self.parse_activity_data(full_activity)
workout = Workout(**workout_data)
self.db.add(workout)
synced_count += 1
logger.debug(f"Added workout {workout.garmin_activity_id} to session.")
# Update sync log
sync_log.status = GarminSyncStatus.COMPLETED
@@ -67,48 +86,58 @@ class WorkoutSyncService:
sync_log.last_sync_time = datetime.now()
await self.db.commit()
logger.info(f"Successfully synced {synced_count} activities")
logger.info(f"Successfully synced {synced_count} activities.")
return synced_count
except GarminAuthError as e:
sync_log.status = GarminSyncStatus.AUTH_FAILED
sync_log.error_message = str(e)
await self.db.commit()
logger.error(f"Garmin authentication failed: {str(e)}")
logger.error(f"Garmin authentication failed during sync: {e}", exc_info=True)
if sync_log:
sync_log.status = GarminSyncStatus.AUTH_FAILED
sync_log.error_message = str(e)
await self.db.commit()
raise
except GarminAPIError as e:
sync_log.status = GarminSyncStatus.FAILED
sync_log.error_message = str(e)
await self.db.commit()
logger.error(f"Garmin API error during sync: {str(e)}")
logger.error(f"Garmin API error during sync: {e}", exc_info=True)
if sync_log:
sync_log.status = GarminSyncStatus.FAILED
sync_log.error_message = str(e)
await self.db.commit()
raise
except Exception as e:
sync_log.status = GarminSyncStatus.FAILED
sync_log.error_message = str(e)
await self.db.commit()
logger.error(f"Unexpected error during sync: {str(e)}")
logger.error(f"Unexpected error during Garmin sync: {e}", exc_info=True)
if sync_log:
sync_log.status = GarminSyncStatus.FAILED
sync_log.error_message = str(e)
await self.db.commit()
raise
async def get_latest_sync_status(self):
"""Get the most recent sync log entry"""
"""Get the most recent sync log entry."""
logger.debug("Fetching latest Garmin sync status.")
result = await self.db.execute(
select(GarminSyncLog)
.order_by(desc(GarminSyncLog.created_at))
.limit(1)
)
return await result.scalar_one_or_none()
status = result.scalar_one_or_none()
logger.debug(f"Latest sync status: {status.status if status else 'None'}")
return status
async def activity_exists(self, garmin_activity_id: str) -> bool:
"""Check if activity already exists in database."""
logger.debug(f"Checking if activity {garmin_activity_id} exists in database.")
result = await self.db.execute(
select(Workout).where(Workout.garmin_activity_id == garmin_activity_id)
)
return result.scalar_one_or_none() is not None # Remove the await here
exists = result.scalar_one_or_none() is not None
logger.debug(f"Activity {garmin_activity_id} exists: {exists}")
return exists
async def parse_activity_data(self, activity: Dict[str, Any]) -> Dict[str, Any]:
"""Parse Garmin activity data into workout model format."""
logger.debug(f"Parsing activity data for Garmin activity ID: {activity.get('activityId')}")
return {
"garmin_activity_id": activity['activityId'],
"garmin_activity_id": str(activity['activityId']),
"activity_type": activity.get('activityType', {}).get('typeKey'),
"start_time": datetime.fromisoformat(activity['startTimeLocal'].replace('Z', '+00:00')),
"duration_seconds": activity.get('duration'),

13
backend/tests/services/test_garmin.py
View File

@@ -1,7 +1,7 @@
import os
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from backend.app.services.garmin import GarminService, GarminAuthError, GarminAPIError
from backend.app.services.garmin import GarminConnectService as GarminService, GarminAuthError, GarminAPIError
from backend.app.models.garmin_sync_log import GarminSyncStatus
from datetime import datetime, timedelta
import garth # Import garth for type hinting
@@ -11,13 +11,12 @@ def mock_env_vars():
with patch.dict(os.environ, {"GARMIN_USERNAME": "test_user", "GARMIN_PASSWORD": "test_password"}):
yield
def create_garth_client_mock():
mock_client_instance = MagicMock(spec=garth.Client)
mock_client_instance.login = AsyncMock(return_value=True)
def create_garmin_client_mock():
mock_client_instance = MagicMock(spec=GarminService) # Use GarminService (which is GarminConnectService)
mock_client_instance.authenticate = AsyncMock(return_value=True)
mock_client_instance.get_activities = AsyncMock(return_value=[])
mock_client_instance.get_activity = AsyncMock(return_value={})
mock_client_instance.load = AsyncMock(side_effect=FileNotFoundError)
mock_client_instance.save = AsyncMock()
mock_client_instance.get_activity_details = AsyncMock(return_value={})
mock_client_instance.is_authenticated = MagicMock(return_value=True)
return mock_client_instance
@pytest.mark.asyncio

39
backend/tests/services/test_garmin_functional.py
View File

@@ -9,7 +9,7 @@ from datetime import datetime, timedelta
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from backend.app.services.garmin import GarminService, GarminAuthError, GarminAPIError
from backend.app.services.garmin import GarminConnectService as GarminService, GarminAuthError, GarminAPIError
from backend.app.services.workout_sync import WorkoutSyncService
from backend.app.models.workout import Workout
from backend.app.models.garmin_sync_log import GarminSyncLog
@@ -36,12 +36,12 @@ class TestGarminAuthentication:
'GARMIN_USERNAME': 'test@example.com',
'GARMIN_PASSWORD': 'testpass123'
})
@patch('garth.Client')
@patch('garminconnect.Garmin')
async def test_successful_authentication(self, mock_client_class, garmin_service):
"""Test successful authentication with valid credentials."""
# Setup mock client
mock_client = MagicMock()
mock_client.login = AsyncMock(return_value=True)
mock_client.login = AsyncMock(return_value=(None, None))
mock_client.save = MagicMock()
mock_client_class.return_value = mock_client
@@ -56,7 +56,7 @@ class TestGarminAuthentication:
'GARMIN_USERNAME': 'invalid@example.com',
'GARMIN_PASSWORD': 'wrongpass'
})
@patch('garth.Client')
@patch('garminconnect.Garmin')
async def test_failed_authentication(self, mock_client_class, garmin_service):
"""Test authentication failure with invalid credentials."""
# Setup mock client to raise exception
@@ -72,21 +72,20 @@ class TestGarminAuthentication:
'GARMIN_USERNAME': 'test@example.com',
'GARMIN_PASSWORD': 'testpass123'
})
@patch('garth.Client')
@patch('garminconnect.Garmin')
async def test_session_reuse(self, mock_client_class, garmin_service):
"""Test that existing sessions are reused."""
# Setup mock client with load method
mock_client = MagicMock()
mock_client.load = MagicMock(return_value=True)
mock_client.login = AsyncMock() # Should not be called
mock_client.login = AsyncMock(return_value=(None, None)) # Login handles loading from tokenstore
mock_client_class.return_value = mock_client
# Test authentication
result = await garmin_service.authenticate()
assert result is True
mock_client.load.assert_called_once()
mock_client.login.assert_not_awaited()
mock_client.login.assert_awaited_once_with(tokenstore=garmin_service.session_dir)
mock_client.save.assert_not_called()
class TestWorkoutSyncing:
@@ -96,7 +95,7 @@ class TestWorkoutSyncing:
'GARMIN_USERNAME': 'test@example.com',
'GARMIN_PASSWORD': 'testpass123'
})
@patch('garth.Client')
@patch('garminconnect.Garmin')
async def test_successful_sync_recent_activities(self, mock_client_class, workout_sync_service, db_session):
"""Test successful synchronization of recent activities."""
# Setup mock Garmin client
@@ -136,8 +135,8 @@ class TestWorkoutSyncing:
'elevationGain': 500.0
}
mock_client.get_activities = MagicMock(return_value=mock_activities)
mock_client.get_activity = MagicMock(return_value=mock_details)
mock_client.get_activities_by_date = MagicMock(return_value=mock_activities)
mock_client.get_activity_details = MagicMock(return_value=mock_details)
mock_client_class.return_value = mock_client
# Test sync
@@ -198,7 +197,7 @@ class TestWorkoutSyncing:
}
]
mock_client.get_activities = MagicMock(return_value=mock_activities)
mock_client.get_activities_by_date = MagicMock(return_value=mock_activities)
mock_client_class.return_value = mock_client
# Test sync
@@ -210,7 +209,7 @@ class TestWorkoutSyncing:
'GARMIN_USERNAME': 'invalid@example.com',
'GARMIN_PASSWORD': 'wrongpass'
})
@patch('garth.Client')
@patch('garminconnect.Garmin')
async def test_sync_with_auth_failure(self, mock_client_class, workout_sync_service, db_session):
"""Test sync failure due to authentication error."""
# Setup mock client to fail authentication
@@ -234,14 +233,14 @@ class TestWorkoutSyncing:
'GARMIN_USERNAME': 'test@example.com',
'GARMIN_PASSWORD': 'testpass123'
})
@patch('garth.Client')
@patch('garminconnect.Garmin')
async def test_sync_with_api_error(self, mock_client_class, workout_sync_service, db_session):
"""Test sync failure due to API error."""
# Setup mock client
mock_client = MagicMock()
mock_client.login = AsyncMock(return_value=True)
mock_client.save = MagicMock()
mock_client.get_activities = MagicMock(side_effect=Exception("API rate limit exceeded"))
mock_client.get_activities_by_date = MagicMock(side_effect=Exception("API rate limit exceeded"))
mock_client_class.return_value = mock_client
# Test sync
@@ -265,7 +264,7 @@ class TestErrorHandling:
'GARMIN_USERNAME': 'test@example.com',
'GARMIN_PASSWORD': 'testpass123'
})
@patch('garth.Client')
@patch('garminconnect.Garmin')
async def test_activity_detail_fetch_retry(self, mock_client_class, workout_sync_service, db_session):
"""Test retry logic when fetching activity details fails."""
# Setup mock client
@@ -283,9 +282,9 @@ class TestErrorHandling:
}
]
mock_client.get_activities = MagicMock(return_value=mock_activities)
mock_client.get_activities_by_date = MagicMock(return_value=mock_activities)
# First two calls fail, third succeeds
mock_client.get_activity = MagicMock(side_effect=[
mock_client.get_activity_details = MagicMock(side_effect=[
Exception("Temporary error"),
Exception("Temporary error"),
{
@@ -305,4 +304,4 @@ class TestErrorHandling:
assert synced_count == 1
# Verify get_activity was called 3 times (initial + 2 retries)
assert mock_client.get_activity.call_count == 3
assert mock_client.get_activity_details.call_count == 3

2
backend/tests/services/test_garmin_sync_functional.py
View File

@@ -5,7 +5,7 @@ from sqlalchemy.pool import StaticPool
from sqlalchemy import select
from backend.app.database import Base
from backend.app.services.workout_sync import WorkoutSyncService
from backend.app.services.garmin import GarminService, GarminAPIError, GarminAuthError
from backend.app.services.garmin import GarminConnectService as GarminService, GarminAPIError, GarminAuthError
from backend.app.models.workout import Workout
from backend.app.models.garmin_sync_log import GarminSyncLog, GarminSyncStatus
from datetime import datetime, timedelta

31241
git_scape_python_garminconnect_digest.txt Executable file
View File

File diff suppressed because it is too large Load Diff

78
main.py
View File

@@ -11,6 +11,7 @@ from pathlib import Path
import sys
from typing import Optional
from datetime import datetime
import os
from textual.app import App, ComposeResult
from textual.containers import Container, Horizontal, Vertical
@@ -83,7 +84,7 @@ class CyclingCoachApp(App):
self.current_view = "dashboard"
self._setup_logging()
def _setup_logging(self):
def _setup_logging(self, level=logging.INFO):
"""Configure logging for the TUI application."""
# Create logs directory
logs_dir = Path("logs")
@@ -91,7 +92,7 @@ class CyclingCoachApp(App):
# Set up logger
logger = logging.getLogger("cycling_coach")
logger.setLevel(logging.INFO)
logger.setLevel(level)
# Add Textual handler for TUI-compatible logging
textual_handler = TextualHandler()
@@ -109,7 +110,6 @@ class CyclingCoachApp(App):
def compose(self) -> ComposeResult:
"""Create the main application layout."""
sys.stdout.write("CyclingCoachApp.compose: START\n")
yield Header()
with Container():
@@ -144,18 +144,14 @@ class CyclingCoachApp(App):
yield RouteView(id="route-view")
yield Footer()
sys.stdout.write("CyclingCoachApp.compose: END\n")
async def on_mount(self) -> None:
"""Initialize the application when mounted."""
sys.stdout.write("CyclingCoachApp.on_mount: START\n")
# Set initial active navigation and tab
self.query_one("#nav-dashboard").add_class("-active")
tabs = self.query_one("#main-tabs", TabbedContent)
if tabs:
tabs.active = "dashboard-tab"
sys.stdout.write("CyclingCoachApp.on_mount: Activated dashboard-tab\n")
sys.stdout.write("CyclingCoachApp.on_mount: END\n")
async def on_button_pressed(self, event: Button.Pressed) -> None:
"""Handle navigation button presses."""
@@ -186,28 +182,45 @@ class CyclingCoachApp(App):
@on(TabbedContent.TabActivated)
async def on_tab_activated(self, event: TabbedContent.TabActivated) -> None:
sys.stdout.write(f"CyclingCoachApp.on_tab_activated: Tab {event.pane.id} activated\n")
"""Handle tab activation to load data for the active tab."""
if event.pane.id == "workouts-tab":
workout_view = self.query_one("#workout-view", WorkoutView)
sys.stdout.write("CyclingCoachApp.on_tab_activated: Calling workout_view.load_data()\n")
workout_view.load_data()
def action_quit(self) -> None:
self.exit()
async def init_db_async():
logger = logging.getLogger("cycling_coach")
try:
await init_db()
sys.stdout.write("Database initialized successfully\n")
logger.info("Database initialized successfully")
except Exception as e:
sys.stdout.write(f"Database initialization failed: {e}\n")
logger.error(f"Database initialization failed: {e}")
sys.exit(1)
async def sync_garmin_activities_cli():
"""Sync Garmin activities in CLI format without starting TUI."""
logger = logging.getLogger("cycling_coach")
try:
logger.info("Initializing database for Garmin sync...")
await init_db_async()
logger.info("Starting Garmin activity sync...")
async with AsyncSessionLocal() as db:
workout_service = WorkoutService(db)
await workout_service.sync_garmin_activities()
logger.info("Garmin activity sync completed successfully.")
except Exception as e:
logger.error(f"Error during Garmin activity sync: {e}")
sys.exit(1)
async def list_workouts_cli():
"""Display workouts in CLI format without starting TUI."""
logger = logging.getLogger("cycling_coach")
try:
# Initialize database
logger.info("Initializing database for listing workouts...")
await init_db_async()
# Get workouts using WorkoutService
@@ -216,14 +229,14 @@ async def list_workouts_cli():
workouts = await workout_service.get_workouts(limit=50)
if not workouts:
print("No workouts found.")
logger.info("No workouts found.")
return
# Print header
print("AI Cycling Coach - Workouts")
print("=" * 80)
print(f"{'Date':<12} {'Type':<15} {'Duration':<10} {'Distance':<10} {'Avg HR':<8} {'Avg Power':<10}")
print("-" * 80)
logger.info("AI Cycling Coach - Workouts")
logger.info("=" * 80)
logger.info(f"{'Date':<12} {'Type':<15} {'Duration':<10} {'Distance':<10} {'Avg HR':<8} {'Avg Power':<10}")
logger.info("-" * 80)
# Print each workout
for workout in workouts:
@@ -257,12 +270,12 @@ async def list_workouts_cli():
if workout.get("avg_power"):
power_str = f"{workout['avg_power']} W"
print(f"{date_str:<12} {workout.get('activity_type', 'Unknown')[:14]:<15} {duration_str:<10} {distance_str:<10} {hr_str:<8} {power_str:<10}")
logger.info(f"{date_str:<12} {workout.get('activity_type', 'Unknown')[:14]:<15} {duration_str:<10} {distance_str:<10} {hr_str:<8} {power_str:<10}")
print(f"\nTotal workouts: {len(workouts)}")
logger.info(f"\nTotal workouts: {len(workouts)}")
except Exception as e:
print(f"Error listing workouts: {e}")
logger.error(f"Error listing workouts: {e}")
sys.exit(1)
def main():
@@ -270,12 +283,30 @@ def main():
parser = argparse.ArgumentParser(description="AI Cycling Coach - Terminal Training Interface")
parser.add_argument("--list-workouts", action="store_true",
help="List all workouts in CLI format and exit")
parser.add_argument("--sync-garmin", action="store_true",
help="Sync Garmin activities and exit")
parser.add_argument("--debug", action="store_true", help="Enable debug logging")
args = parser.parse_args()
log_level = logging.DEBUG if args.debug else logging.INFO
# Handle CLI commands that don't need TUI
if args.list_workouts:
asyncio.run(list_workouts_cli())
if args.list_workouts or args.sync_garmin:
# Configure logging using the app's setup
app = CyclingCoachApp()
app._setup_logging(level=log_level)
# Get the configured logger
cli_logger = logging.getLogger("cycling_coach")
if args.list_workouts:
asyncio.run(list_workouts_cli())
elif args.sync_garmin:
asyncio.run(sync_garmin_activities_cli())
# Exit gracefully after CLI commands
return
return
# Create data directory if it doesn't exist
@@ -288,11 +319,8 @@ def main():
asyncio.run(init_db_async())
# Run the TUI application
sys.stdout.write("main(): Initializing CyclingCoachApp\n")
app = CyclingCoachApp()
sys.stdout.write("main(): CyclingCoachApp initialized. Running app.run()\n")
# Run the TUI application
app.run()
sys.stdout.write("main(): app.run() finished.\n")
if __name__ == "__main__":

2
pyproject.toml
View File

@@ -46,7 +46,7 @@ dependencies = [
"gpxpy>=1.5.0",
# External integrations
"garth==0.4.46",
"garminconnect", # Using python-garminconnect
"httpx==0.25.2",
# Backend framework

2
requirements.txt
View File

@@ -19,7 +19,7 @@ gpxpy # GPX parsing library
aiosqlite==0.20.0 # Async SQLite driver
# External integrations
garth==0.4.46 # Garmin Connect API client
garminconnect # Using python-garminconnect
httpx==0.25.2 # Async HTTP client for OpenRouter API
# Testing

6
tui/services/workout_service.py
View File

@@ -2,6 +2,7 @@
Enhanced workout service with debugging for TUI application.
"""
from typing import Dict, List, Optional
import logging
from sqlalchemy import select, desc, text
from sqlalchemy.ext.asyncio import AsyncSession
@@ -11,6 +12,8 @@ from backend.app.models.garmin_sync_log import GarminSyncLog
from backend.app.services.workout_sync import WorkoutSyncService
from backend.app.services.ai_service import AIService
logger = logging.getLogger(__name__)
class WorkoutService:
"""Service for workout operations."""
@@ -182,11 +185,14 @@ class WorkoutService:
async def sync_garmin_activities(self, days_back: int = 7) -> Dict:
"""Sync Garmin activities."""
logger.debug(f"Initiating Garmin activity sync from TUI with days_back={days_back}.")
try:
sync_service = WorkoutSyncService(self.db)
synced_count = await sync_service.sync_recent_activities(days_back=days_back)
logger.info(f"Garmin activity sync completed successfully from TUI. Synced {synced_count} activities.")
return {"status": "success", "activities_synced": synced_count}
except Exception as e:
logger.error(f"Garmin activity sync failed from TUI: {e}", exc_info=True)
return {"status": "error", "message": str(e)}
async def analyze_workout(self, workout_id: int) -> Dict: