Exercise Selection Algorithms with Equipment Constraints

-- Core exercise catalog
CREATE TABLE exercises (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    description TEXT,
    difficulty_level INTEGER CHECK (difficulty_level BETWEEN 1 AND 5),
    movement_pattern VARCHAR(50),  -- 'push', 'pull', 'hinge', 'squat', 'carry'
    is_compound BOOLEAN DEFAULT false,
    calories_per_minute DECIMAL(4,2)
);

-- Equipment available in the system
CREATE TABLE equipment (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    category VARCHAR(50),  -- 'free_weights', 'machines', 'cables', 'bodyweight'
    is_common_home BOOLEAN DEFAULT false
);

-- Many-to-many: exercises require equipment
CREATE TABLE exercise_equipment (
    exercise_id INTEGER REFERENCES exercises(id),
    equipment_id INTEGER REFERENCES equipment(id),
    is_required BOOLEAN DEFAULT true,  -- false = optional/alternative
    PRIMARY KEY (exercise_id, equipment_id)
);

-- Muscle groups with hierarchy
CREATE TABLE muscle_groups (
    id SERIAL PRIMARY KEY,
    name VARCHAR(50) NOT NULL,
    parent_id INTEGER REFERENCES muscle_groups(id),
    body_region VARCHAR(20)  -- 'upper', 'lower', 'core'
);

-- Many-to-many: exercises target muscles
CREATE TABLE exercise_muscles (
    exercise_id INTEGER REFERENCES exercises(id),
    muscle_group_id INTEGER REFERENCES muscle_groups(id),
    activation_level VARCHAR(10),  -- 'primary', 'secondary', 'stabilizer'
    PRIMARY KEY (exercise_id, muscle_group_id)
);

-- User profiles and their equipment
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    fitness_level INTEGER CHECK (fitness_level BETWEEN 1 AND 5),
    primary_goal VARCHAR(50)  -- 'strength', 'hypertrophy', 'endurance', 'weight_loss'
);

CREATE TABLE user_equipment (
    user_id INTEGER REFERENCES users(id),
    equipment_id INTEGER REFERENCES equipment(id),
    PRIMARY KEY (user_id, equipment_id)
);

-- Track exercise history for personalization
CREATE TABLE exercise_history (
    id SERIAL PRIMARY KEY,
    user_id INTEGER REFERENCES users(id),
    exercise_id INTEGER REFERENCES exercises(id),
    performed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    sets_completed INTEGER,
    rating INTEGER CHECK (rating BETWEEN 1 AND 5)  -- user satisfaction
);

from dataclasses import dataclass
from typing import Set, List
import psycopg2
from psycopg2.extras import RealDictCursor

@dataclass
class Exercise:
    id: int
    name: str
    difficulty_level: int
    movement_pattern: str
    is_compound: bool
    required_equipment: Set[int]
    optional_equipment: Set[int]
    primary_muscles: Set[int]
    secondary_muscles: Set[int]

class ExerciseRepository:
    def __init__(self, db_connection):
        self.conn = db_connection

    def get_all_exercises(self) -> List[Exercise]:
        """Load exercises with their equipment and muscle requirements."""
        with self.conn.cursor(cursor_factory=RealDictCursor) as cur:
            # Get base exercise data
            cur.execute("""
                SELECT id, name, difficulty_level, movement_pattern, is_compound
                FROM exercises
            """)
            exercises_data = {row['id']: row for row in cur.fetchall()}

            # Get equipment requirements
            cur.execute("""
                SELECT exercise_id, equipment_id, is_required
                FROM exercise_equipment
            """)
            for row in cur.fetchall():
                ex_id = row['exercise_id']
                if ex_id not in exercises_data:
                    continue
                if 'required_equipment' not in exercises_data[ex_id]:
                    exercises_data[ex_id]['required_equipment'] = set()
                    exercises_data[ex_id]['optional_equipment'] = set()

                if row['is_required']:
                    exercises_data[ex_id]['required_equipment'].add(row['equipment_id'])
                else:
                    exercises_data[ex_id]['optional_equipment'].add(row['equipment_id'])

            # Get muscle targets
            cur.execute("""
                SELECT exercise_id, muscle_group_id, activation_level
                FROM exercise_muscles
            """)
            for row in cur.fetchall():
                ex_id = row['exercise_id']
                if ex_id not in exercises_data:
                    continue
                if 'primary_muscles' not in exercises_data[ex_id]:
                    exercises_data[ex_id]['primary_muscles'] = set()
                    exercises_data[ex_id]['secondary_muscles'] = set()

                if row['activation_level'] == 'primary':
                    exercises_data[ex_id]['primary_muscles'].add(row['muscle_group_id'])
                else:
                    exercises_data[ex_id]['secondary_muscles'].add(row['muscle_group_id'])

            return [
                Exercise(
                    id=data['id'],
                    name=data['name'],
                    difficulty_level=data['difficulty_level'],
                    movement_pattern=data.get('movement_pattern'),
                    is_compound=data.get('is_compound', False),
                    required_equipment=data.get('required_equipment', set()),
                    optional_equipment=data.get('optional_equipment', set()),
                    primary_muscles=data.get('primary_muscles', set()),
                    secondary_muscles=data.get('secondary_muscles', set())
                )
                for data in exercises_data.values()
            ]

