help@rskworld.in +91 93305 39277
RSK World
  • Home
  • Development
    • Web Development
    • Mobile Apps
    • Software
    • Games
    • Project
  • Technologies
    • Data Science
    • AI Development
    • Cloud Development
    • Blockchain
    • Cyber Security
    • Dev Tools
    • Testing Tools
  • Blog
  • About
  • Contact

Theme Settings

Color Scheme
Display Options
Font Size
100%
Back to Project
RSK World
fitness-coach-bot
/
utils
RSK World
fitness-coach-bot
Fitness Coach Bot - Python + Flask + SQLAlchemy + Workout Plans + Exercise Guidance + Health Tracking + AI Fitness Coach
utils
  • __pycache__
  • __init__.py965 B
  • ai_workout_generator.py18 KB
  • analytics_engine.py27.8 KB
  • fitness_coach.py13.3 KB
  • gamification_system.py30.4 KB
  • nutrition_ai.py23.8 KB
  • smart_recovery.py11 KB
  • social_features.py19.4 KB
  • voice_coach.py10 KB
  • wearable_integration.py27 KB
  • workout_buddy_matcher.py9.5 KB
app.pyai_workout_generator.pywearable_integration.py
app.py
Raw Download
Find: Go to:
"""
Fitness Coach Bot - A comprehensive fitness coaching chatbot
Author: RSK World (https://rskworld.in)
Founded by: Molla Samser
Designer & Tester: Rima Khatun
Contact: help@rskworld.in, +91 93305 39277
Year: 2026
Description: Fitness chatbot for workout plans, exercise guidance, and health tracking
"""

from flask import Flask, render_template, request, jsonify, session
from datetime import datetime
import json
import os
import asyncio
from models.fitness_models import db, User, WorkoutPlan, Exercise, Progress, HealthTip
from utils.fitness_coach import FitnessCoach
from utils.ai_workout_generator import AIWorkoutGenerator
from utils.nutrition_ai import NutritionAI
from utils.social_features import SocialFeatures
from utils.analytics_engine import AdvancedAnalytics
from utils.gamification_system import GamificationSystem
from utils.wearable_integration import WearableIntegration, DeviceType
from utils.voice_coach import VoiceCoach
from utils.workout_buddy_matcher import WorkoutBuddyMatcher
from utils.smart_recovery import SmartRecovery

app = Flask(__name__)
app.config['SECRET_KEY'] = 'fitness-coach-bot-2026-rskworld'
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///fitness_coach.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

# Initialize db with app
db.init_app(app)
fitness_coach = FitnessCoach()
ai_workout_generator = AIWorkoutGenerator()
nutrition_ai = NutritionAI()
social_features = SocialFeatures()
analytics_engine = AdvancedAnalytics()
gamification_system = GamificationSystem()
wearable_integration = WearableIntegration()
voice_coach = VoiceCoach()
buddy_matcher = WorkoutBuddyMatcher()
smart_recovery = SmartRecovery()

@app.route('/')
def index():
    """Main chatbot interface"""
    return render_template('index.html')

