#!/usr/bin/env python3 """ Script to import 10 Point Burn Assessment data into Firebase Maps the DMT burn assessment format to WPA Builder format for Firebase storage """ import json import os from datetime import datetime from typing import Dict, List, Any import firebase_admin from firebase_admin import credentials, firestore def initialize_firebase(): """Initialize Firebase Admin SDK""" # Use the service account key file cred = credentials.Certificate('logit-learning-firebase-adminsdk-fbsvc-3162c5d079.json') firebase_admin.initialize_app(cred) return firestore.client() def load_assessment_data() -> Dict[str, Any]: """Load the 10 Point Burn Assessment JSON file""" file_path = 'assets/data/dmt/10_point_burn_assessment.json' with open(file_path, 'r', encoding='utf-8') as file: return json.load(file) def build_what_to_look_for(guidance: Dict[str, Any]) -> List[str]: """Build comprehensive 'what to look for' list from burn assessment guidance data""" what_to_look_for = [] # Method if 'method' in guidance: what_to_look_for.append(f"METHOD: {guidance['method']}") # Definition if 'definition' in guidance: what_to_look_for.append(f"DEFINITION: {guidance['definition']}") # Assessment focus if 'assessment_focus' in guidance: what_to_look_for.append(f"ASSESSMENT FOCUS: {guidance['assessment_focus']}") # Key elements if 'key_elements' in guidance: what_to_look_for.append("KEY ELEMENTS:") for element in guidance['key_elements']: what_to_look_for.append(f"• {element}") # Burn specific elements if 'burn_specific' in guidance: what_to_look_for.append("BURN SPECIFIC:") for element in guidance['burn_specific']: what_to_look_for.append(f"• {element}") # Burn mechanisms if 'burn_mechanisms' in guidance: what_to_look_for.append("BURN MECHANISMS:") for mechanism in guidance['burn_mechanisms']: what_to_look_for.append(f"• {mechanism}") # Signs of inhalation injury if 'signs_of_inhalation_injury' in guidance: what_to_look_for.append("SIGNS OF INHALATION INJURY:") for sign in guidance['signs_of_inhalation_injury']: what_to_look_for.append(f"• {sign}") # Respiratory assessment if 'respiratory_assessment' in guidance: what_to_look_for.append("RESPIRATORY ASSESSMENT:") for assessment in guidance['respiratory_assessment']: what_to_look_for.append(f"• {assessment}") # Assessment technique if 'assessment_technique' in guidance: what_to_look_for.append("ASSESSMENT TECHNIQUE:") if isinstance(guidance['assessment_technique'], list): for technique in guidance['assessment_technique']: what_to_look_for.append(f"• {technique}") else: what_to_look_for.append(f"• {guidance['assessment_technique']}") # Assessment approach if 'assessment_approach' in guidance: what_to_look_for.append(f"ASSESSMENT APPROACH: {guidance['assessment_approach']}") # Documentation if 'documentation' in guidance: what_to_look_for.append("DOCUMENTATION:") for doc in guidance['documentation']: what_to_look_for.append(f"• {doc}") # Red flags if 'red_flags' in guidance: what_to_look_for.append("RED FLAGS:") for flag in guidance['red_flags']: what_to_look_for.append(f"• {flag}") # Handle nested objects for burn types if 'burn_types' in guidance: what_to_look_for.append("BURN TYPES:") for burn_type, details in guidance['burn_types'].items(): what_to_look_for.append(f"• {burn_type.upper().replace('_', ' ')}: {details['description']}") if 'examples' in details: what_to_look_for.append(f" Examples: {', '.join(details['examples'])}") if 'characteristics' in details: what_to_look_for.append(f" Characteristics: {details['characteristics']}") # Handle burn depths if 'burn_depths' in guidance: what_to_look_for.append("BURN DEPTHS:") try: for depth, details in guidance['burn_depths'].items(): print(f" Processing burn depth: {depth}") print(f" Details keys: {list(details.keys())}") # Safely access description description = details.get('description', 'No description available') what_to_look_for.append(f"• {depth.upper().replace('_', ' ')}: {description}") if 'appearance' in details: what_to_look_for.append(f" Appearance: {details['appearance']}") if 'healing' in details: what_to_look_for.append(f" Healing: {details['healing']}") if 'treatment' in details: what_to_look_for.append(f" Treatment: {details['treatment']}") # Handle nested partial thickness burns if 'superficial' in details: print(f" Processing superficial for {depth}") superficial = details['superficial'] what_to_look_for.append(f" SUPERFICIAL: {superficial.get('description', 'No description')}") what_to_look_for.append(f" Appearance: {superficial.get('appearance', 'No appearance info')}") what_to_look_for.append(f" Healing: {superficial.get('healing', 'No healing info')}") if 'deep' in details: print(f" Processing deep for {depth}") deep = details['deep'] what_to_look_for.append(f" DEEP: {deep.get('description', 'No description')}") what_to_look_for.append(f" Appearance: {deep.get('appearance', 'No appearance info')}") what_to_look_for.append(f" Healing: {deep.get('healing', 'No healing info')}") except Exception as e: print(f"❌ Error processing burn_depths: {e}") print(f" guidance['burn_depths']: {guidance['burn_depths']}") raise # Handle calculation methods for TBSA if 'calculation_methods' in guidance: what_to_look_for.append("CALCULATION METHODS:") for method, details in guidance['calculation_methods'].items(): what_to_look_for.append(f"• {method.upper().replace('_', ' ')}: {details}") if isinstance(details, dict) and 'adult' in details: what_to_look_for.append(f" Adult proportions:") for body_part, percentage in details['adult'].items(): what_to_look_for.append(f" {body_part.replace('_', ' ')}: {percentage}") # Critical burn criteria if 'critical_burn_criteria' in guidance: what_to_look_for.append("CRITICAL BURN CRITERIA:") for criterion in guidance['critical_burn_criteria']: what_to_look_for.append(f"• {criterion}") # Evacuation criteria if 'evacuation_criteria' in guidance: what_to_look_for.append("EVACUATION CRITERIA:") for criterion in guidance['evacuation_criteria']: what_to_look_for.append(f"• {criterion}") # Immediate threats if 'immediate_threats' in guidance: what_to_look_for.append("IMMEDIATE THREATS:") for threat in guidance['immediate_threats']: what_to_look_for.append(f"• {threat}") # PMS components if 'pms_components' in guidance: what_to_look_for.append("PMS COMPONENTS:") for key, value in guidance['pms_components'].items(): what_to_look_for.append(f"• {key.upper()}: {value}") # Burn specific concerns if 'burn_specific_concerns' in guidance: what_to_look_for.append("BURN SPECIFIC CONCERNS:") for concern in guidance['burn_specific_concerns']: what_to_look_for.append(f"• {concern}") # Infection risk factors if 'infection_risk_factors' in guidance: what_to_look_for.append("INFECTION RISK FACTORS:") for factor in guidance['infection_risk_factors']: what_to_look_for.append(f"• {factor}") # High risk burns if 'high_risk_burns' in guidance: what_to_look_for.append("HIGH RISK BURNS:") for risk in guidance['high_risk_burns']: what_to_look_for.append(f"• {risk}") # Prevention measures if 'prevention_measures' in guidance: what_to_look_for.append("PREVENTION MEASURES:") for measure in guidance['prevention_measures']: what_to_look_for.append(f"• {measure}") # Signs of infection if 'signs_of_infection' in guidance: what_to_look_for.append("SIGNS OF INFECTION:") for sign in guidance['signs_of_infection']: what_to_look_for.append(f"• {sign}") # Evacuation indications if 'evacuation_indications' in guidance: what_to_look_for.append("EVACUATION INDICATIONS:") for indication in guidance['evacuation_indications']: what_to_look_for.append(f"• {indication}") # Timing urgency (nested object) if 'timing_urgency' in guidance: what_to_look_for.append("TIMING URGENCY:") for urgency_level, items in guidance['timing_urgency'].items(): what_to_look_for.append(f"• {urgency_level.upper().replace('_', ' ')}:") for item in items: what_to_look_for.append(f" - {item}") # Method selection if 'method_selection' in guidance: what_to_look_for.append("METHOD SELECTION:") for method in guidance['method_selection']: what_to_look_for.append(f"• {method}") # Preparation for transport if 'preparation_for_transport' in guidance: what_to_look_for.append("PREPARATION FOR TRANSPORT:") for prep in guidance['preparation_for_transport']: what_to_look_for.append(f"• {prep}") # Add importance/significance notes if 'notes' in guidance: what_to_look_for.append(f"NOTES: {guidance['notes']}") if 'importance' in guidance: what_to_look_for.append(f"IMPORTANCE: {guidance['importance']}") if 'critical_importance' in guidance: what_to_look_for.append(f"CRITICAL IMPORTANCE: {guidance['critical_importance']}") if 'clinical_significance' in guidance: what_to_look_for.append(f"CLINICAL SIGNIFICANCE: {guidance['clinical_significance']}") return what_to_look_for def create_wpa_assessment_data(original_data: Dict[str, Any]) -> tuple[Dict[str, Any], List[Dict[str, Any]]]: """Convert the original DMT burn assessment to WPA Builder format""" assessment_info = original_data['assessment_info'] burn_points = original_data['burn_assessment_points'] grading = original_data['grading'] additional_reqs = original_data['additional_requirements'] key_learning = original_data['key_learning_points'] burn_management = original_data['burn_management_principles'] # Generate assessment ID in WPA Builder format assessment_id = assessment_info.get('assessment_id', f'WPA_{int(datetime.now().timestamp() * 1000)}') assessment_code = 'WPA-BURN-10PT' # Build comprehensive instruction text instruction_text = f"""{assessment_info['instruction']} TIME LIMIT: {assessment_info['time_limit']} PERFORMANCE LEVEL: {assessment_info['performance_level']} BURN MANAGEMENT PRINCIPLES: • Immediate Care: {', '.join(burn_management['immediate_care'])} • Cooling Guidelines: {', '.join(burn_management['cooling_guidelines'])} • Wound Care: {', '.join(burn_management['wound_care'])} ADDITIONAL REQUIREMENTS: • Radio Medical Advice: {additional_reqs['radio_medical_advice']} • Documentation: {additional_reqs['documentation']} • Retests Allowed: {additional_reqs['retests_allowed']} KEY LEARNING POINTS: • Systematic Approach: {key_learning['systematic_approach']} • Safety First: {key_learning['safety_first']} • Airway Priority: {key_learning['airway_priority']} • TBSA Calculation: {key_learning['tbsa_calculation']} • Cooling Importance: {key_learning['cooling_importance']} • Evacuation Timing: {key_learning['evacuation_timing']} • Documentation: {key_learning['documentation']}""" # Convert burn assessment points to WPA criteria criteria_data = [] for point in burn_points: guidance = point.get('guidance', {}) what_to_look_for = build_what_to_look_for(guidance) criteria = { 'description': point['description'], 'weight': 1, # Equal weight for all criteria 'sort_code': point['sortcode'], 'what_to_look_for': what_to_look_for, 'order_index': (point.get('sortcode', point.get('point', 1))) - 1, # For Firebase ordering } criteria_data.append(criteria) # Create the main assessment document (exactly as WPA Builder would) assessment_data = { # Metadata fields (required for DMT viewer compatibility) 'assessment_id': assessment_id, 'assessment_code': assessment_code, 'total_score_items': len(criteria_data), 'source_file': f'{assessment_code.lower()}.json', # Assessment details (required for DMT viewer compatibility) 'description': assessment_info['description'], 'assessment_type': 'practical_skills', # Based on the practical nature 'norm': 'Diving Medical Emergency Burn Assessment Protocol', 'active': True, 'auto_activate': additional_reqs.get('auto_activate', True), 'number_of_retests': additional_reqs.get('retests_allowed', 1), 'instruction': instruction_text, # WPA Builder specific fields 'created_by': 'system_import', 'created_at': firestore.SERVER_TIMESTAMP, 'last_modified': firestore.SERVER_TIMESTAMP, # System fields 'version': '1.0', 'data_source': 'dmt_burn_import', # Additional metadata for compatibility 'grading': { 'pass_threshold': float(grading['pass_threshold']), 'scoring_type': grading['scoring_type'], 'scoring_options': grading['scoring_options'], }, # Store original DMT data for reference 'original_dmt_data': { 'assessment_id': assessment_info['assessment_id'], 'aspect_id': assessment_info['aspect_id'], 'time_limit': assessment_info['time_limit'], 'performance_level': assessment_info['performance_level'], }, # Store burn-specific management principles 'burn_management_data': burn_management, } return assessment_data, criteria_data def import_to_firebase(db, assessment_data: Dict[str, Any], criteria_data: List[Dict[str, Any]]): """Import the burn assessment and criteria to Firebase""" assessment_id = assessment_data['assessment_id'] try: # Create the main assessment document db.collection('assessments').document(assessment_id).set(assessment_data) print(f'✅ Created main burn assessment document: {assessment_id}') # Add criteria to subcollection for i, criteria in enumerate(criteria_data): db.collection('assessments').document(assessment_id).collection('criteria').add(criteria) print(f'✅ Added {len(criteria_data)} burn assessment criteria to subcollection') # Display summary print(f'\n🔥 Burn Assessment Summary:') print(f' ID: {assessment_id}') print(f' Code: {assessment_data["assessment_code"]}') print(f' Description: {assessment_data["description"]}') print(f' Type: {assessment_data["assessment_type"]}') print(f' Total Criteria: {len(criteria_data)}') print(f' Active: {assessment_data["active"]}') print(f' Auto-activate: {assessment_data["auto_activate"]}') print(f' Retests Allowed: {assessment_data["number_of_retests"]}') except Exception as e: print(f'❌ Error importing burn assessment to Firebase: {e}') raise def main(): """Main execution function""" print("🔥 Starting 10 Point Burn Assessment import to Firebase...") try: # Initialize Firebase print("🔧 Initializing Firebase...") db = initialize_firebase() # Load assessment data print("📖 Loading burn assessment data...") original_data = load_assessment_data() # Convert to WPA format print("🔄 Converting to WPA Builder format...") assessment_data, criteria_data = create_wpa_assessment_data(original_data) # Import to Firebase print("📤 Importing to Firebase...") import_to_firebase(db, assessment_data, criteria_data) print("\n🎉 Successfully imported 10 Point Burn Assessment to Firebase!") print(f" Assessment can now be viewed in the WPA Management Screen") print(f" Assessment ID: {assessment_data['assessment_id']}") print(f" This assessment follows the same format as manual WPA Builder uploads") except FileNotFoundError: print("❌ Error: Could not find the assessment JSON file or Firebase credentials") print(" Make sure the following files exist:") print(" - assets/data/dmt/10_point_burn_assessment.json") print(" - logit-learning-firebase-adminsdk-fbsvc-3162c5d079.json") except Exception as e: print(f"❌ Error: {e}") return 1 return 0 if __name__ == "__main__": exit(main())