def filter_by_equipment(
    exercises: List[Exercise],
    available_equipment: Set[int]
) -> List[Exercise]:
    """
    Filter exercises to only those possible with available equipment.

    An exercise is possible if ALL required equipment is available.
    Optional equipment may enhance the exercise but isn't necessary.
    """
    valid_exercises = []

    for exercise in exercises:
        # Check if all required equipment is available
        if exercise.required_equipment.issubset(available_equipment):
            valid_exercises.append(exercise)

    return valid_exercises

from typing import Dict, List, Set

# Equipment substitution rules
EQUIPMENT_ALTERNATIVES: Dict[int, List[Set[int]]] = {
    # Barbell (id=1) can be replaced by dumbbells (id=2) for many exercises
    1: [{2}],
    # Cable machine (id=5) can be replaced by resistance bands (id=6)
    5: [{6}],
    # Bench (id=3) can be replaced by stability ball (id=7) or floor
    3: [{7}, set()],  # empty set = bodyweight alternative
}

def filter_with_alternatives(
    exercises: List[Exercise],
    available_equipment: Set[int],
    alternatives: Dict[int, List[Set[int]]] = EQUIPMENT_ALTERNATIVES
) -> List[Exercise]:
    """
    Filter exercises considering equipment substitutions.

    Returns exercises where required equipment is available OR
    a valid alternative combination exists.
    """
    valid_exercises = []

    for exercise in exercises:
        if exercise.required_equipment.issubset(available_equipment):
            valid_exercises.append(exercise)
            continue

        # Check if alternatives can fill the gap
        missing = exercise.required_equipment - available_equipment

        can_substitute = True
        for missing_equip in missing:
            if missing_equip not in alternatives:
                can_substitute = False
                break

            # Check if any alternative set is available
            alt_available = any(
                alt_set.issubset(available_equipment)
                for alt_set in alternatives[missing_equip]
            )
            if not alt_available:
                can_substitute = False
                break

        if can_substitute:
            valid_exercises.append(exercise)

    return valid_exercises

from collections import defaultdict
from typing import List, Dict, Tuple
import numpy as np

@dataclass
class MuscleGroupTarget:
    muscle_id: int
    name: str
    target_sets: int  # weekly target
    current_sets: int = 0

