667 lines
28 KiB
Python
667 lines
28 KiB
Python
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), 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]
|
||
# 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),
|
||
'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)
|
||
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_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
|