app.py

import gradio as gr
import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, Dropout, LSTM
import tensorflow as tf
import json
import datetime
import os
import plotly.express as px
import logging
from typing import Dict, List, Optional

# Set up logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class VRTherapySystem:
    def __init__(self):
        """Initialize the VR Therapy System"""
        try:
            self.data_dir = "vr_therapy_data"
            os.makedirs(self.data_dir, exist_ok=True)
            self.session_data = self._load_or_create_session_data()
            self.user_profiles = self._load_or_create_user_profiles()
            logger.info("VR Therapy System initialized successfully")
        except Exception as e:
            logger.error(f"Error initializing VR Therapy System: {str(e)}")
            raise

    def _load_or_create_session_data(self) -> pd.DataFrame:
        """Load existing session data or create new DataFrame"""
        try:
            file_path = os.path.join(self.data_dir, 'session_data.csv')
            if os.path.exists(file_path):
                return pd.read_csv(file_path)
            else:
                df = pd.DataFrame(columns=[
                    'user_id', 'timestamp', 'session_duration',
                    'pain_reduction', 'mobility_improvement'
                ])
                df.to_csv(file_path, index=False)
                return df
        except Exception as e:
            logger.error(f"Error loading session data: {str(e)}")
            return pd.DataFrame()

    def _load_or_create_user_profiles(self) -> pd.DataFrame:
        """Load existing user profiles or create new DataFrame"""
        try:
            file_path = os.path.join(self.data_dir, 'user_profiles.csv')
            if os.path.exists(file_path):
                return pd.read_csv(file_path)
            else:
                df = pd.DataFrame(columns=[
                    'user_id', 'age', 'condition', 'therapy_goals'
                ])
                df.to_csv(file_path, index=False)
                return df
        except Exception as e:
            logger.error(f"Error loading user profiles: {str(e)}")
            return pd.DataFrame()

    def save_user_profile(self, user_id: str, age: int, condition: str,
                         therapy_goals: str) -> str:
        """Save or update user profile"""
        try:
            new_profile = pd.DataFrame([{
                'user_id': user_id,
                'age': age,
                'condition': condition,
                'therapy_goals': therapy_goals
            }])

            # Update existing or append new
            if user_id in self.user_profiles['user_id'].values:
                self.user_profiles.loc[
                    self.user_profiles['user_id'] == user_id
                ] = new_profile.iloc[0]
            else:
                self.user_profiles = pd.concat(
                    [self.user_profiles, new_profile],
                    ignore_index=True
                )

            # Save to CSV
            self.user_profiles.to_csv(
                os.path.join(self.data_dir, 'user_profiles.csv'),
                index=False
            )
            logger.info(f"Profile saved successfully for user {user_id}")
            return "Profile saved successfully"
        except Exception as e:
            error_msg = f"Error saving user profile: {str(e)}"
            logger.error(error_msg)
            return error_msg

    def generate_therapy_session(self, user_id: str, pain_level: int,
                               mobility_score: int) -> str:
        """Generate a personalized therapy session"""
        try:
            difficulty = self._calculate_difficulty(pain_level, mobility_score)
            session = self._create_session_plan(difficulty)
            logger.info(f"Therapy session generated for user {user_id}")
            return json.dumps(session, indent=2)
        except Exception as e:
            error_msg = f"Error generating therapy session: {str(e)}"
            logger.error(error_msg)
            return json.dumps({"error": error_msg})

    def _calculate_difficulty(self, pain_level: int, mobility_score: int) -> str:
        """Calculate session difficulty"""
        try:
            score = (10 - pain_level) * 0.3 + mobility_score * 0.7
            if score < 4:
                return "basic"
            elif score < 7:
                return "intermediate"
            else:
                return "advanced"
        except Exception as e:
            logger.error(f"Error calculating difficulty: {str(e)}")
            return "basic"

    def _create_session_plan(self, difficulty: str) -> Dict:
        """Create a therapy session plan"""
        exercises = {
            "basic": [
                "Guided Breathing",
                "Gentle Stretching",
                "Simple Range of Motion"
            ],
            "intermediate": [
                "Balance Training",
                "Strength Exercises",
                "Coordination Tasks"
            ],
            "advanced": [
                "Complex Movement Patterns",
                "Endurance Training",
                "Dynamic Balance"
            ]
        }

        return {
            "difficulty": difficulty,
            "exercises": exercises.get(difficulty, exercises["basic"]),
            "duration": 30,
            "rest_periods": "As needed",
            "modifications": "Available upon request"
        }

    def log_session_progress(self, user_id: str, session_duration: int,
                           pain_reduction: int, mobility_improvement: int) -> str:
        """Log therapy session progress"""
        try:
            new_session = pd.DataFrame([{
                'user_id': user_id,
                'timestamp': datetime.datetime.now().isoformat(),
                'session_duration': session_duration,
                'pain_reduction': pain_reduction,
                'mobility_improvement': mobility_improvement
            }])

            self.session_data = pd.concat(
                [self.session_data, new_session],
                ignore_index=True
            )

            # Save to CSV
            self.session_data.to_csv(
                os.path.join(self.data_dir, 'session_data.csv'),
                index=False
            )
            logger.info(f"Session progress logged for user {user_id}")
            return "Session progress logged successfully"
        except Exception as e:
            error_msg = f"Error logging session progress: {str(e)}"
            logger.error(error_msg)
            return error_msg

    def get_user_analytics(self, user_id: str) -> str:
        """Generate user analytics"""
        try:
            user_sessions = self.session_data[
                self.session_data['user_id'] == user_id
            ]

            if len(user_sessions) == 0:
                return json.dumps({"message": "No sessions found for this user"})

            analytics = {
                "total_sessions": len(user_sessions),
                "average_duration": user_sessions['session_duration'].mean(),
                "average_pain_reduction": user_sessions['pain_reduction'].mean(),
                "average_mobility_improvement": user_sessions['mobility_improvement'].mean(),
                "progress_trend": user_sessions['mobility_improvement'].tolist()
            }

            logger.info(f"Analytics generated for user {user_id}")
            return json.dumps(analytics, indent=2)
        except Exception as e:
            error_msg = f"Error generating analytics: {str(e)}"
            logger.error(error_msg)
            return json.dumps({"error": error_msg})