class MuscleBalancer:
    """Ensures balanced muscle group coverage in exercise selection."""

    # Recommended weekly sets by muscle group for hypertrophy
    DEFAULT_TARGETS = {
        'chest': 12,
        'back': 15,
        'shoulders': 12,
        'biceps': 9,
        'triceps': 9,
        'quadriceps': 12,
        'hamstrings': 10,
        'glutes': 10,
        'calves': 8,
        'core': 10,
    }

    def __init__(self, muscle_targets: Dict[str, int] = None):
        self.targets = muscle_targets or self.DEFAULT_TARGETS

    def calculate_coverage_score(
        self,
        selected_exercises: List[Exercise],
        candidate: Exercise,
        sets_per_exercise: int = 3
    ) -> float:
        """
        Score how well adding a candidate exercise improves muscle coverage.

        Higher scores indicate the exercise fills gaps in current coverage.
        Lower scores indicate redundant muscle targeting.
        """
        # Calculate current coverage from selected exercises
        current_coverage = defaultdict(int)
        for ex in selected_exercises:
            for muscle_id in ex.primary_muscles:
                current_coverage[muscle_id] += sets_per_exercise
            for muscle_id in ex.secondary_muscles:
                current_coverage[muscle_id] += sets_per_exercise * 0.5

        # Calculate what candidate would add
        candidate_contribution = defaultdict(float)
        for muscle_id in candidate.primary_muscles:
            candidate_contribution[muscle_id] = sets_per_exercise
        for muscle_id in candidate.secondary_muscles:
            candidate_contribution[muscle_id] = sets_per_exercise * 0.5

        # Score based on how much candidate fills gaps vs adds redundancy
        score = 0.0
        for muscle_id, contribution in candidate_contribution.items():
            current = current_coverage.get(muscle_id, 0)
            target = self.targets.get(muscle_id, 10)

            if current < target:
                # Reward filling gaps (diminishing returns near target)
                gap = target - current
                filled = min(contribution, gap)
                score += filled * (gap / target)  # higher weight for bigger gaps
            else:
                # Penalize over-targeting (but don't go negative)
                score -= contribution * 0.3

        return max(0, score)

    def get_coverage_report(
        self,
        selected_exercises: List[Exercise],
        sets_per_exercise: int = 3
    ) -> Dict[str, Tuple[int, int, float]]:
        """
        Generate a coverage report showing current vs target sets.

        Returns dict of {muscle_name: (current_sets, target_sets, percentage)}
        """
        coverage = defaultdict(int)
        for ex in selected_exercises:
            for muscle_id in ex.primary_muscles:
                coverage[muscle_id] += sets_per_exercise
            for muscle_id in ex.secondary_muscles:
                coverage[muscle_id] += int(sets_per_exercise * 0.5)

        report = {}
        for muscle_name, target in self.targets.items():
            current = coverage.get(muscle_name, 0)
            percentage = (current / target * 100) if target > 0 else 0
            report[muscle_name] = (current, target, percentage)

        return report

from dataclasses import dataclass
from typing import List, Dict, Optional
import numpy as np

@dataclass
class UserProfile:
    user_id: int
    fitness_level: int  # 1-5
    primary_goal: str  # 'strength', 'hypertrophy', 'endurance', 'weight_loss'
    available_equipment: Set[int]
    exercise_history: Dict[int, float]  # exercise_id -> avg_rating
    workout_frequency: int  # days per week

