marwan/hospital-management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
hospital-management/emr_data.py
Marwan Alwali 6e643e1cac added gitignore file
2025-10-06 15:25:37 +03:00

2379 lines
95 KiB
Python
Raw Permalink Blame History

import os
import sys
import django
import argparse
# Set up Django environment
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'hospital_management.settings')
django.setup()
import random
from datetime import datetime, date, time, timedelta
from django.utils import timezone as django_timezone
from django.db import transaction
from emr.models import Encounter, VitalSigns, ProblemList, CarePlan, ClinicalNote, NoteTemplate, Icd10, ClinicalRecommendation, AllergyAlert, TreatmentProtocol, ClinicalGuideline, CriticalAlert, DiagnosticSuggestion
from patients.models import PatientProfile
from accounts.models import User
from hr.models import Employee
from appointments.models import AppointmentRequest
from inpatients.models import Admission
from core.models import Tenant
import uuid
from decimal import Decimal
# Optional import for ICD-10 XML parsing
try:
import xmlschema
XMLSCHEMA_AVAILABLE = True
except ImportError:
XMLSCHEMA_AVAILABLE = False
xmlschema = None
# Saudi-specific clinical data
SAUDI_ENCOUNTER_TYPES = [
'CONSULTATION', 'FOLLOW_UP', 'EMERGENCY', 'PROCEDURE', 'SURGERY',
'DIAGNOSTIC', 'THERAPY', 'SCREENING', 'VACCINATION', 'TELEMEDICINE'
]
SAUDI_ENCOUNTER_CLASSES = [
'OUTPATIENT', 'INPATIENT', 'EMERGENCY', 'OBSERVATION',
'DAY_SURGERY', 'TELEMEDICINE', 'HOME_VISIT'
]
SAUDI_CHIEF_COMPLAINTS = [
'Chest pain and shortness of breath',
'Abdominal pain and nausea',
'Headache and dizziness',
'Back pain and stiffness',
'Fever and cough',
'Joint pain and swelling',
'Fatigue and weakness',
'Skin rash and itching',
'Diabetes follow-up',
'Hypertension monitoring',
'Regular health checkup',
'Vaccination appointment',
'Pre-operative consultation',
'Post-operative follow-up',
'Pregnancy consultation',
'Child wellness visit',
'Mental health consultation',
'Physical therapy session',
'Diagnostic imaging',
'Laboratory test follow-up',
'Cardiac symptoms evaluation',
'Respiratory symptoms',
'Gastrointestinal complaints',
'Neurological symptoms',
'Endocrine disorders follow-up'
]
SAUDI_LOCATIONS = [
'Main Building - Floor 1',
'Main Building - Floor 2',
'Main Building - Floor 3',
'Emergency Wing',
'Outpatient Clinic - Wing A',
'Outpatient Clinic - Wing B',
'Surgical Suite - Floor 4',
'Radiology Department',
'Laboratory Building',
'Pediatric Wing',
'ICU - Floor 5',
'Cardiology Unit',
'Maternity Ward',
'Dialysis Center'
]
SAUDI_PROBLEM_TYPES = [
'DIAGNOSIS', 'SYMPTOM', 'FINDING', 'COMPLAINT',
'CONDITION', 'DISORDER', 'SYNDROME', 'INJURY',
'ALLERGY', 'INTOLERANCE', 'RISK_FACTOR'
]
SAUDI_COMMON_PROBLEMS = [
('Diabetes Mellitus Type 2', 'E11.9', 'ICD10'),
('Hypertension', 'I10', 'ICD10'),
('Hyperlipidemia', 'E78.5', 'ICD10'),
('Asthma', 'J45.9', 'ICD10'),
('Chronic Kidney Disease', 'N18.9', 'ICD10'),
('Osteoarthritis', 'M19.9', 'ICD10'),
('Depression', 'F32.9', 'ICD10'),
('Anxiety Disorder', 'F41.9', 'ICD10'),
('Obesity', 'E66.9', 'ICD10'),
('Gastroesophageal Reflux', 'K21.9', 'ICD10'),
('Migraine', 'G43.9', 'ICD10'),
('Allergic Rhinitis', 'J30.9', 'ICD10'),
('Hypothyroidism', 'E03.9', 'ICD10'),
('Coronary Artery Disease', 'I25.9', 'ICD10'),
('Chronic Obstructive Pulmonary Disease', 'J44.9', 'ICD10')
]
SAUDI_CARE_PLAN_TYPES = [
'COMPREHENSIVE', 'DISEASE_SPECIFIC', 'PREVENTIVE', 'CHRONIC_CARE',
'ACUTE_CARE', 'DISCHARGE', 'REHABILITATION', 'PALLIATIVE',
'MENTAL_HEALTH', 'MEDICATION', 'NUTRITION', 'EXERCISE'
]
SAUDI_NOTE_TYPES = [
'CONSULTATION', 'PROGRESS', 'DISCHARGE', 'PROCEDURE',
'SURGERY', 'EMERGENCY', 'NURSING', 'THERAPY',
'DIAGNOSTIC', 'MEDICATION_REVIEW', 'FOLLOW_UP'
]
def get_providers_for_emr(tenants):
"""Get healthcare providers for EMR data creation"""
providers = []
for tenant in tenants:
tenant_providers = User.objects.filter(
tenant=tenant,
is_active=True,
employee_profile__role__in=['PHYSICIAN', 'NURSE_PRACTITIONER', 'PHYSICIAN_ASSISTANT', 'RADIOLOGIST']
)
providers.extend(list(tenant_providers))
# Create mock providers if none exist
if not providers:
providers = create_mock_providers_for_emr(tenants)
return providers
def create_mock_providers_for_emr(tenants):
"""Create mock healthcare providers for EMR data"""
providers = []
mock_providers_data = [
{'first_name': 'Ahmad', 'last_name': 'Al-Rashid', 'role': 'PHYSICIAN'},
{'first_name': 'Fatima', 'last_name': 'Al-Ghamdi', 'role': 'PHYSICIAN'},
{'first_name': 'Mohammed', 'last_name': 'Al-Otaibi', 'role': 'PHYSICIAN'},
{'first_name': 'Sarah', 'last_name': 'Al-Harbi', 'role': 'PHYSICIAN'},
{'first_name': 'Khalid', 'last_name': 'Al-Mutairi', 'role': 'SURGEON'},
{'first_name': 'Nora', 'last_name': 'Al-Zahrani', 'role': 'NURSE_PRACTITIONER'},
{'first_name': 'Omar', 'last_name': 'Al-Dawsari', 'role': 'PHYSICIAN'},
{'first_name': 'Layla', 'last_name': 'Al-Subai', 'role': 'PHYSICIAN'},
]
for tenant in tenants:
for provider_data in mock_providers_data:
try:
email = f"{provider_data['first_name'].lower()}.{provider_data['last_name'].lower().replace('-', '')}@{tenant.domain}"
existing_user = User.objects.filter(email=email).first()
if not existing_user:
user = User.objects.create_user(
email=email,
first_name=provider_data['first_name'],
last_name=provider_data['last_name'],
employee_profile__role=provider_data['role'],
tenant=tenant,
is_active=True,
password='temp_password_123'
)
providers.append(user)
else:
providers.append(existing_user)
except Exception as e:
print(f"Error creating mock provider {provider_data['first_name']} {provider_data['last_name']}: {e}")
continue
return providers
def create_note_templates(tenants):
"""Create note templates for different specialties"""
templates = []
template_configs = [
{
'name': 'General Consultation Note',
'note_type': 'CONSULTATION',
'specialty': 'FAMILY_MEDICINE',
'content': """
CHIEF COMPLAINT: [Chief Complaint]
HISTORY OF PRESENT ILLNESS:
[History details]
PAST MEDICAL HISTORY:
[Past medical history]
MEDICATIONS:
[Current medications]
ALLERGIES:
[Known allergies]
SOCIAL HISTORY:
[Social history]
PHYSICAL EXAMINATION:
General: [General appearance]
Vital Signs: [Vital signs]
HEENT: [Head, eyes, ears, nose, throat]
Cardiovascular: [Heart examination]
Respiratory: [Lung examination]
Abdomen: [Abdominal examination]
Extremities: [Extremity examination]
Neurological: [Neurological examination]
ASSESSMENT:
[Clinical impression]
PLAN:
[Treatment plan]
[Provider Name], [Title]
[Date and Time]
""",
'structured_fields': [
'chief_complaint', 'history_present_illness', 'past_medical_history',
'medications', 'allergies', 'social_history', 'physical_exam',
'assessment', 'plan'
]
},
{
'name': 'Cardiology Consultation',
'note_type': 'CONSULTATION',
'specialty': 'CARDIOLOGY',
'content': """
CHIEF COMPLAINT: [Cardiac symptoms]
HISTORY OF PRESENT ILLNESS:
[Cardiac history details]
CARDIOVASCULAR RISK FACTORS:
[Risk factors assessment]
MEDICATIONS:
[Cardiac medications]
PHYSICAL EXAMINATION:
Blood Pressure: [BP measurements]
Heart Rate: [HR and rhythm]
Cardiovascular: [Detailed cardiac exam]
Peripheral Vascular: [Peripheral circulation]
ECG: [ECG findings]
LABORATORY RESULTS:
[Cardiac markers, lipids, etc.]
IMAGING:
[Echo, stress test, etc.]
ASSESSMENT:
[Cardiac diagnosis]
PLAN:
[Cardiac management plan]
[Cardiologist Name]
[Date and Time]
""",
'structured_fields': [
'chief_complaint', 'cardiac_history', 'risk_factors',
'cardiac_medications', 'cardiac_exam', 'ecg_findings',
'lab_results', 'imaging_results', 'cardiac_assessment', 'cardiac_plan'
]
},
{
'name': 'Emergency Department Note',
'note_type': 'EMERGENCY',
'specialty': 'EMERGENCY',
'content': """
TIME OF ARRIVAL: [Arrival time]
CHIEF COMPLAINT: [Emergency complaint]
TRIAGE LEVEL: [Triage priority]
HISTORY OF PRESENT ILLNESS:
[Emergency presentation details]
PAST MEDICAL HISTORY:
[Relevant medical history]
MEDICATIONS:
[Current medications]
ALLERGIES:
[Known allergies]
PHYSICAL EXAMINATION:
General Appearance: [Patient appearance]
Vital Signs: [Emergency vital signs]
Primary Survey: [ABCDE assessment]
Secondary Survey: [Detailed examination]
EMERGENCY PROCEDURES:
[Procedures performed]
LABORATORY/IMAGING:
[Emergency testing results]
ASSESSMENT:
[Emergency diagnosis]
DISPOSITION:
[Emergency department disposition]
[Emergency Physician Name]
[Date and Time]
""",
'structured_fields': [
'arrival_time', 'chief_complaint', 'triage_level',
'emergency_history', 'emergency_exam', 'procedures_performed',
'emergency_testing', 'emergency_assessment', 'disposition'
]
},
{
'name': 'Surgical Consultation',
'note_type': 'CONSULTATION',
'specialty': 'SURGERY',
'content': """
REASON FOR CONSULTATION: [Surgical consultation reason]
HISTORY OF PRESENT ILLNESS:
[Surgical problem history]
PAST SURGICAL HISTORY:
[Previous surgeries]
MEDICATIONS:
[Current medications]
ALLERGIES:
[Known allergies]
PHYSICAL EXAMINATION:
[Surgical examination findings]
IMAGING STUDIES:
[Relevant imaging]
LABORATORY RESULTS:
[Preoperative labs]
ASSESSMENT:
[Surgical assessment]
RECOMMENDATIONS:
[Surgical recommendations]
SURGICAL PLAN:
[Proposed surgical intervention]
RISKS AND BENEFITS DISCUSSED:
[Risk discussion]
[Surgeon Name]
[Date and Time]
""",
'structured_fields': [
'consultation_reason', 'surgical_history', 'past_surgeries',
'surgical_exam', 'imaging_studies', 'preop_labs',
'surgical_assessment', 'surgical_recommendations', 'surgical_plan'
]
},
{
'name': 'Progress Note',
'note_type': 'PROGRESS',
'specialty': 'INTERNAL_MEDICINE',
'content': """
SUBJECTIVE:
[Patient's reported symptoms and concerns]
OBJECTIVE:
Vital Signs: [Current vital signs]
Physical Examination: [Current physical findings]
Laboratory Results: [Recent lab results]
Imaging Studies: [Recent imaging]
ASSESSMENT:
[Current clinical assessment]
PLAN:
[Updated treatment plan]
[Provider Name]
[Date and Time]
""",
'structured_fields': [
'subjective', 'vital_signs', 'physical_exam',
'lab_results', 'imaging', 'assessment', 'plan'
]
}
]
for tenant in tenants:
for config in template_configs:
try:
template = NoteTemplate.objects.create(
tenant=tenant,
template_id=uuid.uuid4(),
name=config['name'],
description=f"Standard {config['name'].lower()} template",
note_type=config['note_type'],
specialty=config['specialty'],
template_content=config['content'].strip(),
structured_fields=config['structured_fields'],
is_active=True,
is_default=config['name'] == 'General Consultation Note',
usage_count=random.randint(0, 100),
version=1.0,
quality_indicators=[
'completeness', 'accuracy', 'timeliness',
'clarity', 'compliance'
],
compliance_requirements=[
'documentation_complete_within_24h',
'provider_signature_required',
'patient_privacy_maintained'
],
created_at=django_timezone.now() - timedelta(days=random.randint(30, 365)),
updated_at=django_timezone.now() - timedelta(days=random.randint(0, 30))
)
templates.append(template)
except Exception as e:
print(f"Error creating template {config['name']}: {e}")
continue
print(f"Created {len(templates)} note templates")
return templates
def create_encounters(tenants, days_back=30):
"""Create patient encounters"""
encounters = []
# Get patients and providers
patients = list(PatientProfile.objects.filter(tenant__in=tenants))
if not patients:
print("No patients found. Skipping encounter creation.")
return encounters
providers = get_providers_for_emr(tenants)
if not providers:
print("No providers found. Skipping encounter creation.")
return encounters
# Get existing appointments to link with encounters
appointments = list(AppointmentRequest.objects.filter(
tenant__in=tenants,
status__in=['COMPLETED', 'IN_PROGRESS']
))
# Get existing admissions
admissions = list(Admission.objects.filter(tenant__in=tenants))
start_date = django_timezone.now().date() - timedelta(days=days_back)
# Create encounters for the past days_back days
for day_offset in range(days_back):
encounter_date = start_date + timedelta(days=day_offset)
daily_encounters = random.randint(20, 50) # Variable number of encounters per day
for _ in range(daily_encounters):
patient = random.choice(patients)
# Get providers for this patient's tenant
tenant_providers = [p for p in providers if p.tenant == patient.tenant]
if not tenant_providers:
print(f"Warning: No providers found for tenant {patient.tenant.name}. Skipping encounter for {patient.get_full_name()}")
continue
provider = random.choice(tenant_providers)
# Determine encounter type and class
encounter_type = random.choices(
SAUDI_ENCOUNTER_TYPES,
weights=[25, 20, 10, 8, 5, 12, 8, 7, 3, 2]
)[0]
if encounter_type == 'EMERGENCY':
encounter_class = 'EMERGENCY'
elif encounter_type in ['SURGERY', 'PROCEDURE']:
encounter_class = random.choice(['INPATIENT', 'DAY_SURGERY', 'OUTPATIENT'])
elif encounter_type == 'TELEMEDICINE':
encounter_class = 'TELEMEDICINE'
else:
encounter_class = random.choice(['OUTPATIENT', 'INPATIENT'])
# Generate encounter timing
start_hour = random.randint(8, 20)
start_minute = random.choice([0, 15, 30, 45])
start_datetime = django_timezone.make_aware(
datetime.combine(encounter_date, time(start_hour, start_minute))
)
duration_minutes = random.randint(15, 180)
if encounter_type == 'SURGERY':
duration_minutes = random.randint(60, 360)
elif encounter_type == 'EMERGENCY':
duration_minutes = random.randint(30, 240)
end_datetime = start_datetime + timedelta(minutes=duration_minutes)
# Determine status
if encounter_date < django_timezone.now().date():
status = random.choices(
['PLANNED','COMPLETED', 'CANCELLED'],
weights=[30,55, 15]
)[0]
elif encounter_date == django_timezone.now().date():
status = random.choices(
['IN_PROGRESS', 'COMPLETED'],
weights=[30, 70]
)[0]
else:
status = 'SCHEDULED'
# Priority and acuity
if encounter_type == 'EMERGENCY':
priority = random.choices(['EMERGENCY', 'URGENT'], weights=[60, 40])[0]
acuity_level = random.randint(2, 5)
else:
priority = random.choices(['ROUTINE', 'URGENT'], weights=[80, 20])[0]
acuity_level = random.randint(1, 3)
# Link to appointment if available
linked_appointment = None
if appointments:
suitable_appointments = [
apt for apt in appointments
if apt.patient == patient and
apt.scheduled_datetime and
apt.scheduled_datetime.date() == encounter_date
]
if suitable_appointments:
linked_appointment = random.choice(suitable_appointments)
# Link to admission if inpatient
linked_admission = None
if encounter_class == 'INPATIENT' and admissions:
suitable_admissions = [
adm for adm in admissions
if adm.patient == patient and
adm.admission_datetime <= start_datetime
]
if suitable_admissions:
linked_admission = random.choice(suitable_admissions)
# Documentation completion
documentation_complete = status == 'COMPLETED'
signed_off = documentation_complete and random.choice([True, False])
try:
encounter = Encounter.objects.create(
tenant=patient.tenant,
encounter_id=uuid.uuid4(),
patient=patient,
provider=provider,
encounter_type=encounter_type,
encounter_class=encounter_class,
start_datetime=start_datetime,
end_datetime=end_datetime if status in ['COMPLETED', 'CANCELLED'] else None,
status=status,
location=random.choice(SAUDI_LOCATIONS),
room_number=f"Room {random.randint(101, 450)}",
appointment=linked_appointment,
admission=linked_admission,
chief_complaint=random.choice(SAUDI_CHIEF_COMPLAINTS),
reason_for_visit=random.choice(SAUDI_CHIEF_COMPLAINTS),
priority=priority,
acuity_level=acuity_level,
documentation_complete=documentation_complete,
signed_off=signed_off,
signed_by=provider if signed_off else None,
signed_datetime=end_datetime if signed_off else None,
billable=random.choice([True, False]),
billing_codes=[
f"{random.randint(99200, 99499)}",
f"{random.randint(99200, 99499)}"
] if random.choice([True, False]) else [],
quality_measures={
'documentation_score': random.randint(70, 100),
'timeliness_score': random.randint(80, 100),
'completeness_score': random.randint(75, 100)
},
created_at=start_datetime - timedelta(minutes=random.randint(5, 60)),
updated_at=django_timezone.now() - timedelta(minutes=random.randint(0, 30)),
created_by=provider
)
encounters.append(encounter)
except Exception as e:
print(f"Error creating encounter for {patient.get_full_name()}: {e}")
continue
print(f"Created {len(encounters)} encounters")
return encounters
def create_vital_signs(encounters):
"""Create vital signs for encounters"""
vital_signs = []
providers = User.objects.filter(
employee_profile__role__in=['NURSE', 'NURSE_PRACTITIONER', 'PHYSICIAN', 'MEDICAL_ASSISTANT'],
is_active=True
)
if not providers:
# Use encounter providers as fallback
providers = list(set([enc.provider for enc in encounters]))
for encounter in encounters:
# Create 1-3 vital signs measurements per encounter
num_measurements = random.randint(1, 3)
for measurement_num in range(num_measurements):
# Time offset for multiple measurements
time_offset = measurement_num * random.randint(30, 120) # 30-120 minutes apart
measured_datetime = encounter.start_datetime + timedelta(minutes=time_offset)
# Generate realistic vital signs based on patient age
patient_age = (django_timezone.now().date() - encounter.patient.date_of_birth).days // 365
# Temperature (Celsius)
temperature = round(random.uniform(36.0, 39.5), 1)
if temperature > 38.5:
temperature_method = random.choice(['ORAL', 'RECTAL', 'TYMPANIC'])
else:
temperature_method = random.choice(['ORAL', 'AXILLARY', 'TYMPANIC'])
# Blood pressure
if patient_age < 18:
systolic_bp = random.randint(90, 120)
diastolic_bp = random.randint(50, 80)
elif patient_age < 65:
systolic_bp = random.randint(100, 140)
diastolic_bp = random.randint(60, 90)
else:
systolic_bp = random.randint(110, 160)
diastolic_bp = random.randint(70, 100)
bp_position = random.choice(['SITTING', 'LYING', 'STANDING'])
bp_cuff_size = 'REGULAR' if patient_age >= 18 else 'PEDIATRIC'
# Heart rate
if patient_age < 1:
heart_rate = random.randint(100, 160)
elif patient_age < 12:
heart_rate = random.randint(80, 120)
else:
heart_rate = random.randint(60, 100)
heart_rhythm = random.choices(
['REGULAR', 'REGULARLY_IRREGULAR', 'IRREGULARLY_IRREGULAR', 'IRREGULAR_UNSPECIFIED'],
weights=[85, 8, 5, 2]
)[0]
# Respiratory rate
if patient_age < 1:
respiratory_rate = random.randint(30, 60)
elif patient_age < 12:
respiratory_rate = random.randint(18, 30)
else:
respiratory_rate = random.randint(12, 20)
# Oxygen saturation
oxygen_saturation = random.randint(95, 100)
oxygen_delivery = 'ROOM_AIR'
oxygen_flow_rate = None
if oxygen_saturation < 95:
oxygen_delivery = random.choice(['NASAL_CANNULA', 'MASK', 'NON_REBREATHER'])
oxygen_flow_rate = random.randint(2, 15)
# Pain scale
pain_scale = random.randint(0, 10)
pain_location = None
pain_quality = None
if pain_scale > 0:
pain_location = random.choice([
'Head', 'Chest', 'Abdomen', 'Back', 'Extremities',
'Generalized', 'Neck', 'Pelvis'
])
pain_quality = random.choice([
'Sharp', 'Dull', 'Burning', 'Cramping',
'Stabbing', 'Throbbing', 'Aching'
])
# Weight and height
if patient_age < 18:
weight = round(random.uniform(3.0, 70.0), 1)
height = round(random.uniform(50.0, 180.0), 1)
else:
weight = round(random.uniform(45.0, 150.0), 1)
height = round(random.uniform(150.0, 200.0), 1)
# Head circumference for pediatric patients
head_circumference = None
if patient_age < 2:
head_circumference = round(random.uniform(30.0, 50.0), 1)
# Critical values detection
critical_values = []
if temperature > 39.0 or temperature < 35.0:
critical_values.append('temperature')
if systolic_bp > 180 or systolic_bp < 90:
critical_values.append('systolic_bp')
if heart_rate > 120 or heart_rate < 50:
critical_values.append('heart_rate')
if oxygen_saturation < 90:
critical_values.append('oxygen_saturation')
alerts_generated = len(critical_values) > 0
try:
vital_sign = VitalSigns.objects.create(
encounter=encounter,
patient=encounter.patient,
measurement_id=uuid.uuid4(),
measured_datetime=measured_datetime,
temperature=Decimal(str(temperature)),
temperature_method=temperature_method,
systolic_bp=systolic_bp,
diastolic_bp=diastolic_bp,
bp_position=bp_position,
bp_cuff_size=bp_cuff_size,
heart_rate=heart_rate,
heart_rhythm=heart_rhythm,
respiratory_rate=respiratory_rate,
oxygen_saturation=oxygen_saturation,
oxygen_delivery=oxygen_delivery,
oxygen_flow_rate=oxygen_flow_rate,
pain_scale=pain_scale,
pain_location=pain_location,
pain_quality=pain_quality,
weight=Decimal(str(weight)),
height=Decimal(str(height)),
head_circumference=Decimal(str(head_circumference)) if head_circumference else None,
device_used=random.choice(['MANUAL', 'AUTOMATED', 'MONITOR']),
device_calibrated=random.choice([True, False]),
measured_by=random.choice(providers),
verified_by=random.choice(providers) if random.choice([True, False]) else None,
critical_values=critical_values,
alerts_generated=alerts_generated,
notes=f"Routine vital signs measurement #{measurement_num + 1}" if measurement_num > 0 else None,
created_at=measured_datetime,
updated_at=measured_datetime + timedelta(minutes=random.randint(1, 10))
)
vital_signs.append(vital_sign)
except Exception as e:
print(f"Error creating vital signs for encounter {encounter.encounter_id}: {e}")
continue
print(f"Created {len(vital_signs)} vital signs records")
return vital_signs
def create_problem_lists(patients, providers):
"""Create problem lists for patients"""
problems = []
for patient in patients:
# Each patient has 1-5 problems
num_problems = random.randint(1, 5)
for _ in range(num_problems):
problem_data = random.choice(SAUDI_COMMON_PROBLEMS)
problem_name, problem_code, coding_system = problem_data
problem_type = random.choices(
SAUDI_PROBLEM_TYPES,
weights=[60, 20, 10, 5, 2, 2, 1, 1, 1, 1, 1]
)[0]
# Onset date (random date in the past)
onset_date = django_timezone.now().date() - timedelta(
days=random.randint(30, 3650) # 1 month to 10 years ago
)
# Severity
severity = random.choices(
['MILD', 'MODERATE', 'SEVERE', 'CRITICAL'],
weights=[40, 35, 20, 5]
)[0]
# Priority
priority = random.choices(
['HIGH', 'MEDIUM', 'LOW'],
weights=[20, 60, 20]
)[0]
# Status
status = random.choices(
['ACTIVE', 'RESOLVED', 'INACTIVE', 'REMISSION'],
weights=[50, 20, 10, 20]
)[0]
# Resolution date for resolved problems
resolution_date = None
resolution_notes = None
if status == 'RESOLVED':
resolution_date = onset_date + timedelta(
days=random.randint(7, 365)
)
resolution_notes = f"Problem resolved after treatment. Patient improved significantly."
# Provider assignments
tenant_providers = [p for p in providers if p.tenant == patient.tenant]
if not tenant_providers:
print(f"Warning: No providers found for tenant {patient.tenant.name}. Skipping problem for {patient.get_full_name()}")
continue
diagnosing_provider = random.choice(tenant_providers)
managing_provider = diagnosing_provider if random.choice([True, False]) else random.choice(tenant_providers)
# Fix timezone issue
verified_date = None
if random.choice([True, False]):
verified_date = django_timezone.now()
try:
problem = ProblemList.objects.create(
tenant=patient.tenant,
patient=patient,
problem_id=uuid.uuid4(),
problem_name=problem_name,
problem_code=problem_code,
coding_system=coding_system,
problem_type=problem_type,
onset_date=onset_date,
onset_description=f"Patient reports {problem_name.lower()} started gradually",
severity=severity,
priority=priority,
status=status,
resolution_date=resolution_date,
resolution_notes=resolution_notes,
diagnosing_provider=diagnosing_provider,
managing_provider=managing_provider,
body_site=random.choice([
'Chest', 'Abdomen', 'Head', 'Extremities', 'Back', 'Neck'
]) if random.choice([True, False]) else None,
laterality=random.choice(['LEFT', 'RIGHT', 'BILATERAL']) if random.choice([True, False]) else None,
clinical_notes=f"Patient diagnosed with {problem_name}. {severity.lower()} severity. Status: {status.lower()}.",
patient_concerns=f"Patient concerned about {problem_name.lower()} impact on daily activities",
treatment_goals=[
f"Manage {problem_name.lower()} symptoms",
"Improve quality of life",
"Prevent complications"
],
outcome_measures=[
"Symptom improvement",
"Functional status",
"Patient satisfaction"
],
verified=random.choice([True, False]),
verified_by=managing_provider if random.choice([True, False]) else None,
verified_date=verified_date, # Use timezone-aware datetime
created_at=django_timezone.make_aware(
datetime.combine(onset_date, time(random.randint(8, 17), random.randint(0, 59)))
),
updated_at=django_timezone.now() - timedelta(days=random.randint(0, 30)),
created_by=diagnosing_provider
)
problems.append(problem)
except Exception as e:
print(f"Error creating problem for {patient.get_full_name()}: {e}")
continue
print(f"Created {len(problems)} problem list entries")
return problems
def create_care_plans(patients, providers, problems):
"""Create care plans for patients"""
care_plans = []
for patient in patients:
# Each patient has 1-2 care plans
num_plans = random.randint(1, 2)
patient_problems = [p for p in problems if p.patient == patient]
if not patient_problems:
continue
for _ in range(num_plans):
plan_type = random.choice(SAUDI_CARE_PLAN_TYPES)
# Select related problems for this care plan
related_problems_list = random.sample(
patient_problems,
min(random.randint(1, 3), len(patient_problems))
)
# Generate care plan title and description
main_problem = related_problems_list[0].problem_name
title = f"{plan_type.replace('_', ' ').title()} Plan for {main_problem}"
description = f"Comprehensive {plan_type.lower().replace('_', ' ')} plan for managing {main_problem} and related conditions."
# Care plan category
category = random.choice([
'ASSESSMENT', 'TREATMENT', 'EDUCATION', 'COORDINATION',
'PREVENTION', 'LIFESTYLE', 'MEDICATION', 'FOLLOW_UP',
'EMERGENCY', 'SUPPORT'
])
# Timeline
start_date = django_timezone.now().date()
end_date = start_date + timedelta(days=random.randint(30, 365))
target_completion_date = end_date - timedelta(days=random.randint(0, 30))
# Status
status = random.choices(
['ACTIVE', 'COMPLETED', 'ON_HOLD', 'CANCELLED'],
weights=[60, 25, 10, 5]
)[0]
# Priority
priority = random.choices(
['LOW', 'ROUTINE', 'URGENT', 'STAT'],
weights=[25, 50, 20, 5]
)[0]
# Providers
tenant_providers = [p for p in providers if p.tenant == patient.tenant]
if not tenant_providers:
print(f"Warning: No providers found for tenant {patient.tenant.name}. Skipping care plan for {patient.get_full_name()}")
continue
primary_provider = random.choice(tenant_providers)
care_team_list = random.sample(
tenant_providers,
min(random.randint(2, 4), len(tenant_providers))
)
# Goals and objectives
goals = [
f"Manage {main_problem} effectively",
"Improve patient quality of life",
"Prevent disease progression",
"Optimize medication therapy"
]
objectives = [
f"Reduce {main_problem.lower()} symptoms by 50%",
"Achieve target vital signs",
"Improve functional status",
"Enhance patient understanding"
]
# Interventions and activities
interventions = [
"Medication management",
"Patient education",
"Lifestyle modifications",
"Regular monitoring"
]
activities = [
"Daily medication review",
"Weekly patient assessment",
"Monthly provider consultation",
"Quarterly outcome evaluation"
]
# Completion percentage
completion_percentage = random.randint(0, 100) if status in ['ACTIVE', 'COMPLETED'] else 0
if status == 'COMPLETED':
completion_percentage = 100
# Approval
approved = random.choice([True, False])
approved_date = django_timezone.now() if approved else None
try:
care_plan = CarePlan.objects.create(
tenant=patient.tenant,
patient=patient,
care_plan_id=uuid.uuid4(),
title=title,
description=description,
plan_type=plan_type,
category=category,
start_date=start_date,
end_date=end_date,
target_completion_date=target_completion_date,
status=status,
priority=priority,
primary_provider=primary_provider,
goals=goals,
objectives=objectives,
interventions=interventions,
activities=activities,
monitoring_parameters=[
"Vital signs",
"Symptom severity",
"Functional status",
"Medication adherence"
],
evaluation_criteria=[
"Symptom improvement",
"Quality of life measures",
"Patient satisfaction",
"Clinical outcomes"
],
patient_goals=[
"Feel better",
"Return to normal activities",
"Understand my condition",
"Take medications correctly"
],
patient_preferences=[
"Prefer oral medications",
"Morning appointments preferred",
"Family involvement desired"
],
patient_barriers=[
"Transportation challenges",
"Language barriers",
"Financial constraints"
] if random.choice([True, False]) else [],
resources_needed=[
"Medication",
"Medical equipment",
"Educational materials",
"Support services"
],
support_systems=[
"Family support",
"Healthcare team",
"Community resources"
],
progress_notes=f"Patient progressing well with {plan_type.lower().replace('_', ' ')} plan.",
last_reviewed=django_timezone.now().date() - timedelta(days=random.randint(0, 30)),
next_review_date=django_timezone.now().date() + timedelta(days=random.randint(7, 90)),
outcomes_achieved=[
"Symptom reduction achieved",
"Patient education completed"
] if completion_percentage > 50 else ["No outcomes achieved yet"],
completion_percentage=completion_percentage,
approved=approved,
approved_by=primary_provider if approved else None,
approved_date=approved_date, # Use timezone-aware datetime
created_at=django_timezone.make_aware(
datetime.combine(start_date, time(random.randint(8, 17), random.randint(0, 59)))
),
updated_at=django_timezone.now() - timedelta(days=random.randint(0, 7)),
created_by=primary_provider
)
# Set many-to-many relationships after creation
care_plan.care_team.set(care_team_list)
care_plan.related_problems.set(related_problems_list)
care_plans.append(care_plan)
except Exception as e:
print(f"Error creating care plan for {patient.get_full_name()}: {e}")
continue
print(f"Created {len(care_plans)} care plans")
return care_plans
def create_clinical_notes(encounters, templates):
"""Create clinical notes for encounters"""
clinical_notes = []
for encounter in encounters:
if encounter.status not in ['COMPLETED', 'IN_PROGRESS']:
continue
# Each encounter has 1-2 clinical notes
num_notes = random.randint(1, 2)
for note_num in range(num_notes):
# Select appropriate template
suitable_templates = [
t for t in templates
if t.tenant == encounter.tenant and (
t.note_type == encounter.encounter_type or
t.specialty == 'FAMILY_MEDICINE'
)
]
template = random.choice(suitable_templates) if suitable_templates else None
# Note type based on encounter
note_type = encounter.encounter_type if encounter.encounter_type in [nt for nt in
SAUDI_NOTE_TYPES] else 'CONSULTATION'
# Generate note title
titles = {
'CONSULTATION': f"Consultation Note - {encounter.chief_complaint}",
'EMERGENCY': f"Emergency Department Note - {encounter.chief_complaint}",
'PROGRESS': f"Progress Note - Day {note_num + 1}",
'SURGERY': f"Surgical Note - {encounter.chief_complaint}",
'PROCEDURE': f"Procedure Note - {encounter.chief_complaint}",
'DISCHARGE': f"Discharge Summary - {encounter.chief_complaint}"
}
title = titles.get(note_type, f"Clinical Note - {encounter.chief_complaint}")
# Generate note content
if template:
content = template.template_content
# Replace placeholders with actual data
content = content.replace('[Chief Complaint]', encounter.chief_complaint)
content = content.replace('[Provider Name]', encounter.provider.get_full_name())
content = content.replace('[Date and Time]', encounter.start_datetime.strftime('%Y-%m-%d %H:%M'))
content = content.replace('[History details]',
f"Patient presents with {encounter.chief_complaint.lower()}.")
content = content.replace('[Clinical impression]',
f"Impression consistent with {encounter.chief_complaint.lower()}.")
content = content.replace('[Treatment plan]',
"Continue current management plan and follow up as scheduled.")
else:
content = f"""
ENCOUNTER DATE: {encounter.start_datetime.strftime('%Y-%m-%d %H:%M')}
PATIENT: {encounter.patient.get_full_name()}
PROVIDER: {encounter.provider.get_full_name()}
CHIEF COMPLAINT: {encounter.chief_complaint}
HISTORY OF PRESENT ILLNESS:
Patient presents with {encounter.chief_complaint.lower()}. Symptoms have been present for several days.
PHYSICAL EXAMINATION:
Patient appears comfortable. Vital signs stable.
ASSESSMENT:
Clinical impression consistent with {encounter.chief_complaint.lower()}.
PLAN:
Continue current management. Follow up in clinic as needed.
Electronically signed by: {encounter.provider.get_full_name()}
Date: {encounter.start_datetime.strftime('%Y-%m-%d %H:%M')}
""".strip()
# Note timing
note_datetime = encounter.start_datetime + timedelta(minutes=random.randint(15, 120))
# Status and signing
if encounter.status == 'COMPLETED':
note_status = random.choices(
['SIGNED', 'PENDING_SIGNATURE', 'DRAFT'],
weights=[70, 20, 10]
)[0]
else:
note_status = 'DRAFT'
electronically_signed = note_status == 'SIGNED'
signed_datetime = note_datetime + timedelta(
minutes=random.randint(30, 180)) if electronically_signed else None
# Co-signers list
co_signers_list = []
if random.choice([True, False]): # 50% chance of co-signer
potential_cosigners = [
p for p in User.objects.filter(
tenant=encounter.tenant,
employee_profile__role__in=['PHYSICIAN', 'NURSE_PRACTITIONER'],
is_active=True
) if p != encounter.provider
]
if potential_cosigners:
co_signers_list = [random.choice(potential_cosigners)]
# Quality and compliance
quality_score = random.randint(75, 100)
compliance_flags = []
if quality_score < 85:
compliance_flags = ['documentation_incomplete', 'signature_pending']
try:
clinical_note = ClinicalNote.objects.create(
encounter=encounter,
patient=encounter.patient,
note_id=uuid.uuid4(),
note_type=note_type,
title=title,
content=content,
template=template,
structured_data={
'chief_complaint': encounter.chief_complaint,
'encounter_type': encounter.encounter_type,
'provider': encounter.provider.get_full_name(),
'location': encounter.location
},
author=encounter.provider,
status=note_status,
electronically_signed=electronically_signed,
signed_datetime=signed_datetime,
signature_method='ELECTRONIC' if electronically_signed else None,
quality_score=quality_score,
compliance_flags=compliance_flags,
note_datetime=note_datetime,
confidential=random.choice([True, False]),
restricted_access=random.choice([True, False]) if random.random() < 0.1 else False,
access_restrictions=['PROVIDER_ONLY'] if random.choice([True, False]) else [],
created_at=note_datetime,
updated_at=note_datetime + timedelta(minutes=random.randint(5, 60))
)
# Set many-to-many relationship after creation
clinical_note.co_signers.set(co_signers_list)
clinical_notes.append(clinical_note)
except Exception as e:
print(f"Error creating clinical note for encounter {encounter.encounter_id}: {e}")
continue
print(f"Created {len(clinical_notes)} clinical notes")
return clinical_notes
def create_icd10_codes(tenants):
"""Create ICD-10 codes for clinical reference"""
icd10_codes = []
# Sample ICD-10 codes with descriptions
icd10_data = [
# Chapter 1: Certain infectious and parasitic diseases
('A00', 'Cholera', 'Certain infectious and parasitic diseases', 'Cholera and other vibrio infections'),
('A01', 'Typhoid and paratyphoid fevers', 'Certain infectious and parasitic diseases', 'Typhoid and paratyphoid fevers'),
('A02', 'Other salmonella infections', 'Certain infectious and parasitic diseases', 'Other salmonella infections'),
('A03', 'Shigellosis', 'Certain infectious and parasitic diseases', 'Shigellosis'),
('A04', 'Other bacterial intestinal infections', 'Certain infectious and parasitic diseases', 'Other bacterial intestinal infections'),
# Chapter 2: Neoplasms
('C00', 'Malignant neoplasm of lip', 'Neoplasms', 'Malignant neoplasms of lip, oral cavity and pharynx'),
('C01', 'Malignant neoplasm of base of tongue', 'Neoplasms', 'Malignant neoplasms of lip, oral cavity and pharynx'),
('C02', 'Malignant neoplasm of other and unspecified parts of tongue', 'Neoplasms', 'Malignant neoplasms of lip, oral cavity and pharynx'),
('C03', 'Malignant neoplasm of gum', 'Neoplasms', 'Malignant neoplasms of lip, oral cavity and pharynx'),
('C04', 'Malignant neoplasm of floor of mouth', 'Neoplasms', 'Malignant neoplasms of lip, oral cavity and pharynx'),
# Chapter 9: Diseases of the circulatory system
('I00', 'Rheumatic fever without mention of heart involvement', 'Diseases of the circulatory system', 'Acute rheumatic fever'),
('I01', 'Rheumatic fever with heart involvement', 'Diseases of the circulatory system', 'Acute rheumatic fever'),
('I02', 'Rheumatic chorea', 'Diseases of the circulatory system', 'Acute rheumatic fever'),
('I05', 'Rheumatic mitral valve diseases', 'Diseases of the circulatory system', 'Chronic rheumatic heart diseases'),
('I06', 'Rheumatic aortic valve diseases', 'Diseases of the circulatory system', 'Chronic rheumatic heart diseases'),
# Chapter 10: Diseases of the respiratory system
('J00', 'Acute nasopharyngitis [common cold]', 'Diseases of the respiratory system', 'Acute upper respiratory infections'),
('J01', 'Acute sinusitis', 'Diseases of the respiratory system', 'Acute upper respiratory infections'),
('J02', 'Acute pharyngitis', 'Diseases of the respiratory system', 'Acute upper respiratory infections'),
('J03', 'Acute tonsillitis', 'Diseases of the respiratory system', 'Acute upper respiratory infections'),
('J04', 'Acute laryngitis and tracheitis', 'Diseases of the respiratory system', 'Acute upper respiratory infections'),
# Chapter 11: Diseases of the digestive system
('K00', 'Disorders of tooth development and eruption', 'Diseases of the digestive system', 'Disorders of teeth and supporting structures'),
('K01', 'Embedded and impacted teeth', 'Diseases of the digestive system', 'Disorders of teeth and supporting structures'),
('K02', 'Dental caries', 'Diseases of the digestive system', 'Disorders of teeth and supporting structures'),
('K03', 'Other diseases of hard tissues of teeth', 'Diseases of the digestive system', 'Disorders of teeth and supporting structures'),
('K04', 'Diseases of pulp and periapical tissues', 'Diseases of the digestive system', 'Disorders of teeth and supporting structures'),
# Chapter 14: Diseases of the genitourinary system
('N00', 'Acute nephritic syndrome', 'Diseases of the genitourinary system', 'Glomerular diseases'),
('N01', 'Rapidly progressive nephritic syndrome', 'Diseases of the genitourinary system', 'Glomerular diseases'),
('N02', 'Recurrent and persistent hematuria', 'Diseases of the genitourinary system', 'Glomerular diseases'),
('N03', 'Chronic nephritic syndrome', 'Diseases of the genitourinary system', 'Glomerular diseases'),
('N04', 'Nephrotic syndrome', 'Diseases of the genitourinary system', 'Glomerular diseases'),
# Chapter 18: Symptoms, signs and abnormal clinical and laboratory findings
('R00', 'Abnormalities of heart beat', 'Symptoms, signs and abnormal clinical and laboratory findings', 'Symptoms and signs involving the circulatory and respiratory systems'),
('R01', 'Cardiac murmurs and other cardiac sounds', 'Symptoms, signs and abnormal clinical and laboratory findings', 'Symptoms and signs involving the circulatory and respiratory systems'),
('R02', 'Gangrene, not elsewhere classified', 'Symptoms, signs and abnormal clinical and laboratory findings', 'Symptoms and signs involving the circulatory and respiratory systems'),
('R03', 'Abnormal blood-pressure reading, without diagnosis', 'Symptoms, signs and abnormal clinical and laboratory findings', 'Symptoms and signs involving the circulatory and respiratory systems'),
('R04', 'Haemorrhage from respiratory passages', 'Symptoms, signs and abnormal clinical and laboratory findings', 'Symptoms and signs involving the circulatory and respiratory systems'),
]
for tenant in tenants:
for code_data in icd10_data:
code, description, chapter, section = code_data
# Determine if it's a header (shorter codes are typically headers)
is_header = len(code) <= 3
# Create parent relationship for sub-codes
parent = None
if len(code) > 3:
parent_code = code[:3]
try:
parent = Icd10.objects.get(code=parent_code, tenant=tenant)
except Icd10.DoesNotExist:
parent = None
try:
icd10_code = Icd10.objects.create(
tenant=tenant,
code=code,
description=description,
chapter_name=chapter,
section_name=section,
parent=parent,
is_header=is_header,
created_at=django_timezone.now() - timedelta(days=random.randint(30, 365)),
updated_at=django_timezone.now() - timedelta(days=random.randint(0, 30))
)
icd10_codes.append(icd10_code)
except Exception as e:
print(f"Error creating ICD-10 code {code}: {e}")
continue
print(f"Created {len(icd10_codes)} ICD-10 codes")
return icd10_codes
def create_clinical_recommendations(patients, providers, problems, encounters):
"""Create clinical recommendations for patients"""
recommendations = []
recommendation_templates = [
{
'category': 'PREVENTIVE',
'title': 'Annual Health Screening',
'description': 'Patient due for annual comprehensive health screening including blood work, imaging, and preventive counseling.',
'evidence_level': '1A',
'source': 'USPSTF Guidelines 2023',
'rationale': 'Regular screening improves early detection and prevention of chronic diseases.',
'priority': 'MEDIUM'
},
{
'category': 'DIAGNOSTIC',
'title': 'Cardiac Evaluation Recommended',
'description': 'Consider echocardiography and stress testing given patient history and risk factors.',
'evidence_level': '1B',
'source': 'ACC/AHA Guidelines 2022',
'rationale': 'Patient presents with cardiac symptoms requiring further evaluation.',
'priority': 'HIGH'
},
{
'category': 'TREATMENT',
'title': 'Optimize Diabetes Management',
'description': 'Consider adjusting diabetes regimen based on recent HbA1c and glucose readings.',
'evidence_level': '1A',
'source': 'ADA Standards of Care 2023',
'rationale': 'Current glycemic control may be suboptimal requiring treatment intensification.',
'priority': 'HIGH'
},
{
'category': 'MONITORING',
'title': 'Blood Pressure Monitoring',
'description': 'Increase frequency of blood pressure monitoring and consider 24-hour ambulatory monitoring.',
'evidence_level': '1B',
'source': 'JNC 8 Guidelines',
'rationale': 'Recent readings suggest need for closer monitoring and possible treatment adjustment.',
'priority': 'MEDIUM'
},
{
'category': 'LIFESTYLE',
'title': 'Weight Management Program',
'description': 'Recommend enrollment in structured weight management program with dietary counseling.',
'evidence_level': '1A',
'source': 'Obesity Guidelines 2023',
'rationale': 'Patient BMI indicates need for weight reduction to improve health outcomes.',
'priority': 'MEDIUM'
},
{
'category': 'MEDICATION',
'title': 'Statin Therapy Consideration',
'description': 'Consider initiating statin therapy based on cardiovascular risk assessment.',
'evidence_level': '1A',
'source': 'ACC/AHA Cholesterol Guidelines',
'rationale': 'Patient risk factors warrant preventive cardiovascular therapy.',
'priority': 'HIGH'
},
{
'category': 'FOLLOW_UP',
'title': 'Specialist Consultation',
'description': 'Refer to cardiology/endocrinology for comprehensive evaluation.',
'evidence_level': '2A',
'source': 'Clinical judgment',
'rationale': 'Complex medical condition requires specialist input.',
'priority': 'HIGH'
},
{
'category': 'EDUCATION',
'title': 'Diabetes Self-Management Education',
'description': 'Patient would benefit from structured diabetes education program.',
'evidence_level': '1A',
'source': 'ADA Standards',
'rationale': 'Education improves self-management and clinical outcomes.',
'priority': 'MEDIUM'
}
]
for patient in patients:
# Create 1-3 recommendations per patient
num_recommendations = random.randint(1, 3)
patient_problems = [p for p in problems if p.patient == patient]
patient_encounters = [e for e in encounters if e.patient == patient]
for _ in range(num_recommendations):
template = random.choice(recommendation_templates)
# Generate unique title based on patient context
if patient_problems:
main_problem = random.choice(patient_problems).problem_name
title = f"{template['title']} - {main_problem}"
else:
title = template['title']
# Status progression
status = random.choices(
['PENDING', 'ACTIVE', 'ACCEPTED', 'DEFERRED', 'COMPLETED'],
weights=[40, 30, 15, 10, 5]
)[0]
# Provider assignments
tenant_providers = [p for p in providers if p.tenant == patient.tenant]
if not tenant_providers:
print(f"Warning: No providers found for tenant {patient.tenant.name}. Skipping recommendation for {patient.get_full_name()}")
continue
created_by_provider = random.choice(tenant_providers)
accepted_by = None
accepted_at = None
if status in ['ACCEPTED', 'COMPLETED']:
accepted_by = random.choice(tenant_providers)
accepted_at = django_timezone.now() - timedelta(days=random.randint(1, 30))
deferred_by = None
deferred_at = None
if status == 'DEFERRED':
deferred_by = random.choice(tenant_providers)
deferred_at = django_timezone.now() - timedelta(days=random.randint(1, 14))
dismissed_by = None
dismissed_at = None
if status == 'DISMISSED':
dismissed_by = random.choice(tenant_providers)
dismissed_at = django_timezone.now() - timedelta(days=random.randint(1, 7))
# Expiration
expires_at = django_timezone.now() + timedelta(days=random.randint(30, 180))
# Related data
related_problems_list = random.sample(
patient_problems,
min(random.randint(0, 2), len(patient_problems))
) if patient_problems else []
related_encounter = random.choice(patient_encounters) if patient_encounters else None
try:
recommendation = ClinicalRecommendation.objects.create(
tenant=patient.tenant,
patient=patient,
recommendation_id=uuid.uuid4(),
title=title,
description=template['description'],
category=template['category'],
priority=template['priority'],
evidence_level=template['evidence_level'],
source=template['source'],
rationale=template['rationale'],
status=status,
accepted_by=accepted_by,
accepted_at=accepted_at,
deferred_by=deferred_by,
deferred_at=deferred_at,
dismissed_by=dismissed_by,
dismissed_at=dismissed_at,
expires_at=expires_at,
related_encounter=related_encounter,
created_by=created_by_provider,
created_at=django_timezone.now() - timedelta(days=random.randint(1, 60)),
updated_at=django_timezone.now() - timedelta(days=random.randint(0, 7))
)
# Set many-to-many relationships
if related_problems_list:
recommendation.related_problems.set(related_problems_list)
recommendations.append(recommendation)
except Exception as e:
print(f"Error creating clinical recommendation for {patient.get_full_name()}: {e}")
continue
print(f"Created {len(recommendations)} clinical recommendations")
return recommendations
def create_allergy_alerts(patients, providers):
"""Create allergy alerts for patients"""
alerts = []
common_allergens = [
'Penicillin', 'Aspirin', 'Ibuprofen', 'Codeine', 'Morphine', 'Latex',
'Shellfish', 'Peanuts', 'Tree Nuts', 'Eggs', 'Milk', 'Soy',
'Wheat', 'Sulfa Drugs', 'Tetracycline', 'Erythromycin', 'Cephalosporins',
'NSAIDs', 'Contrast Dye', 'Local Anesthetics', 'Insulin', 'ACE Inhibitors'
]
reaction_types = [
'Anaphylaxis', 'Angioedema', 'Urticaria', 'Rash', 'Pruritus',
'Nausea/Vomiting', 'Diarrhea', 'Dyspnea', 'Wheezing', 'Hypotension',
'Tachycardia', 'Bronchospasm', 'Stevens-Johnson Syndrome', 'Toxic Epidermal Necrolysis'
]
severities = ['MILD', 'MODERATE', 'SEVERE', 'LIFE_THREATENING']
for patient in patients:
# 20% of patients have allergies
if random.random() > 0.2:
continue
# Each allergic patient has 1-3 allergies
num_allergies = random.randint(1, 3)
for _ in range(num_allergies):
allergen = random.choice(common_allergens)
# Ensure no duplicate allergens for same patient
existing_allergens = [alert.allergen for alert in alerts if alert.patient == patient]
while allergen in existing_allergens:
allergen = random.choice(common_allergens)
reaction_type = random.choice(reaction_types)
# Severity based on reaction type
if 'Anaphylaxis' in reaction_type or 'Stevens-Johnson' in reaction_type:
severity = random.choices(severities, weights=[0, 10, 30, 60])[0]
elif 'Angioedema' in reaction_type or 'Bronchospasm' in reaction_type:
severity = random.choices(severities, weights=[10, 30, 40, 20])[0]
else:
severity = random.choices(severities, weights=[40, 40, 15, 5])[0]
# Symptoms based on reaction type
symptoms = []
if 'Rash' in reaction_type or 'Urticaria' in reaction_type:
symptoms.extend(['Hives', 'Itching', 'Redness'])
if 'Anaphylaxis' in reaction_type:
symptoms.extend(['Difficulty breathing', 'Swelling of throat', 'Dizziness', 'Nausea'])
if 'Angioedema' in reaction_type:
symptoms.extend(['Facial swelling', 'Tongue swelling', 'Lip swelling'])
if 'Dyspnea' in reaction_type or 'Wheezing' in reaction_type:
symptoms.extend(['Shortness of breath', 'Wheezing', 'Chest tightness'])
symptoms = symptoms[:random.randint(1, 4)] if symptoms else ['Unknown symptoms']
# Onset timing
onset = random.choice([
'Immediate (< 1 hour)', 'Early (1-6 hours)', 'Delayed (6-24 hours)',
'Late (> 24 hours)', 'Unknown'
])
# Status
tenant_providers = [p for p in providers if p.tenant == patient.tenant]
if not tenant_providers:
print(f"Warning: No providers found for tenant {patient.tenant.name}. Skipping allergy alert for {patient.get_full_name()}")
continue
resolved = random.choice([True, False])
resolved_at = django_timezone.now() - timedelta(days=random.randint(1, 365)) if resolved else None
resolved_by = random.choice(tenant_providers) if resolved else None
try:
alert = AllergyAlert.objects.create(
tenant=patient.tenant,
patient=patient,
alert_id=uuid.uuid4(),
allergen=allergen,
reaction_type=reaction_type,
severity=severity,
symptoms=symptoms,
onset=onset,
resolved=resolved,
resolved_at=resolved_at,
resolved_by=resolved_by,
detected_at=django_timezone.now() - timedelta(days=random.randint(1, 3650)),
created_at=django_timezone.now() - timedelta(days=random.randint(1, 3650)),
updated_at=django_timezone.now() - timedelta(days=random.randint(0, 30))
)
alerts.append(alert)
except Exception as e:
print(f"Error creating allergy alert for {patient.get_full_name()}: {e}")
continue
print(f"Created {len(alerts)} allergy alerts")
return alerts
def create_treatment_protocols(tenants, providers):
"""Create treatment protocols"""
protocols = []
protocol_templates = [
{
'name': 'Acute Coronary Syndrome Management',
'indication': 'Management of patients with suspected or confirmed acute coronary syndrome',
'goals': ['Rapid reperfusion', 'Pain control', 'Complication prevention'],
'interventions': ['ECG within 10 minutes', 'Cardiac enzymes', 'Antiplatelet therapy', 'Anticoagulation'],
'monitoring_parameters': ['ECG changes', 'Cardiac enzymes', 'Vital signs', 'Pain assessment'],
'success_rate': 85.0,
'average_duration': 7
},
{
'name': 'Diabetes Mellitus Type 2 Management',
'indication': 'Comprehensive management of type 2 diabetes mellitus',
'goals': ['Glycemic control (HbA1c < 7%)', 'Cardiovascular risk reduction', 'Prevention of complications'],
'interventions': ['Lifestyle modification', 'Oral hypoglycemics', 'Insulin therapy', 'Regular monitoring'],
'monitoring_parameters': ['HbA1c', 'Fasting glucose', 'Blood pressure', 'Lipid profile'],
'success_rate': 75.0,
'average_duration': 365
},
{
'name': 'Community Acquired Pneumonia Treatment',
'indication': 'Treatment of community-acquired pneumonia in adults',
'goals': ['Clinical cure', 'Prevention of complications', 'Appropriate antibiotic use'],
'interventions': ['Appropriate antibiotics', 'Supportive care', 'Vaccination assessment'],
'monitoring_parameters': ['Clinical improvement', 'Fever resolution', 'Oxygen saturation'],
'success_rate': 90.0,
'average_duration': 10
},
{
'name': 'Hypertension Management Protocol',
'indication': 'Management of essential hypertension',
'goals': ['BP control (< 130/80 mmHg)', 'Cardiovascular risk reduction', 'Organ protection'],
'interventions': ['Lifestyle modification', 'Antihypertensive therapy', 'Regular monitoring'],
'monitoring_parameters': ['Blood pressure', 'Electrolyte levels', 'Renal function'],
'success_rate': 70.0,
'average_duration': 180
},
{
'name': 'Acute Asthma Exacerbation Protocol',
'indication': 'Management of acute asthma exacerbations',
'goals': ['Rapid symptom relief', 'Prevention of respiratory failure', 'Hospitalization prevention'],
'interventions': ['Bronchodilators', 'Systemic corticosteroids', 'Oxygen therapy', 'Monitoring'],
'monitoring_parameters': ['Peak flow', 'Oxygen saturation', 'Respiratory rate', 'Clinical symptoms'],
'success_rate': 88.0,
'average_duration': 3
}
]
for tenant in tenants:
for template in protocol_templates:
# Get providers for this tenant
tenant_providers = [p for p in providers if p.tenant == tenant]
if not tenant_providers:
print(f"Warning: No providers found for tenant {tenant.name}. Skipping protocol {template['name']}")
continue
try:
protocol = TreatmentProtocol.objects.create(
tenant=tenant,
protocol_id=uuid.uuid4(),
name=template['name'],
description=f"Standardized protocol for {template['name'].lower()}",
indication=template['indication'],
goals=template['goals'],
interventions=template['interventions'],
monitoring_parameters=template['monitoring_parameters'],
success_rate=Decimal(str(template['success_rate'])),
average_duration=template['average_duration'],
is_active=True,
usage_count=random.randint(10, 200),
created_at=django_timezone.now() - timedelta(days=random.randint(30, 365)),
updated_at=django_timezone.now() - timedelta(days=random.randint(0, 30)),
created_by=random.choice(tenant_providers)
)
protocols.append(protocol)
except Exception as e:
print(f"Error creating treatment protocol {template['name']}: {e}")
continue
print(f"Created {len(protocols)} treatment protocols")
return protocols
def create_clinical_guidelines(tenants):
"""Create clinical guidelines"""
guidelines = []
guideline_templates = [
{
'title': 'Management of Heart Failure',
'organization': 'American Heart Association',
'summary': 'Comprehensive guidelines for the diagnosis and management of heart failure',
'publication_date': date(2022, 4, 1),
'version': '2022',
'keywords': ['heart failure', 'cardiology', 'HFrEF', 'HFpEF'],
'specialties': ['Cardiology', 'Internal Medicine', 'Family Medicine']
},
{
'title': 'Standards of Medical Care in Diabetes',
'organization': 'American Diabetes Association',
'summary': 'Comprehensive guidelines for the diagnosis and management of diabetes mellitus',
'publication_date': date(2023, 1, 1),
'version': '2023',
'keywords': ['diabetes', 'glycemia', 'complications', 'prevention'],
'specialties': ['Endocrinology', 'Internal Medicine', 'Family Medicine']
},
{
'title': 'Guidelines for the Prevention, Detection, Evaluation, and Management of High Blood Pressure',
'organization': 'American College of Cardiology/American Heart Association',
'summary': 'Evidence-based guidelines for hypertension management',
'publication_date': date(2017, 11, 13),
'version': '2017',
'keywords': ['hypertension', 'blood pressure', 'cardiovascular', 'risk'],
'specialties': ['Cardiology', 'Internal Medicine', 'Nephrology']
},
{
'title': 'Global Initiative for Asthma (GINA) Report',
'organization': 'Global Initiative for Asthma',
'summary': 'Global strategy for asthma management and prevention',
'publication_date': date(2023, 5, 1),
'version': '2023',
'keywords': ['asthma', 'respiratory', 'inhalers', 'control'],
'specialties': ['Pulmonology', 'Allergy', 'Pediatrics']
},
{
'title': 'Prevention and Management of Osteoporosis',
'organization': 'World Health Organization',
'summary': 'Guidelines for osteoporosis prevention and treatment',
'publication_date': date(2022, 1, 1),
'version': '2022',
'keywords': ['osteoporosis', 'bone density', 'fracture', 'calcium'],
'specialties': ['Endocrinology', 'Rheumatology', 'Geriatrics']
}
]
for tenant in tenants:
for template in guideline_templates:
try:
guideline = ClinicalGuideline.objects.create(
tenant=tenant,
guideline_id=uuid.uuid4(),
title=template['title'],
organization=template['organization'],
summary=template['summary'],
url=f"https://example.com/guidelines/{template['title'].lower().replace(' ', '_')}",
publication_date=template['publication_date'],
last_updated=django_timezone.now().date() - timedelta(days=random.randint(0, 365)),
version=template['version'],
is_active=True,
keywords=template['keywords'],
specialties=template['specialties'],
created_at=django_timezone.now() - timedelta(days=random.randint(30, 365)),
updated_at=django_timezone.now() - timedelta(days=random.randint(0, 30))
)
guidelines.append(guideline)
except Exception as e:
print(f"Error creating clinical guideline {template['title']}: {e}")
continue
print(f"Created {len(guidelines)} clinical guidelines")
return guidelines
def create_critical_alerts(patients, providers, encounters):
"""Create critical alerts for patients"""
alerts = []
alert_templates = [
{
'title': 'Critical Lab Value Alert',
'description': 'Potassium level critically elevated at 7.2 mEq/L',
'priority': 'CRITICAL',
'recommendation': 'Immediate treatment with calcium gluconate, insulin, and glucose. Transfer to ICU.'
},
{
'title': 'Acute Coronary Syndrome Alert',
'description': 'ST-elevation myocardial infarction detected on ECG',
'priority': 'CRITICAL',
'recommendation': 'Activate cardiac catheterization lab. Administer aspirin, heparin, and prepare for PCI.'
},
{
'title': 'Severe Hypoglycemia Alert',
'description': 'Blood glucose critically low at 35 mg/dL',
'priority': 'CRITICAL',
'recommendation': 'Administer IV dextrose immediately. Recheck glucose in 15 minutes.'
},
{
'title': 'Acute Stroke Alert',
'description': 'Suspected acute ischemic stroke with NIHSS score of 18',
'priority': 'URGENT',
'recommendation': 'Activate stroke team. Obtain CT brain immediately. Prepare for thrombolytic therapy.'
},
{
'title': 'Sepsis Alert',
'description': 'SIRS criteria met with suspected infection. Lactate elevated at 4.2 mmol/L',
'priority': 'URGENT',
'recommendation': 'Obtain blood cultures, start broad-spectrum antibiotics within 1 hour, fluid resuscitation.'
},
{
'title': 'Drug Interaction Alert',
'description': 'Critical drug interaction between warfarin and newly prescribed antibiotic',
'priority': 'HIGH',
'recommendation': 'Monitor INR closely. Consider dose adjustment or alternative antibiotic.'
},
{
'title': 'Allergy Alert - Severe Reaction',
'description': 'Patient with history of anaphylaxis to penicillin now prescribed amoxicillin',
'priority': 'HIGH',
'recommendation': 'Discontinue amoxicillin. Prescribe alternative antibiotic. Consider premedication if necessary.'
}
]
for patient in patients:
# 5% of patients have critical alerts
if random.random() > 0.05:
continue
# Each patient with alerts has 1-2 critical alerts
num_alerts = random.randint(1, 2)
patient_encounters = [e for e in encounters if e.patient == patient]
for _ in range(num_alerts):
template = random.choice(alert_templates)
# Status
tenant_providers = [p for p in providers if p.tenant == patient.tenant]
if not tenant_providers:
print(f"Warning: No providers found for tenant {patient.tenant.name}. Skipping critical alert for {patient.get_full_name()}")
continue
acknowledged = random.choice([True, False])
acknowledged_by = random.choice(tenant_providers) if acknowledged else None
acknowledged_at = django_timezone.now() - timedelta(hours=random.randint(1, 24)) if acknowledged else None
# Expiration
expires_at = django_timezone.now() + timedelta(hours=random.randint(24, 168))
# Related encounter
related_encounter = random.choice(patient_encounters) if patient_encounters else None
try:
alert = CriticalAlert.objects.create(
tenant=patient.tenant,
patient=patient,
alert_id=uuid.uuid4(),
title=template['title'],
description=template['description'],
priority=template['priority'],
recommendation=template['recommendation'],
acknowledged=acknowledged,
acknowledged_by=acknowledged_by,
acknowledged_at=acknowledged_at,
expires_at=expires_at,
related_encounter=related_encounter,
created_by=random.choice(tenant_providers),
created_at=django_timezone.now() - timedelta(hours=random.randint(1, 48)),
updated_at=django_timezone.now() - timedelta(minutes=random.randint(0, 60))
)
alerts.append(alert)
except Exception as e:
print(f"Error creating critical alert for {patient.get_full_name()}: {e}")
continue
print(f"Created {len(alerts)} critical alerts")
return alerts
def create_diagnostic_suggestions(patients, providers, encounters):
"""Create diagnostic suggestions for patients"""
suggestions = []
suggestion_templates = [
{
'test_name': 'Echocardiogram',
'test_code': 'ECHO',
'indication': 'Evaluate cardiac function and structure',
'confidence': 85.0
},
{
'test_name': 'Coronary Angiography',
'test_code': 'CATH',
'indication': 'Assess coronary artery disease',
'confidence': 78.0
},
{
'test_name': 'MRI Brain',
'test_code': 'MRI_BRAIN',
'indication': 'Evaluate neurological symptoms',
'confidence': 72.0
},
{
'test_name': 'CT Chest',
'test_code': 'CT_CHEST',
'indication': 'Assess pulmonary pathology',
'confidence': 80.0
},
{
'test_name': 'Colonoscopy',
'test_code': 'COLON',
'indication': 'Screen for colorectal cancer',
'confidence': 88.0
},
{
'test_name': 'DEXA Scan',
'test_code': 'DEXA',
'indication': 'Assess bone mineral density',
'confidence': 90.0
},
{
'test_name': 'Thyroid Function Tests',
'test_code': 'TFT',
'indication': 'Evaluate thyroid dysfunction',
'confidence': 75.0
},
{
'test_name': 'HbA1c',
'test_code': 'HBA1C',
'indication': 'Assess long-term glycemic control',
'confidence': 95.0
}
]
for patient in patients:
# 15% of patients have diagnostic suggestions
if random.random() > 0.15:
continue
# Each patient with suggestions has 1-3 suggestions
num_suggestions = random.randint(1, 3)
patient_encounters = [e for e in encounters if e.patient == patient]
for _ in range(num_suggestions):
template = random.choice(suggestion_templates)
# Status
status = random.choices(
['PENDING', 'ORDERED', 'COMPLETED'],
weights=[50, 30, 20]
)[0]
# Provider assignments
tenant_providers = [p for p in providers if p.tenant == patient.tenant]
if not tenant_providers:
print(f"Warning: No providers found for tenant {patient.tenant.name}. Skipping diagnostic suggestion for {patient.get_full_name()}")
continue
ordered_by = None
ordered_at = None
if status in ['ORDERED', 'COMPLETED']:
ordered_by = random.choice(tenant_providers)
ordered_at = django_timezone.now() - timedelta(days=random.randint(1, 14))
try:
suggestion = DiagnosticSuggestion.objects.create(
tenant=patient.tenant,
patient=patient,
suggestion_id=uuid.uuid4(),
test_name=template['test_name'],
test_code=template['test_code'],
indication=template['indication'],
confidence=Decimal(str(template['confidence'])),
status=status,
ordered_by=ordered_by,
ordered_at=ordered_at,
created_by=random.choice(tenant_providers),
created_at=django_timezone.now() - timedelta(days=random.randint(1, 30)),
updated_at=django_timezone.now() - timedelta(days=random.randint(0, 7))
)
suggestions.append(suggestion)
except Exception as e:
print(f"Error creating diagnostic suggestion for {patient.get_full_name()}: {e}")
continue
print(f"Created {len(suggestions)} diagnostic suggestions")
return suggestions
# ============================================================================
# ICD-10 XML IMPORT FUNCTIONS (from import_icd10.py management command)
# ============================================================================
def _as_text(val):
"""Convert value to text string"""
if val is None:
return None
if isinstance(val, dict):
return val.get("#text") or val.get("@value") or val.get("value") or str(val)
return str(val)
def _ensure_list(maybe_list):
"""Ensure value is a list"""
if maybe_list is None:
return []
if isinstance(maybe_list, list):
return maybe_list
return [maybe_list]
def _find_first_with_key(data, key):
"""Depth-first search: return the first dict that directly contains `key`"""
if isinstance(data, dict):
if key in data:
return data
for v in data.values():
found = _find_first_with_key(v, key)
if found is not None:
return found
elif isinstance(data, list):
for item in data:
found = _find_first_with_key(item, key)
if found is not None:
return found
return None
def _collect_icd10_rows(chapters, tenants):
"""
Build Icd10 rows + parent links from chapters dict/list.
Expected structure: chapter -> section? -> diag (recursive)
"""
rows = []
parent_links = []
def import_diag(diag, chapter_name, section_name, parent_code, tenant):
code = _as_text(diag.get("name"))
desc = _as_text(diag.get("desc"))
if not code:
return
children = _ensure_list(diag.get("diag"))
is_header = bool(children) and not (desc and desc.strip())
rows.append(Icd10(
tenant=tenant,
code=code,
description=desc,
chapter_name=_as_text(chapter_name),
section_name=_as_text(section_name),
parent=None, # set later
is_header=is_header,
))
if parent_code:
parent_links.append((code, parent_code, tenant))
for child in children:
import_diag(child, chapter_name, section_name, parent_code=code, tenant=tenant)
# Normalize chapters to a list
chapters = _ensure_list(chapters)
# Process for each tenant
for tenant in tenants:
for ch in chapters:
ch_name = _as_text(ch.get("name"))
# Sections may be missing; diags may be directly under chapter
sections = _ensure_list(ch.get("section"))
if sections:
for sec in sections:
sec_name = _as_text(sec.get("name"))
for d in _ensure_list(sec.get("diag")):
import_diag(d, ch_name, sec_name, parent_code=None, tenant=tenant)
else:
# If no sections, look for diags at chapter level
for d in _ensure_list(ch.get("diag")):
import_diag(d, ch_name, None, parent_code=None, tenant=tenant)
return rows, parent_links
def import_icd10_from_xml(xsd_path, xml_path, tenants, truncate=False):
"""
Import ICD-10-CM codes from XML file.
Args:
xsd_path: Path to XSD schema file
xml_path: Path to XML data file
tenants: List of tenant objects to import codes for
truncate: Whether to delete existing codes first
Returns:
Number of codes imported
"""
if not XMLSCHEMA_AVAILABLE:
print("❌ Error: xmlschema library not installed.")
print(" Install it with: pip install xmlschema")
return 0
print(f"\n📥 Importing ICD-10 codes from XML...")
print(f" XSD: {xsd_path}")
print(f" XML: {xml_path}")
try:
xs = xmlschema.XMLSchema(xsd_path)
except Exception as e:
print(f"❌ Failed to load XSD: {e}")
return 0
try:
data = xs.to_dict(xml_path)
except Exception as e:
print(f"❌ Failed to parse XML: {e}")
return 0
# Unwrap root if single-key dict
if isinstance(data, dict) and len(data) == 1:
root_key, root_val = next(iter(data.items()))
root = root_val
else:
root = data
# Find container with "chapter" key
container_with_chapter = _find_first_with_key(root, "chapter")
if not container_with_chapter:
container_with_chapter = _find_first_with_key(root, "chapters")
if container_with_chapter and isinstance(container_with_chapter.get("chapters"), dict):
if "chapter" in container_with_chapter["chapters"]:
container_with_chapter = container_with_chapter["chapters"]
if not container_with_chapter or ("chapter" not in container_with_chapter):
preview_keys = list(root.keys()) if isinstance(root, dict) else type(root)
print(f"❌ Could not locate 'chapter' in XML. Top-level keys: {preview_keys}")
return 0
chapters = container_with_chapter.get("chapter")
if chapters is None:
print("❌ Found container for chapters, but 'chapter' is empty.")
return 0
# Optionally truncate
if truncate:
print("⚠️ Truncating existing ICD-10 data...")
Icd10.objects.all().delete()
# Collect rows + parent links
print("📊 Collecting ICD-10 rows...")
rows, parent_links = _collect_icd10_rows(chapters, tenants)
print(f"✅ Collected {len(rows)} rows. Inserting...")
BATCH = 1000
with transaction.atomic():
for i in range(0, len(rows), BATCH):
Icd10.objects.bulk_create(rows[i:i+BATCH], ignore_conflicts=True)
# Link parents
if parent_links:
print("🔗 Linking parent relationships...")
code_to_obj = {}
for tenant in tenants:
tenant_codes = {o.code: o for o in Icd10.objects.filter(tenant=tenant).only("id", "code")}
code_to_obj[tenant.id] = tenant_codes
updates = []
for child_code, parent_code, tenant in parent_links:
tenant_codes = code_to_obj.get(tenant.id, {})
child = tenant_codes.get(child_code)
parent = tenant_codes.get(parent_code)
if child and parent and child.parent_id != parent.id:
child.parent_id = parent.id
updates.append(child)
for i in range(0, len(updates), BATCH):
Icd10.objects.bulk_update(updates[i:i+BATCH], ["parent"])
print(f"✅ Linked {len(updates)} parent relations.")
print(f"✅ ICD-10 import completed: {len(rows)} codes imported")
return len(rows)
# ============================================================================
# MAIN DATA GENERATION FUNCTIONS
# ============================================================================
def parse_arguments():
"""Parse command-line arguments"""
parser = argparse.ArgumentParser(
description='Generate EMR sample data and optionally import ICD-10 codes from XML'
)
parser.add_argument(
'--import-icd10',
action='store_true',
help='Import full ICD-10 codes from XML files'
)
parser.add_argument(
'--xsd',
type=str,
help='Path to ICD-10 XSD schema file (required with --import-icd10)'
)
parser.add_argument(
'--xml',
type=str,
help='Path to ICD-10 XML data file (required with --import-icd10)'
)
parser.add_argument(
'--icd10-only',
action='store_true',
help='Only import ICD-10 codes, skip other EMR data generation'
)
parser.add_argument(
'--truncate',
action='store_true',
help='Delete existing ICD-10 codes before importing'
)
return parser.parse_args()
def main():
"""Main function to generate all EMR data"""
# Parse command-line arguments
args = parse_arguments()
# Validate ICD-10 import arguments
if args.import_icd10 or args.icd10_only:
if not args.xsd or not args.xml:
print("❌ Error: --xsd and --xml are required when using --import-icd10 or --icd10-only")
sys.exit(1)
if not XMLSCHEMA_AVAILABLE:
print("❌ Error: xmlschema library not installed.")
print(" Install it with: pip install xmlschema")
sys.exit(1)
# Get existing tenants
tenants = list(Tenant.objects.all())
if not tenants:
print("❌ No tenants found. Please run the core data generator first.")
return
# ICD-10 only mode
if args.icd10_only:
print("🏥 ICD-10 Import Mode (skipping other EMR data)")
import_icd10_from_xml(args.xsd, args.xml, tenants, truncate=args.truncate)
return
# Standard EMR data generation
print("Starting Saudi Healthcare EMR Data Generation...")
# Get existing tenants
tenants = list(Tenant.objects.all())
if not tenants:
print("❌ No tenants found. Please run the core data generator first.")
return
# Create note templates
print("\n1. Creating Note Templates...")
templates = create_note_templates(tenants)
# Create encounters
print("\n2. Creating Patient Encounters...")
encounters = create_encounters(tenants, days_back=30)
# Create vital signs
print("\n3. Creating Vital Signs...")
vital_signs = create_vital_signs(encounters)
# Get patients and providers for remaining models
patients = list(PatientProfile.objects.filter(tenant__in=tenants))
providers = get_providers_for_emr(tenants)
# Create problem lists
print("\n4. Creating Problem Lists...")
problems = create_problem_lists(patients, providers)
# Create care plans
print("\n5. Creating Care Plans...")
care_plans = create_care_plans(patients, providers, problems)
# Create clinical notes
print("\n6. Creating Clinical Notes...")
clinical_notes = create_clinical_notes(encounters, templates)
# Create ICD-10 codes (sample or full import)
print("\n7. Creating ICD-10 Codes...")
if args.import_icd10:
# Import full ICD-10 from XML
num_imported = import_icd10_from_xml(args.xsd, args.xml, tenants, truncate=args.truncate)
icd10_codes = list(Icd10.objects.filter(tenant__in=tenants))
else:
# Create sample ICD-10 codes
icd10_codes = create_icd10_codes(tenants)
# Create clinical recommendations
print("\n8. Creating Clinical Recommendations...")
clinical_recommendations = create_clinical_recommendations(patients, providers, problems, encounters)
# Create allergy alerts
print("\n9. Creating Allergy Alerts...")
allergy_alerts = create_allergy_alerts(patients, providers)
# Create treatment protocols
print("\n10. Creating Treatment Protocols...")
treatment_protocols = create_treatment_protocols(tenants, providers)
# Create clinical guidelines
print("\n11. Creating Clinical Guidelines...")
clinical_guidelines = create_clinical_guidelines(tenants)
# Create critical alerts
print("\n12. Creating Critical Alerts...")
critical_alerts = create_critical_alerts(patients, providers, encounters)
# Create diagnostic suggestions
print("\n13. Creating Diagnostic Suggestions...")
diagnostic_suggestions = create_diagnostic_suggestions(patients, providers, encounters)
print(f"\n✅ Saudi Healthcare EMR Data Generation Complete!")
print(f"📊 Summary:")
print(f" - Note Templates: {len(templates)}")
print(f" - Patient Encounters: {len(encounters)}")
print(f" - Vital Signs Records: {len(vital_signs)}")
print(f" - Problem List Entries: {len(problems)}")
print(f" - Care Plans: {len(care_plans)}")
print(f" - Clinical Notes: {len(clinical_notes)}")
print(f" - ICD-10 Codes: {len(icd10_codes)}")
print(f" - Clinical Recommendations: {len(clinical_recommendations)}")
print(f" - Allergy Alerts: {len(allergy_alerts)}")
print(f" - Treatment Protocols: {len(treatment_protocols)}")
print(f" - Clinical Guidelines: {len(clinical_guidelines)}")
print(f" - Critical Alerts: {len(critical_alerts)}")
print(f" - Diagnostic Suggestions: {len(diagnostic_suggestions)}")
# Statistics
if encounters:
encounter_type_counts = {}
for encounter in encounters:
encounter_type_counts[encounter.encounter_type] = encounter_type_counts.get(encounter.encounter_type, 0) + 1
print(f"\n📈 Encounter Type Distribution:")
for enc_type, count in sorted(encounter_type_counts.items(), key=lambda x: x[1], reverse=True)[:10]:
print(f" - {enc_type.replace('_', ' ').title()}: {count}")
if problems:
problem_status_counts = {}
for problem in problems:
problem_status_counts[problem.status] = problem_status_counts.get(problem.status, 0) + 1
print(f"\n📋 Problem Status Distribution:")
for status, count in problem_status_counts.items():
print(f" - {status.title()}: {count}")
return {
'templates': templates,
'encounters': encounters,
'vital_signs': vital_signs,
'problems': problems,
'care_plans': care_plans,
'clinical_notes': clinical_notes,
'icd10_codes': icd10_codes,
'clinical_recommendations': clinical_recommendations,
'allergy_alerts': allergy_alerts,
'treatment_protocols': treatment_protocols,
'clinical_guidelines': clinical_guidelines,
'critical_alerts': critical_alerts,
'diagnostic_suggestions': diagnostic_suggestions
}
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink