Spaces:
Running
Running
| from typing import Optional, Dict, Any | |
| from domain.runner_positioning import RunnerPositioning | |
| from domain.training.weekly_snapshot import WeeklySnapshot | |
| from domain.training.weekly_trend import WeeklyTrend | |
| from application.positioning_service import PositioningEngine | |
| class RunnerPositioningService: | |
| """ | |
| Application service for Runner Positioning. | |
| Enforces architectural boundaries: no LLM, no persistence, no observability here. | |
| """ | |
| def generate( | |
| self, | |
| snapshot: WeeklySnapshot, | |
| trend: WeeklyTrend, | |
| goal_progress: Optional[Dict[str, Any]] = None | |
| ) -> RunnerPositioning: | |
| """ | |
| Aggregates domain data into a RunnerPositioning assessment. | |
| """ | |
| # Extract necessary values for deterministic logic | |
| target_distance = None | |
| if goal_progress: | |
| # goal_service.compute_goal_progress returns a dict | |
| # We assume it has a 'target_value' if applicable | |
| target_distance = goal_progress.get("target_value") | |
| # Call domain logic | |
| engine = PositioningEngine() | |
| training_phase = engine.detect_training_phase(snapshot, trend) | |
| return RunnerPositioning.compute( | |
| week_start=snapshot.week_start_date, | |
| total_distance=snapshot.total_distance_km, | |
| target_distance=target_distance, | |
| consistency_score=snapshot.consistency_score, | |
| pace_delta=trend.pace_delta_s_per_km, | |
| hr_delta=trend.hr_delta, | |
| distance_delta_pct=trend.distance_delta_pct, | |
| comparison_available=trend.comparison_available, | |
| training_phase=training_phase | |
| ) | |