class ExerciseScorer:
    """Score and rank exercises based on user profile and goals."""

    # Goal-specific weights for different exercise attributes
    GOAL_WEIGHTS = {
        'strength': {
            'compound_bonus': 2.0,
            'difficulty_match': 1.5,
            'volume_preference': 'low',  # fewer reps, heavier weight
        },
        'hypertrophy': {
            'compound_bonus': 1.5,
            'difficulty_match': 1.0,
            'volume_preference': 'moderate',
        },
        'endurance': {
            'compound_bonus': 0.5,
            'difficulty_match': 0.5,
            'volume_preference': 'high',
        },
        'weight_loss': {
            'compound_bonus': 1.8,  # compound movements burn more calories
            'difficulty_match': 0.8,
            'volume_preference': 'high',
        },
    }

    def score_exercise(
        self,
        exercise: Exercise,
        user: UserProfile,
        muscle_balance_score: float,
        novelty_weight: float = 0.3
    ) -> float:
        """
        Calculate composite score for an exercise given user profile.

        Combines:
        - Goal alignment (compound vs isolation, difficulty)
        - Muscle balance contribution
        - User history (ratings, novelty)
        - Difficulty appropriateness
        """
        goal_config = self.GOAL_WEIGHTS.get(user.primary_goal, self.GOAL_WEIGHTS['hypertrophy'])

        scores = []
        weights = []

        # 1. Goal alignment score
        goal_score = 0.0
        if exercise.is_compound:
            goal_score += goal_config['compound_bonus']
        else:
            goal_score += 0.5  # isolation exercises still valuable
        scores.append(goal_score)
        weights.append(0.25)

        # 2. Difficulty match score
        difficulty_diff = abs(exercise.difficulty_level - user.fitness_level)
        difficulty_score = max(0, 5 - difficulty_diff) / 5.0
        difficulty_score *= goal_config['difficulty_match']
        scores.append(difficulty_score)
        weights.append(0.20)

        # 3. Muscle balance score (from MuscleBalancer)
        scores.append(muscle_balance_score)
        weights.append(0.30)

        # 4. User history score
        history_score = self._calculate_history_score(exercise, user, novelty_weight)
        scores.append(history_score)
        weights.append(0.25)

        # Weighted combination
        total_score = sum(s * w for s, w in zip(scores, weights))

        return total_score

    def _calculate_history_score(
        self,
        exercise: Exercise,
        user: UserProfile,
        novelty_weight: float
    ) -> float:
        """
        Balance user preferences with exercise variety.

        High-rated exercises get a boost, but we also reward novelty
        to prevent repetitive workouts.
        """
        if exercise.id in user.exercise_history:
            # User has done this exercise before
            avg_rating = user.exercise_history[exercise.id]
            rating_score = avg_rating / 5.0  # normalize to 0-1

            # Apply novelty penalty (done recently = lower novelty)
            novelty_score = 0.3  # base novelty for known exercises
        else:
            # New exercise - no rating but high novelty
            rating_score = 0.6  # neutral-positive assumption
            novelty_score = 1.0

        # Combine rating preference with novelty seeking
        return (rating_score * (1 - novelty_weight)) + (novelty_score * novelty_weight)

    def rank_exercises(
        self,
        exercises: List[Exercise],
        user: UserProfile,
        balancer: MuscleBalancer,
        selected: List[Exercise] = None,
        top_k: int = 10
    ) -> List[Tuple[Exercise, float]]:
        """
        Rank exercises by composite score, returning top K.
        """
        selected = selected or []
        scored = []

        for exercise in exercises:
            balance_score = balancer.calculate_coverage_score(selected, exercise)
            total_score = self.score_exercise(exercise, user, balance_score)
            scored.append((exercise, total_score))

        # Sort by score descending
        scored.sort(key=lambda x: x[1], reverse=True)

        return scored[:top_k]

from sklearn.ensemble import GradientBoostingRegressor
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
import numpy as np
from typing import List, Tuple
import pickle

