Spaces:
Runtime error
Runtime error
| from typing import Optional, Dict, List, Literal | |
| from pydantic import BaseModel, Field | |
| from datetime import date | |
| from domain.training.weekly_snapshot import WeeklySnapshot | |
| from domain.training.weekly_trend import WeeklyTrend | |
| from domain.runner_positioning import TrainingPhase | |
| from _app.presentation.ui_text import get_text | |
| class WeeklyPositioning(BaseModel): | |
| """ | |
| Application-layer model for Positioning Intelligence v1. | |
| Narrative-focused assessment of the runner's current standing. | |
| """ | |
| status: Literal["CONSTRUCTIVE_ADAPTATION", "PRODUCTIVE_LOAD", "STRAIN", "PLATEAU", "BASELINE_BUILDING"] | |
| signal_strength: float = 1.0 # 0.0 to 1.0 | |
| rationale: str | |
| training_phase: TrainingPhase = TrainingPhase.BASE | |
| def status_value(self) -> str: | |
| return self.status | |
| class PositioningEngine: | |
| """ | |
| Engine to compute high-level positioning status from snapshots and trends. | |
| """ | |
| def detect_training_phase(self, snapshot: WeeklySnapshot, trend: WeeklyTrend) -> TrainingPhase: | |
| """ | |
| Infer training phase from workload and trend signals. | |
| """ | |
| distance_delta = trend.distance_delta_pct or 0 | |
| pace_delta = trend.pace_delta_s_per_km or 0 | |
| run_count = snapshot.run_count | |
| # consistency = snapshot.consistency_score or 0 | |
| # Recovery phase (very low activity and sharp drop in load) | |
| if run_count <= 1 and distance_delta < -80: | |
| return TrainingPhase.RECOVERY | |
| # Plateau phase (load dropped or adaptation stalled) | |
| if distance_delta < -20: | |
| return TrainingPhase.PLATEAU | |
| # Build phase (clear volume increase) | |
| if distance_delta > 5: | |
| return TrainingPhase.BUILD | |
| # Peak phase (performance improving without load reduction) | |
| if pace_delta < -5 and distance_delta >= 0: | |
| return TrainingPhase.PEAK | |
| # Default: base training | |
| return TrainingPhase.BASE | |
| def compute(self, snapshot: WeeklySnapshot, trend: WeeklyTrend) -> WeeklyPositioning: | |
| if not trend or not trend.comparison_available: | |
| return WeeklyPositioning( | |
| status="BASELINE_BUILDING", | |
| rationale="positioning_rationale_building", | |
| signal_strength=0.1, | |
| training_phase=TrainingPhase.BASE | |
| ) | |
| # 1. Logic to determine status | |
| # This is a v1 heuristic mapping | |
| status = "PLATEAU" | |
| rationale_key = "positioning_rationale_plateau" | |
| if trend.comparison_available: | |
| # Constructive Adaptation: Increasing load with positive/stable consistency | |
| if trend.distance_delta_pct > 5.0 and trend.consistency_delta >= -5.0: | |
| status = "CONSTRUCTIVE_ADAPTATION" | |
| rationale_key = "positioning_rationale_constructive_adaptation" | |
| # Strain: High load increase or high HR increase | |
| elif trend.distance_delta_pct > 20.0 or (trend.hr_delta and trend.hr_delta > 5.0): | |
| status = "STRAIN" | |
| rationale_key = "positioning_rationale_strain" | |
| # Productive Load: Stable or slightly increasing load | |
| elif trend.distance_delta_pct >= -5.0: | |
| status = "PRODUCTIVE_LOAD" | |
| rationale_key = "positioning_rationale_productive_load" | |
| else: | |
| status = "BASELINE_BUILDING" | |
| rationale_key = "positioning_rationale_building" | |
| training_phase = self.detect_training_phase(snapshot, trend) | |
| return WeeklyPositioning( | |
| status=status, | |
| rationale=rationale_key, # Store key here | |
| signal_strength=1.0 if trend.comparison_available else 0.5, | |
| training_phase=training_phase | |
| ) | |
| def build_positioning_view( | |
| snapshot: WeeklySnapshot, | |
| trend: WeeklyTrend, | |
| positioning: WeeklyPositioning, | |
| goal_progress: Optional[dict] = None, | |
| language: str = "en" | |
| ) -> dict: | |
| """ | |
| Builds a UI-ready positioning view using BOTH absolute state (snapshot) | |
| and relative changes (trend). This restores coaching-quality output. | |
| """ | |
| positioning = PositioningEngine().compute(snapshot, trend) | |
| # --- Absolute context (snapshot) --- | |
| total_distance = snapshot.total_distance_km or 0 | |
| run_count = snapshot.run_count or 0 | |
| avg_pace = snapshot.avg_pace_sec_per_km | |
| avg_hr = snapshot.avg_hr | |
| # --- Relative context (trend) --- | |
| distance_delta = trend.distance_delta_pct or 0 | |
| pace_delta = trend.pace_delta_s_per_km or 0 | |
| hr_delta = trend.hr_delta or 0 | |
| freq_delta = int(trend.frequency_delta or 0) | |
| consistency_delta = trend.consistency_delta or 0 | |
| # --- Headline (context-aware) --- | |
| if run_count <= 1: | |
| headline = get_text("positioning_headline_building", language) | |
| elif total_distance < 5: | |
| headline = f"Your training volume is currently low at {total_distance:.1f} km." | |
| else: | |
| status_lower = positioning.status.lower() | |
| headline = get_text(f"positioning_headline_{status_lower}", language) | |
| # --- State --- | |
| status_icons = { | |
| "CONSTRUCTIVE_ADAPTATION": "🟢", | |
| "PRODUCTIVE_LOAD": "🟡", | |
| "STRAIN": "🔴", | |
| "PLATEAU": "⚪", | |
| "BASELINE_BUILDING": "⚪", | |
| } | |
| status_icon = status_icons.get(positioning.status, "⚪") | |
| status_name = get_text(f"positioning_status_{positioning.status.lower()}", language) | |
| # --- Health --- | |
| if hr_delta > 5: | |
| health_signal = f"🔴 {get_text('health_strain', language)}" | |
| else: | |
| health_signal = f"🟢 {get_text('health_stable', language)}" | |
| # --- Goal trajectory --- | |
| if goal_progress: | |
| if positioning.status == "BASELINE_BUILDING": | |
| goal_trajectory = get_text("trajectory_establishing", language) | |
| elif positioning.status == "CONSTRUCTIVE_ADAPTATION": | |
| goal_trajectory = f"🎯 {get_text('trajectory_improving', language)}" | |
| else: | |
| goal_trajectory = f"🎯 {get_text('trajectory_maintaining', language)}" | |
| else: | |
| goal_trajectory = get_text("trajectory_no_goal", language) | |
| # --- Forward focus --- | |
| forward_focus = get_text(f"positioning_forward_focus_{positioning.status.lower()}", language) | |
| # --- Narrative insight --- | |
| insight = get_text(positioning.rationale, language) | |
| # --- Evidence (enriched with absolute + relative) --- | |
| evidence = { | |
| "distance": f"{total_distance:.1f} km ({distance_delta:+.1f}%)", | |
| "pace": f"{pace_delta:+.1f} sec/km" if pace_delta else None, | |
| "hr": f"{hr_delta:+.1f} bpm" if hr_delta else None, | |
| "frequency": f"{run_count} runs ({freq_delta:+d})", | |
| "consistency": f"{int(consistency_delta):+d}", | |
| } | |
| # Remove empty evidence fields | |
| evidence = {k: v for k, v in evidence.items() if v is not None} | |
| return { | |
| "headline": headline, | |
| "state": f"{status_icon} {status_name}", | |
| "health_signal": health_signal, | |
| "goal_trajectory": goal_trajectory, | |
| "training_phase": positioning.training_phase.value if hasattr(positioning.training_phase, "value") else positioning.training_phase, | |
| "forward_focus": forward_focus, | |
| "trajectory": get_text(f"positioning_trajectory_{positioning.status.lower()}", language), | |
| "insight": insight, | |
| "evidence": evidence if positioning.status != "BASELINE_BUILDING" else None, | |
| } | |