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