1279 lines
56 KiB
Python
1279 lines
56 KiB
Python
# """
|
|
# 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 = [
|
|
"First Operating Room", "Second Operating Room", "Third Operating Room",
|
|
"General Surgery Wing", "Cardiac Surgery Wing", "Neurosurgery Wing",
|
|
"Emergency Surgery Unit", "Robotic Surgery Center", "Orthopedic Surgery Department"
|
|
]
|
|
|
|
# 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.OperatingRoomType.choices]),
|
|
status=random.choice([choice[0] for choice in OperatingRoom.ORStatus.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.SurgicalSpecialty.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.BlockType.choices]),
|
|
primary_surgeon=random.choice(data['surgeons']),
|
|
service=random.choice([choice[0] for choice in ORBlock.ORService.choices]),
|
|
status=random.choice([choice[0] for choice in ORBlock.BlockStatus.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.CaseType.choices]),
|
|
approach=random.choice([choice[0] for choice in SurgicalCase.SurgicalApproach.choices]),
|
|
anesthesia_type=random.choice([choice[0] for choice in SurgicalCase.AnesthesiaType.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.CaseStatus.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.PatientPosition.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.PatientCondition.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.EquipmentType.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.UnitOfMeasure.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!") |