Marwan Alwali b9b8c69129 update
2025-08-31 10:47:23 +03:00

585 lines
24 KiB
Python

"""
Django management command to generate Saudi-influenced Operating Theatre data
Place this file in: operating_theatre/management/commands/generate_saudi_or_data.py
"""
import random
import uuid
from datetime import datetime, timedelta, date, time
from decimal import Decimal
from django.core.management.base import BaseCommand
from django.db import transaction
from django.utils import timezone
from faker import Faker
# Import your models
from operating_theatre.models import (
OperatingRoom,
ORBlock,
SurgicalCase,
SurgicalNote,
EquipmentUsage,
SurgicalNoteTemplate
)
from core.models import Tenant
from patients.models import PatientProfile
from django.contrib.auth import get_user_model
User = get_user_model()
fake = Faker('en_US')
class Command(BaseCommand):
help = 'Generate Saudi-influenced Operating Theatre test data'
# Saudi cultural data
SAUDI_MALE_FIRST_NAMES = [
"Mohammed", "Abdullah", "Abdulrahman", "Khalid", "Fahad",
"Sultan", "Salman", "Saud", "Faisal", "Turki", "Ahmed",
"Omar", "Youssef", "Ibrahim", "Hamad", "Nasser", "Bandar",
"Mansour", "Majed", "Waleed", "Talal", "Rakan", "Yazeed"
]
SAUDI_FEMALE_FIRST_NAMES = [
"Nora", "Fatima", "Aisha", "Mariam", "Sarah", "Reem",
"Lama", "Hind", "Mona", "Amal", "Dalal", "Jawaher",
"Latifa", "Hessa", "Nouf", "Asma", "Khadija", "Layla"
]
SAUDI_FAMILY_NAMES = [
"Al-Saud", "Al-Rasheed", "Al-Qahtani", "Al-Otaibi", "Al-Dossari",
"Al-Harbi", "Al-Zahrani", "Al-Ghamdi", "Al-Shehri", "Al-Asmari",
"Al-Mutairi", "Al-Enezi", "Al-Shamari", "Al-Maliki", "Al-Johani"
]
SAUDI_HOSPITALS = [
"King Faisal Specialist Hospital",
"King Fahad Medical City",
"King Abdulaziz University Hospital",
"Prince Sultan Military Medical City",
"King Saud Medical City"
]
SAUDI_CITIES = [
"Riyadh", "Jeddah", "Mecca", "Medina", "Dammam",
"Khobar", "Dhahran", "Taif", "Tabuk", "Buraidah"
]
SURGICAL_PROCEDURES = {
"GENERAL": [
"Laparoscopic Cholecystectomy",
"Appendectomy",
"Hernia Repair",
"Bowel Resection",
"Gastric Bypass"
],
"CARDIAC": [
"Coronary Artery Bypass Grafting",
"Valve Replacement",
"Pacemaker Insertion"
],
"ORTHOPEDIC": [
"Total Knee Replacement",
"Total Hip Replacement",
"Spinal Fusion",
"ACL Reconstruction"
],
"NEURO": [
"Craniotomy",
"Brain Tumor Resection",
"Spinal Decompression"
]
}
MEDICAL_EQUIPMENT = [
"Da Vinci Surgical Robot",
"C-Arm Fluoroscopy",
"Zeiss Surgical Microscope",
"Harmonic Scalpel",
"LigaSure Device"
]
def add_arguments(self, parser):
parser.add_argument(
'--tenant',
type=str,
help='Tenant ID to use for data generation'
)
parser.add_argument(
'--rooms',
type=int,
default=5,
help='Number of operating rooms to create'
)
parser.add_argument(
'--blocks',
type=int,
default=3,
help='Number of OR blocks per room'
)
parser.add_argument(
'--cases',
type=int,
default=2,
help='Number of cases per block'
)
parser.add_argument(
'--clear',
action='store_true',
help='Clear existing OR data before generating'
)
def handle(self, *args, **options):
self.stdout.write(self.style.SUCCESS('Starting Saudi OR data generation...'))
# Get or create tenant
tenant = self.get_or_create_tenant(options.get('tenant'))
# Clear existing data if requested
if options['clear']:
self.clear_existing_data(tenant)
# Generate data
with transaction.atomic():
# Create users (surgeons, nurses, etc.)
users = self.create_medical_staff(tenant, 20)
# Create patients
patients = self.create_patients(tenant, 30)
# Create surgical note templates
self.create_surgical_note_templates(tenant)
# Create operating rooms
rooms = self.create_operating_rooms(tenant, options['rooms'])
# Create OR blocks and surgical cases
for room in rooms:
self.create_or_blocks_with_cases(
room,
users,
patients,
options['blocks'],
options['cases']
)
self.stdout.write(self.style.SUCCESS('Data generation complete!'))
self.print_summary(tenant)
def get_or_create_tenant(self, tenant_id=1):
"""Get existing tenant or create a new one."""
if tenant_id:
try:
return Tenant.objects.get(id=tenant_id)
except Tenant.DoesNotExist:
self.stdout.write(self.style.WARNING(f'Tenant {tenant_id} not found, creating new one'))
# Create a new tenant
hospital_name = random.choice(self.SAUDI_HOSPITALS)
tenant, created = Tenant.objects.get_or_create(
name=hospital_name,
defaults={
'domain': hospital_name.lower().replace(' ', '-') + '.sa',
'is_active': True,
'settings': {
'country': 'Saudi Arabia',
'city': random.choice(self.SAUDI_CITIES),
'timezone': 'Asia/Riyadh',
'currency': 'SAR'
}
}
)
if created:
self.stdout.write(self.style.SUCCESS(f'Created tenant: {tenant.name}'))
return tenant
def clear_existing_data(self, tenant):
"""Clear existing OR data for the tenant."""
self.stdout.write('Clearing existing data...')
OperatingRoom.objects.filter(tenant=tenant).delete()
SurgicalNoteTemplate.objects.filter(tenant=tenant).delete()
self.stdout.write(self.style.SUCCESS('Existing data cleared'))
def create_medical_staff(self, tenant, count=20):
"""Create medical staff users."""
self.stdout.write('Creating medical staff...')
users = []
for i in range(count):
is_female = i % 3 == 0 # 1/3 female staff
if is_female:
first_name = random.choice(self.SAUDI_FEMALE_FIRST_NAMES)
title = random.choice(['Dr.', 'Nurse'])
else:
first_name = random.choice(self.SAUDI_MALE_FIRST_NAMES)
title = 'Dr.'
last_name = random.choice(self.SAUDI_FAMILY_NAMES)
username = f"{first_name.lower()}.{last_name.lower().replace('-', '')}_{i}"
email = f"{username}@{tenant.domain}"
user, created = User.objects.get_or_create(
username=username,
defaults={
'email': email,
'first_name': first_name,
'last_name': last_name,
'is_active': True
}
)
if created:
user.set_password('password123')
user.save()
users.append(user)
self.stdout.write(self.style.SUCCESS(f'Created {len(users)} medical staff'))
return users
def create_patients(self, tenant, count=30):
"""Create patient profiles."""
self.stdout.write('Creating patients...')
patients = []
for i in range(count):
is_female = random.random() > 0.6
if is_female:
first_name = random.choice(self.SAUDI_FEMALE_FIRST_NAMES)
gender = 'F'
else:
first_name = random.choice(self.SAUDI_MALE_FIRST_NAMES)
gender = 'M'
last_name = random.choice(self.SAUDI_FAMILY_NAMES)
# Create user for patient
username = f"patient_{first_name.lower()}_{i}"
user, _ = User.objects.get_or_create(
username=username,
defaults={
'email': f"{username}@example.com",
'first_name': first_name,
'last_name': last_name
}
)
# Create patient profile
patient, created = PatientProfile.objects.get_or_create(
user=user,
defaults={
'tenant': tenant,
'patient_id': f"P{datetime.now().year}{i:06d}",
'date_of_birth': fake.date_of_birth(minimum_age=18, maximum_age=80),
'gender': gender,
'blood_group': random.choice(['A+', 'A-', 'B+', 'B-', 'O+', 'O-', 'AB+', 'AB-']),
'phone': f"+966{random.randint(500000000, 599999999)}",
'email': user.email,
'address': f"{random.randint(1, 999)} {random.choice(['King Fahd Road', 'Olaya Street', 'Tahlia Street'])}, {random.choice(self.SAUDI_CITIES)}",
'emergency_contact_name': random.choice(self.SAUDI_MALE_FIRST_NAMES) + ' ' + last_name,
'emergency_contact_phone': f"+966{random.randint(500000000, 599999999)}",
'national_id': f"{random.randint(1000000000, 2999999999)}",
'insurance_provider': random.choice(['Bupa', 'Tawuniya', 'MedGulf', 'Allianz', 'AXA']),
'insurance_number': f"INS{random.randint(100000, 999999)}"
}
)
patients.append(patient)
self.stdout.write(self.style.SUCCESS(f'Created {len(patients)} patients'))
return patients
def create_surgical_note_templates(self, tenant):
"""Create surgical note templates."""
self.stdout.write('Creating surgical note templates...')
specialties = ['GENERAL', 'CARDIAC', 'NEURO', 'ORTHOPEDIC', 'OBSTETRIC']
for specialty in specialties:
for template_type in ['Standard', 'Complex', 'Emergency']:
SurgicalNoteTemplate.objects.get_or_create(
tenant=tenant,
name=f"{specialty} - {template_type} Template",
defaults={
'description': f"Standard {template_type.lower()} template for {specialty.lower()} procedures",
'specialty': specialty,
'procedure_type': random.choice(self.SURGICAL_PROCEDURES.get(specialty, ['General'])),
'preoperative_diagnosis_template': "Preoperative Diagnosis: [Enter diagnosis]",
'planned_procedure_template': "Planned Procedure: [Enter procedure]",
'indication_template': "Indication: Patient presents with [symptoms] requiring intervention",
'procedure_performed_template': "Procedure: [Describe procedure performed]",
'surgical_approach_template': "Approach: [Describe approach]",
'findings_template': "Findings: [Describe findings]",
'technique_template': "Technique: Standard surgical technique employed",
'postoperative_diagnosis_template': "Postoperative Diagnosis: [Enter diagnosis]",
'is_active': True,
'is_default': template_type == 'Standard'
}
)
self.stdout.write(self.style.SUCCESS('Created surgical note templates'))
def create_operating_rooms(self, tenant, count=5):
"""Create operating rooms."""
self.stdout.write('Creating operating rooms...')
rooms = []
room_types = ['GENERAL', 'CARDIAC', 'NEURO', 'ORTHOPEDIC', 'OBSTETRIC']
for i in range(1, count + 1):
room_type = room_types[(i-1) % len(room_types)]
room = OperatingRoom.objects.create(
tenant=tenant,
room_number=f"OR-{i:03d}",
room_name=f"{room_type.title()} Operating Room {i}",
room_type=room_type,
status='AVAILABLE',
floor_number=random.randint(1, 4),
room_size=random.uniform(40, 80),
ceiling_height=random.uniform(3.0, 4.5),
temperature_min=18.0,
temperature_max=24.0,
humidity_min=30.0,
humidity_max=60.0,
air_changes_per_hour=random.randint(20, 25),
positive_pressure=True,
equipment_list=self._get_equipment_list(room_type),
special_features=self._get_special_features(room_type),
has_c_arm=room_type in ['ORTHOPEDIC', 'NEURO'],
has_ct=room_type in ['NEURO', 'CARDIAC'] and random.random() > 0.5,
has_mri=room_type == 'NEURO' and random.random() > 0.7,
has_ultrasound=True,
has_neuromonitoring=room_type in ['NEURO', 'ORTHOPEDIC'],
supports_robotic=room_type in ['GENERAL', 'CARDIAC'] and random.random() > 0.4,
supports_laparoscopic=room_type in ['GENERAL', 'OBSTETRIC'],
supports_microscopy=room_type in ['NEURO', 'OPHTHALMOLOGY'],
max_case_duration=random.randint(240, 720),
turnover_time=random.randint(20, 45),
cleaning_time=random.randint(30, 60),
required_nurses=random.randint(2, 4),
required_techs=random.randint(1, 3),
is_active=True,
accepts_emergency=True,
building=random.choice(['Main', 'East Wing', 'West Wing']),
wing=random.choice(['North', 'South', 'Central'])
)
rooms.append(room)
self.stdout.write(self.style.SUCCESS(f'Created {len(rooms)} operating rooms'))
return rooms
def _get_equipment_list(self, room_type):
"""Get equipment list for room type."""
basic = [
"Anesthesia Machine",
"Patient Monitor",
"Surgical Lights",
"Operating Table",
"Electrocautery Unit"
]
specialized = {
'CARDIAC': ["Heart-Lung Machine", "IABP"],
'NEURO': ["Surgical Microscope", "Neuronavigation"],
'ORTHOPEDIC': ["C-Arm", "Power Tools"],
'GENERAL': ["Laparoscopic Tower", "Harmonic Scalpel"]
}
equipment = basic.copy()
if room_type in specialized:
equipment.extend(specialized[room_type])
return equipment
def _get_special_features(self, room_type):
"""Get special features for room type."""
features = ["Laminar Flow", "HEPA Filtration"]
special = {
'CARDIAC': ["Hybrid OR Capability"],
'NEURO': ["Intraoperative MRI Compatible"],
'ORTHOPEDIC': ["Class 100 Clean Room"],
'OBSTETRIC': ["Neonatal Resuscitation Area"]
}
if room_type in special:
features.extend(special[room_type])
return features
def create_or_blocks_with_cases(self, room, users, patients, num_blocks, num_cases):
"""Create OR blocks with surgical cases."""
self.stdout.write(f'Creating blocks for {room.room_number}...')
surgeons = [u for u in users if 'Dr.' in u.first_name or random.random() > 0.5]
nurses = [u for u in users if u not in surgeons]
for block_num in range(num_blocks):
# Create OR block
block_date = timezone.now().date() + timedelta(days=random.randint(1, 30))
start_hour = random.choice([7, 8, 9, 13])
duration = random.randint(2, 6)
block = ORBlock.objects.create(
operating_room=room,
date=block_date,
start_time=time(start_hour, 0),
end_time=time((start_hour + duration) % 24, 0),
block_type='SCHEDULED',
primary_surgeon=random.choice(surgeons),
service=room.room_type,
status=random.choice(['SCHEDULED', 'ACTIVE', 'COMPLETED']),
allocated_minutes=duration * 60,
used_minutes=random.randint(duration * 45, duration * 60),
special_equipment=random.sample(self.MEDICAL_EQUIPMENT, k=random.randint(0, 2)),
notes=f"Block for {room.room_type} procedures"
)
# Add assistant surgeons
if surgeons:
assistants = random.sample(surgeons, k=min(random.randint(0, 2), len(surgeons)))
block.assistant_surgeons.set(assistants)
# Create surgical cases for this block
for case_num in range(random.randint(1, num_cases)):
self.create_surgical_case(block, patients, surgeons, nurses)
def create_surgical_case(self, block, patients, surgeons, nurses):
"""Create a surgical case."""
procedure_type = block.service
procedures = self.SURGICAL_PROCEDURES.get(procedure_type, ['General Surgery'])
case_time = datetime.combine(block.date, block.start_time) + timedelta(minutes=random.randint(0, 120))
duration = random.randint(30, 240)
case = SurgicalCase.objects.create(
or_block=block,
patient=random.choice(patients),
primary_surgeon=block.primary_surgeon,
anesthesiologist=random.choice(surgeons) if surgeons else block.primary_surgeon,
circulating_nurse=random.choice(nurses) if nurses else None,
scrub_nurse=random.choice(nurses) if nurses else None,
primary_procedure=random.choice(procedures),
secondary_procedures=[],
procedure_codes=[f"CPT{random.randint(10000, 99999)}" for _ in range(random.randint(1, 3))],
case_type=random.choice(['ELECTIVE', 'URGENT', 'EMERGENCY']),
approach=random.choice(['OPEN', 'LAPAROSCOPIC', 'ROBOTIC']),
anesthesia_type=random.choice(['GENERAL', 'REGIONAL', 'SPINAL']),
scheduled_start=timezone.make_aware(case_time),
estimated_duration=duration,
status=random.choice(['SCHEDULED', 'COMPLETED', 'IN_PROGRESS']),
diagnosis=self._get_diagnosis(),
diagnosis_codes=[f"ICD10-{random.choice(['K', 'I', 'M'])}{random.randint(10, 99)}.{random.randint(0, 9)}"],
clinical_notes="Patient prepared for surgery as per protocol",
special_equipment=random.sample(self.MEDICAL_EQUIPMENT, k=random.randint(0, 2)),
patient_position=random.choice(['SUPINE', 'PRONE', 'LATERAL']),
estimated_blood_loss=random.randint(10, 500) if random.random() > 0.5 else None
)
# Add assistant surgeons
if surgeons:
assistants = random.sample(surgeons, k=min(random.randint(0, 2), len(surgeons)))
case.assistant_surgeons.set(assistants)
# Create surgical note for completed cases
if case.status == 'COMPLETED':
self.create_surgical_note(case)
# Create equipment usage
self.create_equipment_usage(case)
return case
def _get_diagnosis(self):
"""Get a random diagnosis."""
diagnoses = [
"Acute Appendicitis",
"Cholelithiasis",
"Inguinal Hernia",
"Coronary Artery Disease",
"Spinal Stenosis",
"Osteoarthritis",
"Brain Tumor",
"Breast Cancer"
]
return random.choice(diagnoses)
def create_surgical_note(self, case):
"""Create surgical note for completed case."""
SurgicalNote.objects.create(
surgical_case=case,
surgeon=case.primary_surgeon,
preoperative_diagnosis=case.diagnosis,
planned_procedure=case.primary_procedure,
indication="Symptomatic disease requiring surgical intervention",
procedure_performed=case.primary_procedure,
surgical_approach=f"{case.approach} approach utilized",
findings="As per preoperative diagnosis",
technique="Standard surgical technique employed",
postoperative_diagnosis=case.diagnosis,
condition=random.choice(['STABLE', 'GOOD', 'FAIR']),
disposition=random.choice(['RECOVERY', 'ICU', 'WARD']),
complications="None",
estimated_blood_loss=case.estimated_blood_loss or random.randint(10, 200),
specimens="Sent to pathology" if random.random() > 0.5 else None,
closure="Layered closure with absorbable sutures",
postop_instructions="Standard postoperative care protocol",
follow_up="2 weeks in surgical clinic",
status='SIGNED',
signed_datetime=timezone.now()
)
def create_equipment_usage(self, case):
"""Create equipment usage records."""
equipment_types = [
('Surgical Drape Set', 'DISPOSABLE', 1, 'SET', 150.00),
('Harmonic Scalpel', 'SURGICAL_INSTRUMENT', 1, 'EACH', 2500.00),
('Suture Pack', 'DISPOSABLE', 3, 'PACK', 75.00),
('Surgical Gloves', 'DISPOSABLE', 10, 'PAIR', 5.00)
]
for name, eq_type, qty, unit, cost in random.sample(equipment_types, k=random.randint(2, 4)):
EquipmentUsage.objects.create(
surgical_case=case,
equipment_name=name,
equipment_type=eq_type,
manufacturer=random.choice(['Medtronic', 'Johnson & Johnson', 'Stryker']),
quantity_used=qty,
unit_of_measure=unit,
unit_cost=Decimal(str(cost)),
total_cost=Decimal(str(cost * qty)),
lot_number=f"LOT{random.randint(10000, 99999)}",
expiration_date=timezone.now().date() + timedelta(days=random.randint(180, 730)),
sterilization_date=timezone.now().date() - timedelta(days=random.randint(1, 7)),
recorded_by=case.primary_surgeon
)
def print_summary(self, tenant):
"""Print summary of generated data."""
self.stdout.write("\n" + "="*60)
self.stdout.write(self.style.SUCCESS("Data Generation Summary"))
self.stdout.write("="*60)
rooms = OperatingRoom.objects.filter(tenant=tenant)
blocks = ORBlock.objects.filter(operating_room__tenant=tenant)
cases = SurgicalCase.objects.filter(or_block__operating_room__tenant=tenant)
notes = SurgicalNote.objects.filter(surgical_case__or_block__operating_room__tenant=tenant)
equipment = EquipmentUsage.objects.filter(surgical_case__or_block__operating_room__tenant=tenant)
templates = SurgicalNoteTemplate.objects.filter(tenant=tenant)
self.stdout.write(f"Tenant: {tenant.name}")
self.stdout.write(f"Operating Rooms: {rooms.count()}")
self.stdout.write(f"OR Blocks: {blocks.count()}")
self.stdout.write(f"Surgical Cases: {cases.count()}")
self.stdout.write(f"Surgical Notes: {notes.count()}")
self.stdout.write(f"Equipment Usage Records: {equipment.count()}")
self.stdout.write(f"Surgical Templates: {templates.count()}")
self.stdout.write("="*60)