from dataclasses import dataclass from typing import Optional, Any from datetime import date import logging from domain.training.weekly_trend import WeeklyTrend from domain.training.weekly_snapshot import WeeklySnapshot logger = logging.getLogger(__name__) @dataclass class CardData: week_start_date: date total_distance: float delta_vs_avg: float pace_delta: float hr_delta: Optional[float] consistency_score: float performance_direction: str # IMPROVING, STABLE, RECOVERY, LOAD, RISK, ONBOARDING risk_flag: bool indicator_icon: str comparison_type: str = "none" @dataclass class CardViewModel: data: CardData insight_paragraph: str forward_focus: str language: str llm_used: bool = False class PerformanceCardService: """Service for building and managing the Weekly Performance Card. Stateless and pure.""" def build_card_data( self, snapshot: WeeklySnapshot, trend: WeeklyTrend ) -> CardData: """Builds deterministic metric payload for the performance card using hybrid trends.""" # Determine direction and icon based on hybrid trend if not trend.comparison_available: direction, icon, risk_flag = "ONBOARDING", "🚀", False else: direction, icon, risk_flag = self.compute_indicator( trend.distance_delta_pct, trend.pace_delta_s_per_km, trend.hr_delta, snapshot.total_distance_km ) return CardData( week_start_date=snapshot.week_start_date, total_distance=snapshot.total_distance_km, delta_vs_avg=trend.distance_delta_pct, pace_delta=trend.pace_delta_s_per_km, hr_delta=trend.hr_delta, consistency_score=snapshot.consistency_score, performance_direction=direction, risk_flag=risk_flag, indicator_icon=icon, comparison_type=trend.comparison_type ) def compute_indicator( self, delta_pct: float, pace_delta: float, hr_delta: Optional[float], total_distance: float ) -> tuple[str, str, bool]: """ Deterministic logic for performance indicators. Returns (direction_key, icon, risk_flag) """ # Thresholds STABLE_THRESHOLD = 5.0 # % for distance HR_STABLE_THRESHOLD = 2.0 # bpm VOLUME_SPIKE_THRESHOLD = 25.0 # % HR_SPIKE_THRESHOLD = 5.0 # bpm # 1. Elevated Risk (Red) if (hr_delta is not None and hr_delta > HR_SPIKE_THRESHOLD) or delta_pct > VOLUME_SPIKE_THRESHOLD: return "RISK", "🔴", True # 2. Elevated Load (Orange) if delta_pct > 15.0 or (hr_delta is not None and hr_delta > HR_STABLE_THRESHOLD): return "LOAD", "🟠", False # 3. Improving (Green) # Improving if distance is stable/up and HR is stable/down if delta_pct > 2.0 and (hr_delta is None or hr_delta <= 0): return "IMPROVING", "🟢", False # 4. Recovery (Blue) if delta_pct < -10.0 and (hr_delta is None or hr_delta <= 0): return "RECOVERY", "🔵", False # 5. Stable (Yellow) - Fallback return "STABLE", "🟡", False async def generate( self, snapshot: WeeklySnapshot, trend: WeeklyTrend, brief_service: Any, language: str = "en" ) -> CardViewModel: """ Main entry point for generating a performance card. Decides between LLM insight or deterministic onboarding. Stateless logic only. Observability managed by caller. """ card_data = self.build_card_data(snapshot, trend) llm_used = False if trend.comparison_available: # Proceed with LLM brief generation insight, focus = await brief_service.generate_performance_card_insight( card_data, language=language ) llm_used = True else: # Generate deterministic onboarding message (NO LLM) from _app.presentation.ui_text import get_text insight = get_text("performance_first_week_body", language) focus = get_text("performance_first_week_focus", language) llm_used = False return CardViewModel( data=card_data, insight_paragraph=insight, forward_focus=focus, language=language, llm_used=llm_used )