import logging import math import random import time import uuid from collections import Counter from datetime import timedelta from django.db.models import Q from equipment.models import WorkoutEquipment from generator.models import ( GeneratedWeeklyPlan, GeneratedWorkout, MuscleGroupSplit, MovementPatternOrder, WeeklySplitPattern, WorkoutStructureRule, WorkoutType, ) from generator.rules_engine import ( RuleViolation, UNIVERSAL_RULES, WORKOUT_TYPE_RULES, _normalize_type_key, validate_workout, ) from generator.services.exercise_selector import ExerciseSelector from generator.services.plan_builder import PlanBuilder from generator.services.muscle_normalizer import normalize_muscle_name from generator.services.workout_generation.scaling import apply_fitness_scaling from generator.services.workout_generation.section_builders import ( build_duration_entries, build_section_superset, section_exercise_count, ) from generator.services.workout_generation.modality import ( clamp_duration_bias, plan_superset_modalities, ) from generator.services.workout_generation.pattern_planning import ( merge_pattern_preferences, rotated_muscle_subset, working_position_label, ) from generator.services.workout_generation.entry_rules import ( apply_rep_volume_floor, pick_reps_for_exercise, sort_entries_by_hr, working_rest_seconds, ) from generator.services.workout_generation.focus import ( focus_key_for_exercise, focus_keys_for_exercises, has_duplicate_focus, ) from generator.services.workout_generation.recovery import is_recovery_exercise from muscle.models import ExerciseMuscle 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': False, 'description': 'Exercises within a superset use their native modality (carries get duration, lifts get reps)', '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', } # Prefix used for working superset names — single source of truth. WORKING_PREFIX = "Working" # Final pass retries after full assembly (warmup + working + cooldown) # to guarantee conformance before returning a workout. FINAL_CONFORMANCE_MAX_RETRIES = 2 # ====================================================================== # 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 # Cross-day dedup: clear week state at start of plan generation self.exercise_selector.reset_week() 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, ) # Cross-day dedup: record this day's exercises for future days day_ids, day_names = self._extract_exercise_info_from_spec(workout_spec) self.exercise_selector.accumulate_week_state(day_ids, day_names) 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: difficulties = [c.difficulty for c in recent_completed if c.difficulty is not None] avg_difficulty = sum(difficulties) / len(difficulties) if difficulties else 2.5 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 # Cross-day dedup: clear week state at start of plan generation self.exercise_selector.reset_week() 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, ) # Cross-day dedup: record this day's exercises for future days day_ids, day_names = self._extract_exercise_info_from_spec(workout_spec) self.exercise_selector.accumulate_week_state(day_ids, day_names) # 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, _final_attempt=0, ): """ 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) """ warnings_checkpoint = len(self.warnings) target_muscles = list(muscle_split.get('muscles', [])) label = muscle_split.get('label', 'Workout') duration_minutes = self.duration_override or self.preference.preferred_workout_duration or 45 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)) # Respect workout-type hard duration ceilings (e.g. HIIT <= 30 min). if workout_type: wt_key = _normalize_type_key(getattr(workout_type, 'name', '') or '') wt_rules = WORKOUT_TYPE_RULES.get(wt_key, {}) max_minutes_for_type = wt_rules.get('max_duration_minutes') if max_minutes_for_type: max_duration_sec = min(max_duration_sec, int(max_minutes_for_type) * 60) # 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: 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 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 MAX_WORKING_EXERCISES = self._max_working_exercises_for_type(workout_type) working_supersets = [ ss for ss in workout_spec.get('supersets', []) if ss.get('name', '').startswith(WORKING_PREFIX) ] first_working_superset = working_supersets[0] if working_supersets else None def _min_working_exercises_for_ss(ss): # Allow a first straight set (single main lift) for strength workouts. if is_strength_workout and first_working_superset is not None and ss is first_working_superset: return 1 return 2 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): min_ex_for_ss = _min_working_exercises_for_ss(ss) while excess > 0 and len(ss['exercises']) > min_ex_for_ss: ss['exercises'].pop() excess -= 1 if excess <= 0: break # If everything is already at min size, remove trailing supersets. if excess > 0: for ss in reversed(list(working_supersets)): current_working = [ candidate for candidate in workout_spec.get('supersets', []) if candidate.get('name', '').startswith(WORKING_PREFIX) ] if len(current_working) <= 1 or excess <= 0: break if is_strength_workout and ss is first_working_superset: # Preserve straight-set anchor for strength unless it's the last one left. continue removed_count = len(ss.get('exercises', [])) if removed_count <= 0: continue try: workout_spec['supersets'].remove(ss) except ValueError: continue excess -= removed_count # Remove undersized working supersets. workout_spec['supersets'] = [ ss for ss in workout_spec['supersets'] if ( not ss.get('name', '').startswith(WORKING_PREFIX) or len(ss['exercises']) >= _min_working_exercises_for_ss(ss) ) ] # Enforce minimum exercises per working superset; merge undersized ones. # First strength working set is allowed to be a straight set (1 exercise). all_supersets = workout_spec.get('supersets', []) working_indices = [ i for i, ss in enumerate(all_supersets) if ss.get('name', '').startswith(WORKING_PREFIX) ] first_working_index = working_indices[0] if working_indices else None def _min_working_exercises_for_index(idx): if is_strength_workout and first_working_index is not None and idx == first_working_index: return 1 return 2 undersized = [ i for i in working_indices if len(all_supersets[i]['exercises']) < _min_working_exercises_for_index(i) ] if undersized: # Try to redistribute exercises from undersized into adjacent supersets for idx in reversed(undersized): if len(all_supersets[idx]['exercises']) >= _min_working_exercises_for_index(idx): continue ss = all_supersets[idx] orphan_exercises = ss['exercises'] # Find next working superset to absorb orphans 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) 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_PREFIX) ] first_working_index = working_indices[0] if working_indices else None # Post-build modality validation: ensure each exercise entry has # either reps or duration (not both, not neither), based on the # exercise's native capabilities. min_duration_val = GENERATION_RULES['min_duration']['value'] duration_mult_val = GENERATION_RULES['duration_multiple']['value'] for ss in workout_spec.get('supersets', []): if not ss.get('name', '').startswith(WORKING_PREFIX): continue # Working supersets should always include explicit non-zero rest. rest_between_rounds = ss.get('rest_between_rounds') if not rest_between_rounds or rest_between_rounds <= 0: ss['rest_between_rounds'] = max(15, int(wt_params.get('rest_between_rounds', 45) or 45)) for entry in ss.get('exercises', []): has_reps = entry.get('reps') is not None has_duration = entry.get('duration') is not None # Already has exactly one — good if has_reps != has_duration: continue ex = entry.get('exercise') ex_is_reps = getattr(ex, 'is_reps', False) if ex else True ex_is_duration = getattr(ex, 'is_duration', False) if ex else False if has_reps and has_duration: # Both set — keep the native one if ex_is_duration and not ex_is_reps: entry.pop('reps', None) entry.pop('weight', None) else: entry.pop('duration', None) elif not has_reps and not has_duration: # Neither set — assign based on exercise capabilities if ex_is_duration and not ex_is_reps: duration = random.randint( wt_params['duration_min'], wt_params['duration_max'], ) entry['duration'] = max( min_duration_val, round(duration / duration_mult_val) * duration_mult_val, ) elif ex is not None: entry['reps'] = self._pick_reps_for_exercise(ex, wt_params, workout_type) if getattr(ex, 'is_weight', False): entry['weight'] = None else: entry['reps'] = random.randint(wt_params['rep_min'], wt_params['rep_max']) # Deterministic final-shaping fixes before strict conformance validation. self._enforce_compound_first_order(workout_spec, is_strength_workout=is_strength_workout) self._rebalance_push_pull( workout_spec, target_muscles, wt_params, is_strength_workout, workout_type=workout_type, ) final_violations = self._get_final_conformance_violations( workout_spec, workout_type, target_muscles, ) blocking_final = [ v for v in final_violations if self._is_blocking_final_violation(v) ] if blocking_final: if _final_attempt < FINAL_CONFORMANCE_MAX_RETRIES: logger.info( "Final conformance failed (%d issues) on attempt %d for %s; regenerating", len(blocking_final), _final_attempt + 1, label, ) self.warnings = self.warnings[:warnings_checkpoint] self.exercise_selector.reset() return self.generate_single_workout( muscle_split=muscle_split, workout_type=workout_type, scheduled_date=scheduled_date, _final_attempt=_final_attempt + 1, ) details = '; '.join( f'[{v.severity}] {v.rule_id}: {v.message}' for v in blocking_final[:5] ) raise ValueError( f'Unable to generate a compliant workout for {label}: {details}' ) # Collect warnings from exercise selector if self.exercise_selector.warnings: selector_warnings = list(self.exercise_selector.warnings) if not self._workout_has_side_specific_entries(workout_spec): selector_warnings = [ w for w in selector_warnings if not self._is_side_pair_warning(w) ] self.warnings.extend(selector_warnings) return workout_spec # ================================================================== # Split / scheduling helpers # ================================================================== @staticmethod def _is_side_pair_warning(message): text = (message or '').lower() return ( 'opposite-side' in text or 'side-specific' in text or 'left/right pairing' in text ) @staticmethod def _workout_has_side_specific_entries(workout_spec): for ss in workout_spec.get('supersets', []): for entry in ss.get('exercises', []): ex = entry.get('exercise') side = (getattr(ex, 'side', '') or '').strip() if side: return True return False 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) splits = self._diversify_split_days(splits, clamped_days) 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) splits = self._diversify_split_days(splits, clamped_days) return splits, [] @staticmethod def _split_signature(split_day): """Stable signature for duplicate detection across a week.""" split_type = (split_day.get('split_type') or 'full_body').strip().lower() muscles = tuple(sorted( normalize_muscle_name(m) for m in split_day.get('muscles', []) if m )) return split_type, muscles def _diversify_split_days(self, splits, clamped_days): """ Reduce repetitive split allocation (for example 3+ lower-body days in a 4-day plan) while preserving day count and user constraints. """ if len(splits) < 4: return splits result = [dict(s) for s in splits] max_same_type = 2 max_same_signature = 1 # Candidate pool: defaults first, then common DB splits. candidates = [dict(s) for s in DEFAULT_SPLITS.get(clamped_days, [])] db_candidates = list( MuscleGroupSplit.objects.order_by('-frequency', 'id')[:50] ) for mgs in db_candidates: candidates.append({ 'label': mgs.label or 'Workout', 'muscles': list(mgs.muscle_names or []), 'split_type': mgs.split_type or 'full_body', }) def _first_violation_index(): type_counts = Counter((d.get('split_type') or 'full_body').strip().lower() for d in result) sig_counts = Counter(self._split_signature(d) for d in result) for idx, day in enumerate(result): split_type = (day.get('split_type') or 'full_body').strip().lower() sig = self._split_signature(day) if type_counts[split_type] > max_same_type or sig_counts[sig] > max_same_signature: return idx, type_counts, sig_counts return None, type_counts, sig_counts # A small bounded repair loop prevents pathological endless churn. for _ in range(len(result) * 3): idx, type_counts, sig_counts = _first_violation_index() if idx is None: break replaced = False removed_type = (result[idx].get('split_type') or 'full_body').strip().lower() removed_sig = self._split_signature(result[idx]) for candidate in candidates: candidate_type = (candidate.get('split_type') or 'full_body').strip().lower() candidate_sig = self._split_signature(candidate) if candidate_sig == removed_sig: continue # Account for the removal of the old entry when counting # the new type: subtract 1 for the removed type if it # matches the candidate type, add 1 for the candidate. if candidate_type == removed_type: new_type_count = type_counts[candidate_type] # net zero: -1 removed +1 added else: new_type_count = type_counts[candidate_type] + 1 if new_type_count > max_same_type: continue # Same accounting for signatures: the removed signature # frees a slot, so only block if the candidate sig count # (after removing the old entry) is still at max. effective_sig_count = sig_counts[candidate_sig] if candidate_sig == removed_sig: effective_sig_count -= 1 if effective_sig_count >= max_same_signature: continue result[idx] = dict(candidate) replaced = True break if not replaced: # No safe replacement found; keep original to avoid invalid state. break return result 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 # Normalize to underscore format for consistent matching wt_name = wt.name.strip().lower().replace(' ', '_') if wt_name 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 return apply_fitness_scaling( params, fitness_level=level, scaling_config=FITNESS_LEVEL_SCALING, min_reps=GENERATION_RULES['min_reps']['value'], min_reps_strength=GENERATION_RULES['min_reps_strength']['value'], is_strength=is_strength, ) 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 count = section_exercise_count('warmup', fitness_level, rng=random) # 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 exercise_entries = build_duration_entries( exercises, duration_min=warmup_dur_min, duration_max=warmup_dur_max, min_duration=GENERATION_RULES['min_duration']['value'], duration_multiple=GENERATION_RULES['duration_multiple']['value'], rng=random, ) return build_section_superset('Warm Up', 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) duration_bias = self._clamp_duration_bias_for_type(duration_bias, workout_type) # 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 wt_rules = self._workout_type_rules(workout_type) modality_plan = plan_superset_modalities( num_supersets=num_supersets, duration_bias=duration_bias, duration_bias_range=wt_rules.get('duration_bias_range'), is_strength_workout=is_strength_workout, rng=random, ) 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 = [] previous_focus_keys = set() 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 combined_patterns = merge_pattern_preferences(position_patterns, rule_patterns) # Distribute target muscles across supersets for variety # Each superset focuses on a subset of the target muscles muscle_subset = rotated_muscle_subset(target_muscles, ss_idx) # R9: Decide modality once per superset (all reps or all duration) superset_is_duration = ( modality_plan[ss_idx] if ss_idx < len(modality_plan) else False ) # R6: For strength workouts, prefer weighted exercises prefer_weighted = is_strength_workout # Fix #6: Determine position string for exercise selection position_str = working_position_label(ss_idx, num_supersets) exercises = [] selected_focus_keys = set() for _attempt in range(4): # Select exercises (allow cross-modality so carries/planks # can land in rep supersets with their native duration) 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, allow_cross_modality=True, ) 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, allow_cross_modality=True, ) 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, allow_cross_modality=True, ) 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, ) if self._has_duplicate_focus_in_superset(exercises): continue selected_focus_keys = self._superset_focus_keys(exercises) if previous_focus_keys and selected_focus_keys and selected_focus_keys == previous_focus_keys: continue break if not exercises: continue if self._has_duplicate_focus_in_superset(exercises): logger.warning( "Allowing unresolved duplicate exercise focus in superset %d after retries", ss_idx + 1, ) if not selected_focus_keys: selected_focus_keys = self._superset_focus_keys(exercises) # Build exercise entries with per-exercise modality assignment. # Each exercise gets its native modality: duration-only exercises # (carries, planks) get duration even in a rep-based superset. exercise_entries = [] for i, ex in enumerate(exercises, start=1): if self._is_recovery_exercise(ex): logger.debug("Skipping recovery/stretch exercise %s in working superset", ex.name) continue entry = { 'exercise': ex, 'order': i, } # Determine this exercise's modality based on its capabilities # and the superset's preferred modality. ex_is_reps = getattr(ex, 'is_reps', False) ex_is_duration = getattr(ex, 'is_duration', False) if superset_is_duration: if ex_is_duration: # Matches superset preference use_duration = True elif ex_is_reps and not ex_is_duration: # Rep-only exercise in duration superset → use reps (native) use_duration = False else: use_duration = True else: if ex_is_reps: # Matches superset preference use_duration = False elif ex_is_duration and not ex_is_reps: # Duration-only exercise in rep superset → use duration (native) use_duration = True else: use_duration = False if use_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: entry['reps'] = self._pick_reps_for_exercise(ex, wt_params, workout_type) 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, clamped to workout type rep range apply_rep_volume_floor(exercise_entries, rounds, min_volume) # Clamp volume-floor-boosted reps so they don't exceed the workout type's rep range rep_max_clamp = wt_params.get('rep_max') rep_min_clamp = wt_params.get('rep_min') if rep_max_clamp or rep_min_clamp: for entry in exercise_entries: reps = entry.get('reps') if reps is not None: if rep_max_clamp and reps > rep_max_clamp: entry['reps'] = rep_max_clamp if rep_min_clamp and reps < rep_min_clamp: entry['reps'] = rep_min_clamp working_rest = working_rest_seconds( rest_between_rounds_override, wt_params.get('rest_between_rounds', 45), ) supersets.append({ 'name': f'{WORKING_PREFIX} Set {ss_idx + 1}', 'rounds': rounds, 'rest_between_rounds': working_rest, 'modality': 'duration' if superset_is_duration else 'reps', 'exercises': exercise_entries, }) if selected_focus_keys: previous_focus_keys = selected_focus_keys # Item #6: Modality consistency check — check what was actually assigned # (reps vs duration), not what the exercise is capable of. 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 if ex_entry.get('duration') is not None and ex_entry.get('reps') is None: 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 sort_entries_by_hr(exercises, is_early) return supersets @staticmethod def _exercise_focus_key(exercise): """Classify an exercise into a coarse focus key for variety checks.""" return focus_key_for_exercise(exercise) def _superset_focus_keys(self, exercises): """Return a set of coarse focus keys for a superset.""" return focus_keys_for_exercises(exercises) def _has_duplicate_focus_in_superset(self, exercises): """Prevent same focus from being repeated inside one working superset.""" return has_duplicate_focus(exercises) 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 count = section_exercise_count('cooldown', fitness_level, rng=random) # 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 exercise_entries = build_duration_entries( exercises, duration_min=cooldown_dur_min, duration_max=cooldown_dur_max, min_duration=GENERATION_RULES['min_duration']['value'], duration_multiple=GENERATION_RULES['duration_multiple']['value'], rng=random, ) return build_section_superset('Cool Down', 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, # clamped to the workout type's rep range to avoid violating rep ranges. min_volume = GENERATION_RULES['min_volume']['value'] rep_max_clamp = wt_params.get('rep_max') rep_min_clamp = wt_params.get('rep_min') for ss in workout_spec.get('supersets', []): if not ss.get('name', '').startswith(WORKING_PREFIX): 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) # Clamp to workout type rep range reps = entry.get('reps') if reps is not None: if rep_max_clamp and reps > rep_max_clamp: entry['reps'] = rep_max_clamp if rep_min_clamp and reps < rep_min_clamp: entry['reps'] = rep_min_clamp 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_PREFIX) ] 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_PREFIX) ] 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) duration_bias = self._clamp_duration_bias_for_type(duration_bias, workout_type) # 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'] max_working_exercises = self._max_working_exercises_for_type(workout_type) def _total_working_exercises(): return sum( len(ss.get('exercises', [])) for ss in supersets if ss.get('name', '').startswith(WORKING_PREFIX) ) # 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 ): if _total_working_exercises() >= max_working_exercises: break pad_attempts += 1 # Try adding exercises to existing working supersets first added = False for ss in supersets: if not ss.get('name', '').startswith(WORKING_PREFIX): continue if len(ss['exercises']) >= MAX_EXERCISES_PER_SUPERSET: continue if _total_working_exercises() >= max_working_exercises: break # 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, allow_cross_modality=True, ) if new_exercises: ex = new_exercises[0] if self._is_recovery_exercise(ex): continue new_order = len(ss['exercises']) + 1 entry = {'exercise': ex, 'order': new_order} # Per-exercise modality: use native modality ex_is_reps = getattr(ex, 'is_reps', False) ex_is_duration = getattr(ex, 'is_duration', False) use_duration = ( (ss_is_duration and ex_is_duration) or (not ex_is_reps and ex_is_duration) ) if use_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: entry['reps'] = self._pick_reps_for_exercise(ex, wt_params, workout_type) 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: remaining_capacity = max_working_exercises - _total_working_exercises() if remaining_capacity <= 0: break rounds = random.randint(*wt_params['rounds']) ex_count = random.randint(*wt_params['exercises_per_superset']) min_for_new_superset = GENERATION_RULES['min_exercises_per_superset']['value'] if remaining_capacity < min_for_new_superset: break # R8: Min 2 exercises ex_count = max(min_for_new_superset, ex_count) ex_count = min(ex_count, remaining_capacity) if ex_count <= 0: break # 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: working = [ current for current in supersets if current.get('name', '').startswith(WORKING_PREFIX) ] total_entries = sum(len(current.get('exercises', [])) for current in working) duration_entries = sum( len(current.get('exercises', [])) for current in working if current.get('modality') == 'duration' ) current_ratio = (duration_entries / total_entries) if total_entries else duration_bias if current_ratio < duration_bias - 0.05: ss_is_duration = True elif current_ratio > duration_bias + 0.05: 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, allow_cross_modality=True, ) if not exercises: break exercise_entries = [] for i, ex in enumerate(exercises, start=1): if self._is_recovery_exercise(ex): continue entry = {'exercise': ex, 'order': i} # Per-exercise modality: use native modality ex_is_reps = getattr(ex, 'is_reps', False) ex_is_duration = getattr(ex, 'is_duration', False) use_duration = ( (ss_is_duration and ex_is_duration) or (not ex_is_reps and ex_is_duration) ) if use_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: entry['reps'] = self._pick_reps_for_exercise(ex, wt_params, workout_type) 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 if not exercise_entries: continue # 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_PREFIX) ) new_superset = { 'name': f'{WORKING_PREFIX} Set {working_count + 1}', 'rounds': rounds, 'rest_between_rounds': max(15, int(wt_params.get('rest_between_rounds', 45) or 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 _max_working_exercises_for_type(self, workout_type): """Return the calibrated max working-exercise cap for this workout type.""" fallback = UNIVERSAL_RULES.get('max_exercises_per_workout', 30) if not workout_type: return fallback wt_key = _normalize_type_key(getattr(workout_type, 'name', '') or '') wt_rules = WORKOUT_TYPE_RULES.get(wt_key, {}) return wt_rules.get('max_exercises_per_session', fallback) @staticmethod def _workout_type_rules(workout_type): if not workout_type: return {} wt_key = _normalize_type_key(getattr(workout_type, 'name', '') or '') return WORKOUT_TYPE_RULES.get(wt_key, {}) @staticmethod def _extract_exercise_info_from_spec(workout_spec): """Extract (exercise_ids, exercise_names) from a workout spec dict.""" ids = set() names = set() for ss in workout_spec.get('supersets', []): for entry in ss.get('exercises', []): ex = entry.get('exercise') if ex is not None: ids.add(ex.pk) name = (ex.name or '').lower().strip() if name: names.add(name) # Also handle serialized specs (exercise_id / exercise_name keys) ex_id = entry.get('exercise_id') if ex_id is not None and ex is None: ids.add(ex_id) ex_name = (entry.get('exercise_name', '') or '').lower().strip() if ex_name: names.add(ex_name) return ids, names def _clamp_duration_bias_for_type(self, duration_bias, workout_type): wt_rules = self._workout_type_rules(workout_type) return clamp_duration_bias(duration_bias, wt_rules.get('duration_bias_range')) def _pick_reps_for_exercise(self, exercise, wt_params, workout_type): wt_rules = self._workout_type_rules(workout_type) return pick_reps_for_exercise( exercise, wt_params, wt_rules.get('rep_ranges', {}), rng=random, ) def _plan_superset_modalities(self, num_supersets, duration_bias, workout_type, is_strength_workout): wt_rules = self._workout_type_rules(workout_type) return plan_superset_modalities( num_supersets=num_supersets, duration_bias=duration_bias, duration_bias_range=wt_rules.get('duration_bias_range'), is_strength_workout=is_strength_workout, rng=random, ) @staticmethod def _entry_has_push(entry): ex = entry.get('exercise') if ex is None: return False patterns = (getattr(ex, 'movement_patterns', '') or '').lower() return 'push' in patterns @staticmethod def _entry_has_pull(entry): ex = entry.get('exercise') if ex is None: return False patterns = (getattr(ex, 'movement_patterns', '') or '').lower() return 'pull' in patterns def _enforce_compound_first_order(self, workout_spec, is_strength_workout=False): """Sort working supersets so compound-dominant work appears first.""" supersets = workout_spec.get('supersets', []) working_indices = [ i for i, ss in enumerate(supersets) if ss.get('name', '').startswith(WORKING_PREFIX) ] if not working_indices: return def _is_compound_entry(entry): ex = entry.get('exercise') if ex is None: return False tier = getattr(ex, 'exercise_tier', None) return bool(getattr(ex, 'is_compound', False) and tier in ('primary', 'secondary')) working_sets = [supersets[i] for i in working_indices] for ss in working_sets: exercises = ss.get('exercises', []) exercises.sort( key=lambda entry: ( 0 if _is_compound_entry(entry) else 1, entry.get('order', 0), ) ) for idx, entry in enumerate(exercises, start=1): entry['order'] = idx pinned_first = None sortable_sets = working_sets if is_strength_workout and working_sets: # Preserve the first straight set for strength workouts. pinned_first = working_sets[0] sortable_sets = working_sets[1:] sortable_sets.sort( key=lambda ss: sum( 1 for entry in ss.get('exercises', []) if _is_compound_entry(entry) ), reverse=True, ) if pinned_first is not None: working_sets = [pinned_first] + sortable_sets else: working_sets = sortable_sets for idx, ss in enumerate(working_sets, start=1): ss['name'] = f'{WORKING_PREFIX} Set {idx}' for idx, original_idx in enumerate(working_indices): supersets[original_idx] = working_sets[idx] def _select_pull_replacement(self, target_muscles, is_duration_based, prefer_weighted): """Pick a pull-pattern replacement that still respects user constraints. No hard modality filter — the caller assigns modality per-exercise. """ fitness_level = getattr(self.preference, 'fitness_level', None) def _candidate_pool(muscle_groups): qs = self.exercise_selector._get_filtered_queryset( muscle_groups=muscle_groups, is_duration_based=None, # No hard modality filter fitness_level=fitness_level, ).filter(movement_patterns__icontains='pull') qs = qs.exclude(name__iregex=r'\bstretch(ing|es|ed)?\b') qs = qs.exclude(movement_patterns__icontains='mobility - static') qs = qs.exclude(movement_patterns__icontains='static stretch') qs = qs.exclude(movement_patterns__icontains='yoga') qs = qs.exclude(movement_patterns__icontains='cooldown') # Rebalance replacements should not introduce orphan side-specific entries. qs = qs.filter(Q(side__isnull=True) | Q(side='')) return list(qs[:50]) candidates = _candidate_pool(target_muscles) if not candidates and target_muscles: candidates = _candidate_pool([]) if not candidates: return None if prefer_weighted: weighted = [c for c in candidates if getattr(c, 'is_weight', False)] if weighted: candidates = weighted return random.choice(candidates) @staticmethod def _is_recovery_exercise(ex): """Return True if exercise looks like warm-up/cool-down recovery content.""" return is_recovery_exercise(ex) def _rebalance_push_pull( self, workout_spec, target_muscles, wt_params, is_strength_workout, workout_type=None, ): """Replace push-only entries with pull entries until ratio is compliant.""" working = [ ss for ss in workout_spec.get('supersets', []) if ss.get('name', '').startswith(WORKING_PREFIX) ] if not working: return push_count = 0 pull_count = 0 replaceable = [] for ss in working: for entry in ss.get('exercises', []): has_push = self._entry_has_push(entry) has_pull = self._entry_has_pull(entry) if has_push and has_pull: # Dual pattern — count as neutral to avoid double-counting pass elif has_push: push_count += 1 elif has_pull: pull_count += 1 if has_push and not has_pull: replaceable.append((ss, entry)) if push_count == 0: return if pull_count == 0 and push_count <= 2: return if pull_count >= push_count: return replacements_needed = max(1, math.ceil((push_count - pull_count) / 2)) if not replaceable: return min_duration = GENERATION_RULES['min_duration']['value'] duration_mult = GENERATION_RULES['duration_multiple']['value'] prefer_weighted = is_strength_workout for ss, entry in reversed(replaceable): if replacements_needed <= 0: break is_duration_based = ss.get('modality') == 'duration' replacement = self._select_pull_replacement( target_muscles=target_muscles, is_duration_based=is_duration_based, prefer_weighted=prefer_weighted, ) if replacement is None: continue old_ex = entry.get('exercise') entry['exercise'] = replacement # Per-exercise modality: use the replacement's native modality repl_is_reps = getattr(replacement, 'is_reps', False) repl_is_duration = getattr(replacement, 'is_duration', False) use_duration = ( (is_duration_based and repl_is_duration) or (not repl_is_reps and repl_is_duration) ) if use_duration: entry.pop('reps', None) entry.pop('weight', None) if entry.get('duration') is None: duration = random.randint( wt_params['duration_min'], wt_params['duration_max'], ) entry['duration'] = max( min_duration, round(duration / duration_mult) * duration_mult, ) else: entry.pop('duration', None) if entry.get('reps') is None: entry['reps'] = self._pick_reps_for_exercise( replacement, wt_params, workout_type, ) if getattr(replacement, 'is_weight', False): entry['weight'] = None else: entry.pop('weight', None) if old_ex is not None: self.exercise_selector.used_exercise_ids.discard(old_ex.pk) old_name = (getattr(old_ex, 'name', '') or '').lower().strip() if old_name: self.exercise_selector.used_exercise_names.discard(old_name) self.exercise_selector.used_exercise_ids.add(replacement.pk) replacement_name = (replacement.name or '').lower().strip() if replacement_name: self.exercise_selector.used_exercise_names.add(replacement_name) replacements_needed -= 1 def _get_final_conformance_violations(self, workout_spec, workout_type, target_muscles): """Validate final workout against rules + user-preference conformance.""" workout_type_name = workout_type.name if workout_type else 'unknown_type' goal = getattr(self.preference, 'primary_goal', 'general_fitness') violations = validate_workout(workout_spec, workout_type_name, goal) violations.extend( self._validate_user_preference_alignment(workout_spec, target_muscles) ) return violations def _validate_user_preference_alignment(self, workout_spec, target_muscles): """Validate that final selections still honor explicit user preferences.""" violations = [] supersets = workout_spec.get('supersets', []) all_exercises = [] working_exercises = [] for ss in supersets: is_working = ss.get('name', '').startswith(WORKING_PREFIX) for entry in ss.get('exercises', []): ex = entry.get('exercise') if ex is None: continue all_exercises.append(ex) if is_working: working_exercises.append(ex) if not all_exercises: return violations exercise_ids = {ex.pk for ex in all_exercises} ex_name_map = {ex.pk: (ex.name or f'Exercise {ex.pk}') for ex in all_exercises} # 1) Excluded exercises must never appear. excluded_ids = set( self.preference.excluded_exercises.values_list('pk', flat=True) ) excluded_present = sorted(exercise_ids & excluded_ids) if excluded_present: names = ', '.join(ex_name_map.get(ex_id, str(ex_id)) for ex_id in excluded_present[:3]) violations.append(RuleViolation( rule_id='preference_excluded_exercise', severity='error', message=f'Workout includes excluded exercise(s): {names}.', actual_value=len(excluded_present), )) # 2) Equipment requirements must stay within user-available equipment. available_equipment_ids = set( self.preference.available_equipment.values_list('pk', flat=True) ) equipment_requirements = {} for ex_id, eq_id in WorkoutEquipment.objects.filter( exercise_id__in=exercise_ids, ).values_list('exercise_id', 'equipment_id'): equipment_requirements.setdefault(ex_id, set()).add(eq_id) equipment_mismatch = [] for ex_id, required_equipment in equipment_requirements.items(): if not available_equipment_ids: equipment_mismatch.append(ex_id) continue if not required_equipment.issubset(available_equipment_ids): equipment_mismatch.append(ex_id) if equipment_mismatch: names = ', '.join(ex_name_map.get(ex_id, str(ex_id)) for ex_id in equipment_mismatch[:3]) violations.append(RuleViolation( rule_id='preference_equipment_mismatch', severity='error', message=f'Workout includes equipment beyond user preference: {names}.', actual_value=len(equipment_mismatch), )) # 3) Working exercises should mostly align with target muscles. normalized_targets = { normalize_muscle_name(m) for m in (target_muscles or []) if m } if normalized_targets and working_exercises: working_ids = {ex.pk for ex in working_exercises} exercise_muscles = {} for ex_id, muscle_name in ExerciseMuscle.objects.filter( exercise_id__in=working_ids, ).values_list('exercise_id', 'muscle__name'): exercise_muscles.setdefault(ex_id, set()).add( normalize_muscle_name(muscle_name), ) evaluated = 0 matched = 0 for ex in working_exercises: ex_muscles = exercise_muscles.get(ex.pk) if not ex_muscles: raw = getattr(ex, 'muscle_groups', '') or '' ex_muscles = { normalize_muscle_name(part.strip()) for part in raw.split(',') if part.strip() } if not ex_muscles: continue evaluated += 1 if ex_muscles & normalized_targets: matched += 1 if evaluated > 0: alignment = matched / evaluated min_alignment = 0.7 if alignment < min_alignment: violations.append(RuleViolation( rule_id='preference_target_muscle_alignment', severity='error', message=( f'Target-muscle alignment {alignment:.0%} is below ' f'required {min_alignment:.0%}.' ), actual_value=alignment, expected_range=(min_alignment, 1.0), )) return violations @staticmethod def _is_blocking_final_violation(violation): """Block only hard errors; warnings and info are advisory, not blocking.""" return violation.severity == 'error' 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() wt_key = _normalize_type_key(wt_name_lower) is_strength = wt_name_lower in STRENGTH_WORKOUT_TYPES is_hiit = wt_key == 'high_intensity_interval_training' is_cardio = wt_key == 'cardio' is_core = wt_key == 'core_training' is_flexibility = wt_key == 'flexibility' 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 elif is_hiit: # HIIT: favor high HR, compound, or duration-capable exercises hr = getattr(ex, 'hr_elevation_rating', None) or 0 if hr >= 5 or getattr(ex, 'is_compound', False) or getattr(ex, 'is_duration', False): matching_exercises += 1 elif is_cardio: # Cardio: favor duration-capable or high-HR exercises hr = getattr(ex, 'hr_elevation_rating', None) or 0 if getattr(ex, 'is_duration', False) or hr >= 5: matching_exercises += 1 elif is_core: # Core: check if exercise targets core muscles muscles = (getattr(ex, 'muscle_groups', '') or '').lower() patterns = (getattr(ex, 'movement_patterns', '') or '').lower() if any(tok in muscles for tok in ('core', 'abs', 'oblique')): matching_exercises += 1 elif 'core' in patterns or 'anti' in patterns: matching_exercises += 1 elif is_flexibility: # Flexibility: favor duration-based, stretch/mobility exercises patterns = (getattr(ex, 'movement_patterns', '') or '').lower() if getattr(ex, 'is_duration', False) or any( tok in patterns for tok in ('stretch', 'mobility', 'yoga', 'flexibility') ): matching_exercises += 1 else: # Unknown type -- count all as matching (no false negatives) 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, [])