Spaces:
Running
Running
File size: 7,074 Bytes
557ee65 | 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 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 | 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
}
|