# Create Gradio interface
def create_interface():
    try:
        vr_system = VRTherapySystem()

        with gr.Blocks(title="VR Therapy System") as interface:
            gr.Markdown("# VR Therapy and Rehabilitation System")

            with gr.Tab("User Profile"):
                with gr.Row():
                    user_id = gr.Textbox(label="User ID")
                    age = gr.Number(label="Age")
                    condition = gr.Textbox(label="Medical Condition")
                    therapy_goals = gr.TextArea(label="Therapy Goals")
                save_profile_btn = gr.Button("Save Profile")
                profile_output = gr.Textbox(label="Profile Status")

            with gr.Tab("Therapy Session"):
                with gr.Row():
                    session_user_id = gr.Textbox(label="User ID")
                    pain_level = gr.Slider(1, 10, label="Pain Level")
                    mobility_score = gr.Slider(1, 10, label="Mobility Score")
                generate_btn = gr.Button("Generate Session")
                session_output = gr.JSON(label="Session Plan")

            with gr.Tab("Progress Logging"):
                with gr.Row():
                    log_user_id = gr.Textbox(label="User ID")
                    duration = gr.Number(label="Session Duration (minutes)")
                    pain_reduction = gr.Slider(0, 10, label="Pain Reduction")
                    mobility_improvement = gr.Slider(0, 10, label="Mobility Improvement")
                log_btn = gr.Button("Log Progress")
                log_output = gr.Textbox(label="Logging Status")

            with gr.Tab("Analytics"):
                analytics_user_id = gr.Textbox(label="User ID")
                analytics_btn = gr.Button("Generate Analytics")
                analytics_output = gr.JSON(label="User Analytics")

            # Connect interface functions
            save_profile_btn.click(
                vr_system.save_user_profile,
                inputs=[user_id, age, condition, therapy_goals],
                outputs=profile_output
            )

            generate_btn.click(
                vr_system.generate_therapy_session,
                inputs=[session_user_id, pain_level, mobility_score],
                outputs=session_output
            )

            log_btn.click(
                vr_system.log_session_progress,
                inputs=[log_user_id, duration, pain_reduction, mobility_improvement],
                outputs=log_output
            )

            analytics_btn.click(
                vr_system.get_user_analytics,
                inputs=analytics_user_id,
                outputs=analytics_output
            )

        return interface

    except Exception as e:
        logger.error(f"Error creating interface: {str(e)}")
        raise

# Launch the application
if __name__ == "__main__":
    try:
        interface = create_interface()
        interface.launch(share=True)
        logger.info("VR Therapy System launched successfully")
    except Exception as e:
        logger.error(f"Error launching application: {str(e)}")
        print(f"Error: {str(e)}")