""" workout_list.py - Workout database, PDF parsing, and LLM planning logic """ import os import json import re from typing import Dict, List, Optional, Any from datetime import datetime import pymupdf as fitz from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema import BaseOutputParser from pydantic import BaseModel, Field from dotenv import load_dotenv # Load environment variables load_dotenv() # UI Data Models class ExerciseUI(BaseModel): """Exercise format for UI display""" name: str = Field(..., description="Exercise name") duration: str = Field(..., description="Duration (e.g., '03 Min')") sets: str = Field(..., description="Number of sets (e.g., '3 Sets')") calories: str = Field(..., description="Calories (e.g., '30 Calories')") video_url: Optional[str] = Field(None, description="URL to exercise video") video_status: Optional[str] = Field("pending", description="Status: cached, generated, pending, error") class WorkoutDayUI(BaseModel): """Daily workout format for UI""" day: str = Field(..., description="Day name (e.g., 'Monday')") title: str = Field(..., description="Workout title") duration: str = Field(..., description="Total duration (e.g., '40 Min')") intensity: str = Field(..., description="Intensity level") total_calories: str = Field(..., description="Total calories (e.g., '500 Calories')") category: str = Field(..., description="Workout category") warmup_exercises: List[ExerciseUI] = Field(default_factory=list, description="Warm-up exercises") main_exercises: List[ExerciseUI] = Field(default_factory=list, description="Main workout exercises") cooldown_exercises: List[ExerciseUI] = Field(default_factory=list, description="Cool-down exercises") class WeeklyPlanUI(BaseModel): """Complete 7-day plan for UI""" user_id: str blood_type: str age: Optional[int] = None bmi: Optional[float] = None bmi_category: Optional[str] = None body_shape: Optional[str] = None activity_level: Optional[str] = None main_goal: Optional[str] = None workout_level: Optional[str] = None focus_areas: Optional[List[str]] = None plan_summary: Dict[str, Any] = Field(default_factory=dict) weekly_workouts: List[WorkoutDayUI] videos_generated: int = Field(0, description="Number of videos generated") videos_cached: int = Field(0, description="Number of videos from cache") # User Profile and Fitness Considerations class UserProfile(BaseModel): """User profile for personalized workout generation""" user_id: str blood_type: str age: Optional[int] = None weight: Optional[float] = None height: Optional[float] = None body_shape: Optional[str] = Field(None, description="Body shape: Medium, Flabby, Skinny, Muscular") activity_level: Optional[str] = Field(None, description="Activity level: Sedentary, Lightly Active, Moderately Active, Very Active") workout_level: Optional[str] = Field(None, description="Workout level: Easy To Start, Break A Light Sweat, Challenging") main_goal: Optional[str] = Field(None, description="Main goal: Lose Weight, Gain Muscle, Stay Fit") desired_weight: Optional[float] = Field(None, description="Desired weight in kg") focus_areas: Optional[List[str]] = Field(default_factory=list, description="Focus areas: Arms, Upper Body, Abs, Butt, Legs") generate_videos: bool = Field(True, description="Whether to generate videos") def calculate_bmi(self) -> Optional[float]: if self.height and self.weight: height_m = self.height / 100 return round(self.weight / (height_m ** 2), 1) return None def get_bmi_category(self) -> Optional[str]: bmi = self.calculate_bmi() if bmi is None: return None if bmi < 18.5: return "Underweight" elif bmi < 25: return "Normal" elif bmi < 30: return "Overweight" else: return "Obese" def get_age_category(self) -> Optional[str]: if self.age is None: return None if self.age < 18: return "Youth" elif self.age < 30: return "Young Adult" elif self.age < 40: return "Adult" elif self.age < 50: return "Middle Age" elif self.age < 65: return "Senior" else: return "Elderly" def get_activity_multiplier(self) -> float: """Get calorie multiplier based on activity level""" multipliers = { "Sedentary": 1.2, "Lightly Active": 1.375, "Moderately Active": 1.55, "Very Active": 1.725 } return multipliers.get(self.activity_level, 1.375) def get_intensity_from_workout_level(self) -> str: """Map workout level to intensity""" mapping = { "Easy To Start": "Low", "Break A Light Sweat": "Medium", "Challenging": "High" } return mapping.get(self.workout_level, "Medium") def get_fitness_considerations(self) -> Dict[str, Any]: considerations = { "age_factors": [], "bmi_factors": [], "body_shape_factors": [], "activity_factors": [], "goal_factors": [], "focus_factors": [], "intensity_modifier": 1.0, "volume_modifier": 1.0, "recovery_modifier": 1.0, "calorie_modifier": 1.0 } # Age-based factors if self.age: if self.age < 18: considerations["age_factors"].append("Growing body - focus on technique and foundational movements") considerations["intensity_modifier"] = 0.8 considerations["recovery_modifier"] = 0.9 elif self.age >= 50: considerations["age_factors"].append("Increased recovery needs and joint care") considerations["intensity_modifier"] = 0.85 considerations["recovery_modifier"] = 1.2 elif self.age >= 65: considerations["age_factors"].append("Focus on mobility, balance, and functional fitness") considerations["intensity_modifier"] = 0.7 considerations["recovery_modifier"] = 1.4 # BMI-based factors bmi_category = self.get_bmi_category() if bmi_category == "Underweight": considerations["bmi_factors"].append("Focus on strength building and muscle gain") considerations["volume_modifier"] = 1.2 considerations["calorie_modifier"] = 1.15 elif bmi_category in ["Overweight", "Obese"]: considerations["bmi_factors"].append("Include more cardio and calorie-burning exercises") considerations["volume_modifier"] = 1.1 considerations["calorie_modifier"] = 1.25 # Body shape factors if self.body_shape: if self.body_shape == "Skinny": considerations["body_shape_factors"].append("Emphasize resistance training and compound movements") considerations["volume_modifier"] *= 1.15 elif self.body_shape == "Flabby": considerations["body_shape_factors"].append("Balance cardio with strength training for body recomposition") considerations["calorie_modifier"] *= 1.2 elif self.body_shape == "Muscular": considerations["body_shape_factors"].append("Maintain muscle mass with varied intensity and periodization") considerations["intensity_modifier"] *= 1.1 elif self.body_shape == "Medium": considerations["body_shape_factors"].append("Well-balanced workout approach for overall fitness") # Activity level factors if self.activity_level: if self.activity_level == "Sedentary": considerations["activity_factors"].append("Start gradually, focus on building consistency and habit formation") considerations["intensity_modifier"] *= 0.85 considerations["recovery_modifier"] *= 1.2 elif self.activity_level == "Lightly Active": considerations["activity_factors"].append("Build on current activity with progressive overload") considerations["intensity_modifier"] *= 0.95 elif self.activity_level == "Moderately Active": considerations["activity_factors"].append("Maintain current fitness with varied challenges") elif self.activity_level == "Very Active": considerations["activity_factors"].append("Can handle higher volume and intensity workouts") considerations["intensity_modifier"] *= 1.15 considerations["volume_modifier"] *= 1.1 # Goal-based factors if self.main_goal: if self.main_goal == "Lose Weight": considerations["goal_factors"].append("Prioritize calorie-burning exercises, HIIT, and metabolic conditioning") considerations["calorie_modifier"] *= 1.3 considerations["volume_modifier"] *= 1.15 elif self.main_goal == "Gain Muscle": considerations["goal_factors"].append("Focus on progressive resistance, compound lifts, and hypertrophy training") considerations["volume_modifier"] *= 1.2 considerations["recovery_modifier"] *= 1.1 elif self.main_goal == "Stay Fit": considerations["goal_factors"].append("Balanced approach to maintain overall fitness and health") # Weight difference factor if self.weight and self.desired_weight: weight_diff = self.weight - self.desired_weight if weight_diff > 10: considerations["goal_factors"].append(f"Significant weight loss target: {weight_diff:.1f} kg - sustained caloric deficit needed") considerations["calorie_modifier"] *= 1.2 elif weight_diff > 5: considerations["goal_factors"].append(f"Moderate weight loss target: {weight_diff:.1f} kg") considerations["calorie_modifier"] *= 1.15 elif weight_diff < -5: considerations["goal_factors"].append(f"Weight gain target: {abs(weight_diff):.1f} kg - focus on strength and nutrition") considerations["volume_modifier"] *= 1.15 # Focus area factors if self.focus_areas and len(self.focus_areas) > 0: focus_str = ", ".join(self.focus_areas) considerations["focus_factors"].append(f"Target areas: {focus_str}") considerations["focus_factors"].append("Include specific exercises targeting these muscle groups") return considerations # Workout Database and LLM Planning class WorkoutDatabase: """Manages workout database, PDF parsing, and LLM planning""" def __init__(self): self.parsed_workout_data = {} self.llm = None self.validation_llm = None def extract_exercises_with_font_detection(self, pdf_path: str) -> Dict[str, Dict[str, List[str]]]: """Extract exercises from PDF using font size detection""" try: doc = fitz.open(pdf_path) except Exception as e: print(f"Error opening PDF with fitz: {e}") raise exercises_by_blood_type = {"O": {}, "A": {}, "B": {}, "AB": {}} current_blood_type = None current_category = None for page in doc: try: blocks = page.get_text("dict")["blocks"] except Exception as e: print(f"Error getting text from page: {e}") continue for block in blocks: if "lines" not in block: continue for line in block["lines"]: for span in line["spans"]: text = span["text"].strip() font_size = span["size"] if not text: continue # Blood Type detection if font_size >= 15 and "Type" in text: blood_match = re.search(r'Type\s+([OABAB]{1,2})', text) if blood_match: current_blood_type = blood_match.group(1) current_category = None print(f"Found Blood Type: {current_blood_type}") # Category detection elif font_size >= 13 and font_size < 15 and current_blood_type: categories = [ "Strength", "Combat", "Boxing", "Conditioning", "HIIT", "Endurance", "Sports", "Mobility", "Recovery", "Yoga", "Pilates", "Tai Chi", "Stretching", "Breathing", "Meditation", "Foam Rolling", "Extras", "Stress", "Balance", "Cardio", "Calming", "Flexibility" ] is_category = any(cat.lower() in text.lower() for cat in categories) if is_category: current_category = text print(f" Category: {current_category}") # Exercise detection elif font_size >= 9 and font_size < 12 and current_blood_type and current_category: if text.startswith("•") or text.startswith("-") or text.startswith("*"): exercise_name = text.lstrip("•-* ").strip() if exercise_name and len(exercise_name) > 2: if current_category not in exercises_by_blood_type[current_blood_type]: exercises_by_blood_type[current_blood_type][current_category] = [] exercises_by_blood_type[current_blood_type][current_category].append(exercise_name) doc.close() return exercises_by_blood_type def load_and_parse_all_pdfs(self, folder_path: str = "workout"): """Load and parse all PDFs""" if not os.path.exists(folder_path): print(f"WARNING: Workout folder '{folder_path}' not found. Using empty database.") self.parsed_workout_data = {"O": {}, "A": {}, "B": {}, "AB": {}} return self.parsed_workout_data pdf_files = [f for f in os.listdir(folder_path) if f.lower().endswith('.pdf')] if not pdf_files: print(f"WARNING: No PDF files found in '{folder_path}'. Using empty database.") self.parsed_workout_data = {"O": {}, "A": {}, "B": {}, "AB": {}} return self.parsed_workout_data print(f"Found {len(pdf_files)} PDF files") master_db = {} yoga_db = {} for pdf_file in pdf_files: pdf_path = os.path.join(folder_path, pdf_file) print(f"Processing: {pdf_file}") try: if 'master' in pdf_file.lower() or 'database' in pdf_file.lower(): master_db = self.extract_exercises_with_font_detection(pdf_path) elif 'yoga' in pdf_file.lower() or 'meditation' in pdf_file.lower(): yoga_db = self.extract_exercises_with_font_detection(pdf_path) except Exception as e: print(f"Error processing {pdf_file}: {e}") continue # Merge databases self.parsed_workout_data = {"O": {}, "A": {}, "B": {}, "AB": {}} for blood_type in ["O", "A", "B", "AB"]: for category, exercises in master_db.get(blood_type, {}).items(): if category not in self.parsed_workout_data[blood_type]: self.parsed_workout_data[blood_type][category] = [] self.parsed_workout_data[blood_type][category].extend(exercises) for category, exercises in yoga_db.get(blood_type, {}).items(): if category not in self.parsed_workout_data[blood_type]: self.parsed_workout_data[blood_type][category] = [] self.parsed_workout_data[blood_type][category].extend(exercises) # Print summary for blood_type in ["O", "A", "B", "AB"]: total = sum(len(exs) for exs in self.parsed_workout_data[blood_type].values()) print(f"Blood Type {blood_type}: {total} exercises") return self.parsed_workout_data def create_personalized_context(self, user_profile: UserProfile) -> str: """Create comprehensive context for LLM with all user details""" blood_type = user_profile.blood_type if blood_type not in self.parsed_workout_data or not self.parsed_workout_data[blood_type]: return "WARNING: No exercise data found for this blood type. Use basic exercises." context_parts = [f"PERSONALIZED WORKOUT PLAN FOR USER: {user_profile.user_id}\n"] context_parts.append("="*80 + "\n\n") # User Profile context_parts.append("USER PROFILE:\n") context_parts.append(f"Blood Type: {blood_type}\n") if user_profile.age: context_parts.append(f"Age: {user_profile.age} ({user_profile.get_age_category()})\n") if user_profile.weight and user_profile.height: bmi = user_profile.calculate_bmi() context_parts.append(f"Weight: {user_profile.weight} kg, Height: {user_profile.height} cm\n") context_parts.append(f"BMI: {bmi} ({user_profile.get_bmi_category()})\n") if user_profile.body_shape: context_parts.append(f"Body Shape: {user_profile.body_shape}\n") if user_profile.activity_level: context_parts.append(f"Activity Level: {user_profile.activity_level}\n") context_parts.append(f"Activity Multiplier: {user_profile.get_activity_multiplier()}x\n") if user_profile.workout_level: context_parts.append(f"Workout Level: {user_profile.workout_level}\n") context_parts.append(f"Preferred Intensity: {user_profile.get_intensity_from_workout_level()}\n") if user_profile.main_goal: context_parts.append(f"Main Goal: {user_profile.main_goal}\n") if user_profile.desired_weight and user_profile.weight: weight_diff = user_profile.weight - user_profile.desired_weight context_parts.append(f"Desired Weight: {user_profile.desired_weight} kg (Difference: {weight_diff:+.1f} kg)\n") if user_profile.focus_areas and len(user_profile.focus_areas) > 0: context_parts.append(f"Focus Areas: {', '.join(user_profile.focus_areas)}\n") # Fitness considerations considerations = user_profile.get_fitness_considerations() context_parts.append(f"\nFITNESS CONSIDERATIONS:\n") context_parts.append(f"Intensity Modifier: {considerations['intensity_modifier']:.2f}x\n") context_parts.append(f"Volume Modifier: {considerations['volume_modifier']:.2f}x\n") context_parts.append(f"Recovery Modifier: {considerations['recovery_modifier']:.2f}x\n") context_parts.append(f"Calorie Modifier: {considerations['calorie_modifier']:.2f}x\n") if considerations['age_factors']: context_parts.append(f"\nAge Factors:\n") for factor in considerations['age_factors']: context_parts.append(f" - {factor}\n") if considerations['bmi_factors']: context_parts.append(f"\nBMI Factors:\n") for factor in considerations['bmi_factors']: context_parts.append(f" - {factor}\n") if considerations['body_shape_factors']: context_parts.append(f"\nBody Shape Factors:\n") for factor in considerations['body_shape_factors']: context_parts.append(f" - {factor}\n") if considerations['activity_factors']: context_parts.append(f"\nActivity Factors:\n") for factor in considerations['activity_factors']: context_parts.append(f" - {factor}\n") if considerations['goal_factors']: context_parts.append(f"\nGoal Factors:\n") for factor in considerations['goal_factors']: context_parts.append(f" - {factor}\n") if considerations['focus_factors']: context_parts.append(f"\nFocus Area Requirements:\n") for factor in considerations['focus_factors']: context_parts.append(f" - {factor}\n") # Available exercises context_parts.append("\n" + "="*80 + "\n") context_parts.append("AVAILABLE EXERCISES DATABASE - YOU MUST USE THESE EXACT NAMES:\n") context_parts.append("="*80 + "\n") total_exercises = 0 for category, exercises in self.parsed_workout_data[blood_type].items(): if exercises: context_parts.append(f"\n{category.upper()}:\n") for ex in exercises: context_parts.append(f" • {ex}\n") total_exercises += 1 context_parts.append(f"\nTOTAL AVAILABLE EXERCISES: {total_exercises}\n") context_parts.append("\nIMPORTANT: Use ONLY the exercise names listed above. Do not invent new exercises.\n") return "".join(context_parts) # Create personalized prompt template def create_personalized_prompt(self) -> ChatPromptTemplate: """Create personalized workout prompt with focus area integration""" template = """You are an expert fitness trainer creating a 7-day personalized workout plan. {personalized_context} CRITICAL RULES - MUST FOLLOW EXACTLY: - You MUST use EXACT exercise names from the available exercises database above - Do NOT invent or create new exercise names - Only use exercises listed in the available exercises section - If an exercise category has limited options, use what's available - PRIORITIZE exercises that target the user's focus areas when specified - Apply all modifiers (intensity, volume, calorie, recovery) to create appropriate workouts - Respect the user's workout level preference (Easy To Start, Break A Light Sweat, Challenging) - Align workouts with the user's main goal (Lose Weight, Gain Muscle, Stay Fit) Create a workout plan in this EXACT JSON format: {{ "Monday": {{ "title": "Workout Title Based on Available Exercises and User Goals", "category": "Category from Available Exercises", "intensity": "Low/Medium/High (based on workout level and modifiers)", "duration": "XX Min", "total_calories": "XXX", "warmup": [ {{"name": "EXACT_EXERCISE_NAME_FROM_DATABASE", "duration": "XX Min", "sets": "X Sets", "calories": "XX"}}, {{"name": "EXACT_EXERCISE_NAME_FROM_DATABASE", "duration": "XX Min", "sets": "X Sets", "calories": "XX"}} ], "main": [ {{"name": "EXACT_EXERCISE_NAME_FROM_DATABASE", "duration": "XX Min", "sets": "X Sets", "calories": "XX"}}, {{"name": "EXACT_EXERCISE_NAME_FROM_DATABASE", "duration": "XX Min", "sets": "X Sets", "calories": "XX"}}, {{"name": "EXACT_EXERCISE_NAME_FROM_DATABASE", "duration": "XX Min", "sets": "X Sets", "calories": "XX"}} ], "cooldown": [ {{"name": "EXACT_EXERCISE_NAME_FROM_DATABASE", "duration": "XX Min", "sets": "X Sets", "calories": "XX"}}, {{"name": "EXACT_EXERCISE_NAME_FROM_DATABASE", "duration": "XX Min", "sets": "X Sets", "calories": "XX"}} ] }}, "Tuesday": {{ ... }}, ... for all 7 days }} REQUIREMENTS: - Use ONLY exercises from the available database - this is mandatory - Adapt intensity based on age, BMI, body shape, and activity level factors - Include exercises targeting the user's focus areas (if specified) in main workouts - Apply calorie modifiers for realistic calorie calculations based on goal - Apply volume modifiers to set counts and exercise selection - Vary workout categories across the week for balanced training - Format durations as "XX Min" - Format sets as "X Sets" - Calculate realistic calories using the calorie modifier Generate the complete 7-day plan using ONLY the available exercises:""" return ChatPromptTemplate.from_template(template) class WorkoutPlanOutputParser(BaseOutputParser): """Parse LLM output""" def parse(self, text: str) -> Dict: try: json_match = re.search(r'\{[\s\S]*\}', text) if json_match: return json.loads(json_match.group(0)) raise ValueError("No JSON found") except json.JSONDecodeError as e: print(f"JSON error: {e}") raise def initialize_llms(self): """Initialize LLM instances""" self.llm = ChatOpenAI( model="gpt-4o-mini", temperature=0.7, openai_api_key=os.getenv("OPENAI_API_KEY") ) self.validation_llm = ChatOpenAI( model="gpt-4o-mini", temperature=0.3, openai_api_key=os.getenv("OPENAI_API_KEY") ) return self.llm, self.validation_llm # Helper Functions def parse_duration_to_minutes(duration_str: str) -> int: """Parse duration string like '10 Min' to integer minutes""" match = re.search(r'(\d+)', duration_str) return int(match.group(1)) if match else 0 def parse_calories(calories_str: str) -> int: """Parse calories string like '120 Calories' to integer""" match = re.search(r'(\d+)', calories_str) return int(match.group(1)) if match else 0 def normalize_blood_type(blood_type: str) -> str: """Normalize blood type to uppercase and handle +/- signs""" blood_type = blood_type.strip().upper() return blood_type.split("+")[0].split("-")[0]