Create app.py

app.py

@@ -0,0 +1,290 @@
+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)}")