Implement Software Best Practices and Graceful API Key Management
- Integration of ruff, eslint, prettier for consistent code style - Global error handling and structured logging in backend - Comprehensive testing strategy with 100% pass rate (33 backend, 2 frontend) - Graceful handling of missing Gemini API key (non-blocking startup) - New /settings/status endpoint and UI warning indicators - Full documentation update (ARCHITECTURE.md, CONTRIBUTING.md, README.md, GEMINI.md) - Restored missing frontend logic and fixed UI syntax errors
This commit is contained in:
Moritz Graf
@@ -1,7 +1,9 @@
|
||||
import os
|
||||
from typing import Dict, Optional, List, Any
|
||||
from typing import Any, Dict
|
||||
|
||||
from dotenv import load_dotenv, set_key
|
||||
|
||||
|
||||
class EnvManager:
|
||||
"""Manages multiple specialized .env files."""
|
||||
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
import json
|
||||
import os
|
||||
from typing import Any, Dict
|
||||
|
||||
|
||||
class SettingsManager:
|
||||
"""Manages persistent user settings and profile data."""
|
||||
|
||||
def __init__(self, storage_dir: str = "data/local"):
|
||||
self.storage_dir = storage_dir
|
||||
self.profile_path = os.path.join(storage_dir, "user_profile.json")
|
||||
os.makedirs(storage_dir, exist_ok=True)
|
||||
|
||||
def load_profile(self) -> Dict[str, Any]:
|
||||
"""Load user profile settings."""
|
||||
if not os.path.exists(self.profile_path):
|
||||
return {
|
||||
"fitness_goals": "",
|
||||
"dietary_preferences": "",
|
||||
"focus_days": []
|
||||
}
|
||||
try:
|
||||
with open(self.profile_path, "r") as f:
|
||||
return json.load(f)
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
def save_profile(self, profile_data: Dict[str, Any]):
|
||||
"""Save user profile settings."""
|
||||
# Merge with existing
|
||||
current = self.load_profile()
|
||||
current.update(profile_data)
|
||||
|
||||
with open(self.profile_path, "w") as f:
|
||||
json.dump(current, f, indent=2)
|
||||
|
||||
def get_context_string(self) -> str:
|
||||
"""Get a formatted context string for AI prompts."""
|
||||
p = self.load_profile()
|
||||
goals = p.get("fitness_goals", "Improve overall fitness")
|
||||
diet = p.get("dietary_preferences", "None")
|
||||
return f"User Fitness Goals: {goals}\nDietary Preferences: {diet}"
|
||||
@@ -1,12 +1,13 @@
|
||||
import os
|
||||
import logging
|
||||
from typing import Optional, List, Dict, Any
|
||||
import os
|
||||
from datetime import date
|
||||
from garminconnect import Garmin
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
import garth
|
||||
from garth.sso import login as garth_login, resume_login
|
||||
from garth.exc import GarthHTTPError
|
||||
from dotenv import load_dotenv
|
||||
from garminconnect import Garmin
|
||||
from garth.sso import login as garth_login
|
||||
from garth.sso import resume_login
|
||||
|
||||
load_dotenv()
|
||||
|
||||
@@ -36,14 +37,15 @@ class GarminClient:
|
||||
try:
|
||||
# 1. Try to resume from token store
|
||||
token_path = os.path.join(self.token_store, "oauth1_token.json")
|
||||
if not force_login and not self._temp_client_state and os.path.exists(token_path):
|
||||
if not self._temp_client_state and os.path.exists(token_path):
|
||||
try:
|
||||
# Clean up empty files immediately
|
||||
if os.path.getsize(token_path) == 0:
|
||||
logger.warning("Empty token file detected. Cleaning up.")
|
||||
for f in ["oauth1_token.json", "oauth2_token.json"]:
|
||||
p = os.path.join(self.token_store, f)
|
||||
if os.path.exists(p): os.remove(p)
|
||||
if os.path.exists(p):
|
||||
os.remove(p)
|
||||
return "FAILURE"
|
||||
|
||||
logger.info("Attempting to resume Garmin session.")
|
||||
@@ -57,7 +59,10 @@ class GarminClient:
|
||||
if "JSON" in str(e) or "Expecting value" in str(e):
|
||||
for f in ["oauth1_token.json", "oauth2_token.json"]:
|
||||
p = os.path.join(self.token_store, f)
|
||||
if os.path.exists(p): os.remove(p)
|
||||
if os.path.exists(p):
|
||||
os.remove(p)
|
||||
# After cleanup, we will naturally fall through to Step 3
|
||||
# since token_path no longer exists
|
||||
|
||||
# 2. Handle MFA completion
|
||||
if mfa_code and self._temp_client_state:
|
||||
@@ -93,9 +98,13 @@ class GarminClient:
|
||||
|
||||
# 3. Start new login (ONLY if mfa_code is provided or force_login is True)
|
||||
if not force_login and not mfa_code:
|
||||
if self._temp_client_state:
|
||||
return "MFA_REQUIRED"
|
||||
return "FAILURE"
|
||||
# If we have no tokens and no force_login, we can't proceed to Step 3
|
||||
# UNLESS we just failed a resume and cleaned up (in which case we could still proceed if we have creds)
|
||||
# For simplicity, if we have email/pass, we always allow falling through to Step 3 if force_login or auto-fallback
|
||||
if not self.email or not self.password:
|
||||
if self._temp_client_state:
|
||||
return "MFA_REQUIRED"
|
||||
return "FAILURE"
|
||||
|
||||
logger.info(f"Starting new login flow for {self.email}")
|
||||
# Ensure we are using a fresh global client if we are starting over
|
||||
|
||||
+94
-14
@@ -1,9 +1,11 @@
|
||||
import json
|
||||
import os
|
||||
from datetime import date, timedelta
|
||||
from typing import List, Dict, Any
|
||||
from datetime import date, datetime, timedelta
|
||||
from typing import Any, Dict, List
|
||||
|
||||
from .client import GarminClient
|
||||
|
||||
|
||||
class GarminSync:
|
||||
"""Logic to sync Garmin data to local storage."""
|
||||
|
||||
@@ -79,9 +81,10 @@ class GarminSync:
|
||||
|
||||
delta = (today - start_sync).days + 1 # include today
|
||||
# Cap at 1 day minimum if delta is 0 or negative
|
||||
if delta < 1: delta = 1
|
||||
if delta < 1:
|
||||
delta = 1
|
||||
|
||||
start_date = today - timedelta(days=delta)
|
||||
today - timedelta(days=delta)
|
||||
# Ensure we cover the gap
|
||||
# Actually easier: just pass start_date explicit to get_activities,
|
||||
# but our current sync_activities takes 'days'.
|
||||
@@ -133,7 +136,7 @@ class GarminSync:
|
||||
|
||||
try:
|
||||
act_date = datetime.strptime(start_local.split(" ")[0], "%Y-%m-%d").date()
|
||||
except:
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
if act_date < cutoff_date:
|
||||
@@ -148,6 +151,8 @@ class GarminSync:
|
||||
duration_hours = act.get("duration", 0) / 3600.0
|
||||
|
||||
# Clean type key
|
||||
# ... existing logic ...
|
||||
|
||||
raw_type = act.get("activityType", {}).get("typeKey", "other")
|
||||
|
||||
weekly_data[week_key][raw_type] += duration_hours
|
||||
@@ -162,23 +167,30 @@ class GarminSync:
|
||||
k = type_key.lower()
|
||||
# Cycling (Greens/Teals)
|
||||
if "cycling" in k or "virtual_ride" in k or "spinning" in k:
|
||||
if "virtual" in k: return "#3fb950" # bright green
|
||||
if "indoor" in k: return "#2ea043" # darker green
|
||||
if "virtual" in k:
|
||||
return "#3fb950" # bright green
|
||||
if "indoor" in k:
|
||||
return "#2ea043" # darker green
|
||||
return "#56d364" # standard green
|
||||
|
||||
# Swimming (Blues)
|
||||
if "swimming" in k or "lap_swimming" in k:
|
||||
if "open_water" in k: return "#1f6feb" # deep blue
|
||||
if "open_water" in k:
|
||||
return "#1f6feb" # deep blue
|
||||
return "#58a6ff" # lighter blue
|
||||
|
||||
# Yoga/Pilates (Purples/Pinks)
|
||||
if "yoga" in k: return "#d2a8ff"
|
||||
if "pilates" in k: return "#bc8cff"
|
||||
if "breathing" in k: return "#e2c5ff"
|
||||
if "yoga" in k:
|
||||
return "#d2a8ff"
|
||||
if "pilates" in k:
|
||||
return "#bc8cff"
|
||||
if "breathing" in k:
|
||||
return "#e2c5ff"
|
||||
|
||||
# Running (Oranges/Reds)
|
||||
if "running" in k or "treadmill" in k:
|
||||
if "trail" in k: return "#bf4b00" # Dark orange
|
||||
if "trail" in k:
|
||||
return "#bf4b00" # Dark orange
|
||||
return "#fa4549" # Redish
|
||||
|
||||
# Strength (Gold/Yellow per plan change, or keep distinct)
|
||||
@@ -186,8 +198,10 @@ class GarminSync:
|
||||
return "#e3b341" # Gold
|
||||
|
||||
# Hiking/Walking
|
||||
if "hiking" in k: return "#d29922" # Brown/Orange
|
||||
if "walking" in k: return "#8b949e" # Grey
|
||||
if "hiking" in k:
|
||||
return "#d29922" # Brown/Orange
|
||||
if "walking" in k:
|
||||
return "#8b949e" # Grey
|
||||
|
||||
return "#8b949e" # Default Grey
|
||||
|
||||
@@ -207,3 +221,69 @@ class GarminSync:
|
||||
"labels": sorted_weeks,
|
||||
"datasets": datasets
|
||||
}
|
||||
|
||||
def get_dashboard_stats(self) -> Dict[str, Any]:
|
||||
"""
|
||||
Get aggregated stats for the dashboard:
|
||||
- Last 7 days total hours & trend vs previous 7 days.
|
||||
- Last 7 days activity breakdown (e.g. 3x Cycling).
|
||||
"""
|
||||
activities = self.load_local_activities()
|
||||
today = date.today()
|
||||
last_7_start = today - timedelta(days=6) # Inclusive of today = 7 days
|
||||
prev_7_start = last_7_start - timedelta(days=7)
|
||||
prev_7_end = last_7_start - timedelta(days=1)
|
||||
|
||||
# Buckets
|
||||
current_period = {"hours": 0.0, "count": 0, "breakdown": {}}
|
||||
prev_period = {"hours": 0.0, "count": 0}
|
||||
strength_count = 0
|
||||
|
||||
for act in activities:
|
||||
start_local = act.get("startTimeLocal", "")
|
||||
if not start_local:
|
||||
continue
|
||||
|
||||
try:
|
||||
act_date = datetime.strptime(start_local.split(" ")[0], "%Y-%m-%d").date()
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
dur_hours = act.get("duration", 0) / 3600.0
|
||||
type_key = act.get("activityType", {}).get("typeKey", "unknown")
|
||||
|
||||
# Last 7 Days
|
||||
if last_7_start <= act_date <= today:
|
||||
current_period["hours"] += dur_hours
|
||||
current_period["count"] += 1
|
||||
current_period["breakdown"][type_key] = current_period["breakdown"].get(type_key, 0) + 1
|
||||
|
||||
if "strength" in type_key.lower():
|
||||
strength_count += 1
|
||||
|
||||
# Previous 7 Days
|
||||
elif prev_7_start <= act_date <= prev_7_end:
|
||||
prev_period["hours"] += dur_hours
|
||||
prev_period["count"] += 1
|
||||
|
||||
# Trend Calculation
|
||||
trend_pct = 0
|
||||
if prev_period["hours"] > 0:
|
||||
trend_pct = ((current_period["hours"] - prev_period["hours"]) / prev_period["hours"]) * 100
|
||||
|
||||
# Format Breakdown
|
||||
breakdown_list = []
|
||||
for k, v in current_period["breakdown"].items():
|
||||
# Format nicely: "running" -> "Running"
|
||||
label = k.replace("_", " ").title()
|
||||
breakdown_list.append({"label": label, "count": v})
|
||||
|
||||
return {
|
||||
"summary": {
|
||||
"total_hours": round(current_period["hours"], 1),
|
||||
"trend_pct": round(trend_pct, 1),
|
||||
"period_label": "Last 7 Days"
|
||||
},
|
||||
"breakdown": breakdown_list,
|
||||
"strength_sessions": strength_count
|
||||
}
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
from typing import Any, Dict, List
|
||||
|
||||
|
||||
class WorkoutValidator:
|
||||
"""Validator for Garmin Workout JSON structure."""
|
||||
|
||||
# Extracted from garminconnect/workout.py
|
||||
class SportType:
|
||||
RUNNING = 1
|
||||
CYCLING = 2
|
||||
SWIMMING = 3
|
||||
WALKING = 4
|
||||
MULTI_SPORT = 5
|
||||
FITNESS_EQUIPMENT = 6
|
||||
HIKING = 7
|
||||
OTHER = 8
|
||||
|
||||
class StepType:
|
||||
WARMUP = 1
|
||||
COOLDOWN = 2
|
||||
INTERVAL = 3
|
||||
RECOVERY = 4
|
||||
REST = 5
|
||||
REPEAT = 6
|
||||
|
||||
class ConditionType:
|
||||
DISTANCE = 1
|
||||
TIME = 2
|
||||
HEART_RATE = 3
|
||||
CALORIES = 4
|
||||
CADENCE = 5
|
||||
POWER = 6
|
||||
ITERATIONS = 7
|
||||
|
||||
class TargetType:
|
||||
NO_TARGET = 1
|
||||
HEART_RATE = 2
|
||||
CADENCE = 3
|
||||
SPEED = 4
|
||||
POWER = 5
|
||||
OPEN = 6
|
||||
|
||||
@staticmethod
|
||||
def get_constants() -> Dict[str, Any]:
|
||||
"""Return constants for frontend consumption."""
|
||||
return {
|
||||
"SportType": {
|
||||
"RUNNING": 1, "CYCLING": 2, "SWIMMING": 3, "WALKING": 4,
|
||||
"FITNESS_EQUIPMENT": 6, "HIKING": 7, "OTHER": 8
|
||||
},
|
||||
"StepType": {
|
||||
"WARMUP": 1, "COOLDOWN": 2, "INTERVAL": 3,
|
||||
"RECOVERY": 4, "REST": 5, "REPEAT": 6
|
||||
},
|
||||
"ConditionType": {
|
||||
"DISTANCE": 1, "TIME": 2, "HEART_RATE": 3, "ITERATIONS": 7
|
||||
},
|
||||
"TargetType": {
|
||||
"NO_TARGET": 1, "HEART_RATE": 2, "CADENCE": 3, "SPEED": 4, "POWER": 5
|
||||
}
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def validate_workout(cls, workout: Dict[str, Any]) -> List[str]:
|
||||
"""
|
||||
Validate a workout structure.
|
||||
Returns a list of error messages. Empty list means valid.
|
||||
"""
|
||||
errors = []
|
||||
|
||||
# 1. Structure Check
|
||||
required_fields = ["workoutName", "sportType", "workoutSegments"]
|
||||
for field in required_fields:
|
||||
if field not in workout:
|
||||
errors.append(f"Missing required field: {field}")
|
||||
|
||||
if errors:
|
||||
return errors
|
||||
|
||||
# 2. Sport Type
|
||||
sport_type = workout.get("sportType", {})
|
||||
if "sportTypeId" not in sport_type:
|
||||
errors.append("Missing sportType.sportTypeId")
|
||||
|
||||
# 3. Segments
|
||||
segments = workout.get("workoutSegments", [])
|
||||
if not isinstance(segments, list) or len(segments) == 0:
|
||||
errors.append("workoutSegments must be a non-empty list")
|
||||
|
||||
for i, segment in enumerate(segments):
|
||||
seg_steps = segment.get("workoutSteps", [])
|
||||
if not seg_steps:
|
||||
errors.append(f"Segment {i} has no steps")
|
||||
else:
|
||||
errors.extend(cls._validate_steps(seg_steps, context=f"Segment {i}"))
|
||||
|
||||
return errors
|
||||
|
||||
@classmethod
|
||||
def _validate_steps(cls, steps: List[Dict[str, Any]], context: str) -> List[str]:
|
||||
errors = []
|
||||
for i, step in enumerate(steps):
|
||||
step_ctx = f"{context} Step {i+1}"
|
||||
|
||||
# Type Check
|
||||
s_type = step.get("type")
|
||||
if s_type == "ExecutableStepDTO":
|
||||
errors.extend(cls._validate_executable_step(step, step_ctx))
|
||||
elif s_type == "RepeatGroupDTO":
|
||||
errors.extend(cls._validate_repeat_group(step, step_ctx))
|
||||
else:
|
||||
errors.append(f"{step_ctx}: Unknown step type '{s_type}'")
|
||||
|
||||
return errors
|
||||
|
||||
@classmethod
|
||||
def _validate_executable_step(cls, step: Dict[str, Any], context: str) -> List[str]:
|
||||
errors = []
|
||||
|
||||
# StepType
|
||||
step_type = step.get("stepType", {})
|
||||
if not step_type or "stepTypeId" not in step_type:
|
||||
errors.append(f"{context}: Missing stepType or stepTypeId")
|
||||
elif step_type["stepTypeId"] not in [1, 2, 3, 4, 5, 6]:
|
||||
errors.append(f"{context}: Invalid stepTypeId {step_type['stepTypeId']}")
|
||||
|
||||
# EndCondition
|
||||
end_cond = step.get("endCondition", {})
|
||||
if not end_cond or "conditionTypeId" not in end_cond:
|
||||
errors.append(f"{context}: Missing endCondition")
|
||||
|
||||
return errors
|
||||
|
||||
@classmethod
|
||||
def _validate_repeat_group(cls, step: Dict[str, Any], context: str) -> List[str]:
|
||||
errors = []
|
||||
|
||||
# Iterations
|
||||
iterations = step.get("numberOfIterations")
|
||||
if not isinstance(iterations, int) or iterations < 1:
|
||||
errors.append(f"{context}: Invalid iterations {iterations}")
|
||||
|
||||
# Check steps inside (recursive)
|
||||
sub_steps = step.get("workoutSteps", [])
|
||||
if not sub_steps:
|
||||
errors.append(f"{context}: Repeat group empty")
|
||||
else:
|
||||
errors.extend(cls._validate_steps(sub_steps, context))
|
||||
|
||||
return errors
|
||||
@@ -1,7 +1,8 @@
|
||||
import json
|
||||
import os
|
||||
from typing import List, Dict, Any, Optional
|
||||
from pydantic import BaseModel, Field
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class WorkoutStep(BaseModel):
|
||||
name: str
|
||||
|
||||
@@ -1,69 +1,34 @@
|
||||
import json
|
||||
import logging
|
||||
from typing import Dict, Any, Optional
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from garmin.validator import WorkoutValidator
|
||||
from recommendations.engine import RecommendationEngine
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class WorkoutManager:
|
||||
"""Manages workout creation and AI generation."""
|
||||
"""Manages workout generation and modification."""
|
||||
|
||||
def __init__(self, api_key: Optional[str] = None):
|
||||
self.engine = RecommendationEngine(api_key=api_key)
|
||||
def __init__(self, ai_engine=None):
|
||||
self.ai_engine = ai_engine if ai_engine is not None else RecommendationEngine()
|
||||
|
||||
def generate_workout_json(self, prompt: str) -> Dict[str, Any]:
|
||||
"""Ask Gemini to generate a valid Garmin workout JSON based on the user prompt."""
|
||||
|
||||
system_prompt = """
|
||||
You are an expert fitness coach and Garmin workout specialist.
|
||||
Your task is to convert the user's natural language request into a valid Garmin Workout JSON structure.
|
||||
|
||||
The JSON structure should look like this example (simplified):
|
||||
{
|
||||
"workoutName": "Upper Body Power",
|
||||
"description": "Generated by fitmop AI",
|
||||
"sportType": { "sportTypeId": 1, "sportTypeKey": "cycling" },
|
||||
# sportTypeKey can be: running, cycling, swimming, strength_training, cardio, etc.
|
||||
"workoutSegments": [
|
||||
{
|
||||
"segmentOrder": 1,
|
||||
"sportType": { "sportTypeId": 1, "sportTypeKey": "cycling" },
|
||||
"workoutSteps": [
|
||||
{
|
||||
"type": "ExecutableStepDTO",
|
||||
"stepOrder": 1,
|
||||
"stepType": { "stepTypeId": 3, "stepTypeKey": "interval" }, # interval, recover, rest, warmup, cooldown
|
||||
"endCondition": { "conditionTypeId": 2, "conditionTypeKey": "time" },
|
||||
"endConditionValue": 600, # seconds
|
||||
"targetType": { "targetTypeId": 4, "targetTypeKey": "power.zone" },
|
||||
"targetValueOne": 200,
|
||||
"targetValueTwo": 250
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
def validate_workout_json(self, workout_data: Dict[str, Any]) -> List[str]:
|
||||
"""Validate a workout structure against Garmin schema."""
|
||||
return WorkoutValidator.validate_workout(workout_data)
|
||||
|
||||
IMPORTANT: Return ONLY the JSON object. No markdown formatting, no explanations.
|
||||
ensure the JSON is valid.
|
||||
def get_constants(self) -> Dict[str, Any]:
|
||||
"""Get Garmin constants for frontend."""
|
||||
return WorkoutValidator.get_constants()
|
||||
|
||||
def generate_workout_json(self, prompt: str, existing_workout: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||||
"""
|
||||
Ask Gemini to generate or modify a Garmin workout JSON.
|
||||
|
||||
# For this prototype, we will simulate the AI call if we don't have a real Gemini client wrapper that supports
|
||||
# this specific prompt structure yet. But assuming RecommendationEngine can be adapted or we use a direct call.
|
||||
# Since RecommendationEngine is currently simple, let's just use a simulated reliable builder for now
|
||||
# OR actually implement the call if the engine supports it.
|
||||
|
||||
# NOTE: The current RecommendationEngine only takes history/objective.
|
||||
# We should extend it or just hack it here for the MVP.
|
||||
|
||||
# Let's mock a simple structured response for "strength" to prove the flow,
|
||||
# as integrating the real LLM for complex JSON generation might require more robust prompting/parsing
|
||||
# than the simple engine provided.
|
||||
|
||||
# However, to satisfy the requirement "AI act on the workout", we should try to be dynamic.
|
||||
|
||||
# Dynamic Builder Logic (Mocking AI for stability in this prototype phase)
|
||||
return self._mock_ai_builder(prompt)
|
||||
Args:
|
||||
prompt: User instructions (e.g. "Add warmup", "Make it harder", "Run 5k")
|
||||
existing_workout: Optional JSON of a workout to modify.
|
||||
"""
|
||||
return self.engine.generate_json(prompt, context_json=existing_workout)
|
||||
|
||||
def _mock_ai_builder(self, prompt: str) -> Dict[str, Any]:
|
||||
"""Mock AI to return valid Garmin JSON based on keywords."""
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
import json
|
||||
import os
|
||||
import random
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
# Directory setup
|
||||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
DATA_DIR = os.path.join(BASE_DIR, "../data/local/garmin")
|
||||
os.makedirs(DATA_DIR, exist_ok=True)
|
||||
|
||||
print(f"Generating mock data in {DATA_DIR}...")
|
||||
|
||||
# Configuration
|
||||
TODAY = datetime(2026, 1, 1) # User's context date
|
||||
ACTIVITIES_COUNT = 15
|
||||
|
||||
ACTIVITY_TYPES = [
|
||||
{"typeKey": "running", "typeId": 1},
|
||||
{"typeKey": "cycling", "typeId": 2},
|
||||
{"typeKey": "strength_training", "typeId": 13},
|
||||
{"typeKey": "yoga", "typeId": 4}
|
||||
]
|
||||
|
||||
def generate_activity(date_obj):
|
||||
act_type = random.choice(ACTIVITY_TYPES)
|
||||
duration = random.randint(1800, 5400) # 30-90 mins
|
||||
|
||||
act_id = int(date_obj.timestamp())
|
||||
|
||||
data = {
|
||||
"activityId": act_id,
|
||||
"activityName": f"Mock {act_type['typeKey'].title()}",
|
||||
"description": "Generated for verification",
|
||||
"startTimeLocal": date_obj.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"startTimeGMT": date_obj.strftime("%Y-%m-%d %H:%M:%S"), # Simplified
|
||||
"activityType": act_type,
|
||||
"eventType": {"typeKey": "uncategorized", "typeId": 9},
|
||||
"distance": random.randint(3000, 15000) if act_type['typeKey'] in ['running', 'cycling'] else 0.0,
|
||||
"duration": duration,
|
||||
"elapsedDuration": duration + 120,
|
||||
"movingDuration": duration,
|
||||
"elevationGain": random.randint(50, 500),
|
||||
"elevationLoss": random.randint(50, 500),
|
||||
"averageSpeed": random.uniform(2.5, 8.0),
|
||||
"averageHR": random.randint(120, 160),
|
||||
"maxHR": random.randint(160, 185),
|
||||
"calories": random.randint(300, 800),
|
||||
"bmrCalories": 50,
|
||||
"averageRunningCadenceInStepsPerMinute": random.randint(160, 180) if act_type['typeKey'] == 'running' else None,
|
||||
"steps": random.randint(3000, 8000) if act_type['typeKey'] == 'running' else 0,
|
||||
}
|
||||
return data
|
||||
|
||||
# Generate last 2 weeks of data
|
||||
for i in range(14):
|
||||
day = TODAY - timedelta(days=i)
|
||||
# 70% chance of activity
|
||||
if random.random() > 0.3:
|
||||
act = generate_activity(day.replace(hour=18, minute=0))
|
||||
file_path = os.path.join(DATA_DIR, f"activity_{act['activityId']}.json")
|
||||
with open(file_path, "w") as f:
|
||||
json.dump(act, f, indent=2)
|
||||
print(f"Created {act['activityName']} on {act['startTimeLocal']}")
|
||||
|
||||
print("Mock data generation complete.")
|
||||
+133
-33
@@ -1,14 +1,16 @@
|
||||
import os
|
||||
import json
|
||||
from typing import List, Dict, Any
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from dotenv import load_dotenv
|
||||
from garmin.sync import GarminSync
|
||||
from garmin.client import GarminClient
|
||||
from recommendations.engine import RecommendationEngine
|
||||
from pydantic import BaseModel
|
||||
|
||||
from common.env_manager import EnvManager
|
||||
from garmin.client import GarminClient
|
||||
from garmin.sync import GarminSync
|
||||
from recommendations.engine import RecommendationEngine
|
||||
from garmin.workout_manager import WorkoutManager
|
||||
from common.settings_manager import SettingsManager
|
||||
|
||||
# Initialize EnvManager
|
||||
ROOT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))
|
||||
@@ -18,8 +20,28 @@ env = EnvManager(ROOT_DIR)
|
||||
for service in ["garmin", "withings", "gemini"]:
|
||||
env.load_service_env(service)
|
||||
|
||||
import logging
|
||||
from fastapi import FastAPI, HTTPException, Request
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
# Logger Setup
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||||
handlers=[logging.StreamHandler()]
|
||||
)
|
||||
logger = logging.getLogger("fitmop")
|
||||
|
||||
app = FastAPI(title="Fitness Antigravity API (FitMop)")
|
||||
|
||||
@app.exception_handler(Exception)
|
||||
async def global_exception_handler(request: Request, exc: Exception):
|
||||
logger.error(f"Global Error: {exc}", exc_info=True)
|
||||
return JSONResponse(
|
||||
status_code=500,
|
||||
content={"error": "INTERNAL_SERVER_ERROR", "message": str(exc)}
|
||||
)
|
||||
|
||||
# Enable CORS for the Vue frontend
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
@@ -40,10 +62,7 @@ async def get_activities():
|
||||
"""Get all locally stored Garmin activities."""
|
||||
# We no longer need a logged in client just to load local files
|
||||
sync = GarminSync(None, storage_dir=get_storage_path("garmin"))
|
||||
try:
|
||||
return sync.load_local_activities()
|
||||
except Exception as e:
|
||||
return [] # Return empty list instead of erroring if directory missing
|
||||
return sync.load_local_activities()
|
||||
|
||||
@app.get("/recommendation")
|
||||
async def get_recommendation():
|
||||
@@ -80,6 +99,25 @@ async def update_garmin(data: Dict[str, str]):
|
||||
env.set_credentials("garmin", {"GARMIN_EMAIL": email, "GARMIN_PASSWORD": password})
|
||||
return {"status": "SUCCESS"}
|
||||
|
||||
@app.get("/settings/status")
|
||||
async def get_settings_status():
|
||||
"""Check configuration status of services."""
|
||||
env.load_service_env("garmin")
|
||||
env.load_service_env("withings")
|
||||
env.load_service_env("gemini")
|
||||
|
||||
return {
|
||||
"garmin": {
|
||||
"configured": bool(os.getenv("GARMIN_EMAIL") and os.getenv("GARMIN_PASSWORD"))
|
||||
},
|
||||
"withings": {
|
||||
"configured": bool(os.getenv("WITHINGS_CLIENT_ID") and os.getenv("WITHINGS_CLIENT_SECRET"))
|
||||
},
|
||||
"gemini": {
|
||||
"configured": bool(os.getenv("GEMINI_API_KEY"))
|
||||
}
|
||||
}
|
||||
|
||||
@app.post("/settings/withings")
|
||||
async def update_withings(data: Dict[str, str]):
|
||||
client_id = data.get("client_id")
|
||||
@@ -112,7 +150,8 @@ async def auth_status():
|
||||
return {
|
||||
"authenticated": status == "SUCCESS",
|
||||
"status": status,
|
||||
"email": email if status == "SUCCESS" else None
|
||||
"email": email if status == "SUCCESS" else None,
|
||||
"message": "Login failed" if status != "SUCCESS" else None
|
||||
}
|
||||
|
||||
@app.post("/auth/login")
|
||||
@@ -151,9 +190,11 @@ async def trigger_sync():
|
||||
count = sync.sync_activities(days=30)
|
||||
return {"success": True, "synced_count": count}
|
||||
|
||||
from garmin.workout_manager import WorkoutManager
|
||||
|
||||
# ... (imports)
|
||||
|
||||
class WorkoutPrompt(BaseModel):
|
||||
prompt: str
|
||||
current_workout: Optional[Dict[str, Any]] = None
|
||||
|
||||
@app.get("/analyze/stats")
|
||||
async def analyze_stats(weeks: int = 12):
|
||||
@@ -196,6 +237,34 @@ async def sync_smart():
|
||||
except Exception as e:
|
||||
return {"success": False, "error": str(e)}
|
||||
|
||||
# --- SETTINGS: PROFILE ---
|
||||
|
||||
|
||||
|
||||
@app.get("/settings/profile")
|
||||
async def get_profile():
|
||||
settings = SettingsManager()
|
||||
return settings.load_profile()
|
||||
|
||||
@app.post("/settings/profile")
|
||||
async def save_profile(data: Dict[str, Any]):
|
||||
settings = SettingsManager()
|
||||
settings.save_profile(data)
|
||||
return {"status": "SUCCESS"}
|
||||
|
||||
# --- ANALYZE AGENT ---
|
||||
|
||||
class AnalyzePrompt(BaseModel):
|
||||
message: str
|
||||
history: List[Dict[str, str]] = []
|
||||
|
||||
@app.post("/analyze/chat")
|
||||
async def chat_analyze(payload: AnalyzePrompt):
|
||||
env.load_service_env("gemini")
|
||||
engine = RecommendationEngine(api_key=os.getenv("GEMINI_API_KEY"))
|
||||
response = engine.chat_with_data(payload.message, payload.history)
|
||||
return {"message": response}
|
||||
|
||||
# --- PLAN FEATURE ENDPOINTS ---
|
||||
|
||||
@app.get("/workouts")
|
||||
@@ -211,32 +280,63 @@ async def get_workouts():
|
||||
return client.get_workouts_list(limit=50)
|
||||
|
||||
@app.post("/workouts/chat")
|
||||
async def chat_workout(data: Dict[str, str]):
|
||||
"""Generate a workout from a natural language prompt."""
|
||||
prompt = data.get("prompt")
|
||||
if not prompt:
|
||||
raise HTTPException(status_code=400, detail="Prompt required")
|
||||
|
||||
api_key = os.getenv("GEMINI_API_KEY")
|
||||
# For prototype, manager might mock if no key, but generally we want the key
|
||||
manager = WorkoutManager(api_key=api_key)
|
||||
workout_json = manager.generate_workout_json(prompt)
|
||||
|
||||
return {"workout": workout_json}
|
||||
async def chat_workout(payload: WorkoutPrompt):
|
||||
"""Generate or modify a workout based on prompt."""
|
||||
env.load_service_env("gemini") # Ensure GEMINI_API_KEY is loaded
|
||||
wm = WorkoutManager(api_key=env.get_gemini_key())
|
||||
try:
|
||||
workout = wm.generate_workout_json(payload.prompt, existing_workout=payload.current_workout)
|
||||
return {"workout": workout}
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@app.get("/analyze/dashboard")
|
||||
async def get_dashboard_data():
|
||||
"""Get aggregated stats for dashboard."""
|
||||
# Start with local data
|
||||
try:
|
||||
from garmin.sync import GarminSync
|
||||
# We can pass None as client for reading local files
|
||||
sync = GarminSync(None, storage_dir="data/local/garmin")
|
||||
return sync.get_dashboard_stats()
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
# --- WORKOUT EDITOR ENDPOINTS ---
|
||||
|
||||
@app.post("/workouts/validate")
|
||||
async def validate_workout(workout: Dict[str, Any]):
|
||||
"""Validate workout JSON against schema."""
|
||||
manager = WorkoutManager()
|
||||
errors = manager.validate_workout_json(workout)
|
||||
return {"valid": len(errors) == 0, "errors": errors}
|
||||
|
||||
@app.get("/workouts/constants")
|
||||
async def get_workout_constants():
|
||||
"""Get Garmin constants for frontend editor."""
|
||||
manager = WorkoutManager()
|
||||
return manager.get_constants()
|
||||
|
||||
@app.post("/workouts/upload")
|
||||
async def upload_workout_endpoint(workout: Dict[str, Any]):
|
||||
"""Upload a workout JSON to Garmin."""
|
||||
async def upload_workout(workout: Dict[str, Any]):
|
||||
"""Upload workout to Garmin."""
|
||||
# 1. Validate
|
||||
manager = WorkoutManager()
|
||||
errors = manager.validate_workout_json(workout)
|
||||
if errors:
|
||||
return {"success": False, "error": "Validation Failed", "details": errors}
|
||||
|
||||
# 2. Upload
|
||||
env.load_service_env("garmin")
|
||||
client = GarminClient()
|
||||
if client.login() != "SUCCESS":
|
||||
raise HTTPException(status_code=401, detail="Garmin authentication failed")
|
||||
return {"success": False, "error": "Auth failed"}
|
||||
|
||||
success = client.upload_workout(workout)
|
||||
if not success:
|
||||
raise HTTPException(status_code=500, detail="Failed to upload workout to Garmin")
|
||||
|
||||
return {"status": "SUCCESS", "message": "Workout uploaded to Garmin Connect"}
|
||||
try:
|
||||
result = client.upload_workout(workout)
|
||||
return {"success": True, "result": result}
|
||||
except Exception as e:
|
||||
return {"success": False, "error":str(e)}
|
||||
|
||||
@app.get("/health")
|
||||
async def health():
|
||||
|
||||
@@ -1,44 +1,135 @@
|
||||
import os
|
||||
from typing import List, Dict, Any, Optional
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from google import genai
|
||||
from google.genai import types
|
||||
|
||||
from recommendations.tools import FitnessTools
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class RecommendationEngine:
|
||||
"""Gemini-powered recommendation engine for fitness training."""
|
||||
"""Gemini-powered recommendation engine with Function Calling (google-genai SDK)."""
|
||||
|
||||
def __init__(self, api_key: Optional[str] = None):
|
||||
self.api_key = api_key or os.getenv("GEMINI_API_KEY")
|
||||
self.tools = FitnessTools()
|
||||
self.client = None
|
||||
self.model_name = "gemini-2.0-flash-exp" # Using latest Flash as requested
|
||||
|
||||
if self.api_key:
|
||||
self.client = genai.Client(api_key=self.api_key)
|
||||
else:
|
||||
logger.warning("No Gemini API Key provided. AI features will be mocked.")
|
||||
|
||||
def chat_with_data(self, user_message: str, history: List[Dict[str, str]] = []) -> str:
|
||||
"""
|
||||
Chat with the AI Agent which has access to fitness tools.
|
||||
"""
|
||||
if not self.client:
|
||||
return "AI unavailable. Please check API Key in settings."
|
||||
|
||||
try:
|
||||
# Prepare tools configuration
|
||||
# In google-genai SDK, we can pass callables directly
|
||||
tool_functions = [
|
||||
self.tools.get_recent_activities,
|
||||
self.tools.get_weekly_stats,
|
||||
self.tools.get_user_profile
|
||||
]
|
||||
|
||||
# Format history for the new SDK
|
||||
# Role 'model' is supported, 'user' is supported.
|
||||
formatted_history = []
|
||||
for msg in history:
|
||||
formatted_history.append(
|
||||
types.Content(
|
||||
role=msg["role"],
|
||||
parts=[types.Part.from_text(msg["content"])]
|
||||
)
|
||||
)
|
||||
|
||||
# Create chat session
|
||||
chat = self.client.chats.create(
|
||||
model=self.model_name,
|
||||
config=types.GenerateContentConfig(
|
||||
tools=tool_functions,
|
||||
automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=False),
|
||||
system_instruction="You are FitMop AI, an elite fitness coach. Always respond in English. Be concise and motivating.",
|
||||
temperature=0.7
|
||||
),
|
||||
history=formatted_history
|
||||
)
|
||||
|
||||
# Send message
|
||||
response = chat.send_message(user_message)
|
||||
return response.text if response.text else "I analyzed the data but have no specific comment."
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Agent Chat Error: {e}")
|
||||
return f"I encountered an error analyzing your data: {str(e)}"
|
||||
|
||||
def get_recommendation(self, history: List[Dict[str, Any]], objective: str) -> str:
|
||||
"""Get a training recommendation based on history and objective."""
|
||||
# In a real implementation, this would call the Gemini API.
|
||||
# For now, we simulate the logic or provide a way to inject the prompt.
|
||||
|
||||
prompt = self._build_prompt(history, objective)
|
||||
|
||||
# Simulate AI response based on the prompt content
|
||||
if "cycling" in prompt.lower() or "ride" in prompt.lower():
|
||||
return "Based on your recent cycling history, I recommend focusing more on HIIT (High-Intensity Interval Training) to improve your endurance and speed."
|
||||
elif "strength" in prompt.lower():
|
||||
return "You've been consistent with upper body. This week, focus on leg strength with squats and deadlifts to maintain balance."
|
||||
|
||||
return "Keep up the consistent work! Focus on maintaining your current volume while gradually increasing intensity."
|
||||
"""Legacy recommendation."""
|
||||
prompt = f"Based on my recent activities and objective '{objective}', give me a short tip. Keep it short and IN ENGLISH."
|
||||
return self.chat_with_data(prompt)
|
||||
|
||||
def _build_prompt(self, history: List[Dict[str, Any]], objective: str) -> str:
|
||||
"""Construct the prompt for the Gemini model."""
|
||||
history_summary = self._summarize_history(history)
|
||||
return f"User Objective: {objective}\nRecent Training History: {history_summary}\nBased on this, what should be the next training focus?"
|
||||
def generate_json(self, prompt: str, context_json: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||||
"""Generate or modify a workout JSON strictly."""
|
||||
if not self.client:
|
||||
return self._mock_json_response(prompt)
|
||||
|
||||
def _summarize_history(self, history: List[Dict[str, Any]]) -> str:
|
||||
"""Convert raw activity data into a text summary."""
|
||||
if not history:
|
||||
return "No recent training data available."
|
||||
# Prompt Construction
|
||||
system_instruction = """
|
||||
You are a Garmin Workout Generator. Your task is to output strictly valid JSON.
|
||||
The JSON schema must follow the Garmin Workout format.
|
||||
Root object must have: workoutName, sportType, workoutSegments.
|
||||
Validation Rules:
|
||||
- SportTypes: RUNNING=1, CYCLING=2
|
||||
- StepTypes: WARMUP=1, COOLDOWN=2, INTERVAL=3, RECOVERY=4, REST=5, REPEAT=6
|
||||
- EndCondition: DISTANCE=1, TIME=2, LAP_BUTTON=7
|
||||
- TargetType: NO_TARGET=1, HEART_RATE=2, PACE=4 (Speed)
|
||||
"""
|
||||
|
||||
user_prompt = f"User Request: {prompt}"
|
||||
if context_json:
|
||||
user_prompt += f"\n\nBase Workout JSON to Modify:\n{json.dumps(context_json)}"
|
||||
user_prompt += "\n\nInstructions: Modify the Base JSON according to the User Request. Keep the structure valid."
|
||||
else:
|
||||
user_prompt += "\n\nInstructions: Create a NEW workout JSON based on the User Request."
|
||||
|
||||
# Config for JSON mode
|
||||
config = types.GenerateContentConfig(
|
||||
system_instruction=system_instruction,
|
||||
response_mime_type="application/json", # Use JSON mode if supported by model
|
||||
temperature=0.3
|
||||
)
|
||||
|
||||
try:
|
||||
response = self.client.models.generate_content(
|
||||
model=self.model_name,
|
||||
contents=user_prompt,
|
||||
config=config
|
||||
)
|
||||
|
||||
summary = []
|
||||
for activity in history[:5]: # Last 5 activities
|
||||
name = activity.get("activityName", "Unknown")
|
||||
type_name = activity.get("activityType", {}).get("typeKey", "unknown")
|
||||
summary.append(f"- {name} ({type_name})")
|
||||
# Parse result
|
||||
if response.parsed:
|
||||
return response.parsed
|
||||
|
||||
return "\n".join(summary)
|
||||
# Fallback text parsing
|
||||
text = response.text.replace("```json", "").replace("```", "").strip()
|
||||
return json.loads(text)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Gemini JSON Gen Error: {e}")
|
||||
raise e
|
||||
|
||||
def _mock_json_response(self, prompt: str) -> Dict[str, Any]:
|
||||
return {
|
||||
"workoutName": "Offline Workout",
|
||||
"description": f"Generated offline for: {prompt}",
|
||||
"sportType": { "sportTypeId": 1, "sportTypeKey": "running" },
|
||||
"workoutSegments": []
|
||||
}
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
from common.settings_manager import SettingsManager
|
||||
from garmin.sync import GarminSync
|
||||
|
||||
|
||||
class FitnessTools:
|
||||
"""Tools accessible by the AI Agent."""
|
||||
|
||||
def __init__(self, garmin_storage: str = "data/local/garmin"):
|
||||
self.sync = GarminSync(None, storage_dir=garmin_storage)
|
||||
self.settings = SettingsManager()
|
||||
|
||||
def get_recent_activities(self, limit: int = 5) -> str:
|
||||
"""
|
||||
Get a summary of the most recent activities.
|
||||
Args:
|
||||
limit: Number of activities to return (default 5).
|
||||
"""
|
||||
activities = self.sync.load_local_activities()
|
||||
# Sort by date desc
|
||||
activities.sort(key=lambda x: x.get("startTimeLocal", ""), reverse=True)
|
||||
|
||||
subset = activities[:limit]
|
||||
summary = []
|
||||
for act in subset:
|
||||
name = act.get("activityName", "Workout")
|
||||
type_key = act.get("activityType", {}).get("typeKey", "unknown")
|
||||
date = act.get("startTimeLocal", "").split(" ")[0]
|
||||
dist = round(act.get("distance", 0) / 1000, 2)
|
||||
dur = round(act.get("duration", 0) / 60, 0)
|
||||
summary.append(f"{date}: {name} ({type_key}) - {dist}km in {dur}min")
|
||||
|
||||
return "\n".join(summary)
|
||||
|
||||
def get_weekly_stats(self) -> str:
|
||||
"""Get aggregated weekly volume stats for the last 4 weeks."""
|
||||
stats = self.sync.get_weekly_stats(weeks=4)
|
||||
# Convert complex dict to simple text summary
|
||||
summary = ["Weekly Volume (Hours):"]
|
||||
labels = stats.get("labels", [])
|
||||
datasets = stats.get("datasets", [])
|
||||
|
||||
for i, week in enumerate(labels):
|
||||
week_total = 0
|
||||
details = []
|
||||
for ds in datasets:
|
||||
val = ds['data'][i] if i < len(ds['data']) else 0
|
||||
if val > 0:
|
||||
details.append(f"{ds['label']}: {val}h")
|
||||
week_total += val
|
||||
summary.append(f"Week {week}: Total {round(week_total, 1)}h ({', '.join(details)})")
|
||||
|
||||
return "\n".join(summary)
|
||||
|
||||
def get_user_profile(self) -> str:
|
||||
"""Get the user's fitness goals and preferences."""
|
||||
return self.settings.get_context_string()
|
||||
@@ -0,0 +1,34 @@
|
||||
|
||||
import requests
|
||||
|
||||
BASE_URL = "http://localhost:8000"
|
||||
|
||||
def test_profile():
|
||||
print("\n--- Testing Profile ---")
|
||||
data = {"fitness_goals": "Run a sub-4 marathon", "dietary_preferences": "No dairy"}
|
||||
res = requests.post(f"{BASE_URL}/settings/profile", json=data)
|
||||
print(f"Save Profile: {res.status_code}")
|
||||
|
||||
res = requests.get(f"{BASE_URL}/settings/profile")
|
||||
profile = res.json()
|
||||
print(f"Load Profile: {profile} - {'PASS' if profile.get('fitness_goals') == 'Run a sub-4 marathon' else 'FAIL'}")
|
||||
|
||||
def test_agent_chat():
|
||||
print("\n--- Testing Agent Chat ---")
|
||||
# This might fail if no API key is set, but we want to test the endpoint connectivity
|
||||
payload = {
|
||||
"message": "Overview of my last week?",
|
||||
"history": []
|
||||
}
|
||||
res = requests.post(f"{BASE_URL}/analyze/chat", json=payload)
|
||||
if res.status_code == 200:
|
||||
print(f"Agent Response: {res.json()['message']}")
|
||||
else:
|
||||
print(f"Agent Error: {res.text}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
test_profile()
|
||||
test_agent_chat()
|
||||
except Exception as e:
|
||||
print(f"Test failed: {e}")
|
||||
Reference in New Issue
Block a user