# """ # Saudi-influenced data generator for Operating Theatre models. # Generates realistic test data with Saudi Arabian healthcare context using English. # # Save this file as: operating_theatre/management/commands/generate_saudi_or_data.py # """ # # import random # from datetime import datetime, timedelta # from decimal import Decimal # from django.core.management.base import BaseCommand # from django.contrib.auth import get_user_model # from django.utils import timezone # from faker import Faker # from faker.providers import BaseProvider # # from core.models import Tenant, Department # from accounts.models import User # from patients.models import PatientProfile # from operating_theatre.models import ( # OperatingRoom, ORBlock, SurgicalCase, SurgicalNote, # EquipmentUsage, SurgicalNoteTemplate # ) # # User = get_user_model() # # # Custom Saudi provider for Faker # class SaudiProvider(BaseProvider): # """Custom provider for Saudi Arabian healthcare data in English.""" # # # Saudi male names (transliterated to English) # saudi_male_first_names = [ # 'Mohammed', 'Ahmed', 'Abdullah', 'Abdulrahman', 'Ali', 'Khalid', 'Saad', 'Fahd', 'Abdulaziz', # 'Sultan', 'Turki', 'Bandar', 'Nawaf', 'Mishaal', 'Faisal', 'Salman', 'Abdulmajeed', 'Yasser', # 'Omar', 'Hussam', 'Talal', 'Waleed', 'Majed', 'Rashid', 'Sami', 'Hisham', 'Adel', 'Kareem', # 'Nasser', 'Mansour', 'Hamad', 'Badr', 'Zaid', 'Yazeed', 'Rayan', 'Osama', 'Tariq', 'Mazen' # ] # # # Saudi female names (transliterated to English) # saudi_female_first_names = [ # 'Fatimah', 'Aisha', 'Khadijah', 'Maryam', 'Zainab', 'Sarah', 'Nora', 'Hind', 'Lateefah', 'Mona', # 'Amal', 'Salma', 'Reem', 'Dana', 'Lina', 'Rana', 'Haya', 'Jawaher', 'Shahd', 'Ghada', # 'Raghad', 'Asma', 'Hanan', 'Wafa', 'Iman', 'Samira', 'Nadia', 'Abeer', 'Suad', 'Layla', # 'Nouf', 'Rania', 'Dina', 'Rima', 'Najla', 'Arwa', 'Lamya', 'Thuraya', 'Widad', 'Amina' # ] # # # Saudi family names (transliterated) # saudi_family_names = [ # 'Al Saud', 'Al Otaibi', 'Al Qahtani', 'Al Ghamdi', 'Al Zahrani', 'Al Shehri', 'Al Anzi', 'Al Harbi', # 'Al Mutairi', 'Al Dosari', 'Al Khalidi', 'Al Subaie', 'Al Rashid', 'Al Baqmi', 'Al Juhani', 'Al Asiri', # 'Al Faisal', 'Al Maliki', 'Al Shammari', 'Al Balawi', 'Al Saadi', 'Al Thaqafi', 'Al Ahmadi', 'Al Salmi', # 'Al Hasani', 'Al Omari', 'Al Fahd', 'Al Nuaimi', 'Al Rajhi', 'Al Olayan', 'Al Sudairi', 'Al Tamimi', # 'Bin Laden', 'Bin Mahfouz', 'Al Amoudi', 'Al Gosaibi', 'Al Zamil', 'Al Juffali', 'Al Hokair' # ] # # # Saudi cities # saudi_cities = [ # 'Riyadh', 'Jeddah', 'Mecca', 'Medina', 'Dammam', 'Khobar', 'Taif', # 'Buraidah', 'Tabuk', 'Khamis Mushait', 'Hail', 'Jubail', 'Al Ahsa', 'Najran', 'Yanbu', # 'Abha', 'Arar', 'Sakaka', 'Jazan', 'Qatif', 'Al Bahah', 'Rafha', 'Wadi Al Dawasir', # 'Al Kharj', 'Hafar Al Batin', 'Al Qunfudhah', 'Unaizah', 'Al Majmaah', 'Dhahran' # ] # # # Medical specialties # medical_specialties = [ # 'General Surgery', 'Cardiac Surgery', 'Neurosurgery', 'Orthopedic Surgery', 'Plastic Surgery', # 'Urology', 'Pediatric Surgery', 'Vascular Surgery', 'Thoracic Surgery', 'Ophthalmology', # 'ENT Surgery', 'Gynecology', 'Emergency Medicine', 'Anesthesiology', 'Radiology' # ] # # # Common surgical procedures # surgical_procedures = [ # 'Laparoscopic Cholecystectomy', 'Appendectomy', 'Inguinal Hernia Repair', # 'Thyroidectomy', 'Open Heart Surgery', 'Cardiac Catheterization', 'Knee Replacement', # 'Hip Replacement', 'Mastectomy', 'Cataract Surgery', 'Tonsillectomy', 'Septoplasty', # 'Hysterectomy', 'Cesarean Section', 'Prostatectomy', 'Lumbar Spine Surgery', # 'Discectomy', 'Rhinoplasty', 'Liposuction', 'Hair Transplant', 'Kidney Stone Removal', # 'Varicose Vein Surgery', 'Coronary Artery Bypass', 'Angioplasty', 'Gallbladder Surgery' # ] # # # Medical equipment names # medical_equipment = [ # 'Ventilator', 'Cardiac Monitor', 'Defibrillator', 'Laparoscope', # 'Endoscope', 'Portable X-Ray', 'Infusion Pump', 'Blood Pressure Monitor', # 'Ultrasound Machine', 'Electrocautery Unit', 'Surgical Laser', # 'Arthroscope', 'Dialysis Machine', 'Incubator', 'Physical Therapy Equipment', # 'CT Scanner', 'MRI Machine', 'Anesthesia Machine', 'Operating Microscope' # ] # # # Hospital departments # hospital_departments = [ # 'Emergency Department', 'Surgery Department', 'Internal Medicine', 'Pediatrics', 'Obstetrics & Gynecology', # 'Orthopedics', 'Cardiology', 'Neurology', 'Ophthalmology', 'ENT Department', # 'Dermatology', 'Psychiatry', 'Radiology', 'Laboratory', 'Pharmacy', # 'Intensive Care Unit', 'Operating Theater', 'Outpatient Clinic' # ] # # # Saudi hospital names # saudi_hospitals = [ # 'King Fahd University Hospital', 'King Khalid University Hospital', 'King Abdulaziz Medical City', # 'King Faisal Specialist Hospital', 'Prince Sultan Military Medical City', 'King Saud Medical City', # 'National Guard Health Affairs', 'Saudi German Hospital', 'Dr. Sulaiman Al Habib Medical Group', # 'Dallah Healthcare', 'Mouwasat Medical Services', 'Al Hammadi Hospital' # ] # # def saudi_male_name(self): # """Generate a Saudi male name.""" # first_name = self.random_element(self.saudi_male_first_names) # family_name = self.random_element(self.saudi_family_names) # return f"{first_name} {family_name}" # # def saudi_female_name(self): # """Generate a Saudi female name.""" # first_name = self.random_element(self.saudi_female_first_names) # family_name = self.random_element(self.saudi_family_names) # return f"{first_name} {family_name}" # # def saudi_city(self): # """Generate a Saudi city name.""" # return self.random_element(self.saudi_cities) # # def medical_specialty(self): # """Generate a medical specialty.""" # return self.random_element(self.medical_specialties) # # def surgical_procedure(self): # """Generate a surgical procedure name.""" # return self.random_element(self.surgical_procedures) # # def medical_equipment_name(self): # """Generate medical equipment name.""" # return self.random_element(self.medical_equipment) # # def hospital_department(self): # """Generate hospital department name.""" # return self.random_element(self.hospital_departments) # # def saudi_hospital_name(self): # """Generate Saudi hospital name.""" # return self.random_element(self.saudi_hospitals) # # def saudi_national_id(self): # """Generate a Saudi national ID (10 digits starting with 1 or 2).""" # first_digit = random.choice(['1', '2']) # 1 for Saudi, 2 for resident # remaining_digits = ''.join([str(random.randint(0, 9)) for _ in range(9)]) # return first_digit + remaining_digits # # def saudi_phone_number(self): # """Generate a Saudi phone number.""" # prefixes = ['050', '053', '054', '055', '056', '057', '058', '059'] # prefix = random.choice(prefixes) # number = ''.join([str(random.randint(0, 9)) for _ in range(7)]) # return f"+966{prefix[1:]}{number}" # # def saudi_medical_license(self): # """Generate Saudi medical license number.""" # return f"SCFHS-{random.randint(10000, 99999)}" # # # class Command(BaseCommand): # help = 'Generate Saudi-influenced test data for Operating Theatre models' # # def __init__(self): # super().__init__() # self.fake = Faker(['en_US']) # self.fake.add_provider(SaudiProvider) # # def add_arguments(self, parser): # parser.add_argument( # '--rooms', # type=int, # default=12, # help='Number of operating rooms to create' # ) # parser.add_argument( # '--cases', # type=int, # default=75, # help='Number of surgical cases to create' # ) # parser.add_argument( # '--notes', # type=int, # default=50, # help='Number of surgical notes to create' # ) # parser.add_argument( # '--templates', # type=int, # default=20, # help='Number of surgical note templates to create' # ) # parser.add_argument( # '--blocks', # type=int, # default=30, # help='Number of OR blocks to create' # ) # parser.add_argument( # '--equipment', # type=int, # default=60, # help='Number of equipment usage records to create' # ) # parser.add_argument( # '--tenant', # type=str, # default=None, # help='Tenant name (will use random Saudi hospital if not provided)' # ) # # def handle(self, *args, **options): # self.stdout.write( # self.style.SUCCESS('Starting Saudi-influenced Operating Theatre data generation...') # ) # # # Get or create tenant # tenant_name = options['tenant'] or self.fake.saudi_hospital_name() # tenant = self.get_or_create_tenant(tenant_name) # # # Create users (doctors, nurses, etc.) # users = self.create_users(tenant) # # # Create patients # patients = self.create_patients(tenant) # # # Create operating rooms # operating_rooms = self.create_operating_rooms(tenant, options['rooms']) # # # Create surgical note templates # templates = self.create_surgical_note_templates(tenant, users, options['templates']) # # # Create OR blocks # or_blocks = self.create_or_blocks(tenant, operating_rooms, users, options['blocks']) # # # Create surgical cases # surgical_cases = self.create_surgical_cases( # tenant, patients, users, operating_rooms, options['cases'] # ) # # # Create surgical notes # surgical_notes = self.create_surgical_notes( # tenant, surgical_cases, users, templates, options['notes'] # ) # # # Create equipment usage records # equipment_usage = self.create_equipment_usage( # tenant, surgical_cases, operating_rooms, options['equipment'] # ) # # self.stdout.write( # self.style.SUCCESS( # f'Successfully generated Saudi-influenced data:\n' # f'- {len(operating_rooms)} Operating Rooms\n' # f'- {len(surgical_cases)} Surgical Cases\n' # f'- {len(surgical_notes)} Surgical Notes\n' # f'- {len(templates)} Note Templates\n' # f'- {len(or_blocks)} OR Blocks\n' # f'- {len(equipment_usage)} Equipment Usage Records\n' # f'- Tenant: {tenant.name}' # ) # ) # # def get_or_create_tenant(self, tenant_name): # """Get or create tenant.""" # slug = tenant_name.lower().replace(' ', '-').replace('&', 'and') # tenant, created = Tenant.objects.get_or_create( # name=tenant_name, # defaults={ # 'slug': slug, # 'is_active': True, # 'subscription_plan': 'enterprise', # 'max_users': 1000, # 'country': 'Saudi Arabia', # 'timezone': 'Asia/Riyadh' # } # ) # if created: # self.stdout.write(f'Created tenant: {tenant_name}') # return tenant # # # def create_users(self, tenant): # """Create medical staff users with Saudi names.""" # users = [] # # # Create surgeons # for i in range(20): # gender = random.choice(["M", "F"]) # if gender == "M": # full_name = self.fake.saudi_male_name() # else: # full_name = self.fake.saudi_female_name() # # name_parts = full_name.split() # first_name = name_parts[0] # last_name = " ".join(name_parts[1:]) if len(name_parts) > 1 else name_parts[0] # # username = f"dr_{first_name.lower()}_{i}" # email = f"{username}@{tenant.slug}.sa" # # user = User.objects.create_user( # username=username, # email=email, # first_name=first_name, # last_name=last_name, # tenant=tenant, # is_active=True # ) # # # Create user profile # User.objects.create( # user=user, # title="Dr." if gender == "M" else "Dr.", # specialty=self.fake.medical_specialty(), # license_number=self.fake.saudi_medical_license(), # phone_number=self.fake.saudi_phone_number(), # department=self.fake.hospital_department(), # is_verified=True, # years_of_experience=random.randint(3, 25), # education=f"MD from King Saud University, {random.choice(["Fellowship", "Residency"])} in {self.fake.medical_specialty()}" # ) # # users.append(user) # # # Create anesthesiologists # for i in range(10): # gender = random.choice(["M", "F"]) # if gender == "M": # full_name = self.fake.saudi_male_name() # else: # full_name = self.fake.saudi_female_name() # # name_parts = full_name.split() # first_name = name_parts[0] # last_name = " ".join(name_parts[1:]) if len(name_parts) > 1 else name_parts[0] # # username = f"anesth_{first_name.lower()}_{i}" # email = f"{username}@{tenant.slug}.sa" # # user = User.objects.create_user( # username=username, # email=email, # first_name=first_name, # last_name=last_name, # tenant=tenant, # is_active=True # ) # # User.objects.create( # user=user, # title="Dr.", # specialty="Anesthesiology", # license_number=self.fake.saudi_medical_license(), # phone_number=self.fake.saudi_phone_number(), # department="Anesthesia Department", # is_verified=True, # years_of_experience=random.randint(5, 20), # education=f"MD from {random.choice(["King Saud University", "King Abdulaziz University", "Imam Abdulrahman Bin Faisal University"])}" # ) # # users.append(user) # # self.stdout.write(f"Created {len(users)} medical staff users with Saudi names") # return users # # # def create_patients(self, tenant): # """Create patient records with Saudi demographics.""" # patients = [] # # for i in range(150): # gender = random.choice(["M", "F"]) # if gender == "M": # full_name = self.fake.saudi_male_name() # else: # full_name = self.fake.saudi_female_name() # # name_parts = full_name.split() # first_name = name_parts[0] # last_name = " ".join(name_parts[1:]) if len(name_parts) > 1 else name_parts[0] # # birth_date = self.fake.date_of_birth(minimum_age=1, maximum_age=85) # # patient = PatientProfile.objects.create( # tenant=tenant, # medical_record_number=f"MRN{random.randint(100000, 999999)}", # first_name=first_name, # last_name=last_name, # date_of_birth=birth_date, # gender=gender, # national_id=self.fake.saudi_national_id(), # phone_number=self.fake.saudi_phone_number(), # email=f"patient{i}@email.com", # address=f"{self.fake.street_address()}, {self.fake.saudi_city()}, Saudi Arabia", # emergency_contact_name=self.fake.saudi_male_name() if random.choice( # [True, False]) else self.fake.saudi_female_name(), # emergency_contact_phone=self.fake.saudi_phone_number(), # blood_type=random.choice(["A+", "A-", "B+", "B-", "AB+", "AB-", "O+", "O-"]), # nationality="Saudi Arabian" if random.random() > 0.3 else random.choice( # ["Egyptian", "Pakistani", "Indian", "Bangladeshi", "Filipino"]), # is_active=True # ) # # patients.append(patient) # # self.stdout.write(f"Created {len(patients)} patients with Saudi demographics") # return patients # # # def create_operating_rooms(self, tenant, count): # """Create operating rooms with Saudi hospital context.""" # operating_rooms = [] # # room_types = [ # ("GENERAL", "General Surgery OR"), # ("CARDIAC", "Cardiac Surgery OR"), # ("NEURO", "Neurosurgery OR"), # ("ORTHOPEDIC", "Orthopedic Surgery OR"), # ("PEDIATRIC", "Pediatric Surgery OR"), # ("TRAUMA", "Trauma Surgery OR"), # ("OPHTHALMOLOGY", "Eye Surgery OR"), # ("ENT", "ENT Surgery OR") # ] # # floors = ["2nd Floor - Surgical Wing", "3rd Floor - Cardiac Wing", "4th Floor - Specialty Wing"] # # for i in range(1, count + 1): # room_type, room_name = random.choice(room_types) # # operating_room = OperatingRoom.objects.create( # tenant=tenant, # room_number=f"OR-{i:02d}", # room_name=f"{room_name} {i}", # room_type=room_type, # floor=random.randint(2, 5), # location=random.choice(floors), # capacity=random.randint(10, 18), # status=random.choice(["AVAILABLE", "IN_USE", "MAINTENANCE", "CLEANING"]), # equipment_list=f"Ventilator, Cardiac Monitor, {self.fake.medical_equipment_name()}, Anesthesia Machine", # special_requirements=f"Special requirements for {room_name.lower()}", # is_active=True, # last_maintenance=timezone.now().date() - timedelta(days=random.randint(1, 30)), # next_maintenance=timezone.now().date() + timedelta(days=random.randint(30, 90)) # ) # # operating_rooms.append(operating_room) # # self.stdout.write(f"Created {count} operating rooms") # return operating_rooms # # # def create_surgical_note_templates(self, tenant, users, count): # """Create surgical note templates for common Saudi procedures.""" # templates = [] # # specialties = [ # "General Surgery", "Cardiac Surgery", "Neurosurgery", "Orthopedic Surgery", # "Plastic Surgery", "Urology", "Pediatric Surgery", "Ophthalmology" # ] # # for i in range(count): # specialty = random.choice(specialties) # procedure = self.fake.surgical_procedure() # # template = SurgicalNoteTemplate.objects.create( # tenant=tenant, # template_name=f"{procedure} Template", # procedure_type=procedure, # specialty=specialty, # pre_operative_diagnosis_template=f"Pre-operative diagnosis for {procedure}", # post_operative_diagnosis_template=f"Post-operative diagnosis following {procedure}", # procedure_template=f"Detailed description of {procedure} procedure", # findings_template=f"Expected findings for {procedure}", # technique_template=f"Surgical technique used in {procedure}", # complications_template="No complications noted", # instructions_template=f"Post-operative instructions for {procedure}", # created_by=random.choice(users), # is_active=True, # version=f"v{random.randint(1, 5)}.{random.randint(0, 9)}" # ) # # templates.append(template) # # self.stdout.write(f"Created {count} surgical note templates") # return templates # # # def create_or_blocks(self, tenant, operating_rooms, users, count): # """Create OR blocks (time slots) with Saudi working hours.""" # or_blocks = [] # # # Saudi working hours: typically 7 AM to 3 PM for morning shift, 3 PM to 11 PM for evening # working_hours = [ # (7, 15), # Morning shift # (15, 23), # Evening shift # ] # # for i in range(count): # start_date = timezone.now().date() + timedelta(days=random.randint(-15, 45)) # shift_start, shift_end = random.choice(working_hours) # # start_hour = random.randint(shift_start, shift_end - 4) # start_time = timezone.make_aware( # datetime.combine(start_date, datetime.min.time().replace( # hour=start_hour, # minute=random.choice([0, 30]) # )) # ) # end_time = start_time + timedelta(hours=random.randint(2, 6)) # # or_block = ORBlock.objects.create( # tenant=tenant, # operating_room=random.choice(operating_rooms), # date=start_date, # start_time=start_time.time(), # end_time=end_time.time(), # assigned_surgeon=random.choice(users), # block_type=random.choice(["ELECTIVE", "EMERGENCY", "URGENT"]), # specialty=self.fake.medical_specialty(), # notes=f"OR Block {i + 1} - {self.fake.medical_specialty()}", # is_active=True, # utilization_rate=random.randint(70, 100) # ) # # or_blocks.append(or_block) # # self.stdout.write(f"Created {count} OR blocks") # return or_blocks # # # def create_surgical_cases(self, tenant, patients, users, operating_rooms, count): # """Create surgical cases with Saudi medical context.""" # surgical_cases = [] # # priorities = ["ROUTINE", "URGENT", "EMERGENCY"] # statuses = ["SCHEDULED", "IN_PROGRESS", "COMPLETED", "CANCELLED", "POSTPONED"] # # # Weight the statuses to be more realistic # status_weights = [0.4, 0.1, 0.4, 0.05, 0.05] # # for i in range(count): # scheduled_start = timezone.now() + timedelta( # days=random.randint(-30, 30), # hours=random.randint(7, 20), # minutes=random.choice([0, 15, 30, 45]) # ) # # status = random.choices(statuses, weights=status_weights)[0] # actual_start = None # actual_end = None # # if status in ["IN_PROGRESS", "COMPLETED"]: # actual_start = scheduled_start + timedelta(minutes=random.randint(-15, 45)) # # if status == "COMPLETED": # duration_hours = random.randint(1, 8) # actual_end = actual_start + timedelta( # hours=duration_hours, # minutes=random.randint(0, 59) # ) # # procedure = self.fake.surgical_procedure() # priority = random.choices(priorities, weights=[0.7, 0.2, 0.1])[0] # # # Create case number with Saudi format # case_number = f"CASE-{tenant.slug.upper()}-{timezone.now().year}-{i + 1:05d}" # # surgical_case = SurgicalCase.objects.create( # tenant=tenant, # case_number=case_number, # patient=random.choice(patients), # primary_surgeon=random.choice(users), # anesthesiologist=random.choice(users), # operating_room=random.choice(operating_rooms), # procedure_name=procedure, # scheduled_start_time=scheduled_start, # scheduled_end_time=scheduled_start + timedelta(hours=random.randint(2, 8)), # actual_start_time=actual_start, # actual_end_time=actual_end, # priority=priority, # status=status, # diagnosis=f"Diagnosis requiring {procedure}", # notes=f"Surgical case notes for {procedure} - Patient from {self.fake.saudi_city()}", # estimated_duration=timedelta(hours=random.randint(2, 6)), # is_emergency=(priority == "EMERGENCY"), # insurance_type=random.choice(["Government", "Private", "Self-Pay", "Company Insurance"]), # consent_obtained=True if status != "SCHEDULED" else random.choice([True, False]) # ) # # surgical_cases.append(surgical_case) # # self.stdout.write(f"Created {count} surgical cases") # return surgical_cases # # # def create_surgical_notes(self, tenant, surgical_cases, users, templates, count): # """Create surgical notes with Saudi medical documentation standards.""" # surgical_notes = [] # # # Only create notes for completed or in-progress cases # eligible_cases = [case for case in surgical_cases if case.status in ["COMPLETED", "IN_PROGRESS"]] # # for i in range(min(count, len(eligible_cases))): # case = eligible_cases[i] # template = random.choice(templates) if templates and random.random() > 0.3 else None # # # Create realistic surgical note content # complications = "No complications noted" if random.random() > 0.15 else random.choice([ # "Minor bleeding controlled with electrocautery", # "Slight tissue adhesions noted and carefully dissected", # "Temporary hypotension managed with fluids" # ]) # # surgical_note = SurgicalNote.objects.create( # tenant=tenant, # surgical_case=case, # template=template, # pre_operative_diagnosis=f"Pre-operative diagnosis: {case.diagnosis}", # post_operative_diagnosis=f"Post-operative diagnosis: Successful {case.procedure_name}", # procedure_performed=f"{case.procedure_name} performed according to standard protocol at {tenant.name}", # operative_findings=f"Surgical findings: Procedure completed successfully without major complications", # technique=f"Standard technique used for {case.procedure_name} with modern equipment", # complications=complications, # post_operative_instructions=f"Post-operative care instructions for {case.procedure_name} - Follow up in 1-2 weeks", # additional_notes=f"Additional notes for case {case.case_number} - Patient tolerated procedure well", # created_by=case.primary_surgeon, # status=random.choices(["DRAFT", "SIGNED", "FINALIZED"], weights=[0.2, 0.3, 0.5])[0], # signed_at=timezone.now() - timedelta(hours=random.randint(1, 48)) if random.choice([True, False]) else None, # dictation_time=random.randint(15, 45), # minutes # transcription_time=random.randint(30, 90) # minutes # ) # # surgical_notes.append(surgical_note) # # self.stdout.write(f"Created {len(surgical_notes)} surgical notes") # return surgical_notes # # # def create_equipment_usage(self, tenant, surgical_cases, operating_rooms, count): # """Create equipment usage records with Saudi hospital equipment.""" # equipment_usage_records = [] # # equipment_names = [ # "Ventilator Model SV-300", "Cardiac Monitor CM-2000", "Defibrillator DF-500", # "Laparoscope HD-Pro", "Electrocautery Unit ECU-400", "Ultrasound Machine US-Elite", # "Infusion Pump IP-Smart", "Blood Pressure Monitor BPM-Digital", "Surgical Laser SL-2000", # "Anesthesia Machine AM-Pro", "Operating Microscope OM-HD", "Arthroscope AS-Flex" # ] # # statuses = ["IN_USE", "COMPLETED", "MAINTENANCE", "AVAILABLE"] # status_weights = [0.15, 0.6, 0.1, 0.15] # # for i in range(count): # case = random.choice(surgical_cases) # equipment_name = random.choice(equipment_names) # # start_time = case.actual_start_time or case.scheduled_start_time # end_time = None # status = random.choices(statuses, weights=status_weights)[0] # # if status == "COMPLETED" and case.actual_end_time: # end_time = case.actual_end_time # elif status == "IN_USE": # end_time = None # else: # end_time = start_time + timedelta(hours=random.randint(1, 6)) # # # Generate Saudi equipment serial numbers # serial_prefix = random.choice(["KSA", "RYD", "JED", "DMM"]) # serial_number = f"{serial_prefix}-{random.randint(1000, 9999)}" # # equipment_usage = EquipmentUsage.objects.create( # tenant=tenant, # surgical_case=case, # operating_room=case.operating_room, # equipment_name=equipment_name, # equipment_serial=serial_number, # start_time=start_time, # end_time=end_time, # status=status, # notes=f"Equipment usage for {case.procedure_name} - {equipment_name}", # maintenance_required=random.choice([True, False]), # operator_name=case.primary_surgeon.get_full_name(), # usage_duration=end_time - start_time if end_time else None # ) # # equipment_usage_records.append(equipment_usage) # # self.stdout.write(f"Created {count} equipment usage records") # return equipment_usage_records # # """ Operating Theatre Data Generator Generates realistic test data for operating theatre models in Saudi hospital context. Uses existing tenants, users, and patients from the database. """ import os import django # Set up Django environment os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'hospital_management.settings') django.setup() import random import uuid from datetime import datetime, date, time, timedelta from decimal import Decimal from django.utils import timezone from django.contrib.auth import get_user_model # Import models from operating_theatre.models import ( OperatingRoom, ORBlock, SurgicalCase, SurgicalNote, EquipmentUsage, SurgicalNoteTemplate ) from core.models import Tenant from patients.models import PatientProfile from emr.models import Encounter from inpatients.models import Admission User = get_user_model() class OperatingTheatreDataGenerator: """Generate Saudi-influenced operating theatre data""" def __init__(self): self.setup_saudi_data() def setup_saudi_data(self): """Setup Saudi-specific data for generation""" # Arabic-influenced room names self.room_names = [ "غرفة العمليات الأولى", "غرفة العمليات الثانية", "غرفة العمليات الثالثة", "جناح الجراحة العامة", "جناح جراحة القلب", "جناح جراحة الأعصاب", "وحدة الجراحة الطارئة", "مركز الجراحة الروبوتية", "قسم جراحة العظام" ] # Common Saudi procedures (Arabic + English) self.procedures = [ "استئصال المرارة بالمنظار", "Laparoscopic Cholecystectomy", "جراحة القلب المفتوح", "Open Heart Surgery", "استبدال مفصل الركبة", "Total Knee Replacement", "جراحة الفتق الإربي", "Inguinal Hernia Repair", "استئصال الزائدة الدودية", "Appendectomy", "جراحة المنظار التشخيصي", "Diagnostic Laparoscopy", "استئصال الغدة الدرقية", "Thyroidectomy", "جراحة الساد", "Cataract Surgery", "استئصال اللوزتين", "Tonsillectomy", "جراحة البواسير", "Hemorrhoidectomy" ] # Saudi medical equipment (common in Saudi hospitals) self.equipment = [ "منظار البطن كارل زايس", "Carl Zeiss Laparoscope", "جهاز القطع الكهربائي", "Electrocautery Unit", "جهاز التخدير دريجر", "Drager Anesthesia Machine", "مجهر جراحي ليكا", "Leica Surgical Microscope", "روبوت دافنشي", "da Vinci Surgical Robot", "جهاز الأشعة المقطعية المحمول", "Portable CT Scanner", "منظار المثانة", "Cystoscope", "مضخة القلب الرئة", "Heart-Lung Machine" ] # Common diagnoses in Arabic and English self.diagnoses = [ "التهاب المرارة الحاد", "Acute Cholecystitis", "انسداد الأمعاء", "Bowel Obstruction", "كسر في عظم الفخذ", "Femur Fracture", "أورام الغدة الدرقية", "Thyroid Nodules", "حصى الكلى", "Kidney Stones", "التهاب الزائدة الدودية", "Appendicitis", "الفتق الإربي", "Inguinal Hernia", "سرطان القولون", "Colorectal Cancer" ] def get_existing_data(self): """Retrieve existing data from database""" print("Retrieving existing data from database...") # Get tenants tenants = list(Tenant.objects.all()) if not tenants: raise ValueError("No tenants found in database. Please create tenants first.") # Get users with different roles users = list(User.objects.all()) if not users: raise ValueError("No users found in database. Please create users first.") # Try to categorize users by role/specialty surgeons = list(User.objects.filter( groups__name__icontains='surgeon' ).distinct()) or users[:max(1, len(users) // 3)] anesthesiologists = list(User.objects.filter( groups__name__icontains='anesthesia' ).distinct()) or users[len(users) // 3:2 * len(users) // 3] nurses = list(User.objects.filter( groups__name__icontains='nurse' ).distinct()) or users[2 * len(users) // 3:] # Get patients patients = list(PatientProfile.objects.all()) if not patients: raise ValueError("No patients found in database. Please create patients first.") # Get encounters and admissions if available encounters = list(Encounter.objects.all()) if hasattr(Encounter, 'objects') else [] admissions = list(Admission.objects.all()) if hasattr(Admission, 'objects') else [] print(f"Found: {len(tenants)} tenants, {len(users)} users, {len(patients)} patients") print( f"Categorized: {len(surgeons)} surgeons, {len(anesthesiologists)} anesthesiologists, {len(nurses)} nurses") return { 'tenants': tenants, 'users': users, 'surgeons': surgeons, 'anesthesiologists': anesthesiologists, 'nurses': nurses, 'patients': patients, 'encounters': encounters, 'admissions': admissions } def generate_operating_rooms(self, data, count=15): """Generate operating rooms""" print(f"Generating {count} operating rooms...") operating_rooms = [] for tenant in data['tenants']: rooms_per_tenant = count // len(data['tenants']) + (1 if tenant == data['tenants'][0] else 0) for i in range(rooms_per_tenant): room = OperatingRoom.objects.create( tenant=tenant, room_number=f"OR-{i + 1:02d}", room_name=random.choice(self.room_names), room_type=random.choice([choice[0] for choice in OperatingRoom.ROOM_TYPE_CHOICES]), status=random.choice([choice[0] for choice in OperatingRoom.STATUS_CHOICES]), floor_number=random.randint(2, 8), room_size=random.uniform(25.0, 60.0), ceiling_height=random.uniform(3.0, 4.5), temperature_min=random.uniform(16.0, 20.0), temperature_max=random.uniform(22.0, 28.0), humidity_min=random.uniform(25.0, 35.0), humidity_max=random.uniform(55.0, 65.0), air_changes_per_hour=random.randint(15, 25), positive_pressure=random.choice([True, False]), # equipment_list=random.sample(self.equipment, k=random.randint(3, 8)), special_features=random.sample([ "HEPA Filtration", "Temperature Control", "Humidity Control", "Integrated Monitors", "Surgical Lights", "Emergency Power" ], k=random.randint(2, 4)), has_c_arm=random.choice([True, False]), has_ct=random.choice([True, False]) if random.random() > 0.7 else False, has_mri=random.choice([True, False]) if random.random() > 0.8 else False, has_ultrasound=random.choice([True, False]), has_neuromonitoring=random.choice([True, False]) if random.random() > 0.6 else False, supports_robotic=random.choice([True, False]) if random.random() > 0.7 else False, supports_laparoscopic=random.choice([True, True, True, False]), # More likely True supports_microscopy=random.choice([True, False]), supports_laser=random.choice([True, False]), max_case_duration=random.choice([240, 360, 480, 600, 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=random.choice([True, True, True, False]), # More likely active accepts_emergency=random.choice([True, True, False]), # More likely to accept building=random.choice(["Main Hospital", "Surgical Center", "Medical Tower"]), wing=random.choice(["North Wing", "South Wing", "East Wing", "West Wing"]), created_by=random.choice(data['users']) ) operating_rooms.append(room) print(f"Generated {len(operating_rooms)} operating rooms") return operating_rooms def generate_surgical_note_templates(self, data, count=20): """Generate surgical note templates""" print(f"Generating {count} surgical note templates...") templates = [] template_data = [ { 'name': 'General Laparoscopic Surgery', 'specialty': 'GENERAL', 'procedure_type': 'Laparoscopic Surgery', 'preoperative_diagnosis': 'Acute cholecystitis with cholelithiasis', 'planned_procedure': 'Laparoscopic cholecystectomy', 'indication': 'Symptomatic gallstone disease with acute inflammation' }, { 'name': 'Cardiac Surgery Standard', 'specialty': 'CARDIAC', 'procedure_type': 'Open Heart Surgery', 'preoperative_diagnosis': 'Coronary artery disease, three vessel', 'planned_procedure': 'Coronary artery bypass grafting', 'indication': 'Significant coronary artery stenosis' }, { 'name': 'Orthopedic Joint Replacement', 'specialty': 'ORTHOPEDIC', 'procedure_type': 'Joint Replacement', 'preoperative_diagnosis': 'Severe osteoarthritis of the knee', 'planned_procedure': 'Total knee arthroplasty', 'indication': 'End-stage arthritis with functional limitation' }, { 'name': 'Neurosurgical Craniotomy', 'specialty': 'NEURO', 'procedure_type': 'Craniotomy', 'preoperative_diagnosis': 'Brain tumor, frontal lobe', 'planned_procedure': 'Craniotomy and tumor resection', 'indication': 'Mass lesion with neurological symptoms' }, { 'name': 'Emergency Appendectomy', 'specialty': 'GENERAL', 'procedure_type': 'Emergency Surgery', 'preoperative_diagnosis': 'Acute appendicitis', 'planned_procedure': 'Laparoscopic appendectomy', 'indication': 'Acute inflammatory process of the appendix' } ] for tenant in data['tenants']: for template_info in template_data: template = SurgicalNoteTemplate.objects.create( tenant=tenant, name=template_info['name'], description=f"Standard template for {template_info['procedure_type']}", procedure_type=template_info['procedure_type'], specialty=template_info['specialty'], preoperative_diagnosis_template=template_info['preoperative_diagnosis'], planned_procedure_template=template_info['planned_procedure'], indication_template=template_info['indication'], procedure_performed_template=template_info['planned_procedure'], surgical_approach_template="Standard laparoscopic approach with CO2 insufflation", findings_template="Findings consistent with preoperative diagnosis", technique_template="Standard surgical technique employed", postoperative_diagnosis_template=template_info['preoperative_diagnosis'], complications_template="No intraoperative complications", specimens_template="Specimen sent to pathology for routine examination", implants_template="No implants used", closure_template="Fascia closed with absorbable sutures, skin closed with skin adhesive", postop_instructions_template="NPO until bowel function returns, progressive diet advancement", is_active=True, is_default=random.choice([True, False]), created_by=random.choice(data['users']) ) templates.append(template) # Generate additional random templates for i in range(count - len(template_data)): template = SurgicalNoteTemplate.objects.create( tenant=tenant, name=f"Custom Template {i + 1}", description=f"Custom surgical note template {i + 1}", procedure_type=random.choice(self.procedures), specialty=random.choice([choice[0] for choice in SurgicalNoteTemplate.SPECIALTY_CHOICES]), preoperative_diagnosis_template=random.choice(self.diagnoses), planned_procedure_template=random.choice(self.procedures), indication_template="Standard indication for procedure", procedure_performed_template=random.choice(self.procedures), surgical_approach_template="Standard surgical approach", findings_template="Findings as expected", technique_template="Standard technique employed", postoperative_diagnosis_template=random.choice(self.diagnoses), is_active=random.choice([True, False]), created_by=random.choice(data['users']) ) templates.append(template) print(f"Generated {len(templates)} surgical note templates") return templates def generate_or_blocks(self, operating_rooms, data, count=50): """Generate OR blocks""" print(f"Generating {count} OR blocks...") blocks = [] for i in range(count): # Generate dates for the next 30 days block_date = date.today() + timedelta(days=random.randint(0, 30)) # Generate time slots start_hour = random.randint(6, 20) start_minute = random.choice([0, 30]) start_time = time(start_hour, start_minute) duration_hours = random.randint(2, 8) end_hour = min(23, start_hour + duration_hours) end_time = time(end_hour, start_minute) block = ORBlock.objects.create( operating_room=random.choice(operating_rooms), date=block_date, start_time=start_time, end_time=end_time, block_type=random.choice([choice[0] for choice in ORBlock.BLOCK_TYPE_CHOICES]), primary_surgeon=random.choice(data['surgeons']), service=random.choice([choice[0] for choice in ORBlock.SERVICE_CHOICES]), status=random.choice([choice[0] for choice in ORBlock.STATUS_CHOICES]), special_equipment=random.sample(self.equipment, k=random.randint(1, 4)), special_setup=random.choice([ "Standard setup required", "Special positioning needed", "Additional monitoring required", "Robotic system preparation", None ]), notes=random.choice([ "Standard block", "Priority case", "Complex procedure expected", "Patient has multiple comorbidities", None ]), created_by=random.choice(data['users']) ) # Add assistant surgeons if random.random() > 0.5: assistants = random.sample(data['surgeons'], k=random.randint(1, 2)) block.assistant_surgeons.set(assistants) blocks.append(block) print(f"Generated {len(blocks)} OR blocks") return blocks def generate_surgical_cases(self, or_blocks, data, count=80): """Generate surgical cases""" print(f"Generating {count} surgical cases...") cases = [] for i in range(count): block = random.choice(or_blocks) # Generate procedure information primary_procedure = random.choice(self.procedures) diagnosis = random.choice(self.diagnoses) # Generate timing scheduled_start = datetime.combine( block.date, block.start_time ) + timedelta(minutes=random.randint(0, 60)) estimated_duration = random.randint(30, 360) # Generate actual times for some cases actual_start = None actual_end = None if random.random() > 0.4: # 60% of cases have actual times actual_start = scheduled_start + timedelta(minutes=random.randint(-15, 45)) actual_end = actual_start + timedelta( minutes=estimated_duration + random.randint(-30, 60) ) case = SurgicalCase.objects.create( or_block=block, patient=random.choice(data['patients']), primary_surgeon=block.primary_surgeon, anesthesiologist=random.choice(data['anesthesiologists']) if data['anesthesiologists'] else None, circulating_nurse=random.choice(data['nurses']) if data['nurses'] else None, scrub_nurse=random.choice(data['nurses']) if data['nurses'] else None, primary_procedure=primary_procedure, secondary_procedures=random.sample(self.procedures, k=random.randint(0, 2)), procedure_codes=[f"CPT-{random.randint(10000, 99999)}" for _ in range(random.randint(1, 3))], case_type=random.choice([choice[0] for choice in SurgicalCase.CASE_TYPE_CHOICES]), approach=random.choice([choice[0] for choice in SurgicalCase.APPROACH_CHOICES]), anesthesia_type=random.choice([choice[0] for choice in SurgicalCase.ANESTHESIA_TYPE_CHOICES]), scheduled_start=scheduled_start, estimated_duration=estimated_duration, actual_start=actual_start, actual_end=actual_end, status=random.choice([choice[0] for choice in SurgicalCase.STATUS_CHOICES]), diagnosis=diagnosis, diagnosis_codes=[ f"ICD10-{chr(65 + random.randint(0, 25))}{random.randint(10, 99)}.{random.randint(0, 9)}" for _ in range(random.randint(1, 2))], clinical_notes=f"Patient presents with {diagnosis.lower()}. Surgical intervention indicated.", special_equipment=random.sample(self.equipment, k=random.randint(0, 3)), blood_products=random.sample([ "Packed RBC", "Fresh Frozen Plasma", "Platelets", "Cryoprecipitate" ], k=random.randint(0, 2)) if random.random() > 0.7 else [], implants=random.sample([ "Titanium Plate", "Mesh", "Stent", "Joint Prosthesis", "Screws" ], k=random.randint(0, 2)) if random.random() > 0.6 else [], patient_position=random.choice( [choice[0] for choice in SurgicalCase.PATIENT_POSITION_CHOICES]) if random.random() > 0.2 else None, complications=random.sample([ "Minor bleeding", "Adhesions encountered", "Technical difficulty" ], k=random.randint(0, 1)) if random.random() > 0.8 else [], estimated_blood_loss=random.randint(10, 500) if random.random() > 0.3 else None, encounter=random.choice(data['encounters']) if data['encounters'] and random.random() > 0.5 else None, admission=random.choice(data['admissions']) if data['admissions'] and random.random() > 0.4 else None, created_by=random.choice(data['users']) ) # Add assistant surgeons if random.random() > 0.4: assistants = random.sample(data['surgeons'], k=random.randint(1, 2)) case.assistant_surgeons.set(assistants) cases.append(case) print(f"Generated {len(cases)} surgical cases") return cases def generate_surgical_notes(self, surgical_cases, templates, data): """Generate surgical notes for surgical cases""" print(f"Generating surgical notes...") notes = [] for case in surgical_cases: # Only create notes for completed cases if case.status not in ['COMPLETED', 'IN_PROGRESS'] or random.random() > 0.7: continue # Find appropriate template template = None case_templates = [t for t in templates if t.tenant == case.tenant] if case_templates: template = random.choice(case_templates) note = SurgicalNote.objects.create( surgical_case=case, surgeon=case.primary_surgeon, preoperative_diagnosis=case.diagnosis, planned_procedure=case.primary_procedure, indication=f"Surgical treatment indicated for {case.diagnosis.lower()}", procedure_performed=case.primary_procedure, surgical_approach=f"{case.approach.lower()} approach utilized", findings=random.choice([ "Findings consistent with preoperative diagnosis", "Moderate inflammation noted", "Adhesions present as expected", "Anatomy within normal limits", "Pathology confirmed intraoperatively" ]), technique=f"Standard {case.approach.lower()} technique employed with appropriate surgical instruments", postoperative_diagnosis=case.diagnosis, condition=random.choice([choice[0] for choice in SurgicalNote.CONDITION_CHOICES]), disposition=random.choice([choice[0] for choice in SurgicalNote.DISPOSITION_CHOICES]), complications=random.choice([ "No intraoperative complications", "Minor bleeding encountered and controlled", "Technical difficulty with adhesions", None ]) if case.complications else "No intraoperative complications", estimated_blood_loss=case.estimated_blood_loss, blood_transfusion=random.choice([ None, "No blood transfusion required", "2 units packed RBC transfused" ]) if random.random() > 0.8 else None, specimens=random.choice([ f"Specimen sent to pathology for routine examination", "Tissue specimen sent for histopathological analysis", "No specimens obtained", None ]), implants=", ".join(case.implants) if case.implants else None, drains=random.choice([ "Jackson-Pratt drain placed", "No drains required", "Blake drain inserted", None ]) if random.random() > 0.6 else None, closure=random.choice([ "Fascia closed with interrupted sutures, skin closed with staples", "Layered closure performed, skin closed with absorbable sutures", "Standard closure technique employed" ]), postop_instructions=random.choice([ "NPO until bowel function returns, then clear liquids", "Regular diet as tolerated, ambulate as able", "Standard postoperative care protocol" ]), follow_up="Follow-up in clinic in 2-4 weeks, sooner if concerns", status=random.choice(['COMPLETED', 'SIGNED', 'DRAFT']), signed_datetime=timezone.now() - timedelta( days=random.randint(0, 5)) if random.random() > 0.3 else None, template_used=template ) notes.append(note) print(f"Generated {len(notes)} surgical notes") return notes def generate_equipment_usage(self, surgical_cases, data, count=200): """Generate equipment usage records""" print(f"Generating {count} equipment usage records...") usage_records = [] for _ in range(count): case = random.choice([c for c in surgical_cases if c.status in ['COMPLETED', 'IN_PROGRESS']]) equipment_name = random.choice(self.equipment) # Generate usage timing start_time = None end_time = None if case.actual_start: start_time = case.actual_start + timedelta(minutes=random.randint(0, 60)) end_time = start_time + timedelta(minutes=random.randint(15, 180)) usage = EquipmentUsage.objects.create( surgical_case=case, equipment_name=equipment_name, equipment_type=random.choice([choice[0] for choice in EquipmentUsage.EQUIPMENT_TYPE_CHOICES]), manufacturer=random.choice([ "Medtronic", "Johnson & Johnson", "Stryker", "Olympus", "Karl Storz", "Ethicon", "Boston Scientific", "Abbott" ]), model=f"Model-{random.randint(1000, 9999)}", serial_number=f"SN{random.randint(100000, 999999)}", quantity_used=random.randint(1, 5), unit_of_measure=random.choice([choice[0] for choice in EquipmentUsage.UNIT_OF_MEASURE_CHOICES]), start_time=start_time, end_time=end_time, unit_cost=Decimal(str(random.uniform(10, 1000))).quantize(Decimal('0.01')), lot_number=f"LOT{random.randint(1000, 9999)}" if random.random() > 0.3 else None, expiration_date=date.today() + timedelta( days=random.randint(30, 730)) if random.random() > 0.4 else None, sterilization_date=date.today() - timedelta( days=random.randint(1, 7)) if random.random() > 0.5 else None, notes=random.choice([ "Equipment functioned normally", "Routine maintenance required after use", "Minor technical issue resolved", None ]), recorded_by=random.choice(data['nurses']) if data['nurses'] else random.choice(data['users']) ) usage_records.append(usage) print(f"Generated {len(usage_records)} equipment usage records") return usage_records def generate_all_data(self, or_count=15, block_count=50, case_count=80, equipment_count=200): """Generate all operating theatre data""" print("Starting Operating Theatre data generation...") # Get existing data data = self.get_existing_data() # Generate data in order of dependencies operating_rooms = self.generate_operating_rooms(data, or_count) templates = self.generate_surgical_note_templates(data) or_blocks = self.generate_or_blocks(operating_rooms, data, block_count) surgical_cases = self.generate_surgical_cases(or_blocks, data, case_count) surgical_notes = self.generate_surgical_notes(surgical_cases, templates, data) equipment_usage = self.generate_equipment_usage(surgical_cases, data, equipment_count) summary = { 'operating_rooms': len(operating_rooms), 'surgical_note_templates': len(templates), 'or_blocks': len(or_blocks), 'surgical_cases': len(surgical_cases), 'surgical_notes': len(surgical_notes), 'equipment_usage': len(equipment_usage) } print("\n" + "=" * 50) print("OPERATING THEATRE DATA GENERATION COMPLETE") print("=" * 50) for model, count in summary.items(): print(f"{model.replace('_', ' ').title()}: {count}") print("=" * 50) return summary # Usage example: def generate_operating_theatre_data(): """Main function to generate operating theatre data""" generator = OperatingTheatreDataGenerator() return generator.generate_all_data( or_count=20, # Number of operating rooms block_count=60, # Number of OR blocks case_count=100, # Number of surgical cases equipment_count=250 # Number of equipment usage records ) if __name__ == "__main__": # Run the generator summary = generate_operating_theatre_data() print("\nGeneration completed successfully!")