507 lines
21 KiB
Python
507 lines
21 KiB
Python
import colorsys
|
|
from datetime import datetime, date, timedelta
|
|
from flask import request
|
|
import numpy as np
|
|
|
|
|
|
import math
|
|
from decimal import Decimal
|
|
|
|
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), 26))
|
|
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]
|
|
future_values = intercept + slope * future_x
|
|
|
|
# A 95% prediction band, widened for points further from the observations.
|
|
prediction_error = 1.96 * 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),
|
|
'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 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):
|
|
# 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}
|
|
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))
|
|
|
|
forecast_points = []
|
|
forecast_band_lower = []
|
|
forecast_band_upper = []
|
|
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))
|
|
|
|
# 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,
|
|
'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_points': forecast_points,
|
|
'forecast_band_lower': forecast_band_lower,
|
|
'forecast_band_upper': forecast_band_upper,
|
|
'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)) / (101.3 - 2.67123 * 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
|