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 @property 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_baseline", 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" 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: """ Aggregates snapshot, trend and positioning into a narrative-ready dictionary. """ # Baseline Building deterministic return if positioning.status == "BASELINE_BUILDING": return { "headline": get_text("positioning_headline_baseline", language), "state": "⚪ " + get_text("positioning_status_baseline", language), "health_signal": "🟢 " + get_text("health_stable", language), "goal_trajectory": get_text("trajectory_establishing", language) if goal_progress else get_text("trajectory_no_goal", language), "training_phase": TrainingPhase.BASE.value, "forward_focus": get_text("positioning_forward_focus_baseline", language), "trajectory": get_text("positioning_trajectory_baseline", language), "insight": get_text("positioning_rationale_baseline", language), "evidence": None } # 5. Headline Mapping status_lower = positioning.status.lower() headline = get_text(f"positioning_headline_{status_lower}", language) # 6. Forward Focus Mapping forward_focus = get_text(f"positioning_forward_focus_{status_lower}", language) # 7. Trajectory Narrative Mapping trajectory_mapping = { "CONSTRUCTIVE_ADAPTATION": "building", "PRODUCTIVE_LOAD": "stable", "STRAIN": "fatigue", "PLATEAU": "plateau" } traj_key = trajectory_mapping.get(positioning.status, "plateau") trajectory_narrative = get_text(f"positioning_trajectory_{traj_key}", language) # 8. Icon Mappings (Presentation Step 1) status_icons = { "CONSTRUCTIVE_ADAPTATION": "🟢", "PRODUCTIVE_LOAD": "🟡", "STRAIN": "🔴", "PLATEAU": "⚪" } status_icon = status_icons.get(positioning.status, "⚪") # Get localized status name status_name = get_text(f"positioning_status_{status_lower}", language) # 9. Evidence Extraction evidence = { "distance": trend.distance_delta_pct, "pace": trend.pace_delta_s_per_km, "hr": trend.hr_delta or 0.0, "frequency": int(trend.frequency_delta), "consistency": trend.consistency_delta } # Goal Trajectory logic if goal_progress: goal_traj_text = f"🎯 {get_text('trajectory_improving', language)}" if positioning.status == "CONSTRUCTIVE_ADAPTATION" else f"🎯 {get_text('trajectory_maintaining', language)}" else: goal_traj_text = get_text("trajectory_no_goal", language) # Map to UI names return { "headline": headline, "state": f"{status_icon} {status_name}", "health_signal": f"🟢 {get_text('health_stable', language)}" if positioning.status != "STRAIN" else f"🔴 {get_text('health_strain', language)}", "goal_trajectory": goal_traj_text, "training_phase": positioning.training_phase.value if hasattr(positioning.training_phase, "value") else positioning.training_phase, "forward_focus": forward_focus, "trajectory": trajectory_narrative, "insight": get_text(positioning.rationale, language), # Resolve rationale key "evidence": evidence }