1c61b80731
- Add rules_engine.py with quantitative rules for all 8 workout types - Add quality gate retry loop in generate_single_workout() - Expand calibrate_structure_rules to all 120 combinations (8 types × 5 goals × 3 sections) - Wire WeeklySplitPattern DB records into _pick_weekly_split() - Enforce movement patterns from WorkoutStructureRule in exercise selection - Add straight-set strength support (single main lift, 4-6 rounds) - Add modality consistency check for duration-dominant workout types - Add InjuryStep component to onboarding and preferences - Add sibling exercise exclusion in regenerate and preview_day endpoints - Display generator warnings on dashboard - Expand fix_rep_durations, fix_exercise_flags, fix_movement_pattern_typo - Add audit_exercise_data and check_rules_drift management commands - Add Next.js frontend with dashboard, onboarding, preferences, history pages - Add generator app with ML-powered workout generation pipeline - 96 new tests across 7 test modules Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2303 lines
93 KiB
Python
2303 lines
93 KiB
Python
import logging
|
||
import math
|
||
import random
|
||
import time
|
||
import uuid
|
||
from datetime import timedelta
|
||
|
||
from django.db.models import Q
|
||
|
||
from generator.models import (
|
||
GeneratedWeeklyPlan,
|
||
GeneratedWorkout,
|
||
MuscleGroupSplit,
|
||
MovementPatternOrder,
|
||
WeeklySplitPattern,
|
||
WorkoutStructureRule,
|
||
WorkoutType,
|
||
)
|
||
from generator.rules_engine import validate_workout, RuleViolation
|
||
from generator.services.exercise_selector import ExerciseSelector
|
||
from generator.services.plan_builder import PlanBuilder
|
||
from generator.services.muscle_normalizer import normalize_muscle_name
|
||
from workout.models import CompletedWorkout
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ======================================================================
|
||
# Generation Rules — single source of truth for guardrails + API
|
||
# ======================================================================
|
||
|
||
GENERATION_RULES = {
|
||
'min_reps': {
|
||
'value': 6,
|
||
'description': 'Minimum reps for any exercise',
|
||
'category': 'rep_floors',
|
||
},
|
||
'min_reps_strength': {
|
||
'value': 1,
|
||
'description': 'Minimum reps for strength-type workouts (allows heavy singles)',
|
||
'category': 'rep_floors',
|
||
},
|
||
'fitness_scaling_order': {
|
||
'value': 'scale_then_clamp',
|
||
'description': 'Fitness scaling applied first, then clamped to min reps',
|
||
'category': 'rep_floors',
|
||
},
|
||
'min_duration': {
|
||
'value': 20,
|
||
'description': 'Minimum duration in seconds for any exercise',
|
||
'category': 'duration',
|
||
},
|
||
'duration_multiple': {
|
||
'value': 5,
|
||
'description': 'Durations must be multiples of this value',
|
||
'category': 'duration',
|
||
},
|
||
'strength_working_sets_rep_based': {
|
||
'value': True,
|
||
'description': 'Strength workout working sets must be rep-based',
|
||
'category': 'coherence',
|
||
},
|
||
'strength_prefer_weighted': {
|
||
'value': True,
|
||
'description': 'Strength workouts prefer is_weight=True exercises',
|
||
'category': 'coherence',
|
||
},
|
||
'strength_no_duration_working': {
|
||
'value': True,
|
||
'description': 'Duration exercises only in warmup/cooldown for strength workouts',
|
||
'category': 'coherence',
|
||
},
|
||
'min_exercises_per_superset': {
|
||
'value': 2,
|
||
'description': 'Minimum exercises per working set superset',
|
||
'category': 'superset',
|
||
},
|
||
'superset_same_modality': {
|
||
'value': True,
|
||
'description': 'Exercises within a superset must be same modality (all reps or all duration)',
|
||
'category': 'superset',
|
||
},
|
||
'min_volume': {
|
||
'value': 12,
|
||
'description': 'Minimum reps x rounds per exercise',
|
||
'category': 'superset',
|
||
},
|
||
'cooldown_stretch_only': {
|
||
'value': True,
|
||
'description': 'Cooldown exercises should be stretch/mobility only',
|
||
'category': 'coherence',
|
||
},
|
||
'workout_type_match_pct': {
|
||
'value': 0.6,
|
||
'description': 'At least 60% of working set exercises should match workout type character',
|
||
'category': 'coherence',
|
||
},
|
||
'no_muscle_unrelated_fallback': {
|
||
'value': True,
|
||
'description': 'No muscle-unrelated exercises from fallback paths',
|
||
'category': 'coherence',
|
||
},
|
||
'paired_sides_one_slot': {
|
||
'value': True,
|
||
'description': 'Paired sided exercises (Left/Right) count as 1 slot',
|
||
'category': 'superset',
|
||
},
|
||
'rest_between_supersets': {
|
||
'value': 30,
|
||
'description': 'Transition time between supersets in seconds',
|
||
'category': 'timing',
|
||
},
|
||
'push_pull_ratio_min': {
|
||
'value': 1.0,
|
||
'description': 'Minimum pull:push ratio (1.0 = equal push and pull)',
|
||
'category': 'balance',
|
||
},
|
||
'compound_before_isolation': {
|
||
'value': True,
|
||
'description': 'Compound exercises should precede isolation exercises',
|
||
'category': 'ordering',
|
||
},
|
||
'max_hiit_duration_min': {
|
||
'value': 30,
|
||
'description': 'Maximum recommended HIIT working duration in minutes',
|
||
'category': 'duration',
|
||
},
|
||
}
|
||
|
||
# Workout types that are "strength" character — force rep-based working sets
|
||
# Includes both underscore (DB) and space (display) variants for robustness
|
||
STRENGTH_WORKOUT_TYPES = {
|
||
'traditional strength', 'traditional strength training',
|
||
'traditional_strength', 'traditional_strength_training',
|
||
'functional strength', 'functional strength training',
|
||
'functional_strength', 'functional_strength_training',
|
||
'hypertrophy', 'strength',
|
||
}
|
||
|
||
|
||
# ======================================================================
|
||
# Default fallback data used when ML pattern tables are empty
|
||
# ======================================================================
|
||
|
||
DEFAULT_SPLITS = {
|
||
1: [
|
||
{'label': 'Full Body', 'muscles': ['chest', 'upper back', 'lats', 'deltoids', 'quads', 'hamstrings', 'glutes', 'core'], 'split_type': 'full_body'},
|
||
],
|
||
2: [
|
||
{'label': 'Upper Body', 'muscles': ['chest', 'upper back', 'lats', 'deltoids', 'biceps', 'triceps'], 'split_type': 'upper'},
|
||
{'label': 'Lower Body', 'muscles': ['quads', 'hamstrings', 'glutes', 'calves', 'core'], 'split_type': 'lower'},
|
||
],
|
||
3: [
|
||
{'label': 'Push', 'muscles': ['chest', 'deltoids', 'triceps'], 'split_type': 'push'},
|
||
{'label': 'Pull', 'muscles': ['upper back', 'lats', 'biceps', 'forearms'], 'split_type': 'pull'},
|
||
{'label': 'Legs', 'muscles': ['quads', 'hamstrings', 'glutes', 'calves', 'core'], 'split_type': 'legs'},
|
||
],
|
||
4: [
|
||
{'label': 'Upper Push', 'muscles': ['chest', 'deltoids', 'triceps'], 'split_type': 'push'},
|
||
{'label': 'Lower Body', 'muscles': ['quads', 'hamstrings', 'glutes', 'calves'], 'split_type': 'lower'},
|
||
{'label': 'Upper Pull', 'muscles': ['upper back', 'lats', 'biceps', 'forearms'], 'split_type': 'pull'},
|
||
{'label': 'Full Body', 'muscles': ['chest', 'upper back', 'lats', 'deltoids', 'quads', 'core'], 'split_type': 'full_body'},
|
||
],
|
||
5: [
|
||
{'label': 'Chest + Triceps', 'muscles': ['chest', 'triceps'], 'split_type': 'push'},
|
||
{'label': 'Back + Biceps', 'muscles': ['upper back', 'lats', 'biceps'], 'split_type': 'pull'},
|
||
{'label': 'Legs', 'muscles': ['quads', 'hamstrings', 'glutes', 'calves'], 'split_type': 'legs'},
|
||
{'label': 'Shoulders + Core', 'muscles': ['deltoids', 'core'], 'split_type': 'upper'},
|
||
{'label': 'Full Body', 'muscles': ['chest', 'upper back', 'lats', 'quads', 'core'], 'split_type': 'full_body'},
|
||
],
|
||
6: [
|
||
{'label': 'Push', 'muscles': ['chest', 'deltoids', 'triceps'], 'split_type': 'push'},
|
||
{'label': 'Pull', 'muscles': ['upper back', 'lats', 'biceps', 'forearms'], 'split_type': 'pull'},
|
||
{'label': 'Legs', 'muscles': ['quads', 'hamstrings', 'glutes', 'calves'], 'split_type': 'legs'},
|
||
{'label': 'Push 2', 'muscles': ['chest', 'deltoids', 'triceps'], 'split_type': 'push'},
|
||
{'label': 'Pull 2', 'muscles': ['upper back', 'lats', 'biceps', 'forearms'], 'split_type': 'pull'},
|
||
{'label': 'Legs 2', 'muscles': ['quads', 'hamstrings', 'glutes', 'calves'], 'split_type': 'legs'},
|
||
],
|
||
7: [
|
||
{'label': 'Push', 'muscles': ['chest', 'deltoids', 'triceps'], 'split_type': 'push'},
|
||
{'label': 'Pull', 'muscles': ['upper back', 'lats', 'biceps', 'forearms'], 'split_type': 'pull'},
|
||
{'label': 'Legs', 'muscles': ['quads', 'hamstrings', 'glutes', 'calves'], 'split_type': 'legs'},
|
||
{'label': 'Push 2', 'muscles': ['chest', 'deltoids', 'triceps'], 'split_type': 'push'},
|
||
{'label': 'Pull 2', 'muscles': ['upper back', 'lats', 'biceps', 'forearms'], 'split_type': 'pull'},
|
||
{'label': 'Legs 2', 'muscles': ['quads', 'hamstrings', 'glutes', 'calves'], 'split_type': 'legs'},
|
||
{'label': 'Core + Cardio', 'muscles': ['abs', 'obliques', 'core'], 'split_type': 'core'},
|
||
],
|
||
}
|
||
|
||
# Workout-type-name -> default parameters for working supersets.
|
||
# Keys: num_supersets, rounds, exercises_per_superset, rep_min, rep_max,
|
||
# duration_min, duration_max, duration_bias
|
||
WORKOUT_TYPE_DEFAULTS = {
|
||
'hiit': {
|
||
'num_supersets': (3, 5),
|
||
'rounds': (3, 4),
|
||
'exercises_per_superset': (4, 6),
|
||
'rep_min': 10,
|
||
'rep_max': 20,
|
||
'duration_min': 30,
|
||
'duration_max': 45,
|
||
'duration_bias': 0.7,
|
||
},
|
||
'hypertrophy': {
|
||
'num_supersets': (4, 5),
|
||
'rounds': (3, 4),
|
||
'exercises_per_superset': (2, 3),
|
||
'rep_min': 6,
|
||
'rep_max': 15,
|
||
'duration_min': 30,
|
||
'duration_max': 45,
|
||
'duration_bias': 0.1,
|
||
},
|
||
'traditional strength': {
|
||
'num_supersets': (3, 4),
|
||
'rounds': (4, 5),
|
||
'exercises_per_superset': (1, 3),
|
||
'rep_min': 3,
|
||
'rep_max': 8,
|
||
'duration_min': 30,
|
||
'duration_max': 45,
|
||
'duration_bias': 0.0,
|
||
},
|
||
'strength': {
|
||
'num_supersets': (3, 4),
|
||
'rounds': (4, 5),
|
||
'exercises_per_superset': (1, 3),
|
||
'rep_min': 3,
|
||
'rep_max': 8,
|
||
'duration_min': 30,
|
||
'duration_max': 45,
|
||
'duration_bias': 0.0,
|
||
},
|
||
'functional strength': {
|
||
'num_supersets': (3, 4),
|
||
'rounds': (3, 4),
|
||
'exercises_per_superset': (2, 4),
|
||
'rep_min': 6,
|
||
'rep_max': 12,
|
||
'duration_min': 30,
|
||
'duration_max': 45,
|
||
'duration_bias': 0.15,
|
||
},
|
||
'cardiovascular': {
|
||
'num_supersets': (2, 3),
|
||
'rounds': (2, 3),
|
||
'exercises_per_superset': (3, 5),
|
||
'rep_min': 12,
|
||
'rep_max': 20,
|
||
'duration_min': 45,
|
||
'duration_max': 90,
|
||
'duration_bias': 1.0,
|
||
},
|
||
'cardio': {
|
||
'num_supersets': (2, 3),
|
||
'rounds': (2, 3),
|
||
'exercises_per_superset': (3, 5),
|
||
'rep_min': 12,
|
||
'rep_max': 20,
|
||
'duration_min': 45,
|
||
'duration_max': 90,
|
||
'duration_bias': 1.0,
|
||
},
|
||
'cross training': {
|
||
'num_supersets': (3, 5),
|
||
'rounds': (3, 4),
|
||
'exercises_per_superset': (3, 5),
|
||
'rep_min': 6,
|
||
'rep_max': 15,
|
||
'duration_min': 30,
|
||
'duration_max': 60,
|
||
'duration_bias': 0.4,
|
||
},
|
||
'core training': {
|
||
'num_supersets': (3, 4),
|
||
'rounds': (3, 3),
|
||
'exercises_per_superset': (3, 4),
|
||
'rep_min': 10,
|
||
'rep_max': 20,
|
||
'duration_min': 30,
|
||
'duration_max': 45,
|
||
'duration_bias': 0.5,
|
||
},
|
||
'core': {
|
||
'num_supersets': (3, 4),
|
||
'rounds': (3, 3),
|
||
'exercises_per_superset': (3, 4),
|
||
'rep_min': 10,
|
||
'rep_max': 20,
|
||
'duration_min': 30,
|
||
'duration_max': 45,
|
||
'duration_bias': 0.5,
|
||
},
|
||
'flexibility': {
|
||
'num_supersets': (1, 2),
|
||
'rounds': (1, 1),
|
||
'exercises_per_superset': (5, 8),
|
||
'rep_min': 1,
|
||
'rep_max': 1,
|
||
'duration_min': 30,
|
||
'duration_max': 60,
|
||
'duration_bias': 1.0,
|
||
},
|
||
}
|
||
|
||
# Fallback when workout type name doesn't match any known default
|
||
GENERIC_DEFAULTS = {
|
||
'num_supersets': (3, 4),
|
||
'rounds': (3, 4),
|
||
'exercises_per_superset': (2, 4),
|
||
'rep_min': 8,
|
||
'rep_max': 12,
|
||
'duration_min': 30,
|
||
'duration_max': 45,
|
||
'duration_bias': 0.3,
|
||
}
|
||
|
||
# Aliases mapping DB underscore names to WORKOUT_TYPE_DEFAULTS keys
|
||
_WORKOUT_TYPE_ALIASES = {
|
||
'high_intensity_interval_training': 'hiit',
|
||
'traditional_strength_training': 'traditional strength',
|
||
'functional_strength_training': 'functional strength',
|
||
'cross_training': 'cross training',
|
||
'core_training': 'core training',
|
||
}
|
||
|
||
# Split type -> preferred workout types (affinity matching)
|
||
SPLIT_TYPE_WORKOUT_AFFINITY = {
|
||
'push': {'hypertrophy', 'traditional_strength_training', 'functional_strength_training'},
|
||
'pull': {'hypertrophy', 'traditional_strength_training', 'functional_strength_training'},
|
||
'upper': {'hypertrophy', 'traditional_strength_training', 'cross_training'},
|
||
'lower': {'traditional_strength_training', 'hypertrophy', 'functional_strength_training'},
|
||
'legs': {'traditional_strength_training', 'hypertrophy', 'functional_strength_training'},
|
||
'full_body': {'cross_training', 'functional_strength_training', 'high_intensity_interval_training'},
|
||
'core': {'core_training', 'functional_strength_training', 'high_intensity_interval_training'},
|
||
'cardio': {'cardio', 'high_intensity_interval_training', 'cross_training'},
|
||
}
|
||
|
||
# Fitness-level scaling applied on top of workout-type params.
|
||
# Uses multipliers (not additive) so rep floors clamp correctly.
|
||
# Keys: rep_min_mult, rep_max_mult, rounds_adj, rest_adj
|
||
FITNESS_LEVEL_SCALING = {
|
||
1: {'rep_min_mult': 1.3, 'rep_max_mult': 1.3, 'rounds_adj': -1, 'rest_adj': 15}, # Beginner
|
||
2: {'rep_min_mult': 1.0, 'rep_max_mult': 1.0, 'rounds_adj': 0, 'rest_adj': 0}, # Intermediate
|
||
3: {'rep_min_mult': 0.85, 'rep_max_mult': 1.0, 'rounds_adj': 1, 'rest_adj': -10}, # Advanced
|
||
4: {'rep_min_mult': 0.75, 'rep_max_mult': 1.0, 'rounds_adj': 1, 'rest_adj': -15}, # Elite
|
||
}
|
||
|
||
# Goal-based duration_bias overrides (applied in _build_working_supersets)
|
||
GOAL_DURATION_BIAS = {
|
||
'strength': 0.1,
|
||
'hypertrophy': 0.15,
|
||
'endurance': 0.7,
|
||
'weight_loss': 0.6,
|
||
# 'general_fitness' uses the workout_type default as-is
|
||
}
|
||
|
||
# Active time ratio by fitness level (used in _adjust_to_time_target)
|
||
FITNESS_ACTIVE_TIME_RATIO = {
|
||
1: 0.55, # Beginner - more rest needed
|
||
2: 0.65, # Intermediate - baseline
|
||
3: 0.70, # Advanced
|
||
4: 0.75, # Elite
|
||
}
|
||
|
||
|
||
class WorkoutGenerator:
|
||
"""
|
||
Main generator that orchestrates weekly plan creation.
|
||
|
||
Combines ExerciseSelector (smart exercise picking) and PlanBuilder
|
||
(ORM object creation) with scheduling / split logic from the ML
|
||
pattern tables (or sensible defaults when those are empty).
|
||
"""
|
||
|
||
def __init__(self, user_preference, duration_override=None, rest_day_indices=None,
|
||
day_workout_type_overrides=None):
|
||
self.preference = user_preference
|
||
self.exercise_selector = ExerciseSelector(user_preference)
|
||
self.plan_builder = PlanBuilder(user_preference.registered_user)
|
||
self.warnings = []
|
||
self.duration_override = duration_override # minutes, overrides preferred_workout_duration
|
||
self.rest_day_indices = rest_day_indices # list of weekday ints to force as rest
|
||
self.day_workout_type_overrides = day_workout_type_overrides or {} # {day_index: workout_type_id}
|
||
|
||
# ==================================================================
|
||
# Public API
|
||
# ==================================================================
|
||
|
||
def generate_weekly_preview(self, week_start_date):
|
||
"""
|
||
Generate a preview of a weekly plan as serializable dicts.
|
||
No DB writes occur.
|
||
|
||
Returns
|
||
-------
|
||
dict with keys: week_start_date, days (list of day dicts)
|
||
"""
|
||
split_days, rest_day_positions = self._pick_weekly_split()
|
||
workout_assignments = self._assign_workout_types(split_days)
|
||
schedule = self._build_weekly_schedule(
|
||
week_start_date, workout_assignments, rest_day_positions,
|
||
)
|
||
|
||
# Periodization: detect deload for preview
|
||
recent_plans = list(
|
||
GeneratedWeeklyPlan.objects.filter(
|
||
registered_user=self.preference.registered_user,
|
||
).order_by('-week_start_date')[:4]
|
||
)
|
||
if recent_plans and recent_plans[0].cycle_id:
|
||
cycle_id = recent_plans[0].cycle_id
|
||
consecutive_non_deload = 0
|
||
for p in recent_plans:
|
||
if p.cycle_id == cycle_id and not p.is_deload:
|
||
consecutive_non_deload += 1
|
||
else:
|
||
break
|
||
self._is_deload = consecutive_non_deload >= 3
|
||
else:
|
||
self._is_deload = False
|
||
|
||
# Apply recent exercise exclusion (same as generate_weekly_plan)
|
||
from superset.models import SupersetExercise
|
||
recent_workouts = list(GeneratedWorkout.objects.filter(
|
||
plan__registered_user=self.preference.registered_user,
|
||
is_rest_day=False,
|
||
workout__isnull=False,
|
||
).order_by('-scheduled_date')[:7])
|
||
|
||
hard_workout_ids = [gw.workout_id for gw in recent_workouts[:3] if gw.workout_id]
|
||
soft_workout_ids = [gw.workout_id for gw in recent_workouts[3:] if gw.workout_id]
|
||
|
||
hard_exclude_ids = set(
|
||
SupersetExercise.objects.filter(
|
||
superset__workout_id__in=hard_workout_ids
|
||
).values_list('exercise_id', flat=True)
|
||
) if hard_workout_ids else set()
|
||
|
||
soft_penalty_ids = set(
|
||
SupersetExercise.objects.filter(
|
||
superset__workout_id__in=soft_workout_ids
|
||
).values_list('exercise_id', flat=True)
|
||
) if soft_workout_ids else set()
|
||
|
||
self.exercise_selector.hard_exclude_ids = hard_exclude_ids
|
||
self.exercise_selector.recently_used_ids = soft_penalty_ids
|
||
|
||
days = []
|
||
for day_info in schedule:
|
||
date = day_info['date']
|
||
|
||
if day_info['is_rest_day']:
|
||
days.append({
|
||
'day_of_week': date.weekday(),
|
||
'date': date.isoformat(),
|
||
'is_rest_day': True,
|
||
'focus_area': 'Rest Day',
|
||
'target_muscles': [],
|
||
})
|
||
continue
|
||
|
||
muscle_split = day_info['muscle_split']
|
||
workout_type = day_info['workout_type']
|
||
label = day_info.get('label', muscle_split.get('label', 'Workout'))
|
||
target_muscles = muscle_split.get('muscles', [])
|
||
|
||
self.exercise_selector.reset()
|
||
|
||
workout_spec = self.generate_single_workout(
|
||
muscle_split=muscle_split,
|
||
workout_type=workout_type,
|
||
scheduled_date=date,
|
||
)
|
||
|
||
serialized = self.serialize_workout_spec(workout_spec)
|
||
|
||
days.append({
|
||
'day_of_week': date.weekday(),
|
||
'date': date.isoformat(),
|
||
'is_rest_day': False,
|
||
'focus_area': label,
|
||
'target_muscles': target_muscles,
|
||
'workout_type_id': workout_type.pk if workout_type else None,
|
||
'workout_type_name': workout_type.name if workout_type else None,
|
||
'workout_spec': serialized,
|
||
})
|
||
|
||
result = {
|
||
'week_start_date': week_start_date.isoformat(),
|
||
'is_deload': self._is_deload,
|
||
'days': days,
|
||
}
|
||
if self.warnings:
|
||
result['warnings'] = list(dict.fromkeys(self.warnings)) # deduplicate, preserve order
|
||
return result
|
||
|
||
def generate_single_day_preview(self, muscle_split, workout_type, scheduled_date):
|
||
"""
|
||
Generate a single day preview (no DB writes).
|
||
|
||
Parameters
|
||
----------
|
||
muscle_split : dict {'label': str, 'muscles': list, 'split_type': str}
|
||
workout_type : WorkoutType | None
|
||
scheduled_date : datetime.date
|
||
|
||
Returns
|
||
-------
|
||
dict (single day preview)
|
||
"""
|
||
self.exercise_selector.reset()
|
||
|
||
workout_spec = self.generate_single_workout(
|
||
muscle_split=muscle_split,
|
||
workout_type=workout_type,
|
||
scheduled_date=scheduled_date,
|
||
)
|
||
|
||
serialized = self.serialize_workout_spec(workout_spec)
|
||
label = muscle_split.get('label', 'Workout')
|
||
target_muscles = muscle_split.get('muscles', [])
|
||
|
||
result = {
|
||
'day_of_week': scheduled_date.weekday(),
|
||
'date': scheduled_date.isoformat(),
|
||
'is_rest_day': False,
|
||
'focus_area': label,
|
||
'target_muscles': target_muscles,
|
||
'workout_type_id': workout_type.pk if workout_type else None,
|
||
'workout_type_name': workout_type.name if workout_type else None,
|
||
'workout_spec': serialized,
|
||
}
|
||
if self.warnings:
|
||
result['warnings'] = list(dict.fromkeys(self.warnings))
|
||
return result
|
||
|
||
@staticmethod
|
||
def serialize_workout_spec(workout_spec):
|
||
"""
|
||
Convert a workout_spec (with Exercise ORM objects) into a
|
||
JSON-serializable dict.
|
||
"""
|
||
serialized_supersets = []
|
||
estimated_time = 0
|
||
|
||
for ss in workout_spec.get('supersets', []):
|
||
rounds = ss.get('rounds', 1)
|
||
rest_between = ss.get('rest_between_rounds', 0)
|
||
serialized_exercises = []
|
||
superset_time = 0
|
||
|
||
for ex_entry in ss.get('exercises', []):
|
||
ex = ex_entry.get('exercise')
|
||
if ex is None:
|
||
continue
|
||
|
||
entry = {
|
||
'exercise_id': ex.pk,
|
||
'exercise_name': ex.name,
|
||
'muscle_groups': ex.muscle_groups or '',
|
||
'video_url': ex.video_url(),
|
||
'reps': ex_entry.get('reps'),
|
||
'duration': ex_entry.get('duration'),
|
||
'weight': ex_entry.get('weight'),
|
||
'order': ex_entry.get('order', 1),
|
||
}
|
||
serialized_exercises.append(entry)
|
||
|
||
if ex_entry.get('reps') is not None:
|
||
rep_dur = ex.estimated_rep_duration or 3.0
|
||
superset_time += ex_entry['reps'] * rep_dur
|
||
if ex_entry.get('duration') is not None:
|
||
superset_time += ex_entry['duration']
|
||
|
||
rest_time = rest_between * max(0, rounds - 1)
|
||
estimated_time += (superset_time * rounds) + rest_time
|
||
|
||
serialized_supersets.append({
|
||
'name': ss.get('name', 'Set'),
|
||
'rounds': rounds,
|
||
'rest_between_rounds': rest_between,
|
||
'exercises': serialized_exercises,
|
||
})
|
||
|
||
return {
|
||
'name': workout_spec.get('name', 'Workout'),
|
||
'description': workout_spec.get('description', ''),
|
||
'estimated_time': int(estimated_time),
|
||
'supersets': serialized_supersets,
|
||
}
|
||
|
||
def generate_weekly_plan(self, week_start_date):
|
||
"""
|
||
Generate a complete 7-day plan.
|
||
|
||
Algorithm:
|
||
1. Pick a WeeklySplitPattern matching days_per_week
|
||
2. Assign workout types from user's preferred types
|
||
3. Assign rest days to fill 7 days
|
||
4. For each training day, generate a workout
|
||
5. Create GeneratedWeeklyPlan and GeneratedWorkout records
|
||
|
||
Parameters
|
||
----------
|
||
week_start_date : datetime.date
|
||
Monday of the target week (or any start date).
|
||
|
||
Returns
|
||
-------
|
||
GeneratedWeeklyPlan
|
||
"""
|
||
start_ts = time.monotonic()
|
||
|
||
week_end_date = week_start_date + timedelta(days=6)
|
||
days_per_week = self.preference.days_per_week
|
||
|
||
# 1. Pick split pattern
|
||
split_days, rest_day_positions = self._pick_weekly_split()
|
||
|
||
# 2. Assign workout types
|
||
workout_assignments = self._assign_workout_types(split_days)
|
||
|
||
# 3. Build the 7-day schedule (training + rest)
|
||
schedule = self._build_weekly_schedule(
|
||
week_start_date, workout_assignments, rest_day_positions,
|
||
)
|
||
|
||
# 4. Snapshot preferences for audit trail
|
||
prefs_snapshot = {
|
||
'days_per_week': days_per_week,
|
||
'fitness_level': self.preference.fitness_level,
|
||
'primary_goal': self.preference.primary_goal,
|
||
'secondary_goal': self.preference.secondary_goal,
|
||
'preferred_workout_duration': self.preference.preferred_workout_duration,
|
||
'preferred_days': self.preference.preferred_days,
|
||
'target_muscle_groups': list(
|
||
self.preference.target_muscle_groups.values_list('name', flat=True)
|
||
),
|
||
'available_equipment': list(
|
||
self.preference.available_equipment.values_list('name', flat=True)
|
||
),
|
||
'preferred_workout_types': list(
|
||
self.preference.preferred_workout_types.values_list('name', flat=True)
|
||
),
|
||
'injury_types': self.preference.injury_types or [],
|
||
'excluded_exercises': list(
|
||
self.preference.excluded_exercises.values_list('pk', flat=True)
|
||
),
|
||
}
|
||
|
||
# 5. Periodization: determine week_number, is_deload, cycle_id
|
||
recent_plans = list(
|
||
GeneratedWeeklyPlan.objects.filter(
|
||
registered_user=self.preference.registered_user,
|
||
).order_by('-week_start_date')[:4]
|
||
)
|
||
|
||
# Determine cycle_id and week_number from the most recent plan
|
||
if recent_plans and recent_plans[0].cycle_id:
|
||
cycle_id = recent_plans[0].cycle_id
|
||
week_number = recent_plans[0].week_number + 1
|
||
else:
|
||
cycle_id = uuid.uuid4().hex[:16]
|
||
week_number = 1
|
||
|
||
# Deload: if 3 consecutive non-deload weeks, this week should be a deload
|
||
is_deload = False
|
||
consecutive_non_deload = 0
|
||
for p in recent_plans:
|
||
if p.cycle_id == cycle_id and not p.is_deload:
|
||
consecutive_non_deload += 1
|
||
else:
|
||
break
|
||
if consecutive_non_deload >= 3:
|
||
is_deload = True
|
||
self.warnings.append(
|
||
'This is a deload week — volume and intensity are reduced for recovery.'
|
||
)
|
||
|
||
# After a deload, start a new cycle
|
||
if recent_plans and recent_plans[0].is_deload:
|
||
cycle_id = uuid.uuid4().hex[:16]
|
||
week_number = 1
|
||
|
||
# Store deload flag so generate_single_workout can apply adjustments
|
||
self._is_deload = is_deload
|
||
|
||
# Create the plan record
|
||
plan = GeneratedWeeklyPlan.objects.create(
|
||
registered_user=self.preference.registered_user,
|
||
week_start_date=week_start_date,
|
||
week_end_date=week_end_date,
|
||
status='pending',
|
||
preferences_snapshot=prefs_snapshot,
|
||
week_number=week_number,
|
||
is_deload=is_deload,
|
||
cycle_id=cycle_id,
|
||
)
|
||
|
||
# Query recently used exercise IDs for cross-workout variety
|
||
# Tier 1 (last 3 workouts): hard exclude — prevents direct repetition
|
||
# Tier 2 (workouts 4-7): soft penalty — reduces selection likelihood
|
||
from superset.models import SupersetExercise
|
||
recent_workouts = list(GeneratedWorkout.objects.filter(
|
||
plan__registered_user=self.preference.registered_user,
|
||
is_rest_day=False,
|
||
workout__isnull=False,
|
||
).order_by('-scheduled_date')[:7])
|
||
|
||
hard_workout_ids = [gw.workout_id for gw in recent_workouts[:3] if gw.workout_id]
|
||
soft_workout_ids = [gw.workout_id for gw in recent_workouts[3:] if gw.workout_id]
|
||
|
||
hard_exclude_ids = set(
|
||
SupersetExercise.objects.filter(
|
||
superset__workout_id__in=hard_workout_ids
|
||
).values_list('exercise_id', flat=True)
|
||
) if hard_workout_ids else set()
|
||
|
||
soft_penalty_ids = set(
|
||
SupersetExercise.objects.filter(
|
||
superset__workout_id__in=soft_workout_ids
|
||
).values_list('exercise_id', flat=True)
|
||
) if soft_workout_ids else set()
|
||
|
||
self.exercise_selector.hard_exclude_ids = hard_exclude_ids
|
||
self.exercise_selector.recently_used_ids = soft_penalty_ids
|
||
|
||
# Build progression boost: find exercises that are progressions of recently done ones
|
||
fitness_level = getattr(self.preference, 'fitness_level', 2) or 2
|
||
if fitness_level >= 2:
|
||
all_recent_exercise_ids = set()
|
||
for gw in recent_workouts:
|
||
if gw.workout_id:
|
||
all_recent_exercise_ids.update(
|
||
SupersetExercise.objects.filter(
|
||
superset__workout_id=gw.workout_id
|
||
).values_list('exercise_id', flat=True)
|
||
)
|
||
if all_recent_exercise_ids:
|
||
from exercise.models import Exercise as ExModel
|
||
progression_ids = set(
|
||
ExModel.objects.filter(
|
||
progression_of_id__in=all_recent_exercise_ids
|
||
).values_list('pk', flat=True)
|
||
)
|
||
self.exercise_selector.progression_boost_ids = progression_ids
|
||
|
||
# 5b. Check recent CompletedWorkout difficulty for volume adjustment
|
||
self._volume_adjustment = 0.0 # -0.1 to +0.1
|
||
recent_completed = CompletedWorkout.objects.filter(
|
||
registered_user=self.preference.registered_user,
|
||
).order_by('-created_at')[:4]
|
||
if recent_completed:
|
||
avg_difficulty = sum(c.difficulty for c in recent_completed) / len(recent_completed)
|
||
if avg_difficulty >= 4:
|
||
self._volume_adjustment = -0.10 # reduce 10%
|
||
self.warnings.append(
|
||
'Recent workouts rated as hard — reducing volume by 10%.'
|
||
)
|
||
elif avg_difficulty <= 1:
|
||
self._volume_adjustment = 0.10 # increase 10%
|
||
self.warnings.append(
|
||
'Recent workouts rated as easy — increasing volume by 10%.'
|
||
)
|
||
|
||
# 6. Generate workouts for each day
|
||
for day_info in schedule:
|
||
date = day_info['date']
|
||
day_of_week = date.weekday()
|
||
|
||
if day_info['is_rest_day']:
|
||
GeneratedWorkout.objects.create(
|
||
plan=plan,
|
||
workout=None,
|
||
workout_type=None,
|
||
scheduled_date=date,
|
||
day_of_week=day_of_week,
|
||
is_rest_day=True,
|
||
status='pending',
|
||
focus_area='Rest Day',
|
||
target_muscles=[],
|
||
)
|
||
continue
|
||
|
||
muscle_split = day_info['muscle_split']
|
||
workout_type = day_info['workout_type']
|
||
label = day_info.get('label', muscle_split.get('label', 'Workout'))
|
||
target_muscles = muscle_split.get('muscles', [])
|
||
|
||
# Reset the selector for each new workout
|
||
self.exercise_selector.reset()
|
||
|
||
# Generate the workout spec
|
||
workout_spec = self.generate_single_workout(
|
||
muscle_split=muscle_split,
|
||
workout_type=workout_type,
|
||
scheduled_date=date,
|
||
)
|
||
|
||
# Persist via PlanBuilder
|
||
workout_obj = self.plan_builder.create_workout_from_spec(workout_spec)
|
||
|
||
GeneratedWorkout.objects.create(
|
||
plan=plan,
|
||
workout=workout_obj,
|
||
workout_type=workout_type,
|
||
scheduled_date=date,
|
||
day_of_week=day_of_week,
|
||
is_rest_day=False,
|
||
status='pending',
|
||
focus_area=label,
|
||
target_muscles=target_muscles,
|
||
)
|
||
|
||
elapsed_ms = int((time.monotonic() - start_ts) * 1000)
|
||
plan.generation_time_ms = elapsed_ms
|
||
plan.status = 'completed'
|
||
plan.save()
|
||
|
||
logger.info(
|
||
"Generated weekly plan %s for user %s in %dms",
|
||
plan.pk, self.preference.registered_user.pk, elapsed_ms,
|
||
)
|
||
|
||
return plan
|
||
|
||
def generate_single_workout(self, muscle_split, workout_type, scheduled_date):
|
||
"""
|
||
Generate one workout specification dict.
|
||
|
||
Steps:
|
||
1. Build warm-up superset (duration-based, 1 round, 4-6 exercises)
|
||
2. Build working supersets based on workout_type parameters
|
||
3. Build cool-down superset (duration-based, 1 round, 3-4 exercises)
|
||
4. Calculate estimated time, trim if over user preference
|
||
5. Return workout_spec dict ready for PlanBuilder
|
||
|
||
Parameters
|
||
----------
|
||
muscle_split : dict
|
||
``{'label': str, 'muscles': list[str], 'split_type': str}``
|
||
workout_type : WorkoutType | None
|
||
scheduled_date : datetime.date
|
||
|
||
Returns
|
||
-------
|
||
dict (workout_spec)
|
||
"""
|
||
target_muscles = list(muscle_split.get('muscles', []))
|
||
label = muscle_split.get('label', 'Workout')
|
||
duration_minutes = self.duration_override or self.preference.preferred_workout_duration
|
||
max_duration_sec = duration_minutes * 60
|
||
# Clamp duration to valid range (15-120 minutes)
|
||
max_duration_sec = max(15 * 60, min(120 * 60, max_duration_sec))
|
||
|
||
# Apply volume adjustment from CompletedWorkout feedback loop
|
||
volume_adj = getattr(self, '_volume_adjustment', 0.0)
|
||
if volume_adj:
|
||
max_duration_sec = int(max_duration_sec * (1.0 + volume_adj))
|
||
|
||
# Inject user's target_muscle_groups when relevant to the split type
|
||
split_type = muscle_split.get('split_type', 'full_body')
|
||
user_target_muscles = list(
|
||
self.preference.target_muscle_groups.values_list('name', flat=True)
|
||
)
|
||
if user_target_muscles:
|
||
if split_type == 'full_body':
|
||
relevant_muscles = user_target_muscles
|
||
else:
|
||
# Only inject muscles relevant to this split's muscle groups
|
||
split_muscle_pool = {
|
||
normalize_muscle_name(m)
|
||
for m in self._get_broader_muscles(split_type)
|
||
}
|
||
relevant_muscles = [
|
||
m for m in user_target_muscles
|
||
if normalize_muscle_name(m) in split_muscle_pool
|
||
]
|
||
|
||
for m in relevant_muscles:
|
||
normalized = normalize_muscle_name(m)
|
||
if normalized not in target_muscles:
|
||
target_muscles.append(normalized)
|
||
|
||
if relevant_muscles:
|
||
muscle_split = dict(muscle_split)
|
||
muscle_split['muscles'] = target_muscles
|
||
|
||
# Get workout-type parameters
|
||
wt_params = self._get_workout_type_params(workout_type)
|
||
|
||
# Fix #12: Scale warmup/cooldown by duration
|
||
duration_adj = 0
|
||
if max_duration_sec <= 1800: # <= 30 min
|
||
duration_adj = -1
|
||
elif max_duration_sec >= 4500: # >= 75 min
|
||
duration_adj = 1
|
||
|
||
# 1. Warm-up
|
||
warmup = self._build_warmup(target_muscles, workout_type)
|
||
|
||
if warmup and duration_adj != 0:
|
||
exercises = warmup.get('exercises', [])
|
||
if duration_adj < 0 and len(exercises) > 2:
|
||
exercises.pop() # Remove last warmup exercise
|
||
elif duration_adj > 0:
|
||
# Add one more warmup exercise
|
||
extra = self.exercise_selector.select_warmup_exercises(target_muscles, count=1)
|
||
if extra:
|
||
exercises.append({
|
||
'exercise': extra[0],
|
||
'duration': exercises[-1].get('duration', 30) if exercises else 30,
|
||
'order': len(exercises) + 1,
|
||
})
|
||
|
||
# 2. Working supersets
|
||
working_supersets = self._build_working_supersets(
|
||
muscle_split, workout_type, wt_params,
|
||
)
|
||
|
||
# Quality gate: validate working supersets against rules engine
|
||
MAX_RETRIES = 2
|
||
for attempt in range(MAX_RETRIES + 1):
|
||
violations = self._check_quality_gates(working_supersets, workout_type, wt_params)
|
||
blocking = [v for v in violations if v.severity == 'error']
|
||
if not blocking or attempt == MAX_RETRIES:
|
||
self.warnings.extend([v.message for v in violations])
|
||
break
|
||
logger.info(
|
||
"Quality gate: %d blocking violation(s) on attempt %d, retrying",
|
||
len(blocking), attempt + 1,
|
||
)
|
||
self.exercise_selector.reset()
|
||
working_supersets = self._build_working_supersets(
|
||
muscle_split, workout_type, wt_params,
|
||
)
|
||
|
||
# 3. Cool-down
|
||
cooldown = self._build_cooldown(target_muscles, workout_type)
|
||
|
||
if cooldown and duration_adj != 0:
|
||
exercises = cooldown.get('exercises', [])
|
||
if duration_adj < 0 and len(exercises) > 2:
|
||
exercises.pop()
|
||
elif duration_adj > 0:
|
||
extra = self.exercise_selector.select_cooldown_exercises(target_muscles, count=1)
|
||
if extra:
|
||
exercises.append({
|
||
'exercise': extra[0],
|
||
'duration': exercises[-1].get('duration', 30) if exercises else 30,
|
||
'order': len(exercises) + 1,
|
||
})
|
||
|
||
# Assemble the spec
|
||
all_supersets = []
|
||
if warmup:
|
||
all_supersets.append(warmup)
|
||
all_supersets.extend(working_supersets)
|
||
if cooldown:
|
||
all_supersets.append(cooldown)
|
||
|
||
workout_spec = {
|
||
'name': f"{label} - {scheduled_date.strftime('%b %d')}",
|
||
'description': f"Generated {label.lower()} workout targeting {', '.join(target_muscles[:4])}",
|
||
'supersets': all_supersets,
|
||
}
|
||
|
||
# 4. Estimate total time and trim / pad if needed
|
||
workout_spec = self._adjust_to_time_target(
|
||
workout_spec, max_duration_sec, muscle_split, wt_params,
|
||
workout_type=workout_type,
|
||
)
|
||
|
||
# Hard cap total working exercises to prevent bloated workouts
|
||
MAX_WORKING_EXERCISES = 30
|
||
working_supersets = [
|
||
ss for ss in workout_spec.get('supersets', [])
|
||
if ss.get('name', '').startswith('Working')
|
||
]
|
||
total_working_ex = sum(len(ss['exercises']) for ss in working_supersets)
|
||
if total_working_ex > MAX_WORKING_EXERCISES:
|
||
# Trim from back working supersets
|
||
excess = total_working_ex - MAX_WORKING_EXERCISES
|
||
for ss in reversed(working_supersets):
|
||
while excess > 0 and len(ss['exercises']) > 2:
|
||
ss['exercises'].pop()
|
||
excess -= 1
|
||
if excess <= 0:
|
||
break
|
||
# Remove empty working supersets
|
||
workout_spec['supersets'] = [
|
||
ss for ss in workout_spec['supersets']
|
||
if not ss.get('name', '').startswith('Working') or len(ss['exercises']) >= 2
|
||
]
|
||
|
||
# Enforce min 2 exercises per working superset; merge undersized ones
|
||
all_supersets = workout_spec.get('supersets', [])
|
||
working_indices = [
|
||
i for i, ss in enumerate(all_supersets)
|
||
if ss.get('name', '').startswith('Working')
|
||
]
|
||
undersized = [i for i in working_indices if len(all_supersets[i]['exercises']) < 2]
|
||
if undersized:
|
||
# Try to redistribute exercises from undersized into adjacent supersets
|
||
for idx in reversed(undersized):
|
||
ss = all_supersets[idx]
|
||
orphan_exercises = ss['exercises']
|
||
# Find next working superset to absorb orphans
|
||
absorbed = False
|
||
for other_idx in working_indices:
|
||
if other_idx == idx:
|
||
continue
|
||
other_ss = all_supersets[other_idx]
|
||
if len(other_ss['exercises']) < 6:
|
||
for ex_entry in orphan_exercises:
|
||
if len(other_ss['exercises']) < 6:
|
||
ex_entry['order'] = len(other_ss['exercises']) + 1
|
||
other_ss['exercises'].append(ex_entry)
|
||
absorbed = True
|
||
break
|
||
# Remove the undersized superset
|
||
all_supersets.pop(idx)
|
||
# Refresh working_indices after removal
|
||
working_indices = [
|
||
i for i, ss in enumerate(all_supersets)
|
||
if ss.get('name', '').startswith('Working')
|
||
]
|
||
|
||
# Post-build modality validation: ensure each working superset
|
||
# has consistent modality (all reps or all duration)
|
||
for ss in workout_spec.get('supersets', []):
|
||
if not ss.get('name', '').startswith('Working'):
|
||
continue
|
||
intended = ss.get('modality', 'reps')
|
||
for entry in ss.get('exercises', []):
|
||
if intended == 'duration':
|
||
if entry.get('reps') and not entry.get('duration'):
|
||
ex = entry.get('exercise')
|
||
rep_dur = (ex.estimated_rep_duration or 3.0) if ex else 3.0
|
||
entry['duration'] = max(20, int(entry['reps'] * rep_dur))
|
||
entry.pop('reps', None)
|
||
entry.pop('weight', None)
|
||
logger.debug("Corrected reps->duration for modality consistency in %s", ss.get('name'))
|
||
else:
|
||
if entry.get('duration') and not entry.get('reps'):
|
||
entry['reps'] = random.randint(wt_params['rep_min'], wt_params['rep_max'])
|
||
entry.pop('duration', None)
|
||
logger.debug("Corrected duration->reps for modality consistency in %s", ss.get('name'))
|
||
|
||
# Collect warnings from exercise selector
|
||
if self.exercise_selector.warnings:
|
||
self.warnings.extend(self.exercise_selector.warnings)
|
||
|
||
return workout_spec
|
||
|
||
# ==================================================================
|
||
# Split / scheduling helpers
|
||
# ==================================================================
|
||
|
||
def _pick_weekly_split(self):
|
||
"""
|
||
Select muscle-split dicts for training days, preferring ML-learned
|
||
WeeklySplitPattern records over hardcoded defaults.
|
||
|
||
Returns
|
||
-------
|
||
tuple[list[dict], list[int]]
|
||
(splits, rest_day_positions) — each split dict has keys:
|
||
label, muscles, split_type
|
||
"""
|
||
days = self.preference.days_per_week
|
||
clamped_days = max(1, min(days, 7))
|
||
|
||
# Try DB-learned patterns first
|
||
db_patterns = list(
|
||
WeeklySplitPattern.objects.filter(
|
||
days_per_week=clamped_days
|
||
).order_by('-frequency')
|
||
)
|
||
|
||
if db_patterns:
|
||
# Frequency-weighted random selection
|
||
total_freq = sum(p.frequency for p in db_patterns)
|
||
if total_freq > 0:
|
||
r = random.random() * total_freq
|
||
cumulative = 0
|
||
chosen = db_patterns[0]
|
||
for p in db_patterns:
|
||
cumulative += p.frequency
|
||
if cumulative >= r:
|
||
chosen = p
|
||
break
|
||
else:
|
||
chosen = random.choice(db_patterns)
|
||
|
||
# Resolve MuscleGroupSplit IDs to split dicts
|
||
splits = []
|
||
split_ids = chosen.pattern or []
|
||
labels = chosen.pattern_labels or []
|
||
|
||
for i, split_id in enumerate(split_ids):
|
||
mgs = MuscleGroupSplit.objects.filter(pk=split_id).first()
|
||
if mgs:
|
||
splits.append({
|
||
'label': mgs.label or (labels[i] if i < len(labels) else f'Day {i+1}'),
|
||
'muscles': mgs.muscle_names or [],
|
||
'split_type': mgs.split_type or 'full_body',
|
||
})
|
||
|
||
if splits:
|
||
# Apply user target muscle reordering
|
||
user_target_muscles = list(
|
||
self.preference.target_muscle_groups.values_list('name', flat=True)
|
||
)
|
||
if user_target_muscles and len(splits) > 1:
|
||
target_set = {normalize_muscle_name(m) for m in user_target_muscles}
|
||
|
||
def _target_overlap(split_day):
|
||
day_muscles = {normalize_muscle_name(m) for m in split_day.get('muscles', [])}
|
||
return len(day_muscles & target_set)
|
||
|
||
splits.sort(key=_target_overlap, reverse=True)
|
||
|
||
rest_days = chosen.rest_day_positions or []
|
||
return splits, rest_days
|
||
|
||
# Fallback to DEFAULT_SPLITS
|
||
splits = list(DEFAULT_SPLITS.get(clamped_days, DEFAULT_SPLITS[4]))
|
||
|
||
user_target_muscles = list(
|
||
self.preference.target_muscle_groups.values_list('name', flat=True)
|
||
)
|
||
if user_target_muscles and len(splits) > 1:
|
||
target_set = {normalize_muscle_name(m) for m in user_target_muscles}
|
||
|
||
def _target_overlap(split_day):
|
||
day_muscles = {normalize_muscle_name(m) for m in split_day.get('muscles', [])}
|
||
return len(day_muscles & target_set)
|
||
|
||
splits.sort(key=_target_overlap, reverse=True)
|
||
|
||
return splits, []
|
||
|
||
def _assign_workout_types(self, split_days):
|
||
"""
|
||
Distribute the user's preferred WorkoutTypes across the training
|
||
days represented by *split_days*, using split-type affinity matching.
|
||
|
||
Returns
|
||
-------
|
||
list[dict]
|
||
Each dict has: label, muscles, split_type, workout_type (WorkoutType | None)
|
||
"""
|
||
preferred_types = list(
|
||
self.preference.preferred_workout_types.all()
|
||
)
|
||
if not preferred_types:
|
||
preferred_types = list(WorkoutType.objects.all()[:3])
|
||
|
||
assignments = []
|
||
used_type_indices = set()
|
||
|
||
for i, split_day in enumerate(split_days):
|
||
# Check for API-level workout type override for this day index
|
||
override_wt_id = self.day_workout_type_overrides.get(str(i)) or self.day_workout_type_overrides.get(i)
|
||
if override_wt_id:
|
||
override_wt = WorkoutType.objects.filter(pk=override_wt_id).first()
|
||
if override_wt:
|
||
entry = dict(split_day)
|
||
entry['workout_type'] = override_wt
|
||
assignments.append(entry)
|
||
continue
|
||
|
||
split_type = split_day.get('split_type', 'full_body')
|
||
affinity_names = SPLIT_TYPE_WORKOUT_AFFINITY.get(split_type, set())
|
||
|
||
# Try to find a preferred type that matches the split's affinity
|
||
matched_wt = None
|
||
for j, wt in enumerate(preferred_types):
|
||
if j in used_type_indices:
|
||
continue
|
||
wt_name_lower = wt.name.strip().lower()
|
||
if wt_name_lower in affinity_names:
|
||
matched_wt = wt
|
||
used_type_indices.add(j)
|
||
break
|
||
|
||
# Fall back to round-robin if no affinity match
|
||
if matched_wt is None:
|
||
if preferred_types:
|
||
matched_wt = preferred_types[i % len(preferred_types)]
|
||
else:
|
||
matched_wt = None
|
||
|
||
entry = dict(split_day)
|
||
entry['workout_type'] = matched_wt
|
||
assignments.append(entry)
|
||
|
||
return assignments
|
||
|
||
def _build_weekly_schedule(self, week_start_date, workout_assignments, rest_day_positions=None):
|
||
"""
|
||
Build a 7-day schedule mixing training days and rest days.
|
||
|
||
Uses user's ``preferred_days`` if set; otherwise spaces rest days
|
||
evenly.
|
||
|
||
Parameters
|
||
----------
|
||
week_start_date : datetime.date
|
||
workout_assignments : list[dict]
|
||
Training day assignments (label, muscles, split_type, workout_type).
|
||
|
||
Returns
|
||
-------
|
||
list[dict]
|
||
7 entries, each with keys: date, is_rest_day, and optionally
|
||
muscle_split, workout_type, label.
|
||
"""
|
||
num_training = len(workout_assignments)
|
||
num_rest = 7 - num_training
|
||
preferred_days = self.preference.preferred_days or []
|
||
|
||
# Apply API-level rest day overrides if provided
|
||
if self.rest_day_indices:
|
||
rest_set = set(self.rest_day_indices)
|
||
training_day_indices = sorted([d for d in range(7) if d not in rest_set])[:num_training]
|
||
elif preferred_days and len(preferred_days) >= num_training:
|
||
training_day_indices = sorted(preferred_days[:num_training])
|
||
elif rest_day_positions:
|
||
# Use rest day positions from the split pattern
|
||
rest_set = set(rest_day_positions)
|
||
training_day_indices = [d for d in range(7) if d not in rest_set][:num_training]
|
||
else:
|
||
# Space training days evenly across the week
|
||
training_day_indices = self._evenly_space_days(num_training)
|
||
|
||
# Build the 7-day list
|
||
schedule = []
|
||
training_idx = 0
|
||
for day_offset in range(7):
|
||
date = week_start_date + timedelta(days=day_offset)
|
||
weekday = date.weekday()
|
||
|
||
if weekday in training_day_indices and training_idx < num_training:
|
||
assignment = workout_assignments[training_idx]
|
||
schedule.append({
|
||
'date': date,
|
||
'is_rest_day': False,
|
||
'muscle_split': {
|
||
'label': assignment.get('label', 'Workout'),
|
||
'muscles': assignment.get('muscles', []),
|
||
'split_type': assignment.get('split_type', 'full_body'),
|
||
},
|
||
'workout_type': assignment.get('workout_type'),
|
||
'label': assignment.get('label', 'Workout'),
|
||
})
|
||
training_idx += 1
|
||
else:
|
||
schedule.append({
|
||
'date': date,
|
||
'is_rest_day': True,
|
||
})
|
||
|
||
# If some training days weren't placed (preferred_days mismatch),
|
||
# fill them into remaining rest slots
|
||
while training_idx < num_training:
|
||
for i, entry in enumerate(schedule):
|
||
if entry['is_rest_day'] and training_idx < num_training:
|
||
assignment = workout_assignments[training_idx]
|
||
schedule[i] = {
|
||
'date': entry['date'],
|
||
'is_rest_day': False,
|
||
'muscle_split': {
|
||
'label': assignment.get('label', 'Workout'),
|
||
'muscles': assignment.get('muscles', []),
|
||
'split_type': assignment.get('split_type', 'full_body'),
|
||
},
|
||
'workout_type': assignment.get('workout_type'),
|
||
'label': assignment.get('label', 'Workout'),
|
||
}
|
||
training_idx += 1
|
||
# Safety: break to avoid infinite loop if 7 days filled
|
||
if training_idx < num_training:
|
||
break
|
||
|
||
return schedule
|
||
|
||
@staticmethod
|
||
def _evenly_space_days(num_training):
|
||
"""
|
||
Return a list of weekday indices (0-6) evenly spaced for
|
||
*num_training* days.
|
||
"""
|
||
if num_training <= 0:
|
||
return []
|
||
if num_training >= 7:
|
||
return list(range(7))
|
||
|
||
spacing = 7 / num_training
|
||
return [int(round(i * spacing)) % 7 for i in range(num_training)]
|
||
|
||
# ==================================================================
|
||
# Workout construction helpers
|
||
# ==================================================================
|
||
|
||
def _get_workout_type_params(self, workout_type):
|
||
"""
|
||
Get parameters for building supersets, either from
|
||
WorkoutStructureRule (DB) or from WORKOUT_TYPE_DEFAULTS.
|
||
|
||
Filters WorkoutStructureRule by the user's primary_goal when
|
||
possible, and applies fitness-level scaling to reps, rounds,
|
||
and rest times.
|
||
|
||
Returns
|
||
-------
|
||
dict with keys matching WORKOUT_TYPE_DEFAULTS value shape
|
||
"""
|
||
# Try DB structure rules first
|
||
if workout_type:
|
||
# Prefer rules matching the user's primary goal, then secondary, then unfiltered
|
||
goal = getattr(self.preference, 'primary_goal', '') or ''
|
||
secondary_goal = getattr(self.preference, 'secondary_goal', '') or ''
|
||
rules = list(
|
||
WorkoutStructureRule.objects.filter(
|
||
workout_type=workout_type,
|
||
section_type='working',
|
||
goal_type=goal,
|
||
)
|
||
) if goal else []
|
||
|
||
# Fall back to secondary goal
|
||
if not rules and secondary_goal:
|
||
rules = list(
|
||
WorkoutStructureRule.objects.filter(
|
||
workout_type=workout_type,
|
||
section_type='working',
|
||
goal_type=secondary_goal,
|
||
)
|
||
)
|
||
|
||
# Fall back to unfiltered query
|
||
if not rules:
|
||
rules = list(
|
||
WorkoutStructureRule.objects.filter(
|
||
workout_type=workout_type,
|
||
section_type='working',
|
||
)
|
||
)
|
||
|
||
if rules:
|
||
# Use the first matching rule (could later combine multiple)
|
||
rule = rules[0]
|
||
# Source num_supersets from WORKOUT_TYPE_DEFAULTS (not exercises_per_superset)
|
||
name_lower = workout_type.name.strip().lower()
|
||
resolved_name = _WORKOUT_TYPE_ALIASES.get(name_lower, name_lower)
|
||
type_defaults = WORKOUT_TYPE_DEFAULTS.get(resolved_name, GENERIC_DEFAULTS)
|
||
params = {
|
||
'num_supersets': type_defaults['num_supersets'],
|
||
'rounds': (rule.typical_rounds, rule.typical_rounds),
|
||
'exercises_per_superset': (
|
||
max(2, rule.typical_exercises_per_superset - 1),
|
||
min(6, rule.typical_exercises_per_superset + 1),
|
||
),
|
||
'rep_min': rule.typical_rep_range_min,
|
||
'rep_max': rule.typical_rep_range_max,
|
||
'duration_min': rule.typical_duration_range_min,
|
||
'duration_max': rule.typical_duration_range_max,
|
||
'duration_bias': workout_type.duration_bias,
|
||
'movement_patterns': rule.movement_patterns or [],
|
||
'rest_between_rounds': workout_type.typical_rest_between_sets,
|
||
}
|
||
return self._apply_fitness_scaling(params, is_strength=resolved_name in STRENGTH_WORKOUT_TYPES)
|
||
|
||
# DB rule not found, but we have the WorkoutType model fields
|
||
name_lower = workout_type.name.strip().lower()
|
||
resolved = _WORKOUT_TYPE_ALIASES.get(name_lower, name_lower)
|
||
defaults = WORKOUT_TYPE_DEFAULTS.get(resolved, GENERIC_DEFAULTS)
|
||
params = {
|
||
'num_supersets': defaults['num_supersets'],
|
||
'rounds': (workout_type.round_range_min, workout_type.round_range_max),
|
||
'exercises_per_superset': (
|
||
workout_type.superset_size_min,
|
||
workout_type.superset_size_max,
|
||
),
|
||
'rep_min': workout_type.rep_range_min,
|
||
'rep_max': workout_type.rep_range_max,
|
||
'duration_min': defaults['duration_min'],
|
||
'duration_max': defaults['duration_max'],
|
||
'duration_bias': workout_type.duration_bias,
|
||
'movement_patterns': defaults.get('movement_patterns', []),
|
||
'rest_between_rounds': workout_type.typical_rest_between_sets,
|
||
}
|
||
return self._apply_fitness_scaling(params, is_strength=resolved in STRENGTH_WORKOUT_TYPES)
|
||
|
||
defaults = dict(GENERIC_DEFAULTS)
|
||
defaults['rest_between_rounds'] = 45
|
||
return self._apply_fitness_scaling(defaults)
|
||
|
||
def _apply_fitness_scaling(self, params, is_strength=False):
|
||
"""
|
||
Adjust workout params based on the user's fitness_level.
|
||
|
||
Applies percentage-based scaling first, then clamps to the global
|
||
minimum of GENERATION_RULES['min_reps'] (R1 + R2).
|
||
For strength workouts, uses min_reps_strength (allows heavy singles).
|
||
|
||
Beginners get higher reps, fewer rounds, and more rest.
|
||
Advanced/Elite get lower rep minimums, more rounds, and less rest.
|
||
"""
|
||
level = getattr(self.preference, 'fitness_level', 2) or 2
|
||
scaling = FITNESS_LEVEL_SCALING.get(level, FITNESS_LEVEL_SCALING[2])
|
||
if is_strength:
|
||
min_reps = GENERATION_RULES['min_reps_strength']['value']
|
||
else:
|
||
min_reps = GENERATION_RULES['min_reps']['value']
|
||
|
||
# 1. Apply percentage scaling, 2. Clamp to global minimum
|
||
params['rep_min'] = max(min_reps, int(params['rep_min'] * scaling['rep_min_mult']))
|
||
params['rep_max'] = max(params['rep_min'], int(params['rep_max'] * scaling['rep_max_mult']))
|
||
|
||
# Scale rounds (tuple of min, max)
|
||
r_min, r_max = params['rounds']
|
||
r_min = max(1, r_min + scaling['rounds_adj'])
|
||
r_max = max(r_min, r_max + scaling['rounds_adj'])
|
||
params['rounds'] = (r_min, r_max)
|
||
|
||
# Scale rest between rounds
|
||
rest = params.get('rest_between_rounds', 45)
|
||
params['rest_between_rounds'] = max(15, rest + scaling['rest_adj'])
|
||
|
||
# Fix #14: Beginners should not do triples — clamp rep_min to 5 for strength
|
||
if level <= 1 and is_strength:
|
||
params['rep_min'] = max(5, params['rep_min'])
|
||
params['rep_max'] = max(params['rep_min'], params['rep_max'])
|
||
|
||
return params
|
||
|
||
def _build_warmup(self, target_muscles, workout_type=None):
|
||
"""
|
||
Build a warm-up superset spec: duration-based, 1 round.
|
||
|
||
Exercise count scaled by fitness level:
|
||
- Beginner: 5-7 (more preparation needed)
|
||
- Intermediate: 4-6 (default)
|
||
- Advanced/Elite: 3-5 (less warm-up needed)
|
||
|
||
Returns
|
||
-------
|
||
dict | None
|
||
Superset spec dict, or None if no exercises available.
|
||
"""
|
||
fitness_level = getattr(self.preference, 'fitness_level', 2) or 2
|
||
if fitness_level <= 1:
|
||
count = random.randint(5, 7)
|
||
elif fitness_level >= 3:
|
||
count = random.randint(3, 5)
|
||
else:
|
||
count = random.randint(4, 6)
|
||
|
||
# Try to get duration range from DB structure rules
|
||
warmup_dur_min = 20
|
||
warmup_dur_max = 40
|
||
if workout_type:
|
||
warmup_rules = list(
|
||
WorkoutStructureRule.objects.filter(
|
||
workout_type=workout_type,
|
||
section_type='warm_up',
|
||
).order_by('-id')[:1]
|
||
)
|
||
if warmup_rules:
|
||
warmup_dur_min = warmup_rules[0].typical_duration_range_min
|
||
warmup_dur_max = warmup_rules[0].typical_duration_range_max
|
||
|
||
exercises = self.exercise_selector.select_warmup_exercises(
|
||
target_muscles, count=count,
|
||
)
|
||
|
||
if not exercises:
|
||
return None
|
||
|
||
min_duration = GENERATION_RULES['min_duration']['value']
|
||
duration_mult = GENERATION_RULES['duration_multiple']['value']
|
||
|
||
exercise_entries = []
|
||
for i, ex in enumerate(exercises, start=1):
|
||
duration = random.randint(warmup_dur_min, warmup_dur_max)
|
||
# R4: Round to multiple of 5, clamp to min 20
|
||
duration = max(min_duration, round(duration / duration_mult) * duration_mult)
|
||
exercise_entries.append({
|
||
'exercise': ex,
|
||
'duration': duration,
|
||
'order': i,
|
||
})
|
||
|
||
return {
|
||
'name': 'Warm Up',
|
||
'rounds': 1,
|
||
'rest_between_rounds': 0,
|
||
'exercises': exercise_entries,
|
||
}
|
||
|
||
def _build_working_supersets(self, muscle_split, workout_type, wt_params):
|
||
"""
|
||
Build the main working superset specs based on workout-type
|
||
parameters.
|
||
|
||
Uses MovementPatternOrder to put compound movements early and
|
||
isolation movements late.
|
||
|
||
Returns
|
||
-------
|
||
list[dict]
|
||
"""
|
||
target_muscles = muscle_split.get('muscles', [])
|
||
num_supersets = random.randint(*wt_params['num_supersets'])
|
||
duration_bias = wt_params.get('duration_bias', 0.3)
|
||
|
||
# Blend goal-based duration bias with workout type's native bias.
|
||
# The workout type's character stays dominant (70%) so a strength
|
||
# workout remains mostly rep-based even for endurance/weight-loss goals.
|
||
goal = getattr(self.preference, 'primary_goal', '') or ''
|
||
secondary_goal = getattr(self.preference, 'secondary_goal', '') or ''
|
||
if goal in GOAL_DURATION_BIAS:
|
||
goal_bias = GOAL_DURATION_BIAS[goal]
|
||
if secondary_goal and secondary_goal in GOAL_DURATION_BIAS:
|
||
secondary_bias = GOAL_DURATION_BIAS[secondary_goal]
|
||
goal_bias = (goal_bias * 0.7) + (secondary_bias * 0.3)
|
||
duration_bias = (duration_bias * 0.7) + (goal_bias * 0.3)
|
||
|
||
# Apply secondary goal influence on rep ranges (30% weight)
|
||
if secondary_goal:
|
||
rep_adjustments = {
|
||
'strength': (-2, -1), # Lower reps
|
||
'hypertrophy': (0, 2), # Wider range
|
||
'endurance': (2, 4), # Higher reps
|
||
'weight_loss': (1, 2), # Slightly higher
|
||
}
|
||
adj = rep_adjustments.get(secondary_goal)
|
||
if adj:
|
||
wt_params = dict(wt_params) # Don't mutate the original
|
||
wt_params['rep_min'] = max(GENERATION_RULES['min_reps']['value'],
|
||
wt_params['rep_min'] + round(adj[0] * 0.3))
|
||
wt_params['rep_max'] = max(wt_params['rep_min'],
|
||
wt_params['rep_max'] + round(adj[1] * 0.3))
|
||
|
||
# Scale exercise counts by fitness level
|
||
fitness_level = getattr(self.preference, 'fitness_level', 2) or 2
|
||
exercises_per_superset = wt_params['exercises_per_superset']
|
||
if fitness_level == 1: # Beginner: cap supersets and exercises
|
||
num_supersets = min(num_supersets, 3)
|
||
exercises_per_superset = (
|
||
exercises_per_superset[0],
|
||
min(exercises_per_superset[1], 3),
|
||
)
|
||
elif fitness_level == 4: # Elite: allow more
|
||
num_supersets = min(num_supersets + 1, wt_params['num_supersets'][1] + 1)
|
||
exercises_per_superset = (
|
||
exercises_per_superset[0],
|
||
exercises_per_superset[1] + 1,
|
||
)
|
||
|
||
# Deload adjustments: reduce volume for recovery week
|
||
if getattr(self, '_is_deload', False):
|
||
num_supersets = max(1, num_supersets - 1)
|
||
wt_params = dict(wt_params) # Don't mutate the original
|
||
wt_params['rounds'] = (
|
||
max(1, wt_params['rounds'][0] - 1),
|
||
max(1, wt_params['rounds'][1] - 1),
|
||
)
|
||
wt_params['rest_between_rounds'] = wt_params.get('rest_between_rounds', 45) + 15
|
||
|
||
# Get movement pattern ordering from DB (if available)
|
||
early_patterns = list(
|
||
MovementPatternOrder.objects.filter(
|
||
position='early', section_type='working',
|
||
).order_by('-frequency').values_list('movement_pattern', flat=True)
|
||
)
|
||
middle_patterns = list(
|
||
MovementPatternOrder.objects.filter(
|
||
position='middle', section_type='working',
|
||
).order_by('-frequency').values_list('movement_pattern', flat=True)
|
||
)
|
||
late_patterns = list(
|
||
MovementPatternOrder.objects.filter(
|
||
position='late', section_type='working',
|
||
).order_by('-frequency').values_list('movement_pattern', flat=True)
|
||
)
|
||
|
||
# Exercise science: plyometrics should be early when CNS is fresh
|
||
if 'plyometric' in late_patterns:
|
||
late_patterns = [p for p in late_patterns if p != 'plyometric']
|
||
if 'plyometric' not in early_patterns:
|
||
early_patterns.append('plyometric')
|
||
|
||
# Item #4: Merge movement patterns from WorkoutStructureRule (if available)
|
||
rule_patterns = []
|
||
if workout_type:
|
||
goal = getattr(self.preference, 'primary_goal', '') or 'general_fitness'
|
||
structure_rule = WorkoutStructureRule.objects.filter(
|
||
workout_type=workout_type,
|
||
section_type='working',
|
||
goal_type=goal,
|
||
).first()
|
||
if not structure_rule:
|
||
# Fallback: try general_fitness if specific goal not found
|
||
structure_rule = WorkoutStructureRule.objects.filter(
|
||
workout_type=workout_type,
|
||
section_type='working',
|
||
goal_type='general_fitness',
|
||
).first()
|
||
if structure_rule and structure_rule.movement_patterns:
|
||
rule_patterns = structure_rule.movement_patterns
|
||
|
||
# R5/R7: Determine if this is a strength-type workout
|
||
is_strength_workout = False
|
||
if workout_type:
|
||
wt_name_lower = workout_type.name.strip().lower()
|
||
if wt_name_lower in STRENGTH_WORKOUT_TYPES:
|
||
is_strength_workout = True
|
||
|
||
min_duration = GENERATION_RULES['min_duration']['value']
|
||
duration_mult = GENERATION_RULES['duration_multiple']['value']
|
||
min_volume = GENERATION_RULES['min_volume']['value']
|
||
min_ex_per_ss = GENERATION_RULES['min_exercises_per_superset']['value']
|
||
|
||
supersets = []
|
||
|
||
for ss_idx in range(num_supersets):
|
||
rounds = random.randint(*wt_params['rounds'])
|
||
ex_count = random.randint(*exercises_per_superset)
|
||
|
||
# Item #7: First working superset in strength workouts = single main lift
|
||
if is_strength_workout and ss_idx == 0:
|
||
ex_count = 1
|
||
rounds = random.randint(4, 6)
|
||
rest_between_rounds_override = getattr(workout_type, 'typical_rest_between_sets', 120)
|
||
else:
|
||
# R8: Minimum 2 exercises per working superset (non-first-strength only)
|
||
ex_count = max(min_ex_per_ss, ex_count)
|
||
rest_between_rounds_override = None
|
||
|
||
# Determine movement pattern preference based on position
|
||
if num_supersets <= 1:
|
||
position_patterns = early_patterns or None
|
||
elif ss_idx == 0:
|
||
position_patterns = early_patterns or None
|
||
elif ss_idx >= num_supersets - 1:
|
||
position_patterns = late_patterns or None
|
||
else:
|
||
position_patterns = middle_patterns or None
|
||
|
||
# Item #4: Merge position patterns with structure rule patterns
|
||
if rule_patterns and position_patterns:
|
||
combined_patterns = [p for p in position_patterns if p in rule_patterns] or rule_patterns[:3]
|
||
elif rule_patterns:
|
||
combined_patterns = rule_patterns[:3]
|
||
else:
|
||
combined_patterns = position_patterns
|
||
|
||
# Distribute target muscles across supersets for variety
|
||
# Each superset focuses on a subset of the target muscles
|
||
if len(target_muscles) > 1:
|
||
# Rotate which muscles are emphasised per superset
|
||
start = ss_idx % len(target_muscles)
|
||
muscle_subset = (
|
||
target_muscles[start:]
|
||
+ target_muscles[:start]
|
||
)
|
||
else:
|
||
muscle_subset = target_muscles
|
||
|
||
# R9: Decide modality once per superset (all reps or all duration)
|
||
# R5/R7: For strength workouts, force rep-based in working sets
|
||
if is_strength_workout:
|
||
superset_is_duration = False
|
||
else:
|
||
superset_is_duration = random.random() < duration_bias
|
||
|
||
# R6: For strength workouts, prefer weighted exercises
|
||
prefer_weighted = is_strength_workout
|
||
|
||
# Fix #6: Determine position string for exercise selection
|
||
if num_supersets <= 1:
|
||
position_str = 'early'
|
||
elif ss_idx == 0:
|
||
position_str = 'early'
|
||
elif ss_idx >= num_supersets - 1:
|
||
position_str = 'late'
|
||
else:
|
||
position_str = 'middle'
|
||
|
||
# Select exercises
|
||
exercises = self.exercise_selector.select_exercises(
|
||
muscle_groups=muscle_subset,
|
||
count=ex_count,
|
||
is_duration_based=superset_is_duration,
|
||
movement_pattern_preference=combined_patterns,
|
||
prefer_weighted=prefer_weighted,
|
||
superset_position=position_str,
|
||
)
|
||
|
||
if not exercises:
|
||
# R13: Try broader muscles for this split type before going fully unfiltered
|
||
logger.warning(
|
||
"No exercises found for muscles %s, falling back to broader muscles",
|
||
muscle_subset,
|
||
)
|
||
broader_muscles = self._get_broader_muscles(muscle_split.get('split_type', 'full_body'))
|
||
exercises = self.exercise_selector.select_exercises(
|
||
muscle_groups=broader_muscles,
|
||
count=ex_count,
|
||
is_duration_based=superset_is_duration,
|
||
movement_pattern_preference=combined_patterns,
|
||
prefer_weighted=prefer_weighted,
|
||
)
|
||
|
||
if not exercises:
|
||
# Final fallback: any exercises matching modality (no muscle filter)
|
||
logger.warning("Broader muscles also failed for superset %d, trying unfiltered", ss_idx)
|
||
exercises = self.exercise_selector.select_exercises(
|
||
muscle_groups=[],
|
||
count=ex_count,
|
||
is_duration_based=superset_is_duration,
|
||
prefer_weighted=prefer_weighted,
|
||
)
|
||
|
||
if not exercises:
|
||
continue
|
||
|
||
# Balance stretch positions for all goals (not just hypertrophy)
|
||
if len(exercises) >= 3:
|
||
exercises = self.exercise_selector.balance_stretch_positions(
|
||
exercises, muscle_groups=muscle_subset, fitness_level=fitness_level,
|
||
)
|
||
|
||
# Build exercise entries with reps/duration
|
||
exercise_entries = []
|
||
for i, ex in enumerate(exercises, start=1):
|
||
entry = {
|
||
'exercise': ex,
|
||
'order': i,
|
||
}
|
||
|
||
if superset_is_duration:
|
||
if ex.is_duration:
|
||
duration = random.randint(
|
||
wt_params['duration_min'],
|
||
wt_params['duration_max'],
|
||
)
|
||
# R4: Round to multiple of 5, clamp to min 20
|
||
duration = max(min_duration, round(duration / duration_mult) * duration_mult)
|
||
entry['duration'] = duration
|
||
else:
|
||
# Non-duration exercise leaked through fallback -- skip to preserve R9
|
||
logger.debug("Skipping non-duration exercise %s in duration superset", ex.name)
|
||
continue
|
||
else:
|
||
# R9: When superset is rep-based, always assign reps
|
||
# even if the exercise is duration-capable
|
||
entry['reps'] = random.randint(
|
||
wt_params['rep_min'],
|
||
wt_params['rep_max'],
|
||
)
|
||
if ex.is_weight:
|
||
entry['weight'] = None # user fills in their weight
|
||
|
||
exercise_entries.append(entry)
|
||
|
||
# Re-number orders after filtering
|
||
for idx, entry in enumerate(exercise_entries, start=1):
|
||
entry['order'] = idx
|
||
|
||
# R10: Volume floor — reps × rounds >= 12
|
||
for entry in exercise_entries:
|
||
if entry.get('reps') and entry['reps'] * rounds < min_volume:
|
||
entry['reps'] = max(entry['reps'], math.ceil(min_volume / rounds))
|
||
|
||
supersets.append({
|
||
'name': f'Working Set {ss_idx + 1}',
|
||
'rounds': rounds,
|
||
'rest_between_rounds': rest_between_rounds_override or wt_params.get('rest_between_rounds', 45),
|
||
'modality': 'duration' if superset_is_duration else 'reps',
|
||
'exercises': exercise_entries,
|
||
})
|
||
|
||
# Item #6: Modality consistency check
|
||
if wt_params.get('duration_bias', 0) >= 0.6:
|
||
total_exercises = 0
|
||
duration_exercises = 0
|
||
for ss in supersets:
|
||
for ex_entry in ss.get('exercises', []):
|
||
total_exercises += 1
|
||
ex = ex_entry.get('exercise')
|
||
if ex and hasattr(ex, 'is_duration') and ex.is_duration:
|
||
duration_exercises += 1
|
||
if total_exercises > 0:
|
||
duration_ratio = duration_exercises / total_exercises
|
||
if duration_ratio < 0.5:
|
||
self.warnings.append(
|
||
f"Modality mismatch: {duration_ratio:.0%} duration exercises "
|
||
f"in a duration-dominant workout type (expected >= 50%)"
|
||
)
|
||
|
||
# Sort exercises within supersets by HR elevation: higher HR early, lower late
|
||
for idx, ss in enumerate(supersets):
|
||
exercises = ss.get('exercises', [])
|
||
if len(exercises) <= 1:
|
||
continue
|
||
is_early = idx < len(supersets) / 2
|
||
exercises.sort(
|
||
key=lambda e: getattr(e.get('exercise'), 'hr_elevation_rating', 5) or 5,
|
||
reverse=is_early, # Descending for early (high HR first), ascending for late
|
||
)
|
||
# Re-number orders after sorting
|
||
for i, entry in enumerate(exercises, start=1):
|
||
entry['order'] = i
|
||
|
||
return supersets
|
||
|
||
def _build_cooldown(self, target_muscles, workout_type=None):
|
||
"""
|
||
Build a cool-down superset spec: duration-based, 1 round.
|
||
|
||
Exercise count scaled by fitness level:
|
||
- Beginner: 4-5 (more recovery)
|
||
- Intermediate: 3-4 (default)
|
||
- Advanced/Elite: 2-3
|
||
|
||
Returns
|
||
-------
|
||
dict | None
|
||
"""
|
||
fitness_level = getattr(self.preference, 'fitness_level', 2) or 2
|
||
if fitness_level <= 1:
|
||
count = random.randint(4, 5)
|
||
elif fitness_level >= 3:
|
||
count = random.randint(2, 3)
|
||
else:
|
||
count = random.randint(3, 4)
|
||
|
||
# Try to get duration range from DB structure rules
|
||
cooldown_dur_min = 25
|
||
cooldown_dur_max = 45
|
||
if workout_type:
|
||
cooldown_rules = list(
|
||
WorkoutStructureRule.objects.filter(
|
||
workout_type=workout_type,
|
||
section_type='cool_down',
|
||
).order_by('-id')[:1]
|
||
)
|
||
if cooldown_rules:
|
||
cooldown_dur_min = cooldown_rules[0].typical_duration_range_min
|
||
cooldown_dur_max = cooldown_rules[0].typical_duration_range_max
|
||
|
||
exercises = self.exercise_selector.select_cooldown_exercises(
|
||
target_muscles, count=count,
|
||
)
|
||
|
||
if not exercises:
|
||
return None
|
||
|
||
min_duration = GENERATION_RULES['min_duration']['value']
|
||
duration_mult = GENERATION_RULES['duration_multiple']['value']
|
||
|
||
exercise_entries = []
|
||
for i, ex in enumerate(exercises, start=1):
|
||
duration = random.randint(cooldown_dur_min, cooldown_dur_max)
|
||
# R4: Round to multiple of 5, clamp to min 20
|
||
duration = max(min_duration, round(duration / duration_mult) * duration_mult)
|
||
exercise_entries.append({
|
||
'exercise': ex,
|
||
'duration': duration,
|
||
'order': i,
|
||
})
|
||
|
||
return {
|
||
'name': 'Cool Down',
|
||
'rounds': 1,
|
||
'rest_between_rounds': 0,
|
||
'exercises': exercise_entries,
|
||
}
|
||
|
||
# ==================================================================
|
||
# Time adjustment
|
||
# ==================================================================
|
||
|
||
def _adjust_to_time_target(self, workout_spec, max_duration_sec, muscle_split, wt_params, workout_type=None):
|
||
"""
|
||
Estimate workout duration and trim or pad to stay close to the
|
||
user's preferred_workout_duration.
|
||
|
||
Uses the same formula as ``add_workout``:
|
||
reps * estimated_rep_duration + duration values
|
||
|
||
The estimated_time stored on Workout represents "active time" (no
|
||
rest between sets). Historical data shows active time is roughly
|
||
60-70% of wall-clock time, so we target ~65% of the user's
|
||
preferred duration.
|
||
"""
|
||
# Target active time as a fraction of wall-clock preference,
|
||
# scaled by fitness level (beginners rest more, elite rest less)
|
||
fitness_level = getattr(self.preference, 'fitness_level', 2) or 2
|
||
active_ratio = FITNESS_ACTIVE_TIME_RATIO.get(fitness_level, 0.65)
|
||
active_target_sec = max_duration_sec * active_ratio
|
||
estimated = self._estimate_total_time(workout_spec)
|
||
|
||
# Trim if over budget (tightened from 1.15 to 1.10)
|
||
if estimated > active_target_sec * 1.10:
|
||
workout_spec = self._trim_to_fit(workout_spec, active_target_sec)
|
||
|
||
# Pad if significantly under budget (tightened from 0.80 to 0.85)
|
||
elif estimated < active_target_sec * 0.85:
|
||
workout_spec = self._pad_to_fill(
|
||
workout_spec, active_target_sec, muscle_split, wt_params,
|
||
workout_type=workout_type,
|
||
)
|
||
|
||
# R10: Re-enforce volume floor after any trimming/padding
|
||
min_volume = GENERATION_RULES['min_volume']['value']
|
||
for ss in workout_spec.get('supersets', []):
|
||
if not ss.get('name', '').startswith('Working'):
|
||
continue
|
||
rounds = ss.get('rounds', 1)
|
||
for entry in ss.get('exercises', []):
|
||
if entry.get('reps') and entry['reps'] * rounds < min_volume:
|
||
entry['reps'] = math.ceil(min_volume / rounds)
|
||
|
||
return workout_spec
|
||
|
||
def _estimate_total_time(self, workout_spec):
|
||
"""
|
||
Calculate estimated total time in seconds for a workout_spec,
|
||
following the same logic as plan_builder.
|
||
"""
|
||
total = 0
|
||
for ss in workout_spec.get('supersets', []):
|
||
rounds = ss.get('rounds', 1)
|
||
superset_time = 0
|
||
for ex_entry in ss.get('exercises', []):
|
||
ex = ex_entry.get('exercise')
|
||
if ex_entry.get('reps') is not None and ex is not None:
|
||
rep_dur = ex.estimated_rep_duration or 3.0
|
||
superset_time += ex_entry['reps'] * rep_dur
|
||
if ex_entry.get('duration') is not None:
|
||
superset_time += ex_entry['duration']
|
||
rest_between = ss.get('rest_between_rounds', 45)
|
||
rest_time = rest_between * max(0, rounds - 1)
|
||
total += (superset_time * rounds) + rest_time
|
||
|
||
# Add transition time between supersets
|
||
supersets = workout_spec.get('supersets', [])
|
||
if len(supersets) > 1:
|
||
rest_between_supersets = GENERATION_RULES['rest_between_supersets']['value']
|
||
total += rest_between_supersets * (len(supersets) - 1)
|
||
|
||
return total
|
||
|
||
def _trim_to_fit(self, workout_spec, max_duration_sec):
|
||
"""
|
||
Remove exercises from working supersets (back-to-front) until
|
||
estimated time is within budget. Always preserves at least one
|
||
working superset with minimal configuration.
|
||
"""
|
||
supersets = workout_spec.get('supersets', [])
|
||
|
||
# Identify working supersets (not Warm Up / Cool Down)
|
||
working_indices = [
|
||
i for i, ss in enumerate(supersets)
|
||
if ss.get('name', '').startswith('Working')
|
||
]
|
||
|
||
min_ex_per_ss = GENERATION_RULES['min_exercises_per_superset']['value']
|
||
removed_supersets = []
|
||
|
||
# Remove exercises from the last working superset first
|
||
for idx in reversed(working_indices):
|
||
ss = supersets[idx]
|
||
# R8: Don't trim below min exercises per superset
|
||
while len(ss['exercises']) > min_ex_per_ss:
|
||
ss['exercises'].pop()
|
||
if self._estimate_total_time(workout_spec) <= max_duration_sec:
|
||
return workout_spec
|
||
|
||
# If still over, reduce rounds
|
||
while ss['rounds'] > 1:
|
||
ss['rounds'] -= 1
|
||
if self._estimate_total_time(workout_spec) <= max_duration_sec:
|
||
return workout_spec
|
||
|
||
# If still over, remove the entire superset (better than leaving
|
||
# a superset with too few exercises)
|
||
if self._estimate_total_time(workout_spec) > max_duration_sec:
|
||
removed = supersets.pop(idx)
|
||
removed_supersets.append(removed)
|
||
if self._estimate_total_time(workout_spec) <= max_duration_sec:
|
||
break
|
||
|
||
# Ensure at least 1 working superset remains
|
||
remaining_working = [
|
||
ss for ss in supersets
|
||
if ss.get('name', '').startswith('Working')
|
||
]
|
||
if not remaining_working and removed_supersets:
|
||
# Re-add the last removed superset with minimal config
|
||
minimal = removed_supersets[-1]
|
||
minimal['exercises'] = minimal['exercises'][:min_ex_per_ss]
|
||
minimal['rounds'] = 2
|
||
# Insert before Cool Down if present
|
||
cooldown_idx = next(
|
||
(i for i, ss in enumerate(supersets) if ss.get('name') == 'Cool Down'),
|
||
len(supersets),
|
||
)
|
||
supersets.insert(cooldown_idx, minimal)
|
||
logger.info('Re-added minimal working superset to prevent empty workout')
|
||
|
||
return workout_spec
|
||
|
||
def _pad_to_fill(self, workout_spec, max_duration_sec, muscle_split, wt_params, workout_type=None):
|
||
"""
|
||
Add exercises to working supersets or add a new superset to fill
|
||
time closer to the target.
|
||
"""
|
||
target_muscles = muscle_split.get('muscles', [])
|
||
supersets = workout_spec.get('supersets', [])
|
||
duration_bias = wt_params.get('duration_bias', 0.3)
|
||
|
||
# Derive strength context for workout-type-aware padding
|
||
is_strength_workout = False
|
||
if workout_type:
|
||
wt_name_lower = workout_type.name.strip().lower()
|
||
is_strength_workout = wt_name_lower in STRENGTH_WORKOUT_TYPES
|
||
prefer_weighted = is_strength_workout
|
||
min_duration = GENERATION_RULES['min_duration']['value']
|
||
duration_mult = GENERATION_RULES['duration_multiple']['value']
|
||
min_volume = GENERATION_RULES['min_volume']['value']
|
||
|
||
# Find the insertion point: before Cool Down if it exists, else at end
|
||
insert_idx = len(supersets)
|
||
for i, ss in enumerate(supersets):
|
||
if ss.get('name', '') == 'Cool Down':
|
||
insert_idx = i
|
||
break
|
||
|
||
MAX_EXERCISES_PER_SUPERSET = 6
|
||
max_pad_attempts = 8
|
||
pad_attempts = 0
|
||
|
||
while (
|
||
self._estimate_total_time(workout_spec) < max_duration_sec * 0.9
|
||
and pad_attempts < max_pad_attempts
|
||
):
|
||
pad_attempts += 1
|
||
|
||
# Try adding exercises to existing working supersets first
|
||
added = False
|
||
for ss in supersets:
|
||
if not ss.get('name', '').startswith('Working'):
|
||
continue
|
||
if len(ss['exercises']) >= MAX_EXERCISES_PER_SUPERSET:
|
||
continue
|
||
|
||
# R9: Use stored modality from superset spec
|
||
ss_is_duration = ss.get('modality') == 'duration'
|
||
|
||
new_exercises = self.exercise_selector.select_exercises(
|
||
muscle_groups=target_muscles,
|
||
count=1,
|
||
is_duration_based=ss_is_duration,
|
||
prefer_weighted=prefer_weighted,
|
||
)
|
||
if new_exercises:
|
||
ex = new_exercises[0]
|
||
new_order = len(ss['exercises']) + 1
|
||
entry = {'exercise': ex, 'order': new_order}
|
||
|
||
if ss_is_duration:
|
||
if ex.is_duration:
|
||
duration = random.randint(
|
||
wt_params['duration_min'],
|
||
wt_params['duration_max'],
|
||
)
|
||
duration = max(min_duration, round(duration / duration_mult) * duration_mult)
|
||
entry['duration'] = duration
|
||
else:
|
||
# Skip non-duration exercise in duration superset (R9)
|
||
continue
|
||
else:
|
||
entry['reps'] = random.randint(
|
||
wt_params['rep_min'],
|
||
wt_params['rep_max'],
|
||
)
|
||
if ex.is_weight:
|
||
entry['weight'] = None
|
||
# R10: Volume floor
|
||
rounds = ss.get('rounds', 1)
|
||
if entry['reps'] * rounds < min_volume:
|
||
entry['reps'] = max(entry['reps'], math.ceil(min_volume / rounds))
|
||
|
||
ss['exercises'].append(entry)
|
||
added = True
|
||
|
||
# Check immediately after adding to prevent overshooting
|
||
if self._estimate_total_time(workout_spec) >= max_duration_sec * 0.9:
|
||
break
|
||
|
||
if self._estimate_total_time(workout_spec) >= max_duration_sec * 0.9:
|
||
break
|
||
|
||
# If we couldn't add to existing, create a new working superset
|
||
if not added:
|
||
rounds = random.randint(*wt_params['rounds'])
|
||
ex_count = random.randint(*wt_params['exercises_per_superset'])
|
||
# R8: Min 2 exercises
|
||
ex_count = max(GENERATION_RULES['min_exercises_per_superset']['value'], ex_count)
|
||
# R9: Decide modality once for the new superset
|
||
# R5/R7: For strength workouts, force rep-based
|
||
if is_strength_workout:
|
||
ss_is_duration = False
|
||
else:
|
||
ss_is_duration = random.random() < duration_bias
|
||
|
||
exercises = self.exercise_selector.select_exercises(
|
||
muscle_groups=target_muscles,
|
||
count=ex_count,
|
||
is_duration_based=ss_is_duration,
|
||
prefer_weighted=prefer_weighted,
|
||
)
|
||
if not exercises:
|
||
break
|
||
|
||
exercise_entries = []
|
||
for i, ex in enumerate(exercises, start=1):
|
||
entry = {'exercise': ex, 'order': i}
|
||
if ss_is_duration:
|
||
if ex.is_duration:
|
||
duration = random.randint(
|
||
wt_params['duration_min'],
|
||
wt_params['duration_max'],
|
||
)
|
||
duration = max(min_duration, round(duration / duration_mult) * duration_mult)
|
||
entry['duration'] = duration
|
||
else:
|
||
# Skip non-duration exercise in duration superset (R9)
|
||
continue
|
||
else:
|
||
entry['reps'] = random.randint(
|
||
wt_params['rep_min'],
|
||
wt_params['rep_max'],
|
||
)
|
||
if ex.is_weight:
|
||
entry['weight'] = None
|
||
exercise_entries.append(entry)
|
||
|
||
# Re-number orders after filtering
|
||
for idx, entry in enumerate(exercise_entries, start=1):
|
||
entry['order'] = idx
|
||
|
||
# R10: Volume floor for new superset
|
||
for entry in exercise_entries:
|
||
if entry.get('reps') and entry['reps'] * rounds < min_volume:
|
||
entry['reps'] = max(entry['reps'], math.ceil(min_volume / rounds))
|
||
|
||
working_count = sum(
|
||
1 for ss in supersets
|
||
if ss.get('name', '').startswith('Working')
|
||
)
|
||
new_superset = {
|
||
'name': f'Working Set {working_count + 1}',
|
||
'rounds': rounds,
|
||
'rest_between_rounds': wt_params.get('rest_between_rounds', 45),
|
||
'modality': 'duration' if ss_is_duration else 'reps',
|
||
'exercises': exercise_entries,
|
||
}
|
||
supersets.insert(insert_idx, new_superset)
|
||
insert_idx += 1
|
||
|
||
# Early exit if we've reached 90% of target after adding new superset
|
||
if self._estimate_total_time(workout_spec) >= max_duration_sec * 0.9:
|
||
break
|
||
|
||
return workout_spec
|
||
|
||
def _check_quality_gates(self, working_supersets, workout_type, wt_params):
|
||
"""Run quality gate validation on working supersets.
|
||
|
||
Combines the rules engine validation with the legacy workout-type
|
||
match check. Returns a list of RuleViolation objects.
|
||
|
||
Parameters
|
||
----------
|
||
working_supersets : list[dict]
|
||
The working supersets (no warmup/cooldown).
|
||
workout_type : WorkoutType | None
|
||
wt_params : dict
|
||
Workout type parameters from _get_workout_type_params().
|
||
|
||
Returns
|
||
-------
|
||
list[RuleViolation]
|
||
"""
|
||
if not workout_type:
|
||
return []
|
||
|
||
wt_name = workout_type.name.strip()
|
||
|
||
# Build a temporary workout_spec for the rules engine
|
||
# (just the working supersets — warmup/cooldown added later)
|
||
temp_spec = {
|
||
'supersets': list(working_supersets),
|
||
}
|
||
|
||
# Run the rules engine validation (skips warmup/cooldown checks
|
||
# since those aren't built yet at this point)
|
||
goal = getattr(self.preference, 'primary_goal', 'general_fitness')
|
||
violations = validate_workout(temp_spec, wt_name, goal)
|
||
|
||
# Filter out warmup/cooldown violations since they haven't been
|
||
# added yet at this stage
|
||
violations = [
|
||
v for v in violations
|
||
if v.rule_id not in ('warmup_missing', 'cooldown_missing')
|
||
]
|
||
|
||
# Legacy workout-type match check (now returns violations instead of logging)
|
||
legacy_violations = self._validate_workout_type_match_violations(
|
||
working_supersets, workout_type,
|
||
)
|
||
violations.extend(legacy_violations)
|
||
|
||
return violations
|
||
|
||
def _validate_workout_type_match_violations(self, working_supersets, workout_type):
|
||
"""Check workout type match percentage, returning RuleViolation objects.
|
||
|
||
Refactored from _validate_workout_type_match to return structured
|
||
violations instead of just logging.
|
||
"""
|
||
if not workout_type:
|
||
return []
|
||
|
||
wt_name_lower = workout_type.name.strip().lower()
|
||
is_strength = wt_name_lower in STRENGTH_WORKOUT_TYPES
|
||
threshold = GENERATION_RULES['workout_type_match_pct']['value']
|
||
|
||
total_exercises = 0
|
||
matching_exercises = 0
|
||
for ss in working_supersets:
|
||
for entry in ss.get('exercises', []):
|
||
total_exercises += 1
|
||
ex = entry.get('exercise')
|
||
if ex is None:
|
||
continue
|
||
if is_strength:
|
||
if getattr(ex, 'is_weight', False) or getattr(ex, 'is_compound', False):
|
||
matching_exercises += 1
|
||
else:
|
||
matching_exercises += 1
|
||
|
||
violations = []
|
||
if total_exercises > 0:
|
||
match_pct = matching_exercises / total_exercises
|
||
if match_pct < threshold:
|
||
logger.warning(
|
||
"Workout type match %.0f%% below threshold %.0f%% for %s",
|
||
match_pct * 100, threshold * 100, wt_name_lower,
|
||
)
|
||
violations.append(RuleViolation(
|
||
rule_id='workout_type_match_legacy',
|
||
severity='error',
|
||
message=(
|
||
f'Workout type match {match_pct:.0%} below '
|
||
f'threshold {threshold:.0%} for {wt_name_lower}.'
|
||
),
|
||
actual_value=match_pct,
|
||
expected_range=(threshold, 1.0),
|
||
))
|
||
return violations
|
||
|
||
def _validate_workout_type_match(self, working_supersets, workout_type):
|
||
"""Legacy method kept for backward compatibility. Now delegates to violations version."""
|
||
self._validate_workout_type_match_violations(working_supersets, workout_type)
|
||
|
||
@staticmethod
|
||
def _get_broader_muscles(split_type):
|
||
"""Get broader muscle list for a split type when specific muscles can't find exercises."""
|
||
from generator.services.muscle_normalizer import MUSCLE_GROUP_CATEGORIES
|
||
broader = {
|
||
'push': MUSCLE_GROUP_CATEGORIES.get('upper_push', []),
|
||
'pull': MUSCLE_GROUP_CATEGORIES.get('upper_pull', []),
|
||
'upper': MUSCLE_GROUP_CATEGORIES.get('upper_push', []) + MUSCLE_GROUP_CATEGORIES.get('upper_pull', []),
|
||
'lower': MUSCLE_GROUP_CATEGORIES.get('lower_push', []) + MUSCLE_GROUP_CATEGORIES.get('lower_pull', []),
|
||
'legs': MUSCLE_GROUP_CATEGORIES.get('lower_push', []) + MUSCLE_GROUP_CATEGORIES.get('lower_pull', []),
|
||
'core': MUSCLE_GROUP_CATEGORIES.get('core', []),
|
||
'full_body': (MUSCLE_GROUP_CATEGORIES.get('upper_push', []) +
|
||
MUSCLE_GROUP_CATEGORIES.get('upper_pull', []) +
|
||
MUSCLE_GROUP_CATEGORIES.get('lower_push', []) +
|
||
MUSCLE_GROUP_CATEGORIES.get('core', [])),
|
||
}
|
||
return broader.get(split_type, [])
|