runner-ai-intelligence / src /services /performance_card_service.py
avfranco's picture
HF Space deploy snapshot (minimal allow-list)
d64fd55
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
)