Spaces:
Running
Running
| from datetime import date, timedelta | |
| from typing import Optional, Dict, Any | |
| from _app.presentation.ui_text import get_text | |
| def _get_val(obj, key, default=None): | |
| if obj is None: | |
| return default | |
| # Mapping for new DTO keys against domain models | |
| ATTR_MAP = { | |
| "week_start": "week_start_date", | |
| "num_runs": "run_count", | |
| "weekly_distance_km": "total_distance_km" | |
| } | |
| def _extract(target, k): | |
| if target is None: | |
| return None, False | |
| if isinstance(target, dict): | |
| if k in target: | |
| return target[k], True | |
| return None, False | |
| from unittest.mock import MagicMock | |
| if isinstance(target, MagicMock): | |
| # Only return if specifically set (in __dict__) | |
| if k in target.__dict__: | |
| return getattr(target, k), True | |
| return None, False | |
| # Regular object | |
| if hasattr(target, k): | |
| return getattr(target, k), True | |
| return None, False | |
| # 1. Try exact key | |
| val, found = _extract(obj, key) | |
| if found: | |
| return val | |
| # 2. Try mapped key | |
| mapped = ATTR_MAP.get(key) | |
| if mapped: | |
| val, found = _extract(obj, mapped) | |
| if found: | |
| return val | |
| # 3. Final fallback | |
| if isinstance(obj, dict): | |
| return obj.get(key, default) | |
| # For non-mocks, try getattr one last time | |
| from unittest.mock import MagicMock | |
| if not isinstance(obj, MagicMock): | |
| return getattr(obj, key, default) | |
| return default | |
| def is_current_week(week_start) -> bool: | |
| """Detect if the given week start date corresponds to the current week.""" | |
| if not week_start: | |
| return False | |
| from datetime import date | |
| if isinstance(week_start, str): | |
| try: | |
| week_start = date.fromisoformat(week_start) | |
| except: | |
| return False | |
| today = date.today() | |
| # Monday of the current week | |
| current_monday = today - timedelta(days=today.weekday()) | |
| return week_start == current_monday | |
| def format_positioning_metrics(snapshot, language: str = "en") -> Dict[str, Any]: | |
| """ | |
| Returns absolute metrics for the current week to avoid misleading percentages. | |
| Returns delta metrics for previous weeks. | |
| """ | |
| if not snapshot: | |
| return {} | |
| w_start = _get_val(snapshot, "week_start") | |
| is_current = is_current_week(w_start) | |
| if is_current: | |
| dist_val = _get_val(snapshot, "weekly_distance_km", 0.0) | |
| dist_str = get_text("so_far_this_week", language).format(val=f"{dist_val:.1f} km") | |
| runs_val = _get_val(snapshot, "num_runs", 0) | |
| runs_str = f"{runs_val} " + (get_text("unit_runs", language) if runs_val != 1 else get_text("lbl_runs", language).lower()[:-1] if language == "en" else "corrida") | |
| # Add "building consistency" message if only 1 run | |
| consistency_msg = "" | |
| if runs_val <= 1: | |
| consistency_msg = get_text("building_consistency", language) | |
| return { | |
| "distance": dist_str, | |
| "runs": runs_str, | |
| "consistency_msg": consistency_msg, | |
| "mode": "absolute" | |
| } | |
| trend_val = _get_val(snapshot, "trend") | |
| if trend_val: | |
| dist_delta = _get_val(trend_val, "distance_delta_pct", 0.0) | |
| run_delta = _get_val(trend_val, "frequency_delta", 0) | |
| else: | |
| # Fallback to direct attributes for domain models (WeeklySnapshot doesn't have trend attached usually) | |
| dist_delta = _get_val(snapshot, "distance_delta_pct", 0.0) | |
| run_delta = _get_val(snapshot, "run_delta", 0) | |
| return { | |
| "distance": f"{dist_delta:+.1f}%", | |
| "runs": f"{run_delta:+d}", | |
| "consistency_msg": "", | |
| "mode": "delta" | |
| } | |
| def get_baseline_aware_target(current_km: float, baseline: float) -> float: | |
| """ | |
| Computes a safe target volume based on historical baseline. | |
| Rule: | |
| - If current < 60% of baseline -> Rebuild to 60% | |
| - If current < 80% of baseline -> Rebuild to 80% | |
| - Otherwise -> target baseline | |
| - Minimum safety floor: 8.0 km | |
| """ | |
| if not baseline or baseline <= 0: | |
| return max(current_km, 8.0) | |
| drop_ratio = current_km / baseline | |
| if drop_ratio < 0.6: | |
| target_km = baseline * 0.6 | |
| elif drop_ratio < 0.8: | |
| target_km = baseline * 0.8 | |
| else: | |
| target_km = baseline | |
| return max(target_km, 8.0) | |
| def interpret_week(snapshot, baseline: Optional[float], language: str = "en") -> Optional[str]: | |
| """Add simple interpretation rules for the narrative layer.""" | |
| if not snapshot: | |
| return None | |
| run_count = _get_val(snapshot, "num_runs", 0) | |
| if run_count <= 1: | |
| return get_text("early_week_building", language) | |
| dist_km = _get_val(snapshot, "weekly_distance_km", 0.0) | |
| if baseline and dist_km < baseline * 0.6: | |
| return get_text("rebuild_phase", language) | |
| return None | |