8e14fd5774
Additional fixes from parallel hardening streams: - exercise/serializers: remove unused WorkoutEquipment import, add prefetch docs - generator/serializers: N+1 fix in GeneratedWorkoutDetailSerializer (inline workout dict, prefetch-aware supersets) - generator/services/plan_builder: eliminate redundant .save() after .create() via single create_kwargs dict - generator/services/workout_generator: proper type-match validation for HIIT/cardio/core/flexibility; fix diversify type count to account for removed entry - generator/views: request-level caching for get_registered_user helper; prefetch chain for accept_workout - superset/serializers: guard against dangling FK in SupersetExerciseSerializer - workout/helpers: use prefetched data instead of re-querying per superset Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
3092 lines
127 KiB
Python
3092 lines
127 KiB
Python
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 = 4
|
||
|
||
|
||
# ======================================================================
|
||
# 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, [])
|