class PersonalizedExerciseRanker:
    """
    Learn personalized exercise preferences from user history.

    Features: exercise attributes + user profile
    Target: user rating (1-5) or completion rate
    """

    def __init__(self):
        self.model = GradientBoostingRegressor(
            n_estimators=100,
            max_depth=4,
            learning_rate=0.1,
            random_state=42
        )
        self.scaler = StandardScaler()
        self.is_trained = False

    def extract_features(
        self,
        exercise: Exercise,
        user: UserProfile
    ) -> np.ndarray:
        """
        Convert exercise + user context into feature vector.
        """
        features = [
            # Exercise attributes
            exercise.difficulty_level,
            1 if exercise.is_compound else 0,
            len(exercise.primary_muscles),
            len(exercise.secondary_muscles),
            len(exercise.required_equipment),

            # User context
            user.fitness_level,
            1 if user.primary_goal == 'strength' else 0,
            1 if user.primary_goal == 'hypertrophy' else 0,
            1 if user.primary_goal == 'endurance' else 0,
            1 if user.primary_goal == 'weight_loss' else 0,
            user.workout_frequency,

            # Interaction features
            abs(exercise.difficulty_level - user.fitness_level),
            1 if exercise.id in user.exercise_history else 0,
            user.exercise_history.get(exercise.id, 3.0),  # default neutral rating
        ]

        return np.array(features)

    def train(
        self,
        training_data: List[Tuple[Exercise, UserProfile, float]]
    ) -> Dict[str, float]:
        """
        Train the model on historical exercise ratings.

        Args:
            training_data: List of (exercise, user, rating) tuples

        Returns:
            Training metrics (R2 score, MSE)
        """
        X = np.array([
            self.extract_features(ex, user)
            for ex, user, _ in training_data
        ])
        y = np.array([rating for _, _, rating in training_data])

        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=42
        )

        # Scale features
        X_train_scaled = self.scaler.fit_transform(X_train)
        X_test_scaled = self.scaler.transform(X_test)

        # Train model
        self.model.fit(X_train_scaled, y_train)
        self.is_trained = True

        # Evaluate
        train_score = self.model.score(X_train_scaled, y_train)
        test_score = self.model.score(X_test_scaled, y_test)

        return {
            'train_r2': train_score,
            'test_r2': test_score,
            'feature_importance': dict(zip(
                self._feature_names(),
                self.model.feature_importances_
            ))
        }

    def predict_rating(
        self,
        exercise: Exercise,
        user: UserProfile
    ) -> float:
        """Predict user's rating for an exercise."""
        if not self.is_trained:
            raise ValueError("Model must be trained before prediction")

        features = self.extract_features(exercise, user).reshape(1, -1)
        features_scaled = self.scaler.transform(features)

        prediction = self.model.predict(features_scaled)[0]
        return np.clip(prediction, 1.0, 5.0)  # keep in valid range

    def rank_exercises(
        self,
        exercises: List[Exercise],
        user: UserProfile,
        top_k: int = 10
    ) -> List[Tuple[Exercise, float]]:
        """Rank exercises by predicted rating."""
        predictions = [
            (ex, self.predict_rating(ex, user))
            for ex in exercises
        ]
        predictions.sort(key=lambda x: x[1], reverse=True)
        return predictions[:top_k]

    def _feature_names(self) -> List[str]:
        return [
            'difficulty_level', 'is_compound', 'n_primary_muscles',
            'n_secondary_muscles', 'n_required_equipment', 'user_fitness_level',
            'goal_strength', 'goal_hypertrophy', 'goal_endurance',
            'goal_weight_loss', 'workout_frequency', 'difficulty_mismatch',
            'previously_done', 'historical_rating'
        ]

    def save(self, path: str):
        """Persist trained model to disk."""
        with open(path, 'wb') as f:
            pickle.dump({'model': self.model, 'scaler': self.scaler}, f)

    def load(self, path: str):
        """Load trained model from disk."""
        with open(path, 'rb') as f:
            data = pickle.load(f)
            self.model = data['model']
            self.scaler = data['scaler']
            self.is_trained = True

class ExerciseSelector:
    """Complete exercise selection pipeline."""

    def __init__(
        self,
        repository: ExerciseRepository,
        balancer: MuscleBalancer,
        scorer: ExerciseScorer,
        ml_ranker: Optional[PersonalizedExerciseRanker] = None
    ):
        self.repository = repository
        self.balancer = balancer
        self.scorer = scorer
        self.ml_ranker = ml_ranker
        self._all_exercises = None

    def select_exercises(
        self,
        user: UserProfile,
        target_count: int = 6,
        use_ml: bool = True
    ) -> List[Exercise]:
        """
        Select balanced, personalized exercises for a user.

        Pipeline:
        1. Filter by equipment availability
        2. Score by goal alignment and balance
        3. Optionally re-rank with ML model
        4. Select top exercises ensuring variety
        """
        # Load exercises if not cached
        if self._all_exercises is None:
            self._all_exercises = self.repository.get_all_exercises()

        # Step 1: Hard constraint filtering
        feasible = filter_with_alternatives(
            self._all_exercises,
            user.available_equipment
        )

        if len(feasible) < target_count:
            # Not enough exercises available
            return feasible

        # Step 2 & 3: Rank by composite score or ML
        if use_ml and self.ml_ranker and self.ml_ranker.is_trained:
            ranked = self.ml_ranker.rank_exercises(feasible, user, top_k=target_count * 2)
        else:
            ranked = self.scorer.rank_exercises(
                feasible, user, self.balancer, top_k=target_count * 2
            )

        # Step 4: Greedy selection with balance checking
        selected = []
        for exercise, score in ranked:
            if len(selected) >= target_count:
                break

            # Check if this adds value to balance
            balance_score = self.balancer.calculate_coverage_score(selected, exercise)
            if balance_score > 0.1 or len(selected) < 3:  # always take first few
                selected.append(exercise)

        return selected