@app.route('/api/chat', methods=['POST'])
def chat():
    """Handle chat messages and provide fitness coaching responses"""
    try:
        data = request.get_json()
        user_message = data.get('message', '')
        user_id = session.get('user_id', 'anonymous')
        
        # Get AI response
        response = fitness_coach.get_response(user_message, user_id)
        
        return jsonify({
            'success': True,
            'response': response,
            'timestamp': datetime.now().isoformat()
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/workout-plans', methods=['GET'])
def get_workout_plans():
    """Get available workout plans"""
    try:
        plans = WorkoutPlan.query.all()
        return jsonify({
            'success': True,
            'plans': [plan.to_dict() for plan in plans]
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/exercises', methods=['GET'])
def get_exercises():
    """Get exercises by category or muscle group"""
    try:
        category = request.args.get('category')
        muscle_group = request.args.get('muscle_group')
        
        query = Exercise.query
        if category:
            query = query.filter_by(category=category)
        if muscle_group:
            query = query.filter_by(muscle_group=muscle_group)
            
        exercises = query.all()
        return jsonify({
            'success': True,
            'exercises': [exercise.to_dict() for exercise in exercises]
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/progress', methods=['POST'])
def save_progress():
    """Save user workout progress"""
    try:
        data = request.get_json()
        user_id = session.get('user_id', 'anonymous')
        
        progress = Progress(
            user_id=user_id,
            exercise_id=data.get('exercise_id'),
            sets_completed=data.get('sets_completed'),
            reps_completed=data.get('reps_completed'),
            weight_used=data.get('weight', data.get('weight_used')),
            notes=data.get('notes', ''),
            date=datetime.now()
        )
        
        db.session.add(progress)
        db.session.commit()
        
        return jsonify({
            'success': True,
            'message': 'Progress saved successfully'
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/health-tips', methods=['GET'])
def get_health_tips():
    """Get health and fitness tips"""
    try:
        tips = HealthTip.query.order_by(db.func.random()).limit(5).all()
        return jsonify({
            'success': True,
            'tips': [tip.to_dict() for tip in tips]
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/user/profile', methods=['GET', 'POST'])
def user_profile():
    """Get or update user profile"""
    if request.method == 'GET':
        try:
            user_id = session.get('user_id')
            if user_id:
                user = User.query.get(user_id)
                if user:
                    return jsonify({
                        'success': True,
                        'user': user.to_dict()
                    })
            return jsonify({
                'success': False,
                'message': 'User not found'
            })
        except Exception as e:
            return jsonify({
                'success': False,
                'error': str(e)
            }), 500
    
    elif request.method == 'POST':
        try:
            data = request.get_json()
            user_id = session.get('user_id', 'anonymous')
            
            user = User.query.get(user_id)
            if not user:
                user = User(id=user_id)
                db.session.add(user)
            
            user.name = data.get('name', user.name)
            user.age = data.get('age', user.age)
            user.weight = data.get('weight', user.weight)
            user.height = data.get('height', user.height)
            user.fitness_goal = data.get('fitness_goal', user.fitness_goal)
            user.activity_level = data.get('activity_level', user.activity_level)
            
            db.session.commit()
            
            return jsonify({
                'success': True,
                'message': 'Profile updated successfully',
                'user': user.to_dict()
            })
        except Exception as e:
            return jsonify({
                'success': False,
                'error': str(e)
            }), 500

# Advanced Features API Endpoints

@app.route('/api/ai-workout', methods=['POST'])
def generate_ai_workout():
    """Generate AI-powered personalized workout"""
    try:
        data = request.get_json()
        user_profile = data.get('user_profile', {})
        preferences = data.get('preferences', {})
        
        workout = ai_workout_generator.generate_personalized_workout(user_profile, preferences)
        
        return jsonify({
            'success': True,
            'workout': workout
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/nutrition-analyze', methods=['POST'])
def analyze_nutrition():
    """Analyze nutrition from food image"""
    try:
        data = request.get_json()
        image_data = data.get('image_data', '')
        user_profile = data.get('user_profile', {})
        
        # Run async function in sync context
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            analysis = loop.run_until_complete(nutrition_ai.analyze_food_image(image_data, user_profile))
        finally:
            loop.close()
        
        return jsonify({
            'success': True,
            'analysis': analysis
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/social-challenge', methods=['POST'])
def create_challenge():
    """Create social fitness challenge"""
    try:
        data = request.get_json()
        challenge_data = data.get('challenge', {})
        
        challenge = social_features.create_challenge(challenge_data)
        
        return jsonify({
            'success': True,
            'challenge': challenge
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/analytics', methods=['GET'])
def get_analytics():
    """Get comprehensive analytics dashboard"""
    try:
        user_id = session.get('user_id', 'anonymous')
        timeframe = request.args.get('timeframe', 30)
        
        analytics = analytics_engine.calculate_comprehensive_analytics(user_id, timeframe)
        
        return jsonify({
            'success': True,
            'analytics': analytics
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/gamification/profile', methods=['GET'])
def get_gamification_profile():
    """Get user gamification profile"""
    try:
        user_id = session.get('user_id', 'anonymous')
        profile = gamification_system.get_user_stats_summary(user_id)
        
        return jsonify({
            'success': True,
            'profile': profile
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/wearable/connect', methods=['POST'])
def connect_wearable():
    """Connect wearable device"""
    try:
        data = request.get_json()
        device_type_str = data.get('device_type')
        user_id = session.get('user_id', 'anonymous')
        auth_data = data.get('auth_data', {})
        
        # Convert string to DeviceType enum
        try:
            device_type = DeviceType(device_type_str)
        except (ValueError, TypeError):
            return jsonify({
                'success': False,
                'error': f'Invalid device type: {device_type_str}'
            }), 400
        
        # Run async function in sync context
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            result = loop.run_until_complete(wearable_integration.connect_device(device_type, user_id, auth_data))
        finally:
            loop.close()
        
        return jsonify(result)
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/wearable/sync/<device_id>', methods=['POST'])
def sync_wearable(device_id):
    """Sync data from wearable device"""
    try:
        # Run async function in sync context
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            result = loop.run_until_complete(wearable_integration.sync_device_data(device_id))
        finally:
            loop.close()
        
        return jsonify(result)
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

# Advanced Features API Endpoints

@app.route('/api/voice/command', methods=['POST'])
def voice_command():
    """Process voice command"""
    try:
        data = request.get_json()
        transcript = data.get('transcript', '')
        
        result = voice_coach.process_voice_command(transcript)
        
        return jsonify(result)
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/voice/workout-guidance', methods=['POST'])
def workout_guidance():
    """Get voice guidance for workout"""
    try:
        data = request.get_json()
        exercise_name = data.get('exercise_name')
        set_number = data.get('set_number', 1)
        
        guidance = voice_coach.generate_workout_guidance(exercise_name, set_number)
        
        return jsonify(guidance)
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/buddy/profile', methods=['GET', 'POST'])
def buddy_profile():
    """Get or create workout buddy profile"""
    try:
        if request.method == 'POST':
            data = request.get_json()
            data['user_id'] = session.get('user_id', 'anonymous')
            result = buddy_matcher.create_buddy_profile(data)
            return jsonify(result)
        else:
            user_id = session.get('user_id', 'anonymous')
            matches = buddy_matcher.find_matches(user_id)
            return jsonify({
                'success': True,
                'matches': matches
            })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/buddy/find-matches', methods=['GET'])
def find_buddies():
    """Find workout buddy matches"""
    try:
        user_id = session.get('user_id', 'anonymous')
        limit = request.args.get('limit', 5, type=int)
        
        matches = buddy_matcher.find_matches(user_id, limit)
        
        return jsonify({
            'success': True,
            'matches': matches
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/buddy/request', methods=['POST'])
def send_buddy_request():
    """Send workout buddy request"""
    try:
        data = request.get_json()
        from_user_id = session.get('user_id', 'anonymous')
        to_user_id = data.get('to_user_id')
        message = data.get('message')
        
        result = buddy_matcher.create_buddy_request(from_user_id, to_user_id, message)
        
        return jsonify(result)
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/recovery/calculate', methods=['POST'])
def calculate_recovery():
    """Calculate recovery score and recommendations"""
    try:
        data = request.get_json()
        
        result = smart_recovery.calculate_recovery_score(data)
        
        return jsonify({
            'success': True,
            'recovery': result
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/recovery/activities', methods=['GET'])
def recovery_activities():
    """Get suggested recovery activities"""
    try:
        status = request.args.get('status', 'good')
        
        activities = smart_recovery.suggest_recovery_activities(status)
        
        return jsonify({
            'success': True,
            'activities': activities
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

@app.route('/api/pose-workout', methods=['POST'])
def save_pose_workout():
    """Save pose detection workout data"""
    try:
        data = request.get_json()
        user_id = session.get('user_id', 'anonymous')
        
        # Save to database
        progress = Progress(
            user_id=user_id,
            exercise_id=data.get('exercise_id', 1),
            sets_completed=1,
            reps_completed=data.get('reps', 0),
            weight_used=0,
            duration_seconds=data.get('duration', 0) // 1000,
            calories_burned=data.get('calories_burned', 0),
            notes=f"Pose detection workout - Form Score: {data.get('avgFormScore', 0)}%",
            date=datetime.now()
        )
        
        db.session.add(progress)
        db.session.commit()
        
        return jsonify({
            'success': True,
            'message': 'Workout saved successfully'
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
    app.run(debug=True, host='0.0.0.0', port=5000)
532 lines•16.5 KB
python
utils/ai_workout_generator.py
Raw Download
Find: Go to:
"""
AI Workout Generator - Advanced Personalized Workout Creation
Author: RSK World (https://rskworld.in)
Founded by: Molla Samser
Designer & Tester: Rima Khatun
Contact: help@rskworld.in, +91 93305 39277
Year: 2026
"""

import random
import json
from datetime import datetime, timedelta
from typing import Dict, List, Any

class AIWorkoutGenerator:
    """Advanced AI-powered workout generator with personalization"""
    
    def __init__(self):
        self.exercise_database = self._load_exercise_database()
        self.workout_templates = self._load_workout_templates()
        
    def _load_exercise_database(self) -> Dict:
        """Load comprehensive exercise database"""
        return {
            "chest": {
                "beginner": ["Push-ups", "Wall Push-ups", "Incline Push-ups", "Chest Press Machine"],
                "intermediate": ["Bench Press", "Dumbbell Flyes", "Incline Dumbbell Press", "Cable Crossovers"],
                "advanced": ["Weighted Dips", "Explosive Push-ups", "One-arm Push-ups", "Chain Bench Press"]
            },
            "back": {
                "beginner": ["Supermans", "Bird Dogs", "Resistance Band Rows", "Lat Pulldown Machine"],
                "intermediate": ["Pull-ups", "Bent-over Rows", "T-bar Rows", "Seated Cable Rows"],
                "advanced": ["Muscle-ups", "Deadlifts", "Weighted Pull-ups", "Kettlebell Swings"]
            },
            "legs": {
                "beginner": ["Bodyweight Squats", "Lunges", "Glute Bridges", "Calf Raises"],
                "intermediate": ["Barbell Squats", "Leg Press", "Romanian Deadlifts", "Bulgarian Split Squats"],
                "advanced": ["Pistol Squats", "Box Jumps", "Weighted Walking Lunges", "Front Squats"]
            },
            "shoulders": {
                "beginner": ["Arm Circles", "Wall Angels", "Resistance Band Press", "Dumbbell Lateral Raises"],
                "intermediate": ["Overhead Press", "Arnold Press", "Upright Rows", "Face Pulls"],
                "advanced": ["Handstand Push-ups", "Barbell Push Press", "Snatch", "Kettlebell Press"]
            },
            "core": {
                "beginner": ["Plank", "Dead Bug", "Bird Dog", "Crunches"],
                "intermediate": ["Hanging Leg Raises", "Russian Twists", "Ab Rollouts", "Dragon Flags"],
                "advanced": ["Front Lever", "Planche", "Human Flag", "Weighted Planks"]
            },
            "cardio": {
                "beginner": ["Walking", "Light Jogging", "Cycling (moderate)", "Elliptical"],
                "intermediate": ["HIIT Running", "Rowing", "Stair Climber", "Jump Rope"],
                "advanced": ["Sprinting Intervals", "Battle Ropes", "Boxing", "CrossFit WODs"]
            }
        }
    
    def _load_workout_templates(self) -> Dict:
        """Load workout templates for different goals"""
        return {
            "weight_loss": {
                "focus": ["cardio", "full_body"],
                "intensity": "moderate_to_high",
                "duration": 45,
                "frequency": 5
            },
            "muscle_gain": {
                "focus": ["strength", "hypertrophy"],
                "intensity": "high",
                "duration": 60,
                "frequency": 4
            },
            "endurance": {
                "focus": ["cardio", "functional"],
                "intensity": "moderate",
                "duration": 50,
                "frequency": 6
            },
            "strength": {
                "focus": ["compound", "heavy"],
                "intensity": "high",
                "duration": 70,
                "frequency": 3
            },
            "general_fitness": {
                "focus": ["balanced", "functional"],
                "intensity": "moderate",
                "duration": 45,
                "frequency": 4
            }
        }
    
    def generate_personalized_workout(self, user_profile: Dict, preferences: Dict = None) -> Dict:
        """Generate AI-powered personalized workout"""
        
        # Extract user information
        fitness_level = user_profile.get('fitness_level', 'beginner')
        goal = user_profile.get('goal', 'general_fitness')
        available_time = user_profile.get('available_time', 45)
        equipment = user_profile.get('equipment', ['bodyweight'])
        limitations = user_profile.get('limitations', [])
        
        # Get workout template
        template = self.workout_templates.get(goal, self.workout_templates['general_fitness'])
        
        # Generate workout structure
        workout = {
            "user_id": user_profile.get('id'),
            "date": datetime.now().isoformat(),
            "goal": goal,
            "fitness_level": fitness_level,
            "estimated_duration": available_time,
            "estimated_calories": self._estimate_calories(goal, fitness_level, available_time),
            "warmup": self._generate_warmup(fitness_level, 5),
            "main_workout": self._generate_main_workout(
                template, fitness_level, available_time, equipment, limitations
            ),
            "cool_down": self._generate_cool_down(fitness_level, 5),
            "tips": self._generate_workout_tips(goal, fitness_level),
            "progression_plan": self._generate_progression_plan(goal, fitness_level)
        }
        
        return workout
    
    def _generate_main_workout(self, template: Dict, fitness_level: str, 
                             available_time: int, equipment: List[str], 
                             limitations: List[str]) -> List[Dict]:
        """Generate main workout exercises"""
        
        focus_areas = template['focus']
        exercises = []
        time_per_exercise = (available_time - 10) / 6  # Account for warmup/cooldown
        
        # Select exercises based on focus areas
        if "full_body" in focus_areas or "balanced" in focus_areas:
            muscle_groups = ["chest", "back", "legs", "shoulders", "core"]
        elif "strength" in focus_areas or "hypertrophy" in focus_areas:
            muscle_groups = ["chest", "back", "legs", "shoulders"]
        else:
            muscle_groups = random.sample(["chest", "back", "legs", "shoulders", "core"], 3)
        
        for muscle_group in muscle_groups:
            if muscle_group in self.exercise_database:
                # Select appropriate exercise
                level_exercises = self.exercise_database[muscle_group][fitness_level]
                available_exercises = [ex for ex in level_exercises 
                                    if self._check_equipment_compatibility(ex, equipment)]
                
                if available_exercises:
                    exercise = random.choice(available_exercises)
                    exercises.append({
                        "name": exercise,
                        "muscle_group": muscle_group,
                        "sets": self._determine_sets(fitness_level, template['intensity']),
                        "reps": self._determine_reps(fitness_level, muscle_group),
                        "rest_time": self._determine_rest_time(fitness_level, template['intensity']),
                        "duration": time_per_exercise,
                        "instructions": self._get_exercise_instructions(exercise),
                        "form_tips": self._get_form_tips(exercise)
                    })
        
        return exercises
    
    def _generate_warmup(self, fitness_level: str, duration: int) -> List[Dict]:
        """Generate warmup routine"""
        warmup_exercises = [
            "Jumping Jacks", "High Knees", "Arm Circles", "Leg Swings", 
            "Hip Circles", "Torso Twists", "Light Jogging"
        ]
        
        return [
            {
                "name": random.choice(warmup_exercises),
                "duration": duration // len(warmup_exercises),
                "intensity": "light"
            } for _ in range(min(4, len(warmup_exercises)))
        ]
    
    def _generate_cool_down(self, fitness_level: str, duration: int) -> List[Dict]:
        """Generate cool-down routine"""
        stretches = [
            "Hamstring Stretch", "Quad Stretch", "Chest Stretch", 
            "Back Stretch", "Shoulder Stretch", "Triceps Stretch"
        ]
        
        return [
            {
                "name": random.choice(stretches),
                "duration": duration // len(stretches),
                "type": "static_stretch"
            } for _ in range(min(4, len(stretches)))
        ]
    
    def _estimate_calories(self, goal: str, fitness_level: str, duration: int) -> int:
        """Estimate calories burned"""
        base_calories = {
            "beginner": 5,
            "intermediate": 8,
            "advanced": 12
        }
        
        goal_multiplier = {
            "weight_loss": 1.3,
            "muscle_gain": 1.1,
            "endurance": 1.4,
            "strength": 0.9,
            "general_fitness": 1.0
        }
        
        return int(base_calories[fitness_level] * duration * goal_multiplier.get(goal, 1.0))
    
    def _determine_sets(self, fitness_level: str, intensity: str) -> int:
        """Determine number of sets"""
        base_sets = {"beginner": 2, "intermediate": 3, "advanced": 4}
        intensity_multiplier = {"moderate": 1, "high": 1.2, "moderate_to_high": 1.1}
        
        return int(base_sets[fitness_level] * intensity_multiplier.get(intensity, 1.0))
    
    def _determine_reps(self, fitness_level: str, muscle_group: str) -> str:
        """Determine rep range"""
        if muscle_group == "cardio":
            return "30-60 seconds"
        
        rep_ranges = {
            "beginner": "8-12",
            "intermediate": "10-15",
            "advanced": "12-20"
        }
        
        return rep_ranges[fitness_level]
    
    def _determine_rest_time(self, fitness_level: str, intensity: str) -> str:
        """Determine rest time between sets"""
        if intensity == "high":
            return "60-90 seconds"
        elif intensity == "moderate_to_high":
            return "45-60 seconds"
        else:
            return "30-45 seconds"
    
    def _check_equipment_compatibility(self, exercise: str, available_equipment: List[str]) -> bool:
        """Check if exercise is compatible with available equipment"""
        exercise_requirements = {
            "Bench Press": ["barbell", "bench"],
            "Pull-ups": ["pull_up_bar"],
            "Barbell Squats": ["barbell", "squat_rack"],
            "Deadlifts": ["barbell"],
            "Weighted Dips": ["dip_station", "weight_belt"],
            "Muscle-ups": ["pull_up_bar", "rings"],
            "Handstand Push-ups": [],
            "Front Lever": ["pull_up_bar"],
            "Planche": [],
            "Human Flag": ["pole"]
        }
        
        requirements = exercise_requirements.get(exercise, [])
        return all(req in available_equipment for req in requirements) or not requirements
    
    def _get_exercise_instructions(self, exercise: str) -> str:
        """Get exercise instructions"""
        instructions_db = {
            "Push-ups": "Start in plank position, lower body until chest nearly touches floor, push back up",
            "Squats": "Stand with feet shoulder-width apart, lower hips back and down, return to standing",
            "Pull-ups": "Hang from pull-up bar, pull body up until chin clears bar, lower slowly",
            "Plank": "Hold push-up position with body straight, engage core muscles"
        }
        
        return instructions_db.get(exercise, "Follow proper form and technique for this exercise")
    
    def _get_form_tips(self, exercise: str) -> List[str]:
        """Get form tips for exercise"""
        tips_db = {
            "Push-ups": ["Keep core tight", "Maintain straight line from head to heels", "Lower chest to floor"],
            "Squats": ["Keep chest up", "Knees behind toes", "Go to parallel or lower"],
            "Pull-ups": ["Engage back muscles", "Full range of motion", "Control negative portion"]
        }
        
        return tips_db.get(exercise, ["Maintain proper form", "Breathe correctly", "Start with light weight"])
    
    def _generate_workout_tips(self, goal: str, fitness_level: str) -> List[str]:
        """Generate workout-specific tips"""
        tips = []
        
        if goal == "weight_loss":
            tips.extend([
                "Focus on compound movements for maximum calorie burn",
                "Keep rest periods short to maintain heart rate",
                "Combine with proper nutrition for best results"
            ])
        elif goal == "muscle_gain":
            tips.extend([
                "Progressive overload is key - increase weight/reps over time",
                "Ensure adequate protein intake (1.6-2.2g per kg)",
                "Allow 48 hours recovery between training same muscle groups"
            ])
        
        if fitness_level == "beginner":
            tips.append("Focus on form before increasing intensity")
        
        return tips
    
    def _generate_progression_plan(self, goal: str, fitness_level: str) -> Dict:
        """Generate 4-week progression plan"""
        return {
            "week_1": {
                "focus": "Technique and adaptation",
                "intensity": "60% of max effort",
                "volume": "3 sets x 10-12 reps"
            },
            "week_2": {
                "focus": "Building foundation",
                "intensity": "70% of max effort",
                "volume": "3-4 sets x 8-12 reps"
            },
            "week_3": {
                "focus": "Increasing intensity",
                "intensity": "80% of max effort",
                "volume": "4 sets x 8-10 reps"
            },
            "week_4": {
                "focus": "Peak performance",
                "intensity": "85-90% of max effort",
                "volume": "4-5 sets x 6-8 reps"
            }
        }
    
    def generate_ai_insights(self, workout_history: List[Dict]) -> Dict:
        """Generate AI-powered insights from workout history"""
        if not workout_history:
            return {"message": "Start working out to get personalized insights!"}
        
        # Analyze patterns
        total_workouts = len(workout_history)
        avg_duration = sum(w.get('duration', 0) for w in workout_history) / total_workouts
        most_trained = self._analyze_most_trained(workout_history)
        consistency_score = self._calculate_consistency(workout_history)
        
        insights = {
            "total_workouts": total_workouts,
            "average_duration": round(avg_duration, 1),
            "consistency_score": consistency_score,
            "most_trained_muscle_groups": most_trained,
            "recommendations": self._generate_recommendations(workout_history),
            "achievement_badges": self._check_achievements(workout_history)
        }
        
        return insights
    
    def _analyze_most_trained(self, workout_history: List[Dict]) -> List[str]:
        """Analyze most trained muscle groups"""
        muscle_counts = {}
        for workout in workout_history:
            for exercise in workout.get('exercises', []):
                muscle = exercise.get('muscle_group', 'unknown')
                muscle_counts[muscle] = muscle_counts.get(muscle, 0) + 1
        
        return sorted(muscle_counts.keys(), key=lambda x: muscle_counts[x], reverse=True)[:3]
    
    def _calculate_consistency(self, workout_history: List[Dict]) -> float:
        """Calculate workout consistency score"""
        if len(workout_history) < 2:
            return 0.0
        
        dates = [datetime.fromisoformat(w.get('date', '')) for w in workout_history]
        dates.sort()
        
        # Calculate average days between workouts
        intervals = [(dates[i] - dates[i-1]).days for i in range(1, len(dates))]
        avg_interval = sum(intervals) / len(intervals)
        
        # Ideal is 2-3 days between workouts
        if 2 <= avg_interval <= 3:
            return 100.0
        elif avg_interval <= 4:
            return 80.0
        else:
            return max(0, 60.0 - (avg_interval - 4) * 10)
    
    def _generate_recommendations(self, workout_history: List[Dict]) -> List[str]:
        """Generate personalized recommendations"""
        recommendations = []
        
        if len(workout_history) < 4:
            recommendations.append("Try to work out at least 3 times per week for best results")
        
        muscle_groups = self._analyze_most_trained(workout_history)
        if len(muscle_groups) < 3:
            recommendations.append("Add variety to target different muscle groups")
        
        avg_duration = sum(w.get('duration', 0) for w in workout_history) / len(workout_history)
        if avg_duration < 30:
            recommendations.append("Consider extending workouts to 30-45 minutes")
        
        return recommendations
    
    def _check_achievements(self, workout_history: List[Dict]) -> List[Dict]:
        """Check for achievement badges"""
        achievements = []
        
        if len(workout_history) >= 10:
            achievements.append({
                "name": "Dedicated Athlete",
                "description": "Completed 10 workouts",
                "icon": "🏆"
            })
        
        if len(workout_history) >= 30:
            achievements.append({
                "name": "Fitness Warrior",
                "description": "Completed 30 workouts",
                "icon": "⚔️"
            })
        
        total_duration = sum(w.get('duration', 0) for w in workout_history)
        if total_duration >= 1000:  # ~16.6 hours
            achievements.append({
                "name": "Time Champion",
                "description": "1000+ minutes of exercise",
                "icon": "⏰"
            })
        
        return achievements
422 lines•18 KB
python
utils/wearable_integration.py
Raw Download
Find: Go to:
"""
Wearable Device Integration for Fitness Coach Bot
Author: RSK World (https://rskworld.in)
Founded by: Molla Samser
Designer & Tester: Rima Khatun
Contact: help@rskworld.in, +91 93305 39277
Year: 2026
"""

import json
import requests
import asyncio
from datetime import datetime, timedelta
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from enum import Enum

class DeviceType(Enum):
    FITBIT = "fitbit"
    GARMIN = "garmin"
    APPLE_WATCH = "apple_watch"
    WHOOP = "whoop"
    OURA = "oura"
    POLAR = "polar"
    CUSTOM = "custom"

@dataclass
class WearableDevice:
    device_id: str
    device_type: DeviceType
    user_id: str
    access_token: str
    refresh_token: Optional[str] = None
    last_sync: Optional[datetime] = None
    is_active: bool = True
    capabilities: List[str] = None

class WearableIntegration:
    """Comprehensive wearable device integration system"""
    
    def __init__(self):
        self.connected_devices = {}
        self.api_endpoints = self._load_api_endpoints()
        self.data_cache = {}
        self.sync_queue = []
        
    def _load_api_endpoints(self) -> Dict:
        """Load API endpoints for different wearable platforms"""
        return {
            DeviceType.FITBIT: {
                "base_url": "https://api.fitbit.com/1/user",
                "auth_url": "https://www.fitbit.com/oauth2/authorize",
                "token_url": "https://api.fitbit.com/oauth2/token",
                "scopes": ["activity", "heartrate", "sleep", "weight", "nutrition"]
            },
            DeviceType.GARMIN: {
                "base_url": "https://healthapi.garmin.com",
                "auth_url": "https://connect.garmin.com/oauthConfirm",
                "scopes": ["read:activities", "read:heartrate", "read:sleep"]
            },
            DeviceType.APPLE_WATCH: {
                "base_url": "https://healthkit.apple.com",
                "auth_url": "applehealth://authorize",
                "scopes": ["step_count", "heart_rate", "sleep_analysis", "workout"]
            },
            DeviceType.WHOOP: {
                "base_url": "https://api.whoop.com/developer/v1",
                "auth_url": "https://api.whoop.com/oauth/oauth2/auth",
                "scopes": ["recovery", "sleep", "workout", "biometric"]
            },
            DeviceType.OURA: {
                "base_url": "https://api.ouraring.com/v2",
                "auth_url": "https://cloud.ouraring.com/oauth/authorize",
                "scopes": ["personal", "daily", "heartrate", "sleep"]
            }
        }
    
    async def connect_device(self, device_type: DeviceType, user_id: str, 
                           auth_data: Dict) -> Dict:
        """Connect a new wearable device"""
        try:
            device = WearableDevice(
                device_id=f"{device_type.value}_{user_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
                device_type=device_type,
                user_id=user_id,
                access_token=auth_data.get("access_token"),
                refresh_token=auth_data.get("refresh_token"),
                capabilities=self._get_device_capabilities(device_type)
            )
            
            # Test connection
            test_result = await self._test_device_connection(device)
            if not test_result["success"]:
                return {"success": False, "error": test_result["error"]}
            
            # Store device
            self.connected_devices[device.device_id] = device
            
            # Initial sync
            await self.sync_device_data(device.device_id)
            
            return {
                "success": True,
                "device_id": device.device_id,
                "message": f"Successfully connected {device_type.value} device"
            }
            
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    async def _test_device_connection(self, device: WearableDevice) -> Dict:
        """Test connection to wearable device"""
        try:
            endpoint = self.api_endpoints.get(device.device_type)
            if not endpoint:
                return {"success": False, "error": "Unsupported device type"}
            
            # Make test API call
            headers = {"Authorization": f"Bearer {device.access_token}"}
            
            if device.device_type == DeviceType.FITBIT:
                response = requests.get(
                    f"{endpoint['base_url']}/-/profile.json",
                    headers=headers,
                    timeout=10
                )
                return {"success": response.status_code == 200}
            
            elif device.device_type == DeviceType.WHOOP:
                response = requests.get(
                    f"{endpoint['base_url']}/user/measurement/recovery",
                    headers=headers,
                    timeout=10
                )
                return {"success": response.status_code == 200}
            
            # Add other device types as needed
            
            return {"success": True}
            
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    async def sync_device_data(self, device_id: str) -> Dict:
        """Sync data from wearable device"""
        if device_id not in self.connected_devices:
            return {"success": False, "error": "Device not connected"}
        
        device = self.connected_devices[device_id]
        
        try:
            sync_data = {}
            
            # Sync based on device capabilities
            if "steps" in device.capabilities:
                sync_data["steps"] = await self._sync_steps_data(device)
            
            if "heart_rate" in device.capabilities:
                sync_data["heart_rate"] = await self._sync_heart_rate_data(device)
            
            if "sleep" in device.capabilities:
                sync_data["sleep"] = await self._sync_sleep_data(device)
            
            if "workouts" in device.capabilities:
                sync_data["workouts"] = await self._sync_workout_data(device)
            
            if "weight" in device.capabilities:
                sync_data["weight"] = await self._sync_weight_data(device)
            
            if "recovery" in device.capabilities:
                sync_data["recovery"] = await self._sync_recovery_data(device)
            
            # Update last sync time
            device.last_sync = datetime.now()
            
            # Cache data
            self.data_cache[device_id] = {
                "data": sync_data,
                "last_sync": device.last_sync
            }
            
            return {
                "success": True,
                "device_id": device_id,
                "sync_data": sync_data,
                "sync_time": device.last_sync.isoformat()
            }
            
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    async def _sync_steps_data(self, device: WearableDevice) -> Dict:
        """Sync steps data from device"""
        if device.device_type == DeviceType.FITBIT:
            return await self._sync_fitbit_steps(device)
        elif device.device_type == DeviceType.APPLE_WATCH:
            return await self._sync_apple_health_steps(device)
        elif device.device_type == DeviceType.WHOOP:
            return await self._sync_whoop_activity(device)
        
        return {"steps": 0, "date": datetime.now().isoformat()}
    
    async def _sync_fitbit_steps(self, device: WearableDevice) -> Dict:
        """Sync steps from Fitbit"""
        try:
            headers = {"Authorization": f"Bearer {device.access_token}"}
            endpoint = self.api_endpoints[DeviceType.FITBIT]
            
            # Get today's steps
            today = datetime.now().strftime("%Y-%m-%d")
            response = requests.get(
                f"{endpoint['base_url']}/-/activities/date/{today}.json",
                headers=headers
            )
            
            if response.status_code == 200:
                data = response.json()
                steps = data.get("steps", [])
                
                return {
                    "source": "fitbit",
                    "steps": steps[-1] if steps else 0,  # Latest step count
                    "date": today,
                    "details": data
                }
            
            return {"steps": 0, "date": today}
            
        except Exception as e:
            print(f"Error syncing Fitbit steps: {e}")
            return {"steps": 0, "date": datetime.now().strftime("%Y-%m-%d")}
    
    async def _sync_heart_rate_data(self, device: WearableDevice) -> Dict:
        """Sync heart rate data from device"""
        if device.device_type == DeviceType.FITBIT:
            return await self._sync_fitbit_heart_rate(device)
        elif device.device_type == DeviceType.APPLE_WATCH:
            return await self._sync_apple_health_heart_rate(device)
        
        return {"heart_rate": 0, "date": datetime.now().isoformat()}
    
    async def _sync_fitbit_heart_rate(self, device: WearableDevice) -> Dict:
        """Sync heart rate from Fitbit"""
        try:
            headers = {"Authorization": f"Bearer {device.access_token}"}
            endpoint = self.api_endpoints[DeviceType.FITBIT]
            
            # Get today's heart rate
            today = datetime.now().strftime("%Y-%m-%d")
            response = requests.get(
                f"{endpoint['base_url']}/-/activities/heart/date/{today}/1d.json",
                headers=headers
            )
            
            if response.status_code == 200:
                data = response.json()
                heart_rate_data = data.get("activities-heart", [])
                
                if heart_rate_data:
                    hr_data = heart_rate_data[0]
                    return {
                        "source": "fitbit",
                        "resting_hr": hr_data.get("value", {}).get("restingHeartRate", 0),
                        "average_hr": hr_data.get("value", {}).get("heartRateZones", [{}])[0].get("max", 0),
                        "date": today,
                        "details": hr_data
                    }
            
            return {"resting_hr": 0, "average_hr": 0, "date": today}
            
        except Exception as e:
            print(f"Error syncing Fitbit heart rate: {e}")
            return {"resting_hr": 0, "average_hr": 0, "date": datetime.now().strftime("%Y-%m-%d")}
    
    async def _sync_sleep_data(self, device: WearableDevice) -> Dict:
        """Sync sleep data from device"""
        if device.device_type == DeviceType.FITBIT:
            return await self._sync_fitbit_sleep(device)
        elif device.device_type == DeviceType.WHOOP:
            return await self._sync_whoop_sleep(device)
        
        return {"sleep_score": 0, "date": datetime.now().isoformat()}
    
    async def _sync_fitbit_sleep(self, device: WearableDevice) -> Dict:
        """Sync sleep from Fitbit"""
        try:
            headers = {"Authorization": f"Bearer {device.access_token}"}
            endpoint = self.api_endpoints[DeviceType.FITBIT]
            
            # Get last night's sleep
            yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
            response = requests.get(
                f"{endpoint['base_url']}/-/sleep/date/{yesterday}.json",
                headers=headers
            )
            
            if response.status_code == 200:
                data = response.json()
                sleep_data = data.get("sleep", [])
                
                if sleep_data:
                    main_sleep = sleep_data[0]
                    return {
                        "source": "fitbit",
                        "sleep_score": main_sleep.get("efficiency", 0),
                        "duration_minutes": main_sleep.get("minutesAsleep", 0),
                        "deep_sleep": main_sleep.get("minutesDeepSleep", 0),
                        "rem_sleep": main_sleep.get("minutesREM", 0),
                        "date": yesterday,
                        "details": main_sleep
                    }
            
            return {"sleep_score": 0, "duration_minutes": 0, "date": yesterday}
            
        except Exception as e:
            print(f"Error syncing Fitbit sleep: {e}")
            return {"sleep_score": 0, "duration_minutes": 0, "date": (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")}
    
    async def _sync_workout_data(self, device: WearableDevice) -> Dict:
        """Sync workout data from device"""
        if device.device_type == DeviceType.FITBIT:
            return await self._sync_fitbit_workouts(device)
        elif device.device_type == DeviceType.GARMIN:
            return await self._sync_garmin_workouts(device)
        
        return {"workouts": [], "date": datetime.now().isoformat()}
    
    async def _sync_fitbit_workouts(self, device: WearableDevice) -> Dict:
        """Sync workouts from Fitbit"""
        try:
            headers = {"Authorization": f"Bearer {device.access_token}"}
            endpoint = self.api_endpoints[DeviceType.FITBIT]
            
            # Get recent workouts
            response = requests.get(
                f"{endpoint['base_url']}/-/activities/list.json?limit=10&sort=desc",
                headers=headers
            )
            
            if response.status_code == 200:
                data = response.json()
                workouts = data.get("activities", [])
                
                processed_workouts = []
                for workout in workouts:
                    processed_workouts.append({
                        "name": workout.get("name", ""),
                        "type": workout.get("activityName", ""),
                        "duration_minutes": workout.get("duration", 0) // 60000,  # Convert ms to minutes
                        "calories": workout.get("calories", 0),
                        "distance": workout.get("distance", 0),
                        "date": workout.get("startTime", ""),
                        "steps": workout.get("steps", 0)
                    })
                
                return {
                    "source": "fitbit",
                    "workouts": processed_workouts,
                    "count": len(processed_workouts),
                    "date": datetime.now().isoformat()
                }
            
            return {"workouts": [], "count": 0, "date": datetime.now().isoformat()}
            
        except Exception as e:
            print(f"Error syncing Fitbit workouts: {e}")
            return {"workouts": [], "count": 0, "date": datetime.now().isoformat()}
    
    async def _sync_weight_data(self, device: WearableDevice) -> Dict:
        """Sync weight data from device"""
        if device.device_type == DeviceType.FITBIT:
            return await self._sync_fitbit_weight(device)
        
        return {"weight": 0, "date": datetime.now().isoformat()}
    
    async def _sync_fitbit_weight(self, device: WearableDevice) -> Dict:
        """Sync weight from Fitbit"""
        try:
            headers = {"Authorization": f"Bearer {device.access_token}"}
            endpoint = self.api_endpoints[DeviceType.FITBIT]
            
            # Get recent weight
            response = requests.get(
                f"{endpoint['base_url']}/-/body/log/weight/list.json?limit=1&sort=desc",
                headers=headers
            )
            
            if response.status_code == 200:
                data = response.json()
                weight_data = data.get("weight", [])
                
                if weight_data:
                    latest_weight = weight_data[0]
                    return {
                        "source": "fitbit",
                        "weight": latest_weight.get("weight", 0),
                        "bmi": latest_weight.get("bmi", 0),
                        "fat": latest_weight.get("fat", 0),
                        "date": latest_weight.get("date", ""),
                        "details": latest_weight
                    }
            
            return {"weight": 0, "date": datetime.now().isoformat()}
            
        except Exception as e:
            print(f"Error syncing Fitbit weight: {e}")
            return {"weight": 0, "date": datetime.now().isoformat()}
    
    async def _sync_recovery_data(self, device: WearableDevice) -> Dict:
        """Sync recovery data from device"""
        if device.device_type == DeviceType.WHOOP:
            return await self._sync_whoop_recovery(device)
        
        return {"recovery_score": 0, "date": datetime.now().isoformat()}
    
    async def _sync_whoop_recovery(self, device: WearableDevice) -> Dict:
        """Sync recovery from Whoop"""
        try:
            headers = {"Authorization": f"Bearer {device.access_token}"}
            endpoint = self.api_endpoints[DeviceType.WHOOP]
            
            # Get today's recovery
            response = requests.get(
                f"{endpoint['base_url']}/user/measurement/recovery",
                headers=headers
            )
            
            if response.status_code == 200:
                data = response.json()
                if data and len(data) > 0:
                    recovery = data[0]
                    return {
                        "source": "whoop",
                        "recovery_score": recovery.get("recovery_score", 0),
                        "resting_hr": recovery.get("resting_heart_rate", 0),
                        "hrv": recovery.get("hrv_rmssd", 0),
                        "sleep_performance": recovery.get("sleep_performance", 0),
                        "date": recovery.get("created_at", ""),
                        "details": recovery
                    }
            
            return {"recovery_score": 0, "date": datetime.now().isoformat()}
            
        except Exception as e:
            print(f"Error syncing Whoop recovery: {e}")
            return {"recovery_score": 0, "date": datetime.now().isoformat()}
    
    def get_device_data(self, device_id: str, data_type: str, 
                      start_date: Optional[str] = None, end_date: Optional[str] = None) -> Dict:
        """Get specific type of data from device"""
        if device_id not in self.connected_devices:
            return {"success": False, "error": "Device not connected"}
        
        if device_id not in self.data_cache:
            return {"success": False, "error": "No data available"}
        
        cached_data = self.data_cache[device_id]["data"]
        
        if data_type in cached_data:
            return {
                "success": True,
                "data": cached_data[data_type],
                "last_sync": self.data_cache[device_id]["last_sync"]
            }
        
        return {"success": False, "error": f"Data type {data_type} not available"}
    
    def get_combined_dashboard(self, user_id: str) -> Dict:
        """Get combined dashboard from all connected devices"""
        user_devices = [d for d in self.connected_devices.values() if d.user_id == user_id]
        
        dashboard = {
            "user_id": user_id,
            "connected_devices": len(user_devices),
            "last_sync": None,
            "metrics": {},
            "alerts": []
        }
        
        # Combine data from all devices
        all_steps = []
        all_heart_rate = []
        all_sleep = []
        all_workouts = []
        
        for device in user_devices:
            if device.device_id in self.data_cache:
                device_data = self.data_cache[device.device_id]["data"]
                
                # Collect steps
                if "steps" in device_data:
                    all_steps.append(device_data["steps"])
                
                # Collect heart rate
                if "heart_rate" in device_data:
                    all_heart_rate.append(device_data["heart_rate"])
                
                # Collect sleep
                if "sleep" in device_data:
                    all_sleep.append(device_data["sleep"])
                
                # Collect workouts
                if "workouts" in device_data:
                    all_workouts.extend(device_data["workouts"]["workouts"])
                
                # Update last sync
                if device.last_sync:
                    if not dashboard["last_sync"] or device.last_sync > dashboard["last_sync"]:
                        dashboard["last_sync"] = device.last_sync
        
        # Calculate combined metrics
        dashboard["metrics"] = {
            "total_steps": sum(step.get("steps", 0) for step in all_steps),
            "avg_heart_rate": sum(hr.get("average_hr", 0) for hr in all_heart_rate) / len(all_heart_rate) if all_heart_rate else 0,
            "resting_heart_rate": min(hr.get("resting_hr", 999) for hr in all_heart_rate) if all_heart_rate else 0,
            "avg_sleep_score": sum(sleep.get("sleep_score", 0) for sleep in all_sleep) / len(all_sleep) if all_sleep else 0,
            "total_workouts": len(all_workouts),
            "total_calories": sum(workout.get("calories", 0) for workout in all_workouts)
        }
        
        # Generate alerts
        dashboard["alerts"] = self._generate_health_alerts(dashboard["metrics"])
        
        return dashboard
    
    def _generate_health_alerts(self, metrics: Dict) -> List[Dict]:
        """Generate health alerts based on metrics"""
        alerts = []
        
        # Resting heart rate alert
        if metrics.get("resting_heart_rate", 0) > 80:
            alerts.append({
                "type": "warning",
                "message": "Elevated resting heart rate detected",
                "recommendation": "Consider stress management techniques or consult a healthcare provider"
            })
        
        # Sleep alert
        if metrics.get("avg_sleep_score", 100) < 70:
            alerts.append({
                "type": "warning",
                "message": "Poor sleep quality detected",
                "recommendation": "Focus on improving sleep hygiene and bedtime routine"
            })
        
        # Activity alert
        if metrics.get("total_steps", 0) < 5000:
            alerts.append({
                "type": "info",
                "message": "Low activity level today",
                "recommendation": "Try to take a walk or do some light exercise"
            })
        
        return alerts
    
    def _get_device_capabilities(self, device_type: DeviceType) -> List[str]:
        """Get capabilities for device type"""
        capabilities = {
            DeviceType.FITBIT: ["steps", "heart_rate", "sleep", "workouts", "weight"],
            DeviceType.GARMIN: ["steps", "heart_rate", "sleep", "workouts", "weight"],
            DeviceType.APPLE_WATCH: ["steps", "heart_rate", "sleep", "workouts"],
            DeviceType.WHOOP: ["heart_rate", "sleep", "recovery", "workouts"],
            DeviceType.OURA: ["sleep", "heart_rate", "recovery"],
            DeviceType.POLAR: ["heart_rate", "workouts", "steps"]
        }
        
        return capabilities.get(device_type, [])
    
    async def disconnect_device(self, device_id: str) -> Dict:
        """Disconnect wearable device"""
        if device_id not in self.connected_devices:
            return {"success": False, "error": "Device not found"}
        
        # Remove from connected devices
        del self.connected_devices[device_id]
        
        # Clear cached data
        if device_id in self.data_cache:
            del self.data_cache[device_id]
        
        return {
            "success": True,
            "message": "Device disconnected successfully"
        }
    
    def get_available_devices(self) -> List[Dict]:
        """Get list of available wearable device types"""
        return [
            {
                "type": DeviceType.FITBIT.value,
                "name": "Fitbit",
                "capabilities": ["steps", "heart_rate", "sleep", "workouts", "weight"],
                "description": "Comprehensive fitness tracking with sleep analysis"
            },
            {
                "type": DeviceType.GARMIN.value,
                "name": "Garmin",
                "capabilities": ["steps", "heart_rate", "sleep", "workouts", "weight"],
                "description": "Advanced sports watch with GPS and performance metrics"
            },
            {
                "type": DeviceType.APPLE_WATCH.value,
                "name": "Apple Watch",
                "capabilities": ["steps", "heart_rate", "sleep", "workouts"],
                "description": "Smartwatch with health monitoring and fitness tracking"
            },
            {
                "type": DeviceType.WHOOP.value,
                "name": "Whoop",
                "capabilities": ["heart_rate", "sleep", "recovery", "workouts"],
                "description": "24/7 health and performance monitoring with recovery insights"
            },
            {
                "type": DeviceType.OURA.value,
                "name": "Oura Ring",
                "capabilities": ["sleep", "heart_rate", "recovery"],
                "description": "Smart ring with advanced sleep and recovery tracking"
            }
        ]
    
    def export_device_data(self, device_id: str, format: str = "json") -> Dict:
        """Export device data in specified format"""
        if device_id not in self.data_cache:
            return {"success": False, "error": "No data available"}
        
        data = self.data_cache[device_id]
        
        if format == "json":
            return {
                "success": True,
                "data": data,
                "format": "json"
            }
        elif format == "csv":
            # Convert to CSV format
            return {
                "success": True,
                "data": self._convert_to_csv(data),
                "format": "csv"
            }
        
        return {"success": False, "error": "Unsupported format"}
    
    def _convert_to_csv(self, data: Dict) -> str:
        """Convert data to CSV format"""
        # Simplified CSV conversion
        csv_lines = ["Date,Metric,Value,Source"]
        
        device_data = data.get("data", {})
        for metric_type, metric_data in device_data.items():
            if isinstance(metric_data, dict):
                for key, value in metric_data.items():
                    if isinstance(value, (int, float)):
                        csv_lines.append(f"{datetime.now().strftime('%Y-%m-%d')},{metric_type}_{key},{value},wearable")
        
        return "\n".join(csv_lines)
657 lines•27 KB
python
🚀 Support RSK World

Subscribe to our YouTube channel for latest tutorials & updates!



Click subscribe & support our work ❤️

About RSK World

Founded by Molla Samser, with Designer & Tester Rima Khatun, RSK World is your one-stop destination for free programming resources, source code, and development tools.

Founder: Molla Samser
Designer & Tester: Rima Khatun

Development

  • Game Development
  • Web Development
  • Mobile Development
  • AI Development
  • Development Tools

Legal

  • Terms & Conditions
  • Privacy Policy
  • Disclaimer

Contact Info

Nutanhat, Mongolkote
Purba Burdwan, West Bengal
India, 713147

+91 93305 39277

hello@rskworld.in
support@rskworld.in

© 2026 RSK World. All rights reserved.

Content used for educational purposes only. View Disclaimer