""" Testing Lab Module - система для тестування нових пацієнтів """ import json import os from datetime import datetime from typing import Dict, List, Optional, Tuple from dataclasses import dataclass, asdict import csv @dataclass class TestSession: """Клас для збереження результатів тестової сесії""" session_id: str patient_name: str timestamp: str total_messages: int medical_messages: int lifestyle_messages: int escalations_count: int controller_decisions: List[Dict] response_times: List[float] session_duration_minutes: float final_profile_state: Dict notes: str = "" @dataclass class TestingMetrics: """Метрики для аналізу тестування""" session_id: str accuracy_score: float # % правильних рішень Controller response_quality_score: float # суб'єктивна оцінка medical_safety_score: float # % правильно виявлених red flags lifestyle_personalization_score: float # % врахування обмежень user_experience_score: float # загальна оцінка UX class TestingDataManager: """Клас для управління тестовими даними та результатами""" def __init__(self): self.results_dir = "testing_results" self.ensure_results_directory() def ensure_results_directory(self): """Створює директорії для збереження результатів""" if not os.path.exists(self.results_dir): os.makedirs(self.results_dir) # Піддиректорії subdirs = ["sessions", "patients", "reports", "exports"] for subdir in subdirs: path = os.path.join(self.results_dir, subdir) if not os.path.exists(path): os.makedirs(path) def validate_clinical_background(self, json_data: dict) -> Tuple[bool, List[str]]: """Валідує структуру clinical_background.json""" errors = [] required_fields = [ "patient_summary", "vital_signs_and_measurements", "assessment_and_plan" ] for field in required_fields: if field not in json_data: errors.append(f"Відсутнє обов'язкове поле: {field}") # Перевірка patient_summary if "patient_summary" in json_data: patient_summary = json_data["patient_summary"] required_sub_fields = ["active_problems", "current_medications"] for field in required_sub_fields: if field not in patient_summary: errors.append(f"Відсутнє поле в patient_summary: {field}") return len(errors) == 0, errors def validate_lifestyle_profile(self, json_data: dict) -> Tuple[bool, List[str]]: """Валідує структуру lifestyle_profile.json""" errors = [] required_fields = [ "patient_name", "patient_age", "conditions", "primary_goal", "exercise_limitations" ] for field in required_fields: if field not in json_data: errors.append(f"Відсутнє обов'язкове поле: {field}") # Перевірка типів даних if "conditions" in json_data and not isinstance(json_data["conditions"], list): errors.append("Поле 'conditions' має бути списком") if "exercise_limitations" in json_data and not isinstance(json_data["exercise_limitations"], list): errors.append("Поле 'exercise_limitations' має бути списком") return len(errors) == 0, errors def save_patient_profile(self, clinical_data: dict, lifestyle_data: dict) -> str: """Зберігає профіль пацієнта для тестування""" patient_name = lifestyle_data.get("patient_name", "Unknown") timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") patient_id = f"{patient_name}_{timestamp}" # Зберігаємо в окремих файлах clinical_path = os.path.join(self.results_dir, "patients", f"{patient_id}_clinical.json") lifestyle_path = os.path.join(self.results_dir, "patients", f"{patient_id}_lifestyle.json") with open(clinical_path, 'w', encoding='utf-8') as f: json.dump(clinical_data, f, indent=2, ensure_ascii=False) with open(lifestyle_path, 'w', encoding='utf-8') as f: json.dump(lifestyle_data, f, indent=2, ensure_ascii=False) return patient_id def save_test_session(self, session: TestSession) -> str: """Зберігає результати тестової сесії""" filename = f"session_{session.session_id}.json" filepath = os.path.join(self.results_dir, "sessions", filename) with open(filepath, 'w', encoding='utf-8') as f: json.dump(asdict(session), f, indent=2, ensure_ascii=False) return filepath def save_testing_metrics(self, metrics: TestingMetrics) -> str: """Зберігає метрики тестування""" filename = f"metrics_{metrics.session_id}.json" filepath = os.path.join(self.results_dir, "sessions", filename) with open(filepath, 'w', encoding='utf-8') as f: json.dump(asdict(metrics), f, indent=2, ensure_ascii=False) return filepath def get_all_test_sessions(self) -> List[Dict]: """Повертає всі збережені тестові сесії""" sessions_dir = os.path.join(self.results_dir, "sessions") sessions = [] for filename in os.listdir(sessions_dir): if filename.startswith("session_") and filename.endswith(".json"): filepath = os.path.join(sessions_dir, filename) try: with open(filepath, 'r', encoding='utf-8') as f: session_data = json.load(f) sessions.append(session_data) except Exception as e: print(f"Помилка читання сесії {filename}: {e}") return sorted(sessions, key=lambda x: x.get('timestamp', ''), reverse=True) def export_results_to_csv(self, sessions: List[Dict]) -> str: """Експортує результати в CSV формат""" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"testing_results_export_{timestamp}.csv" filepath = os.path.join(self.results_dir, "exports", filename) if not sessions: return "" # Визначаємо поля для CSV fieldnames = [ 'session_id', 'patient_name', 'timestamp', 'total_messages', 'medical_messages', 'lifestyle_messages', 'escalations_count', 'session_duration_minutes', 'notes' ] with open(filepath, 'w', newline='', encoding='utf-8') as csvfile: writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() for session in sessions: # Фільтруємо тільки потрібні поля filtered_session = {key: session.get(key, '') for key in fieldnames} writer.writerow(filtered_session) return filepath def generate_summary_report(self, sessions: List[Dict]) -> str: """Генерує звітний текст по результатах тестування""" if not sessions: return "Немає даних для звіту" total_sessions = len(sessions) total_messages = sum(session.get('total_messages', 0) for session in sessions) total_medical = sum(session.get('medical_messages', 0) for session in sessions) total_lifestyle = sum(session.get('lifestyle_messages', 0) for session in sessions) total_escalations = sum(session.get('escalations_count', 0) for session in sessions) # Середні показники avg_messages_per_session = total_messages / total_sessions if total_sessions > 0 else 0 avg_duration = sum(session.get('session_duration_minutes', 0) for session in sessions) / total_sessions # Розподіл по режимах medical_percentage = (total_medical / total_messages * 100) if total_messages > 0 else 0 lifestyle_percentage = (total_lifestyle / total_messages * 100) if total_messages > 0 else 0 escalation_rate = (total_escalations / total_messages * 100) if total_messages > 0 else 0 report = f""" 📊 ЗВІТ ПО ТЕСТУВАННЮ LIFESTYLE JOURNEY {'='*50} 📈 ЗАГАЛЬНА СТАТИСТИКА: • Всього тестових сесій: {total_sessions} • Загальна кількість повідомлень: {total_messages} • Середня тривалість сесії: {avg_duration:.1f} хв • Середня кількість повідомлень на сесію: {avg_messages_per_session:.1f} 🔄 РОЗПОДІЛ ПО РЕЖИМАХ: • Medical режим: {total_medical} ({medical_percentage:.1f}%) • Lifestyle режим: {total_lifestyle} ({lifestyle_percentage:.1f}%) • Ескалації: {total_escalations} ({escalation_rate:.1f}%) 👥 ПАЦІЄНТИ В ТЕСТУВАННІ: """ # Додаємо інформацію про пацієнтів patients = {} for session in sessions: patient_name = session.get('patient_name', 'Unknown') if patient_name not in patients: patients[patient_name] = { 'sessions': 0, 'messages': 0, 'escalations': 0 } patients[patient_name]['sessions'] += 1 patients[patient_name]['messages'] += session.get('total_messages', 0) patients[patient_name]['escalations'] += session.get('escalations_count', 0) for patient_name, stats in patients.items(): report += f"• {patient_name}: {stats['sessions']} сесій, {stats['messages']} повідомлень, {stats['escalations']} ескалацій\n" report += f"\n📅 Період тестування: {sessions[-1].get('timestamp', 'N/A')} - {sessions[0].get('timestamp', 'N/A')}" return report class PatientTestingInterface: """Інтерфейс для тестування нових пацієнтів""" def __init__(self, testing_manager: TestingDataManager): self.testing_manager = testing_manager self.current_session: Optional[TestSession] = None self.session_start_time: Optional[datetime] = None def start_test_session(self, patient_name: str) -> str: """Початок нової тестової сесії""" self.session_start_time = datetime.now() session_id = f"{patient_name}_{self.session_start_time.strftime('%Y%m%d_%H%M%S')}" self.current_session = TestSession( session_id=session_id, patient_name=patient_name, timestamp=self.session_start_time.isoformat(), total_messages=0, medical_messages=0, lifestyle_messages=0, escalations_count=0, controller_decisions=[], response_times=[], session_duration_minutes=0.0, final_profile_state={} ) return f"🧪 Почато тестову сесію: {session_id}" def log_message_interaction(self, mode: str, decision: Dict, response_time: float, escalation: bool): """Логує взаємодію в поточній сесії""" if not self.current_session: return self.current_session.total_messages += 1 if mode == "medical": self.current_session.medical_messages += 1 elif mode == "lifestyle": self.current_session.lifestyle_messages += 1 if escalation: self.current_session.escalations_count += 1 self.current_session.controller_decisions.append({ "timestamp": datetime.now().isoformat(), "mode": mode, "decision": decision, "escalation": escalation }) self.current_session.response_times.append(response_time) def end_test_session(self, final_profile: Dict, notes: str = "") -> str: """Завершення тестової сесії""" if not self.current_session or not self.session_start_time: return "Немає активної сесії для завершення" end_time = datetime.now() duration = (end_time - self.session_start_time).total_seconds() / 60 self.current_session.session_duration_minutes = duration self.current_session.final_profile_state = final_profile self.current_session.notes = notes # Зберігаємо сесію filepath = self.testing_manager.save_test_session(self.current_session) session_id = self.current_session.session_id # Скидаємо поточну сесію self.current_session = None self.session_start_time = None return f"✅ Сесію завершено та збережено: {session_id}\n📁 Файл: {filepath}"