""" Body Metrics Data Layer Provides structured data for body composition and measurements. Functions: - get_weight_trend_data(): Weight trend with slope and direction - get_body_composition_data(): Body fat percentage and lean mass - get_circumference_summary_data(): Latest circumference measurements All functions return structured data (dict) without formatting. Use placeholder_resolver.py for formatted strings for AI. Phase 0c: Multi-Layer Architecture Version: 1.0 """ from typing import Dict, List, Optional, Tuple from datetime import datetime, timedelta, date from db import get_db, get_cursor, r2d from data_layer.utils import calculate_confidence, safe_float def get_weight_trend_data( profile_id: str, days: int = 28 ) -> Dict: """ Calculate weight trend with slope and direction. Args: profile_id: User profile ID days: Analysis window (default 28) Returns: { "first_value": float, "last_value": float, "delta": float, # kg change "direction": str, # "increasing" | "decreasing" | "stable" "data_points": int, "confidence": str, "days_analyzed": int, "first_date": date, "last_date": date } Confidence Rules: - high: >= 18 points (28d) or >= 4 points (7d) - medium: >= 12 points (28d) or >= 3 points (7d) - low: >= 8 points (28d) or >= 2 points (7d) - insufficient: < thresholds Migration from Phase 0b: OLD: get_weight_trend() returned formatted string NEW: Returns structured data for reuse in charts + AI """ with get_db() as conn: cur = get_cursor(conn) cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') cur.execute( """SELECT weight, date FROM weight_log WHERE profile_id=%s AND date >= %s ORDER BY date""", (profile_id, cutoff) ) rows = [r2d(r) for r in cur.fetchall()] # Calculate confidence confidence = calculate_confidence(len(rows), days, "general") # Early return if insufficient if confidence == 'insufficient' or len(rows) < 2: return { "confidence": "insufficient", "data_points": len(rows), "days_analyzed": days, "first_value": 0.0, "last_value": 0.0, "delta": 0.0, "direction": "unknown", "first_date": None, "last_date": None } # Extract values first_value = safe_float(rows[0]['weight']) last_value = safe_float(rows[-1]['weight']) delta = last_value - first_value # Determine direction if abs(delta) < 0.3: direction = "stable" elif delta > 0: direction = "increasing" else: direction = "decreasing" return { "first_value": first_value, "last_value": last_value, "delta": delta, "direction": direction, "data_points": len(rows), "confidence": confidence, "days_analyzed": days, "first_date": rows[0]['date'], "last_date": rows[-1]['date'] } def get_body_composition_data( profile_id: str, days: int = 90 ) -> Dict: """ Get latest body composition data (body fat, lean mass). Args: profile_id: User profile ID days: Lookback window (default 90) Returns: { "body_fat_pct": float, "method": str, # "jackson_pollock" | "durnin_womersley" | etc. "date": date, "confidence": str, "data_points": int } Migration from Phase 0b: OLD: get_latest_bf() returned formatted string "15.2%" NEW: Returns structured data {"body_fat_pct": 15.2, ...} """ with get_db() as conn: cur = get_cursor(conn) cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') cur.execute( """SELECT body_fat_pct, sf_method, date FROM caliper_log WHERE profile_id=%s AND body_fat_pct IS NOT NULL AND date >= %s ORDER BY date DESC LIMIT 1""", (profile_id, cutoff) ) row = r2d(cur.fetchone()) if cur.rowcount > 0 else None if not row: return { "confidence": "insufficient", "data_points": 0, "body_fat_pct": 0.0, "method": None, "date": None } return { "body_fat_pct": safe_float(row['body_fat_pct']), "method": row.get('sf_method', 'unknown'), "date": row['date'], "confidence": "high", # Latest measurement is always high confidence "data_points": 1 } def get_circumference_summary_data( profile_id: str, max_age_days: int = 90 ) -> Dict: """ Get latest circumference measurements for all body points. For each measurement point, fetches the most recent value (even if from different dates). Returns measurements with age in days for each point. Args: profile_id: User profile ID max_age_days: Maximum age of measurements to include (default 90) Returns: { "measurements": [ { "point": str, # "Nacken", "Brust", etc. "field": str, # "c_neck", "c_chest", etc. "value": float, # cm "date": date, "age_days": int }, ... ], "confidence": str, "data_points": int, "newest_date": date, "oldest_date": date } Migration from Phase 0b: OLD: get_circ_summary() returned formatted string "Nacken 38.0cm (vor 2 Tagen), ..." NEW: Returns structured array for charts + AI formatting """ with get_db() as conn: cur = get_cursor(conn) # Define all circumference points fields = [ ('c_neck', 'Nacken'), ('c_chest', 'Brust'), ('c_waist', 'Taille'), ('c_belly', 'Bauch'), ('c_hip', 'Hüfte'), ('c_thigh', 'Oberschenkel'), ('c_calf', 'Wade'), ('c_arm', 'Arm') ] measurements = [] today = datetime.now().date() # Get latest value for each field individually for field_name, label in fields: cur.execute( f"""SELECT {field_name}, date, CURRENT_DATE - date AS age_days FROM circumference_log WHERE profile_id=%s AND {field_name} IS NOT NULL AND date >= %s ORDER BY date DESC LIMIT 1""", (profile_id, (today - timedelta(days=max_age_days)).isoformat()) ) row = r2d(cur.fetchone()) if cur.rowcount > 0 else None if row: measurements.append({ "point": label, "field": field_name, "value": safe_float(row[field_name]), "date": row['date'], "age_days": row['age_days'] }) # Calculate confidence based on how many points we have confidence = calculate_confidence(len(measurements), 8, "general") if not measurements: return { "measurements": [], "confidence": "insufficient", "data_points": 0, "newest_date": None, "oldest_date": None } # Find newest and oldest dates dates = [m['date'] for m in measurements] newest_date = max(dates) oldest_date = min(dates) return { "measurements": measurements, "confidence": confidence, "data_points": len(measurements), "newest_date": newest_date, "oldest_date": oldest_date }