#!/usr/bin/env python3 """ Script to import 10 Point Wound Assessment data into Firebase Maps the DMT 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 Wound Assessment JSON file""" file_path = 'assets/data/dmt/10_point_wound_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 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']}") # 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}") # Commands to use if 'commands_to_use' in guidance: what_to_look_for.append("COMMANDS TO USE:") for command in guidance['commands_to_use']: what_to_look_for.append(f'• "{command}"') # Examples if 'examples' in guidance: what_to_look_for.append("EXAMPLES:") for example in guidance['examples']: what_to_look_for.append(f"• {example}") # Assessment approach if 'assessment_approach' in guidance: what_to_look_for.append("ASSESSMENT APPROACH:") for approach in guidance['assessment_approach']: what_to_look_for.append(f"• {approach}") # Testing approach if 'testing_approach' in guidance: what_to_look_for.append("TESTING APPROACH:") for approach in guidance['testing_approach']: what_to_look_for.append(f"• {approach}") # 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}") # 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}") # Timing considerations if 'timing_considerations' in guidance: what_to_look_for.append("TIMING CONSIDERATIONS:") for timing in guidance['timing_considerations']: what_to_look_for.append(f"• {timing}") # 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}") # Documentation requirements if 'documentation_requirements' in guidance: what_to_look_for.append("DOCUMENTATION REQUIREMENTS:") for req in guidance['documentation_requirements']: what_to_look_for.append(f"• {req}") # Measurement technique if 'measurement_technique' in guidance: what_to_look_for.append("MEASUREMENT TECHNIQUE:") for technique in guidance['measurement_technique']: what_to_look_for.append(f"• {technique}") # Documentation format if 'documentation_format' in guidance: what_to_look_for.append("DOCUMENTATION FORMAT:") for format_item in guidance['documentation_format']: what_to_look_for.append(f"• {format_item}") # Structures to assess if 'structures_to_assess' in guidance: what_to_look_for.append("STRUCTURES TO ASSESS:") for structure in guidance['structures_to_assess']: what_to_look_for.append(f"• {structure}") # Assessment methods if 'assessment_methods' in guidance: what_to_look_for.append("ASSESSMENT METHODS:") for method in guidance['assessment_methods']: what_to_look_for.append(f"• {method}") # 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}") # Time critical factors if 'time_critical_factors' in guidance: what_to_look_for.append("TIME CRITICAL FACTORS:") for factor in guidance['time_critical_factors']: what_to_look_for.append(f"• {factor}") # 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}") # High risk indicators if 'high_risk_indicators' in guidance: what_to_look_for.append("HIGH RISK INDICATORS:") for indicator in guidance['high_risk_indicators']: what_to_look_for.append(f"• {indicator}") # Handle nested objects # 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}") # Three components if 'three_components' in guidance: what_to_look_for.append("THREE COMPONENTS:") for key, value in guidance['three_components'].items(): what_to_look_for.append(f"• {key.upper()}: {value}") # Wound types if 'wound_types' in guidance: what_to_look_for.append("WOUND TYPES:") for wound_type in guidance['wound_types']: what_to_look_for.append(f"• {wound_type}") # Classification factors if 'classification_factors' in guidance: what_to_look_for.append("CLASSIFICATION FACTORS:") for factor in guidance['classification_factors']: what_to_look_for.append(f"• {factor}") # Anatomical description if 'anatomical_description' in guidance: what_to_look_for.append("ANATOMICAL DESCRIPTION:") for desc in guidance['anatomical_description']: what_to_look_for.append(f"• {desc}") # Wound photography if 'wound_photography' in guidance: what_to_look_for.append("WOUND PHOTOGRAPHY:") for photo in guidance['wound_photography']: what_to_look_for.append(f"• {photo}") # 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 'clinical_significance' in guidance: what_to_look_for.append(f"CLINICAL SIGNIFICANCE: {guidance['clinical_significance']}") if 'clinical_importance' in guidance: what_to_look_for.append(f"CLINICAL IMPORTANCE: {guidance['clinical_importance']}") if 'documentation_importance' in guidance: what_to_look_for.append(f"DOCUMENTATION IMPORTANCE: {guidance['documentation_importance']}") return what_to_look_for def create_wpa_assessment_data(original_data: Dict[str, Any]): """Convert the original DMT assessment to WPA Builder format""" assessment_info = original_data['assessment_info'] wound_points = original_data['wound_assessment_points'] grading = original_data['grading'] additional_reqs = original_data['additional_requirements'] key_learning = original_data['key_learning_points'] # Generate assessment ID in WPA Builder format assessment_id = assessment_info.get('assessment_id', f'WPA_{int(datetime.now().timestamp() * 1000)}') assessment_code = 'WPA-WOUND-10PT' # Build comprehensive instruction text instruction_text = f"""{assessment_info['instruction']} TIME LIMIT: {assessment_info['time_limit']} PERFORMANCE LEVEL: {assessment_info['performance_level']} 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']} • Communication: {key_learning['communication']} • Time Management: {key_learning['time_management']} • Safety First: {key_learning['safety_first']} • Documentation: {key_learning['documentation']}""" # Convert wound assessment points to WPA criteria criteria_data = [] for point in wound_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 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_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'], }, } return assessment_data, criteria_data def import_to_firebase(db, assessment_data: Dict[str, Any], criteria_data: List[Dict[str, Any]]): """Import the 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 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)} criteria to subcollection') # Display summary print(f'\n📋 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 to Firebase: {e}') raise def main(): """Main execution function""" print("🚀 Starting 10 Point Wound Assessment import to Firebase...") try: # Initialize Firebase print("🔧 Initializing Firebase...") db = initialize_firebase() # Load assessment data print("📖 Loading 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 Wound Assessment to Firebase!") print(f" Assessment can now be viewed in the WPA Management Screen") print(f" Assessment ID: {assessment_data['assessment_id']}") 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_wound_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())