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
RSK World
fitness-coach-bot
Fitness Coach Bot - Python + Flask + SQLAlchemy + Workout Plans + Exercise Guidance + Health Tracking + AI Fitness Coach
fitness-coach-bot
  • __pycache__
  • data
  • instance
  • models
  • static
  • templates
  • utils
  • .gitignore564 B
  • ADVANCED_FEATURES.md7.1 KB
  • HOW_TO_CREATE_RELEASE.md4.1 KB
  • LICENSE1.1 KB
  • PROJECT_CHECK_SUMMARY.md5.5 KB
  • README.md8.5 KB
  • RELEASE_NOTES_v1.0.0.md6.9 KB
  • app.py16.5 KB
  • config.py1.5 KB
  • demo_data.py2.5 KB
  • init_db.py12.7 KB
  • requirements.txt442 B
README.mdapp.pyai_workout_generator.pywearable_integration.py
README.md
Raw Download

README.md

# Fitness Coach Bot

**AI-powered fitness coaching chatbot for workout plans, exercise guidance, and health tracking**

---

## 🏋️ Project Information

**Author:** RSK World (https://rskworld.in)
**Founded by:** Molla Samser
**Designer & Tester:** Rima Khatun
**Contact:** help@rskworld.in | +91 93305 39277
**Year:** 2026
**Category:** Custom Chatbots
**Difficulty:** Intermediate

---

## 📋 Project Description

Fitness Coach Bot is a comprehensive AI-powered chatbot that provides personalized fitness coaching, including workout plans, exercise guidance, progress tracking, and health tips. Perfect for fitness apps and health platforms, this bot serves as your personal AI fitness trainer available 24/7.

---

## ✨ Features

- 🏋️ **Workout Plans**: Personalized workout routines for all fitness levels
- 🏃 **Exercise Guidance**: Proper form instructions and technique tips
- 📊 **Progress Tracking**: Monitor your fitness journey and achievements
- 💡 **Health Tips**: Nutrition advice and wellness recommendations
- 🎯 **Goal Setting**: Set and achieve your fitness objectives
- 💬 **Interactive Chat**: Natural conversation interface
- 📱 **Responsive Design**: Works on all devices
- 🎨 **Modern UI**: Beautiful and intuitive interface

---

## 🛠️ Technologies Used

- **Backend**: Python 3.8+
- **Framework**: Flask
- **Database**: SQLite (with SQLAlchemy ORM)
- **Frontend**: HTML5, CSS3, JavaScript (ES6+)
- **UI Framework**: Bootstrap 5
- **Icons**: Font Awesome
- **AI**: Custom fitness coaching engine with pattern recognition

---

## 📦 Installation & Setup

### Prerequisites

- Python 3.8 or higher
- pip (Python package manager)
- Git

### Step 1: Clone the Repository

```bash
git clone <repository-url>
cd fitness-coach-bot
```

### Step 2: Create Virtual Environment

```bash
# Create virtual environment
python -m venv venv

# Activate virtual environment
# On Windows:
venv\Scripts\activate
# On macOS/Linux:
source venv/bin/activate
```

### Step 3: Install Dependencies

```bash
pip install -r requirements.txt
```

### Step 4: Initialize Database

```bash
python init_db.py
```

This will create the SQLite database and populate it with sample data including exercises, workout plans, and health tips.

### Step 5: Run the Application

```bash
python app.py
```

The application will be available at `http://localhost:5000`

---

## 🏗️ Project Structure

```
fitness-coach-bot/
├── app.py # Main Flask application
├── init_db.py # Database initialization script
├── requirements.txt # Python dependencies
├── README.md # Project documentation
├── models/
│ └── fitness_models.py # Database models
├── utils/
│ └── fitness_coach.py # AI coaching engine
├── templates/
│ └── index.html # Main HTML template
├── static/
│ ├── css/
│ │ └── style.css # Custom styles
│ └── js/
│ └── app.js # Frontend JavaScript
└── data/
└── fitness_coach.db # SQLite database (created after init)
```

---

## 🎯 Usage Guide

### Getting Started

1. **Open the Application**: Navigate to `http://localhost:5000` in your browser
2. **Set Up Your Profile**: Fill in your fitness goals and personal information
3. **Start Chatting**: Ask the bot about workouts, nutrition, or fitness advice

### Sample Conversations

**Workout Plans:**
```
You: Create a beginner workout plan
Bot: Great decision to start your fitness journey! Here's a beginner-friendly approach...
```

**Nutrition Advice:**
```
You: What should I eat for muscle gain?
Bot: Protein for Fitness: Daily Needs: 1.6-2.2g per kg bodyweight...
```

**Progress Tracking:**
```
You: How do I track my fitness progress?
Bot: Tracking Your Fitness Progress: What to Track: Weight, Body Measurements...
```

### Features Explained

#### Chat Interface
- **Natural Conversation**: Chat with the AI coach like talking to a real trainer
- **Quick Suggestions**: Use preset buttons for common questions
- **Real-time Responses**: Get instant fitness advice and guidance

#### User Profile
- **Personal Information**: Age, weight, height
- **Fitness Goals**: Weight loss, muscle gain, endurance, etc.
- **Activity Level**: From sedentary to very active

#### Progress Tracking
- **Workout Count**: Total number of workouts completed
- **Streak Counter**: Consecutive days of activity
- **Calories Burned**: Estimated calories burned
- **Goals Met**: Number of fitness goals achieved

#### Health Tips
- **Randomized Tips**: Get fresh health advice with each refresh
- **Categories**: Nutrition, exercise, recovery, motivation, safety
- **Expert Advice**: Tips from fitness professionals

---

## 🔧 Customization

### Adding New Exercises

```python
# In init_db.py or via admin interface
new_exercise = Exercise(
name="Exercise Name",
description="Exercise description",
category="strength",
muscle_group="target_muscle",
equipment="required_equipment",
difficulty="beginner",
instructions="Step-by-step instructions",
tips="Helpful tips for proper form"
)
db.session.add(new_exercise)
db.session.commit()
```

### Modifying AI Responses

Edit the `utils/fitness_coach.py` file to customize:
- Response patterns
- Exercise recommendations
- Nutrition advice
- Motivational messages

### Styling Changes

Modify `static/css/style.css` to:
- Change color schemes
- Adjust layouts
- Add animations
- Responsive design tweaks

---

## 🔌 API Endpoints

### Chat API
```
POST /api/chat
Content-Type: application/json

{
"message": "Your fitness question here"
}

Response:
{
"success": true,
"response": "AI coach response",
"timestamp": "2026-01-10T21:48:00"
}
```

### Workout Plans API
```
GET /api/workout-plans

Response:
{
"success": true,
"plans": [...]
}
```

### User Profile API
```
GET /api/user/profile
POST /api/user/profile

Content-Type: application/json
{
"name": "John Doe",
"age": 30,
"weight": 70,
"fitness_goal": "muscle_gain"
}
```

---

## 🧪 Testing

### Running Tests

```bash
# Install test dependencies
pip install pytest pytest-flask

# Run tests
pytest tests/
```

### Manual Testing

1. **Chat Functionality**: Test various user inputs and responses
2. **Profile Management**: Test saving and loading user profiles
3. **Progress Tracking**: Verify stats updates
4. **Responsive Design**: Test on different screen sizes

---

## 🚀 Deployment

### Production Deployment

1. **Set Environment Variables**:
```bash
export FLASK_ENV=production
export SECRET_KEY=your-secret-key
```

2. **Use Gunicorn**:
```bash
pip install gunicorn
gunicorn -w 4 -b 0.0.0.0:5000 app:app
```

3. **Database Setup**: Ensure proper database configuration for production

### Docker Deployment

```dockerfile
FROM python:3.9-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .
EXPOSE 5000

CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:5000", "app:app"]
```

---

## 🤝 Contributing

We welcome contributions! Please follow these steps:

1. Fork the repository
2. Create a feature branch (`git checkout -b feature/AmazingFeature`)
3. Commit your changes (`git commit -m 'Add some AmazingFeature'`)
4. Push to the branch (`git push origin feature/AmazingFeature`)
5. Open a Pull Request

---

## 📄 License

This project is part of RSK World's free programming resources. Content used for educational purposes only.

---

## 📞 Support & Contact

**RSK World**
- **Email**: help@rskworld.in
- **Phone**: +91 93305 39277
- **Website**: https://rskworld.in
- **Address**: Nutanhat, Mongolkote, Purba Burdwan, West Bengal, India, 713147

---

## 🙏 Acknowledgments

- **Molla Samser** - Founder & Developer
- **Rima Khatun** - Designer & Tester
- **RSK World Community** - For continuous support and feedback

---

## 📈 Future Enhancements

- [ ] Integration with fitness wearables
- [ ] Video exercise demonstrations
- [ ] Mobile app version
- [ ] Social features and community
- [ ] Advanced analytics and insights
- [ ] Multi-language support
- [ ] Voice interaction capabilities

---

**© 2026 RSK World. All rights reserved.**
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