Spaces:
Running
Running
| 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)}") |