118 lines
4.9 KiB
Python
118 lines
4.9 KiB
Python
"""Generate AI activity summaries using Gemini."""
|
|
|
|
from backend.app.models.activity import Activity, ActivityMetrics
|
|
from backend.app.services.gemini_client import chat_async
|
|
|
|
SYSTEM_PROMPT = """You are VeloBrain, an AI cycling coach.
|
|
Analyze the cycling activity data and provide a concise, insightful summary in 3-5 sentences.
|
|
Focus on: performance highlights, pacing strategy, areas for improvement, and training effect.
|
|
Be specific with numbers. Use a friendly, coaching tone.
|
|
Respond in Russian.
|
|
Use a HTML text formatting."""
|
|
|
|
SYSTEM_PROMPT_WITH_PLAN = """You are VeloBrain, an AI cycling coach.
|
|
The athlete completed a workout that was part of their training plan.
|
|
|
|
Your analysis MUST include two sections:
|
|
|
|
1. **Анализ тренировки** — General performance summary (2-3 sentences): highlights, pacing, training effect.
|
|
2. **Соответствие плану** — How well the actual workout matched the planned workout (2-4 sentences). Compare:
|
|
- Actual duration vs planned duration
|
|
- Actual intensity (IF, TSS, avg power) vs planned targets
|
|
- Workout type match (was it the right type of effort?)
|
|
- Give a compliance verdict: fully matched / partially matched / deviated significantly
|
|
- If deviated, explain what was different and whether the deviation was acceptable or needs attention.
|
|
|
|
Be specific with numbers. Use a friendly, coaching tone.
|
|
Respond in Russian.
|
|
Use HTML text formatting."""
|
|
|
|
|
|
def _build_activity_prompt(
|
|
activity: Activity,
|
|
rider_ftp: float | None = None,
|
|
planned_workout: dict | None = None,
|
|
) -> str:
|
|
m: ActivityMetrics | None = activity.metrics
|
|
lines = [
|
|
f"Activity: {activity.name or 'Workout'}",
|
|
f"Type: {activity.activity_type}",
|
|
f"Duration: {activity.duration // 3600}h {(activity.duration % 3600) // 60}m",
|
|
]
|
|
|
|
if activity.distance:
|
|
lines.append(f"Distance: {activity.distance / 1000:.1f} km")
|
|
if activity.elevation_gain:
|
|
lines.append(f"Elevation: {activity.elevation_gain:.0f} m")
|
|
|
|
if m:
|
|
if m.avg_power:
|
|
lines.append(f"Avg Power: {m.avg_power:.0f} W")
|
|
if m.normalized_power:
|
|
lines.append(f"Normalized Power: {m.normalized_power:.0f} W")
|
|
if m.tss:
|
|
lines.append(f"TSS: {m.tss:.0f}")
|
|
if m.intensity_factor:
|
|
lines.append(f"IF: {m.intensity_factor:.2f}")
|
|
if m.variability_index:
|
|
lines.append(f"VI: {m.variability_index:.2f}")
|
|
if m.avg_hr:
|
|
lines.append(f"Avg HR: {m.avg_hr} bpm")
|
|
if m.max_hr:
|
|
lines.append(f"Max HR: {m.max_hr} bpm")
|
|
if m.avg_cadence:
|
|
lines.append(f"Avg Cadence: {m.avg_cadence} rpm")
|
|
if m.calories:
|
|
lines.append(f"Calories: {m.calories} kcal")
|
|
|
|
if rider_ftp:
|
|
lines.append(f"Rider FTP: {rider_ftp:.0f} W")
|
|
|
|
# Exercise sets for strength workouts
|
|
if activity.exercise_sets:
|
|
lines.append(f"\nExercise sets ({len(activity.exercise_sets)} total):")
|
|
for s in activity.exercise_sets:
|
|
parts = [s.get("exercise_name", "Unknown")]
|
|
if s.get("repetitions"):
|
|
parts.append(f"{s['repetitions']} reps")
|
|
if s.get("weight"):
|
|
parts.append(f"{s['weight']} kg")
|
|
if s.get("duration") and not s.get("repetitions"):
|
|
parts.append(f"{s['duration']:.0f}s")
|
|
lines.append(f" - {' / '.join(parts)}")
|
|
|
|
intervals = activity.intervals or []
|
|
work_intervals = [i for i in intervals if i.interval_type == "work"]
|
|
if work_intervals:
|
|
lines.append(f"Work intervals: {len(work_intervals)}")
|
|
powers = [i.avg_power for i in work_intervals if i.avg_power]
|
|
if powers:
|
|
lines.append(f"Interval avg powers: {', '.join(f'{p:.0f}W' for p in powers)}")
|
|
|
|
# Planned workout from training plan
|
|
if planned_workout:
|
|
lines.append("\n--- PLANNED WORKOUT ---")
|
|
lines.append(f"Planned type: {planned_workout.get('workout_type', '?')}")
|
|
lines.append(f"Planned title: {planned_workout.get('title', '?')}")
|
|
if planned_workout.get("description"):
|
|
lines.append(f"Planned description: {planned_workout['description']}")
|
|
if planned_workout.get("duration_minutes"):
|
|
lines.append(f"Planned duration: {planned_workout['duration_minutes']} min")
|
|
if planned_workout.get("target_tss"):
|
|
lines.append(f"Planned TSS: {planned_workout['target_tss']}")
|
|
if planned_workout.get("target_if"):
|
|
lines.append(f"Planned IF: {planned_workout['target_if']}")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
async def generate_summary(
|
|
activity: Activity,
|
|
rider_ftp: float | None = None,
|
|
planned_workout: dict | None = None,
|
|
) -> str:
|
|
prompt = _build_activity_prompt(activity, rider_ftp, planned_workout)
|
|
messages = [{"role": "user", "text": prompt}]
|
|
system = SYSTEM_PROMPT_WITH_PLAN if planned_workout else SYSTEM_PROMPT
|
|
return await chat_async(messages, system_instruction=system, temperature=0.5)
|