Spaces:
Running
Running
File size: 4,612 Bytes
d64fd55 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | 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
)
|