# lifestyle_app.py - Main application class import os import json import time from datetime import datetime from dataclasses import asdict from typing import List, Dict, Optional, Tuple from core_classes import ( ClinicalBackground, LifestyleProfile, ChatMessage, SessionState, AIClientManager, PatientDataLoader, MedicalAssistant, # Active classifiers EntryClassifier, TriageExitClassifier, LifestyleSessionManager, # Main Lifestyle Assistant MainLifestyleAssistant, # Soft medical triage SoftMedicalTriage ) from testing_lab import TestingDataManager, PatientTestingInterface, TestSession from test_patients import TestPatientData from file_utils import FileHandler class ExtendedLifestyleJourneyApp: """Extended version of the app with Testing Lab functionality""" def __init__(self): self.api = AIClientManager() # Active classifiers self.entry_classifier = EntryClassifier(self.api) self.triage_exit_classifier = TriageExitClassifier(self.api) # LifestyleExitClassifier removed - functionality moved to MainLifestyleAssistant # Assistants self.medical_assistant = MedicalAssistant(self.api) self.main_lifestyle_assistant = MainLifestyleAssistant(self.api) self.soft_medical_triage = SoftMedicalTriage(self.api) # Lifecycle manager self.lifestyle_session_manager = LifestyleSessionManager(self.api) # Testing Lab components self.testing_manager = TestingDataManager() self.testing_interface = PatientTestingInterface(self.testing_manager) # Loading standard data print("πŸ”„ Loading standard patient data...") self.clinical_background = PatientDataLoader.load_clinical_background() self.lifestyle_profile = PatientDataLoader.load_lifestyle_profile() print(f"βœ… Loaded standard profile: {self.clinical_background.patient_name}") # App state self.chat_history: List[ChatMessage] = [] self.session_state = SessionState( current_mode="none", is_active_session=False, session_start_time=None, last_controller_decision={} ) # Testing states self.test_mode_active = False self.current_test_patient = None def load_test_patient(self, clinical_file, lifestyle_file) -> Tuple[str, str, List, str]: """Loads test patient from files""" try: # Read clinical background clinical_content, error = FileHandler.read_uploaded_file(clinical_file, "clinical_background.json") if error: return error, "", [], self._get_status_info() clinical_data, error = FileHandler.parse_json_file(clinical_content, "clinical_background.json") if error: return error, "", [], self._get_status_info() # Read lifestyle profile lifestyle_content, error = FileHandler.read_uploaded_file(lifestyle_file, "lifestyle_profile.json") if error: return error, "", [], self._get_status_info() lifestyle_data, error = FileHandler.parse_json_file(lifestyle_content, "lifestyle_profile.json") if error: return error, "", [], self._get_status_info() # Use common processing method return self._process_patient_data(clinical_data, lifestyle_data, "") except Exception as e: return f"❌ File loading error: {str(e)}", "", [], self._get_status_info() def load_quick_test_patient(self, patient_type: str) -> Tuple[str, str, List, str]: """Loads built-in test data for quick testing""" patient_type_names = TestPatientData.get_patient_types() try: clinical_data, lifestyle_data = TestPatientData.get_patient_data(patient_type) test_type_description = patient_type_names.get(patient_type, "") result = self._process_patient_data( clinical_data, lifestyle_data, f"⚑ **Quick test:** {test_type_description}" ) return result except ValueError as e: return f"❌ {str(e)}", "", [], self._get_status_info() except Exception as e: return f"❌ Quick test loading error: {str(e)}", "", [], self._get_status_info() def _process_patient_data(self, clinical_data: dict, lifestyle_data: dict, test_type_info: str = "") -> Tuple[str, str, List, str]: """Common code for processing patient data""" debug_enabled = os.getenv("LOG_PROMPTS", "false").lower() == "true" if debug_enabled: print(f"πŸ”„ _process_patient_data called with test_type_info: '{test_type_info}'") # STEP 1: End previous test session if active if self.test_mode_active and self.testing_interface.current_session: if debug_enabled: print("πŸ”„ Ending previous test session...") self.end_test_session("Automatically ended - new patient loaded") # Clinical data validation is_valid, errors = self.testing_manager.validate_clinical_background(clinical_data) if not is_valid: return f"❌ Clinical background validation error:\n" + "\n".join(errors), "", [], self._get_status_info() # Lifestyle data validation is_valid, errors = self.testing_manager.validate_lifestyle_profile(lifestyle_data) if not is_valid: return f"❌ Lifestyle profile validation error:\n" + "\n".join(errors), "", [], self._get_status_info() # Create objects self.clinical_background = ClinicalBackground( patient_id="test_patient", patient_name=lifestyle_data.get("patient_name", "Test Patient"), patient_age=lifestyle_data.get("patient_age", "unknown"), active_problems=clinical_data.get("patient_summary", {}).get("active_problems", []), past_medical_history=clinical_data.get("patient_summary", {}).get("past_medical_history", []), current_medications=clinical_data.get("patient_summary", {}).get("current_medications", []), allergies=clinical_data.get("patient_summary", {}).get("allergies", ""), vital_signs_and_measurements=clinical_data.get("vital_signs_and_measurements", []), laboratory_results=clinical_data.get("laboratory_results", []), assessment_and_plan=clinical_data.get("assessment_and_plan", ""), critical_alerts=clinical_data.get("critical_alerts", []), social_history=clinical_data.get("social_history", {}), recent_clinical_events=clinical_data.get("recent_clinical_events_and_encounters", []) ) self.lifestyle_profile = LifestyleProfile( patient_name=lifestyle_data.get("patient_name", "Test Patient"), patient_age=lifestyle_data.get("patient_age", "unknown"), conditions=lifestyle_data.get("conditions", []), primary_goal=lifestyle_data.get("primary_goal", ""), exercise_preferences=lifestyle_data.get("exercise_preferences", []), exercise_limitations=lifestyle_data.get("exercise_limitations", []), dietary_notes=lifestyle_data.get("dietary_notes", []), personal_preferences=lifestyle_data.get("personal_preferences", []), journey_summary=lifestyle_data.get("journey_summary", ""), last_session_summary=lifestyle_data.get("last_session_summary", ""), next_check_in=lifestyle_data.get("next_check_in", "not set"), progress_metrics=lifestyle_data.get("progress_metrics", {}) ) # Save test patient profile patient_id = self.testing_manager.save_patient_profile(clinical_data, lifestyle_data) self.current_test_patient = patient_id # Activate test mode self.test_mode_active = True # STEP 2: COMPLETELY RESET CHAT STATE self.chat_history = [] self.session_state = SessionState( current_mode="none", is_active_session=False, session_start_time=None, last_controller_decision={} ) # Start test session session_start_msg = self.testing_interface.start_test_session( self.lifestyle_profile.patient_name ) # Create initial chat message about new patient welcome_content = f"πŸ§ͺ **New test patient loaded: {self.lifestyle_profile.patient_name}**" if test_type_info: welcome_content += f"\n{test_type_info}" welcome_content += "\n\nYou can start the dialogue. All interactions will be logged for analysis." welcome_message = { "role": "assistant", "content": welcome_content } if debug_enabled: print(f"βœ… Created new patient: {self.lifestyle_profile.patient_name}") print(f"πŸ’¬ Welcome message: {welcome_content[:100]}...") success_msg = f"""βœ… **NEW TEST PATIENT LOADED** πŸ‘€ **Patient:** {self.lifestyle_profile.patient_name} ({self.lifestyle_profile.patient_age} years old) πŸ₯ **Active problems:** {len(self.clinical_background.active_problems)} πŸ’Š **Medications:** {len(self.clinical_background.current_medications)} 🎯 **Lifestyle goal:** {self.lifestyle_profile.primary_goal[:100]}... πŸ“‹ **Patient ID:** {patient_id} {session_start_msg} πŸ§ͺ **TEST MODE ACTIVATED** - all interactions will be logged. πŸ’¬ **CHAT RESET** - you can start a new conversation!""" preview = self._generate_patient_preview() # Return: result, preview, CHAT WITH WELCOME MESSAGE, UPDATED STATUS if debug_enabled: print(f"πŸ“€ Returning 4 values: success_msg, preview, chat=[1 message], status") return success_msg, preview, [welcome_message], self._get_status_info() def _generate_patient_preview(self) -> str: """Generates preview of loaded patient""" if not self.clinical_background or not self.lifestyle_profile: return "Patient data not loaded" # Shortened lists for convenient viewing active_problems = self.clinical_background.active_problems[:5] medications = self.clinical_background.current_medications[:8] conditions = self.lifestyle_profile.conditions[:5] preview = f""" πŸ“‹ **MEDICAL PROFILE** πŸ‘€ **Name:** {self.clinical_background.patient_name} πŸŽ‚ **Age:** {self.lifestyle_profile.patient_age} πŸ₯ **Active problems ({len(self.clinical_background.active_problems)}):** {chr(10).join([f"β€’ {problem}" for problem in active_problems])} {"..." if len(self.clinical_background.active_problems) > 5 else ""} πŸ’Š **Medications ({len(self.clinical_background.current_medications)}):** {chr(10).join([f"β€’ {med}" for med in medications])} {"..." if len(self.clinical_background.current_medications) > 8 else ""} 🚨 **Critical alerts:** {len(self.clinical_background.critical_alerts)} πŸ§ͺ **Laboratory results:** {len(self.clinical_background.laboratory_results)} πŸ’š **LIFESTYLE PROFILE** 🎯 **Primary goal:** {self.lifestyle_profile.primary_goal} πŸƒ **Conditions:** {', '.join(conditions)} {"..." if len(self.lifestyle_profile.conditions) > 5 else ""} ⚠️ **Limitations:** {len(self.lifestyle_profile.exercise_limitations)} 🍽️ **Nutrition:** {len(self.lifestyle_profile.dietary_notes)} notes πŸ“ˆ **Progress metrics:** {len(self.lifestyle_profile.progress_metrics)} indicators """ return preview def process_message(self, message: str, history) -> Tuple[List, str]: """New message processing logic with three classifiers""" start_time = time.time() if not message.strip(): return history, self._get_status_info() # Add user message to history user_msg = ChatMessage( timestamp=datetime.now().strftime("%H:%M"), role="user", message=message, mode="pending" # Will be updated after classification ) self.chat_history.append(user_msg) # NEW LOGIC: Determine current state and process accordingly response = "" final_mode = "none" if self.session_state.current_mode == "lifestyle": # If already in lifestyle mode, check if need to exit response, final_mode = self._handle_lifestyle_mode(message) else: # If not in lifestyle mode, use Entry Classifier response, final_mode = self._handle_entry_classification(message) # Update mode in user message user_msg.mode = final_mode # Add assistant response assistant_msg = ChatMessage( timestamp=datetime.now().strftime("%H:%M"), role="assistant", message=response, mode=final_mode ) self.chat_history.append(assistant_msg) # Update session state self.session_state.current_mode = final_mode self.session_state.is_active_session = final_mode != "none" # Logging for testing response_time = time.time() - start_time if self.test_mode_active and self.testing_interface.current_session: self.testing_interface.log_message_interaction( final_mode, {"mode": final_mode, "reasoning": "new_logic"}, response_time, False ) # Update Gradio history if not history: history = [] history.append({"role": "user", "content": message}) history.append({"role": "assistant", "content": response}) return history, self._get_status_info() def _handle_entry_classification(self, message: str) -> Tuple[str, str]: """Processes message through Entry Classifier with new K/V/T format""" # 1. Classify message classification = self.entry_classifier.classify(message, self.clinical_background) self.session_state.entry_classification = classification lifestyle_mode = classification.get("V", "off") if lifestyle_mode == "off": response = self.soft_medical_triage.conduct_triage( message, self.clinical_background, self.chat_history ) return response, "medical" elif lifestyle_mode == "on": # Direct to lifestyle mode self.session_state.lifestyle_session_length = 1 result = self.main_lifestyle_assistant.process_message( message, self.chat_history, self.clinical_background, self.lifestyle_profile, 1 ) return result.get("message", "How are you feeling?"), "lifestyle" elif lifestyle_mode == "hybrid": # Hybrid flow: medical triage + possible lifestyle return self._handle_hybrid_flow(message, classification) else: # Fallback to medical mode with soft triage response = self.soft_medical_triage.conduct_triage( message, self.clinical_background, self.chat_history # Π”ΠΎΠ΄Π°Π½ΠΎ! ) return response, "medical" def _handle_hybrid_flow(self, message: str, classification: Dict) -> Tuple[str, str]: """Handles HYBRID messages: medical triage + lifestyle assessment""" # 1. Medical triage (use regular medical assistant for hybrid) medical_response = self.medical_assistant.generate_response( message, self.chat_history, self.clinical_background ) # Save triage result if medical_response: self.session_state.last_triage_summary = medical_response[:200] + "..." else: self.session_state.last_triage_summary = "Medical assessment completed" # 2. Assess readiness for lifestyle triage_assessment = self.triage_exit_classifier.assess_readiness( self.clinical_background, self.session_state.last_triage_summary, message ) if triage_assessment.get("ready_for_lifestyle", False): # Switch to lifestyle mode with new assistant self.session_state.lifestyle_session_length = 1 result = self.main_lifestyle_assistant.process_message( message, self.chat_history, self.clinical_background, self.lifestyle_profile, 1 ) # Combine responses combined_response = f"{medical_response}\n\n---\n\nπŸ’š **Lifestyle coaching:**\n{result.get('message', 'How are you feeling?')}" return combined_response, "lifestyle" else: # Stay in medical mode return medical_response, "medical" def _handle_lifestyle_mode(self, message: str) -> Tuple[str, str]: """Handles messages in lifestyle mode with new Main Lifestyle Assistant""" # Use new Main Lifestyle Assistant result = self.main_lifestyle_assistant.process_message( message, self.chat_history, self.clinical_background, self.lifestyle_profile, self.session_state.lifestyle_session_length ) action = result.get("action", "lifestyle_dialog") response_message = result.get("message", "How are you feeling?") if action == "close": # End lifestyle session and update profile with LLM analysis self.lifestyle_profile = self.lifestyle_session_manager.update_profile_after_session( self.lifestyle_profile, self.chat_history, f"Automatic session end: {result.get('reasoning', 'MainLifestyleAssistant decided to close')}", save_to_disk=True ) # Switch to medical mode medical_response = self.medical_assistant.generate_response( message, self.chat_history, self.clinical_background ) # Reset lifestyle counter self.session_state.lifestyle_session_length = 0 return f"πŸ’š **Lifestyle session completed.** {result.get('reasoning', '')}\n\n---\n\n{medical_response}", "medical" else: # Continue lifestyle mode (gather_info or lifestyle_dialog) self.session_state.lifestyle_session_length += 1 return response_message, "lifestyle" def end_test_session(self, notes: str = "") -> str: """Ends current test session""" if not self.test_mode_active or not self.testing_interface.current_session: return "❌ No active test session to end" # Get current profile state final_profile = { "clinical_background": asdict(self.clinical_background), "lifestyle_profile": asdict(self.lifestyle_profile), "chat_history_length": len(self.chat_history) } result = self.testing_interface.end_test_session(final_profile, notes) # Turn off test mode self.test_mode_active = False self.current_test_patient = None return result def get_test_results_summary(self) -> Tuple[str, List]: """Returns summary of all test results""" sessions = self.testing_manager.get_all_test_sessions() if not sessions: return "πŸ“­ No saved test sessions", [] # Generate report summary = self.testing_manager.generate_summary_report(sessions) # Create detailed table of recent sessions latest_sessions = sessions[:10] # Last 10 sessions table_data = [] for session in latest_sessions: table_data.append([ session.get('patient_name', 'N/A'), session.get('timestamp', 'N/A')[:16], # Date and time only session.get('total_messages', 0), session.get('medical_messages', 0), session.get('lifestyle_messages', 0), session.get('escalations_count', 0), f"{session.get('session_duration_minutes', 0):.1f} min", session.get('notes', '')[:50] + "..." if len(session.get('notes', '')) > 50 else session.get('notes', '') ]) return summary, table_data def export_test_results(self) -> str: """Exports test results""" sessions = self.testing_manager.get_all_test_sessions() if not sessions: return "❌ No data to export" csv_path = self.testing_manager.export_results_to_csv(sessions) if csv_path and os.path.exists(csv_path): return f"βœ… Data exported to: {csv_path}" else: return "❌ Data export error" def _get_ai_providers_status(self) -> str: """Get detailed AI providers status""" try: clients_info = self.api.get_all_clients_info() status_lines = [] status_lines.append(f"πŸ€– **AI PROVIDERS STATUS:**") status_lines.append(f"β€’ Total API calls: {clients_info['total_calls']}") status_lines.append(f"β€’ Active clients: {clients_info['active_clients']}") if clients_info['clients']: status_lines.append("β€’ Client details:") for agent, info in clients_info['clients'].items(): if 'error' not in info: provider = info['provider'] model = info['model'] fallback = " (fallback)" if info['using_fallback'] else "" status_lines.append(f" - {agent}: {provider} ({model}){fallback}") else: status_lines.append(f" - {agent}: Error - {info['error']}") return "\n".join(status_lines) except Exception as e: return f"πŸ€– **AI PROVIDERS STATUS:** Error - {e}" def _get_status_info(self) -> str: """Extended status information with new logic""" log_prompts_enabled = os.getenv("LOG_PROMPTS", "false").lower() == "true" # Basic information active_problems = self.clinical_background.active_problems[:3] if self.clinical_background.active_problems else ["No data"] problems_text = "; ".join(active_problems) if len(self.clinical_background.active_problems) > 3: problems_text += f" and {len(self.clinical_background.active_problems) - 3} more..." # K/V/T classification information entry_info = "" if self.session_state.entry_classification: classification = self.session_state.entry_classification entry_info = f""" πŸ” **LAST CLASSIFICATION (K/V/T):** β€’ K: {classification.get('K', 'N/A')} β€’ V: {classification.get('V', 'N/A')} β€’ T: {classification.get('T', 'N/A')}""" # Lifestyle session information lifestyle_info = "" if self.session_state.current_mode == "lifestyle": lifestyle_info = f""" πŸ’š **LIFESTYLE SESSION:** β€’ Messages in session: {self.session_state.lifestyle_session_length} β€’ Last summary: {self.lifestyle_profile.last_session_summary[:100]}... """ # Test information test_status = "" if self.test_mode_active: test_status += f"\nπŸ‘€ **ACTIVE TEST PATIENT: {self.lifestyle_profile.patient_name}**" current_session = self.testing_interface.current_session if current_session: test_status += f""" πŸ§ͺ **TEST SESSION ACTIVE** β€’ ID: {current_session.session_id} β€’ Messages: {current_session.total_messages} β€’ Medical: {current_session.medical_messages} | Lifestyle: {current_session.lifestyle_messages} β€’ Escalations: {current_session.escalations_count} """ else: test_status += f"\nπŸ“ Test session not active (loaded but not started)" status = f""" πŸ“Š **SESSION STATE (NEW LOGIC)** β€’ Mode: {self.session_state.current_mode.upper()} β€’ Active: {'βœ…' if self.session_state.is_active_session else '❌'} β€’ Logging: {'πŸ“ ACTIVE' if log_prompts_enabled else '❌ DISABLED'} {entry_info} {lifestyle_info} πŸ‘€ **PATIENT: {self.clinical_background.patient_name}**{' (TEST)' if self.test_mode_active else ''} β€’ Age: {self.lifestyle_profile.patient_age} β€’ Active problems: {problems_text} β€’ Lifestyle goal: {self.lifestyle_profile.primary_goal} πŸ₯ **MEDICAL CONTEXT:** β€’ Medications: {len(self.clinical_background.current_medications)} β€’ Critical alerts: {len(self.clinical_background.critical_alerts)} β€’ Recent vitals: {len(self.clinical_background.vital_signs_and_measurements)} πŸ”§ **AI STATISTICS:** β€’ Total API calls: {self.api.call_counter} β€’ Active AI clients: {len(self.api._clients)} {self._get_ai_providers_status()} {test_status}""" return status def reset_session(self) -> Tuple[List, str]: """Session reset with new logic""" # If test mode is active, end session if self.test_mode_active and self.testing_interface.current_session: self.end_test_session("Session reset by user") # If there was an active lifestyle session, update profile if self.session_state.current_mode == "lifestyle" and self.session_state.lifestyle_session_length > 0: self.lifestyle_profile = self.lifestyle_session_manager.update_profile_after_session( self.lifestyle_profile, self.chat_history, "Session reset by user", save_to_disk=True ) self.chat_history = [] self.session_state = SessionState( current_mode="none", is_active_session=False, session_start_time=None, last_controller_decision={}, lifestyle_session_length=0, last_triage_summary="", entry_classification={} ) return [], self._get_status_info() def end_conversation_with_profile_update(self) -> Tuple[List, str, str]: """Ends conversation with intelligent profile update and saves to disk""" result_message = "" # Check if there's an active lifestyle session to update if (self.session_state.current_mode == "lifestyle" and self.session_state.lifestyle_session_length > 0 and len(self.chat_history) > 0): try: print("πŸ”„ User initiated conversation end - updating lifestyle profile...") # Update profile with LLM analysis and save to disk self.lifestyle_profile = self.lifestyle_session_manager.update_profile_after_session( self.lifestyle_profile, self.chat_history, "User initiated conversation end", save_to_disk=True ) result_message = f"""βœ… **Conversation ended successfully** 🧠 **Profile Analysis Complete**: Lifestyle profile has been intelligently updated based on your session πŸ’Ύ **Saved to Disk**: Changes have been permanently saved to lifestyle_profile.json πŸ“Š **Session Summary**: {len([m for m in self.chat_history if m.mode == 'lifestyle'])} lifestyle messages analyzed Your progress and preferences have been recorded for future sessions.""" except Exception as e: print(f"❌ Error updating profile on conversation end: {e}") result_message = f"⚠️ **Conversation ended** but there was an error updating your profile: {str(e)}" else: result_message = "βœ… **Conversation ended** - No active lifestyle session to update" # If active test mode, end test session if self.test_mode_active and self.testing_interface.current_session: self.end_test_session("User ended conversation manually") # Reset session state self.chat_history = [] self.session_state = SessionState( current_mode="none", is_active_session=False, session_start_time=None, last_controller_decision={}, lifestyle_session_length=0, last_triage_summary="", entry_classification={} ) return [], self._get_status_info(), result_message def sync_custom_prompts_from_session(self, session_data): """Π‘ΠΈΠ½Ρ…Ρ€ΠΎΠ½Ρ–Π·ΡƒΡ” кастомні ΠΏΡ€ΠΎΠΌΠΏΡ‚ΠΈ Π· SessionData""" from prompts import SYSTEM_PROMPT_MAIN_LIFESTYLE if hasattr(session_data, 'custom_prompts') and session_data.custom_prompts: main_lifestyle_prompt = session_data.custom_prompts.get('main_lifestyle') if main_lifestyle_prompt and main_lifestyle_prompt != SYSTEM_PROMPT_MAIN_LIFESTYLE: self.main_lifestyle_assistant.set_custom_system_prompt(main_lifestyle_prompt) else: self.main_lifestyle_assistant.reset_to_default_prompt() def get_current_prompt_info(self) -> Dict[str, str]: """ΠžΡ‚Ρ€ΠΈΠΌΡƒΡ” Ρ–Π½Ρ„ΠΎΡ€ΠΌΠ°Ρ†Ρ–ΡŽ ΠΏΡ€ΠΎ ΠΏΠΎΡ‚ΠΎΡ‡Π½Ρ– ΠΏΡ€ΠΎΠΌΠΏΡ‚ΠΈ""" current_prompt = self.main_lifestyle_assistant.get_current_system_prompt() is_custom = self.main_lifestyle_assistant.custom_system_prompt is not None return { "is_custom": is_custom, "prompt_length": len(current_prompt), "prompt_preview": current_prompt[:100] + "..." if len(current_prompt) > 100 else current_prompt, "status": "Custom prompt active" if is_custom else "Default prompt active" }