import colorsys from datetime import datetime, date, timedelta from flask import request import numpy as np import math from decimal import Decimal E1RM_BASE = 101.3 E1RM_REP_FACTOR = 2.67123 def _format_kg(value): """Format a load without noisy trailing zeroes.""" numeric_value = float(value) if numeric_value.is_integer(): return str(int(numeric_value)) return f"{numeric_value:.1f}".rstrip('0').rstrip('.') def convert_str_to_date(date_str, format='%Y-%m-%d'): try: return datetime.strptime(date_str, format).date() except ValueError: return None except TypeError: return None def build_forecast(values, dates, horizon_weeks=8, metric='E1RM'): """Build a conservative linear forecast from dated exercise observations.""" try: horizon_weeks = max(4, min(int(horizon_weeks), 52)) except (TypeError, ValueError): horizon_weeks = 8 observations = [] for value, point_date in zip(values, dates): if value is None or point_date is None: continue try: numeric_value = float(value) except (TypeError, ValueError): continue if isinstance(point_date, datetime): point_date = point_date.date() observations.append((point_date, numeric_value)) # Multiple sets on one date should not make the forecast look more certain. # Use the best observed value for that date, which matches the E1RM use case. best_by_date = {} for point_date, numeric_value in observations: best_by_date[point_date] = max(best_by_date.get(point_date, numeric_value), numeric_value) observations = sorted(best_by_date.items()) if len(observations) < 4: return { 'available': False, 'reason': 'Add at least 4 sessions to generate a forecast.', 'metric': metric, 'horizon_weeks': horizon_weeks, 'observed_points': len(observations) } observed_dates = [point_date for point_date, _ in observations] observed_values = np.array([value for _, value in observations], dtype=float) first_date = observed_dates[0] last_date = observed_dates[-1] x = np.array([(point_date - first_date).days for point_date in observed_dates], dtype=float) if len(np.unique(x)) < 3 or x[-1] <= 0: return { 'available': False, 'reason': 'Use at least 3 different workout dates to generate a forecast.', 'metric': metric, 'horizon_weeks': horizon_weeks, 'observed_points': len(observations) } try: slope, intercept = np.polyfit(x, observed_values, 1) except (np.linalg.LinAlgError, ValueError, TypeError): return { 'available': False, 'reason': 'There is not enough consistent data to generate a forecast.', 'metric': metric, 'horizon_weeks': horizon_weeks, 'observed_points': len(observations) } fitted = intercept + slope * x residuals = observed_values - fitted degrees_of_freedom = max(len(observed_values) - 2, 1) residual_std = float(np.sqrt(np.sum(residuals ** 2) / degrees_of_freedom)) mean_x = float(np.mean(x)) sum_squared_x = float(np.sum((x - mean_x) ** 2)) or 1.0 horizon_days = horizon_weeks * 7 future_x = np.arange(0, horizon_days + 1, 7, dtype=float) + x[-1] future_dates = [last_date + timedelta(days=int(day - x[-1])) for day in future_x] # Anchor the projection to the most recent observed value. A pure # regression line can sit noticeably above or below the latest workout, # which makes the forecast appear to jump when it starts. future_values = observed_values[-1] + slope * (future_x - x[-1]) # A restrained 90% prediction band, widened for points further from the # observations so the uncertainty remains readable on the small chart. prediction_error = 1.645 * residual_std * np.sqrt( 1 + (1 / len(observed_values)) + ((future_x - mean_x) ** 2 / sum_squared_x) ) lower_values = np.maximum(future_values - prediction_error, 0) upper_values = future_values + prediction_error total_variation = float(np.sum((observed_values - np.mean(observed_values)) ** 2)) residual_variation = float(np.sum(residuals ** 2)) r_squared = None if total_variation == 0 else max(0.0, 1 - residual_variation / total_variation) weekly_change = float(slope * 7) confidence = 'High' if len(observed_values) >= 8 and (r_squared is None or r_squared >= 0.6) else 'Moderate' if len(observed_values) < 6 or (r_squared is not None and r_squared < 0.25): confidence = 'Low' points = [] for point_date, predicted, lower, upper in zip(future_dates, future_values, lower_values, upper_values): points.append({ 'date': point_date, 'value': round(float(predicted), 1), 'lower': round(float(lower), 1), 'upper': round(float(upper), 1) }) return { 'available': True, 'metric': metric, 'horizon_weeks': horizon_weeks, 'observed_points': len(observations), 'last_date': last_date, 'last_value': round(float(observed_values[-1]), 1), 'projected_value': round(float(future_values[-1]), 1), 'projected_change': round(float(future_values[-1] - observed_values[-1]), 1), 'weekly_change': round(weekly_change, 1), 'r_squared': None if r_squared is None else round(r_squared, 2), 'confidence': confidence, 'end_date': future_dates[-1], 'points': points } def infer_weight_increment(weights): """Infer a sensible display increment from the exercise's recorded weights.""" numeric_weights = sorted({round(float(weight), 4) for weight in weights if weight is not None}) differences = [ round(current - previous, 4) for previous, current in zip(numeric_weights, numeric_weights[1:]) if current > previous ] if not differences: return 2.5 common_increments = [0.5, 1.0, 1.25, 2.0, 2.5, 5.0, 10.0] smallest_difference = min(differences) return min(common_increments, key=lambda increment: abs(increment - smallest_difference)) def _round_weight(value, increment, direction='nearest'): if value is None: return None scaled = max(float(value), 0) / increment if direction == 'floor': rounded = math.floor(scaled) * increment elif direction == 'ceil': rounded = math.ceil(scaled) * increment else: rounded = round(scaled) * increment return round(rounded, 2) def infer_rep_targets(repetitions, buffer=5): """Return an expanded rep range based on the currently visible history.""" numeric_reps = sorted({int(round(float(value))) for value in repetitions if value is not None}) if not numeric_reps: return [1, 2, 3, 4, 5, 6, 7, 8, 9, 10], 5, None, None observed_min = min(numeric_reps) observed_max = max(numeric_reps) target_min = max(1, observed_min - buffer) target_max = min(20, observed_max + buffer) target_reps = list(range(target_min, target_max + 1)) centered_reps = int(round(float(np.median(numeric_reps)))) centered_reps = max(target_min, min(target_max, centered_reps)) return target_reps, centered_reps, observed_min, observed_max def _top_set_estimates(e1rm, lower_e1rm, upper_e1rm, target_reps, increment): estimates = [] for repetitions in target_reps: conversion = (E1RM_BASE - E1RM_REP_FACTOR * repetitions) / 100 if conversion <= 0: continue expected_weight = e1rm * conversion low_weight = lower_e1rm * conversion high_weight = upper_e1rm * conversion expected = _round_weight(expected_weight, increment) lower = _round_weight(low_weight, increment, 'floor') upper = _round_weight(high_weight, increment, 'ceil') estimates.append({ 'repetitions': repetitions, 'weight': expected, 'weight_display': _format_kg(expected), 'lower_weight': lower, 'lower_weight_display': _format_kg(lower), 'upper_weight': upper, 'upper_weight_display': _format_kg(upper), 'e1rm_display': _format_kg(e1rm) }) return estimates def add_top_set_forecast(forecast, historical_weights, historical_repetitions, target_reps=None): """Attach projected reps x weight combinations to each forecast point.""" if not forecast.get('available'): return forecast target_reps_list, centered_reps, observed_rep_min, observed_rep_max = infer_rep_targets(historical_repetitions) try: selected_reps = centered_reps if target_reps is None else int(target_reps) except (TypeError, ValueError): selected_reps = centered_reps selected_reps = max(target_reps_list[0], min(target_reps_list[-1], selected_reps)) increment = infer_weight_increment(historical_weights) for point in forecast['points']: point['top_sets'] = _top_set_estimates( point['value'], point['lower'], point['upper'], target_reps_list, increment ) forecast['target_reps'] = target_reps_list forecast['target_weight_increment'] = increment forecast['target_weight_increment_display'] = _format_kg(increment) forecast['observed_rep_min'] = observed_rep_min forecast['observed_rep_max'] = observed_rep_max forecast['rep_buffer'] = 5 forecast['top_sets'] = forecast['points'][-1]['top_sets'] forecast['selected_reps'] = selected_reps forecast['selected_top_set'] = next( (top_set for top_set in forecast['top_sets'] if top_set['repetitions'] == selected_reps), None ) return forecast def get_exercise_graph_model(title, estimated_1rm, repetitions, weight, start_dates, messages, epoch, person_id, exercise_id, min_date=None, max_date=None, degree=1, forecast_enabled=False, forecast_weeks=8, forecast_target_reps=None): # Precompute ranges. Forecast values are included so the projected line and # confidence band have room inside the same SVG coordinate system. observed_min_date, observed_max_date = min(start_dates), max(start_dates) forecast = build_forecast(estimated_1rm, start_dates, forecast_weeks, metric='E1RM') if forecast_enabled else {'available': False} forecast = add_top_set_forecast(forecast, weight, repetitions, forecast_target_reps) forecast_horizons = [(4, '4w'), (8, '8w'), (12, '3m'), (26, '6m'), (52, '1y')] chart_max_date = forecast.get('end_date', observed_max_date) if forecast.get('available') else observed_max_date total_span = (chart_max_date - observed_min_date).days or 1 e1rm_values_for_scale = list(estimated_1rm) if forecast.get('available'): e1rm_values_for_scale.extend([point['lower'] for point in forecast['points']]) e1rm_values_for_scale.extend([point['upper'] for point in forecast['points']]) min_e1rm, max_e1rm = min(e1rm_values_for_scale), max(e1rm_values_for_scale) min_reps, max_reps = min(repetitions), max(repetitions) min_weight, max_weight = min(weight), max(weight) e1rm_range = max_e1rm - min_e1rm or 1 reps_range = max_reps - min_reps or 1 weight_range = max_weight - min_weight or 1 # Calculate viewBox dimensions vb_width, vb_height = total_span, e1rm_range vb_width *= 200 / vb_width # Scale to 200px width vb_height *= 75 / vb_height # Scale to 75px height # Use NumPy arrays for efficient scaling relative_positions = np.round(np.array([(date - observed_min_date).days / total_span for date in start_dates]), 3) estimated_1rm_scaled = np.round(((np.array(estimated_1rm) - min_e1rm) / e1rm_range) * vb_height, 1) repetitions_scaled = np.round(((np.array(repetitions) - min_reps) / reps_range) * vb_height, 1) weight_scaled = np.round(((np.array(weight) - min_weight) / weight_range) * vb_height, 1) # Calculate slope and line of best fit slope_kg_per_day = e1rm_range / total_span best_fit_formula = { 'kg_per_week': round(slope_kg_per_day * 7, 1), 'kg_per_month': round(slope_kg_per_day * 30, 1) } if forecast.get('available'): best_fit_formula = { 'kg_per_week': forecast['weekly_change'], 'kg_per_month': round(forecast['weekly_change'] * 4.345, 1) } best_fit_points = [] try: # Filter out NaNs if any (though scaled values shouldn't have them if ranges are correct) mask = ~np.isnan(estimated_1rm_scaled) x_fit = relative_positions[mask] y_fit = estimated_1rm_scaled[mask] # Ensure we have enough unique X positions for the given degree if len(np.unique(x_fit)) > degree: coeffs = np.polyfit(x_fit, y_fit, degree) poly_fit = np.poly1d(coeffs) y_best_fit = np.round(poly_fit(relative_positions), 1) best_fit_points = list(zip(y_best_fit.tolist(), relative_positions.tolist())) else: best_fit_points = [] except (np.linalg.LinAlgError, ValueError, TypeError) as e: # Handle cases where polyfit fails or input is invalid best_fit_points = [] # Prepare data for plots repetitions_data = { 'label': 'Reps', 'color': '#388fed', 'points': list(zip(repetitions_scaled.tolist(), relative_positions.tolist())) } weight_data = { 'label': 'Weight', 'color': '#bd3178', 'points': list(zip(weight_scaled.tolist(), relative_positions.tolist())) } estimated_1rm_data = { 'label': 'E1RM', 'color': '#2ca02c', 'points': list(zip(estimated_1rm_scaled.tolist(), relative_positions.tolist())) } # Prepare plot labels plot_labels = list(zip(relative_positions.tolist(), messages)) plot_details = [] for position, point_date, point_e1rm, point_reps, point_weight in zip( relative_positions.tolist(), start_dates, estimated_1rm, repetitions, weight ): plot_details.append({ 'position': position, 'date': point_date.strftime('%d %b %y'), 'kind': 'Recorded top set', 'e1rm': f"{_format_kg(point_e1rm)}kg", 'top_set': f"{_format_kg(point_weight)}kg × {int(point_reps)} reps", 'range': 'Recorded value' }) forecast_points = [] forecast_band_lower = [] forecast_band_upper = [] forecast_labels = [] forecast_details = [] forecast_start_position = None if forecast.get('available'): forecast_start_position = round((observed_max_date - observed_min_date).days / total_span, 3) for point in forecast['points']: position = round((point['date'] - observed_min_date).days / total_span, 3) forecast_points.append((round(((point['value'] - min_e1rm) / e1rm_range) * vb_height, 1), position)) forecast_band_lower.append((round(((point['lower'] - min_e1rm) / e1rm_range) * vb_height, 1), position)) forecast_band_upper.append((round(((point['upper'] - min_e1rm) / e1rm_range) * vb_height, 1), position)) selected_top_set = next( (top_set for top_set in point['top_sets'] if top_set['repetitions'] == forecast['selected_reps']), None ) top_set_message = '' if selected_top_set: top_set_message = ( f" · Est. {selected_top_set['weight_display']}kg × {selected_top_set['repetitions']} " f"(range {selected_top_set['lower_weight_display']}–{selected_top_set['upper_weight_display']}kg)" ) forecast_details.append({ 'position': position, 'date': point['date'].strftime('%d %b %y'), 'kind': 'Projected top set', 'e1rm': f"{_format_kg(point['value'])}kg", 'top_set': f"{selected_top_set['weight_display']}kg × {selected_top_set['repetitions']} reps", 'range': f"{selected_top_set['lower_weight_display']}–{selected_top_set['upper_weight_display']}kg", 'confidence': forecast['confidence'] }) forecast_labels.append(( position, f"{point['date'].strftime('%d %b %y')} · Projected E1RM: {point['value']}kg " f"(range {point['lower']}–{point['upper']}kg){top_set_message}" )) # Return exercise data with SVG dimensions and data points return { 'title': title, 'vb_width': vb_width, 'vb_height': vb_height, 'plots': [repetitions_data, weight_data, estimated_1rm_data], 'best_fit_points': best_fit_points, 'best_fit_formula': best_fit_formula, 'plot_labels': plot_labels, 'plot_details': plot_details, 'epochs': ['Custom', '1M', '3M', '6M', 'All'], 'selected_epoch': epoch, 'person_id': person_id, 'exercise_id': exercise_id, 'min_date': observed_min_date, 'max_date': observed_max_date, 'degree': degree, 'forecast': forecast, 'forecast_enabled': forecast_enabled, 'forecast_weeks': forecast_weeks, 'forecast_target_reps': forecast.get('selected_reps', forecast_target_reps or 5), 'forecast_horizons': forecast_horizons, 'forecast_points': forecast_points, 'forecast_band_lower': forecast_band_lower, 'forecast_band_upper': forecast_band_upper, 'forecast_labels': forecast_labels, 'forecast_details': forecast_details, 'forecast_start_position': forecast_start_position } def get_distinct_colors(n): colors = [] for i in range(n): # Divide the color wheel into n parts hue = i / n # Convert HSL (Hue, Saturation, Lightness) to RGB and then to a Hex string rgb = colorsys.hls_to_rgb(hue, 0.6, 0.4) # Fixed lightness and saturation hex_color = '#{:02x}{:02x}{:02x}'.format(int(rgb[0]*255), int(rgb[1]*255), int(rgb[2]*255)) colors.append(hex_color) return colors def calculate_estimated_1rm(weight, repetitions): # Ensure the inputs are numeric if repetitions == 0: # Avoid division by zero return 0 estimated_1rm = round((100 * int(weight)) / (E1RM_BASE - E1RM_REP_FACTOR * repetitions), 0) return int(estimated_1rm) def _is_numeric(val): """Check if a value is numeric (int, float, Decimal).""" return isinstance(val, (int, float, Decimal)) def _is_datetime(val): """Check if a value is a date or datetime object.""" return isinstance(val, (date, datetime)) def _get_column_type(results, column_name): """Determine the effective type of a column (numeric, datetime, categorical).""" numeric_count = 0 datetime_count = 0 total_count = 0 for row in results: val = row.get(column_name) if val is not None: total_count += 1 if _is_numeric(val): numeric_count += 1 elif _is_datetime(val): datetime_count += 1 if total_count == 0: return 'categorical' # Default if all null or empty if numeric_count / total_count > 0.8: return 'numeric' # Allow some non-numeric noise if datetime_count / total_count > 0.8: return 'datetime' return 'categorical' def _normalize_value(value, min_val, range_val, target_max): """Normalize a value to a target range (e.g., SVG coordinate).""" if range_val == 0: return target_max / 2 # Avoid division by zero, place in middle return ((value - min_val) / range_val) * target_max def prepare_svg_plot_data(results, columns, title): """ Prepares data from raw SQL results for SVG plotting. Determines plot type and scales data. """ if not results: raise ValueError("No data provided for plotting.") num_columns = len(columns) plot_type = 'table' # Default if no suitable plot found plot_data = {} x_col, y_col = None, None x_type, y_type = None, None # --- Determine Plot Type and Columns --- if num_columns == 1: x_col = columns[0] x_type = _get_column_type(results, x_col) if x_type == 'numeric': plot_type = 'histogram' else: plot_type = 'bar_count' # Bar chart of value counts elif num_columns >= 2: # Prioritize common patterns x_col, y_col = columns[0], columns[1] x_type = _get_column_type(results, x_col) y_type = _get_column_type(results, y_col) if x_type == 'numeric' and y_type == 'numeric': plot_type = 'scatter' elif x_type == 'datetime' and y_type == 'numeric': plot_type = 'line' # Treat datetime as numeric for position elif x_type == 'categorical' and y_type == 'numeric': plot_type = 'bar' elif x_type == 'numeric' and y_type == 'categorical': # Could do horizontal bar, but let's stick to vertical for now plot_type = 'bar' # Treat numeric as category label, categorical as value (count?) - less common # Or maybe swap? Let's assume categorical X, numeric Y is more likely intended x_col, y_col = columns[1], columns[0] # Try swapping x_type, y_type = y_type, x_type if not (x_type == 'categorical' and y_type == 'numeric'): plot_type = 'table' # Revert if swap didn't help else: # Other combinations (datetime/cat, cat/cat, etc.) default to table plot_type = 'table' # --- Basic SVG Setup --- vb_width = 500 vb_height = 300 margin = {'top': 20, 'right': 20, 'bottom': 50, 'left': 60} # Increased bottom/left for labels/axes draw_width = vb_width - margin['left'] - margin['right'] draw_height = vb_height - margin['top'] - margin['bottom'] plot_data = { 'title': title, 'plot_type': plot_type, 'vb_width': vb_width, 'vb_height': vb_height, 'margin': margin, 'draw_width': draw_width, 'draw_height': draw_height, 'x_axis_label': x_col or '', 'y_axis_label': y_col or '', 'plots': [], 'x_ticks': [], 'y_ticks': [], 'original_results': results, # Keep original for table fallback 'original_columns': columns } if plot_type == 'table': return plot_data # No further processing needed for table fallback # --- Data Extraction and Scaling (Specific to Plot Type) --- points = [] x_values_raw = [] y_values_raw = [] # Extract relevant data, handling potential type issues for row in results: x_val_raw = row.get(x_col) y_val_raw = row.get(y_col) # Convert datetimes to numeric representation (e.g., days since min date) if x_type == 'datetime': x_values_raw.append(x_val_raw) # Keep original dates for range calculation elif _is_numeric(x_val_raw): x_values_raw.append(float(x_val_raw)) # Convert Decimal to float # Add handling for categorical X if needed (e.g., bar chart) if y_type == 'numeric': if _is_numeric(y_val_raw): y_values_raw.append(float(y_val_raw)) else: y_values_raw.append(None) # Mark non-numeric Y as None # Add handling for categorical Y if needed if not x_values_raw or not y_values_raw: plot_data['plot_type'] = 'table' # Fallback if essential data is missing return plot_data # Calculate ranges (handle datetime separately) if x_type == 'datetime': valid_dates = [d for d in x_values_raw if d is not None] if not valid_dates: plot_data['plot_type'] = 'table'; return plot_data min_x_dt, max_x_dt = min(valid_dates), max(valid_dates) # Convert dates to days since min_date for numerical scaling total_days = (max_x_dt - min_x_dt).days x_values_numeric = [(d - min_x_dt).days if d is not None else None for d in x_values_raw] min_x, max_x = 0, total_days else: # Numeric or Categorical (treat categorical index as numeric for now) valid_x = [x for x in x_values_raw if x is not None] if not valid_x: plot_data['plot_type'] = 'table'; return plot_data min_x, max_x = min(valid_x), max(valid_x) x_values_numeric = x_values_raw # Already numeric (or will be treated as such) valid_y = [y for y in y_values_raw if y is not None] if not valid_y: plot_data['plot_type'] = 'table'; return plot_data min_y, max_y = min(valid_y), max(valid_y) range_x = max_x - min_x range_y = max_y - min_y # Scale points for i, row in enumerate(results): x_num = x_values_numeric[i] y_num = y_values_raw[i] # Use original list which might have None if x_num is None or y_num is None: continue # Skip points with missing essential data # Scale X to drawing width, Y to drawing height (inverted Y for SVG) scaled_x = margin['left'] + _normalize_value(x_num, min_x, range_x, draw_width) scaled_y = margin['top'] + draw_height - _normalize_value(y_num, min_y, range_y, draw_height) points.append({ 'x': scaled_x, 'y': scaled_y, 'original': row # Store original row data for tooltips }) # --- Generate Ticks --- num_ticks = 5 # Desired number of ticks # X Ticks x_ticks = [] if range_x >= 0: step_x = (max_x - min_x) / (num_ticks -1) if num_ticks > 1 and range_x > 0 else 0 for i in range(num_ticks): tick_val_raw = min_x + i * step_x tick_pos = margin['left'] + _normalize_value(tick_val_raw, min_x, range_x, draw_width) label = "" if x_type == 'datetime': tick_date = min_x_dt + timedelta(days=tick_val_raw) label = tick_date.strftime('%Y-%m-%d') # Format date label else: # Numeric label = f"{tick_val_raw:.1f}" if isinstance(tick_val_raw, float) else str(tick_val_raw) x_ticks.append({'value': tick_val_raw, 'label': label, 'position': tick_pos}) # Y Ticks y_ticks = [] if range_y >= 0: step_y = (max_y - min_y) / (num_ticks - 1) if num_ticks > 1 and range_y > 0 else 0 for i in range(num_ticks): tick_val = min_y + i * step_y tick_pos = margin['top'] + draw_height - _normalize_value(tick_val, min_y, range_y, draw_height) label = f"{tick_val:.1f}" if isinstance(tick_val, float) else str(tick_val) y_ticks.append({'value': tick_val, 'label': label, 'position': tick_pos}) # --- Finalize Plot Data --- # For now, put all points into one series plot_data['plots'].append({ 'label': f'{y_col} vs {x_col}', 'color': '#388fed', # Default color 'points': points }) plot_data['x_ticks'] = x_ticks plot_data['y_ticks'] = y_ticks # Add specific adjustments for plot types if needed (e.g., bar width) if plot_type == 'bar': # Calculate bar width based on number of bars/categories # This needs more refinement based on how categorical X is handled plot_data['bar_width'] = draw_width / len(points) * 0.8 if points else 10 return plot_data def get_client_ip(): """Get real client IP address, checking proxy headers first""" # Check common proxy headers in order of preference if request.headers.get('X-Forwarded-For'): # X-Forwarded-For can contain multiple IPs, get the first (original client) return request.headers.get('X-Forwarded-For').split(',')[0].strip() elif request.headers.get('X-Real-IP'): return request.headers.get('X-Real-IP') elif request.headers.get('CF-Connecting-IP'): # Cloudflare return request.headers.get('CF-Connecting-IP') else: # Fallback to direct connection IP return request.remote_addr