866 lines
37 KiB
Python
866 lines
37 KiB
Python
"""
|
|
Saudi-influenced Operating Theatre Data Generator
|
|
Generates realistic test data for hospital operating room management system
|
|
with Saudi Arabian cultural context and medical practices.
|
|
"""
|
|
|
|
import random
|
|
import uuid
|
|
from datetime import datetime, timedelta, date, time
|
|
from decimal import Decimal
|
|
from faker import Faker
|
|
from typing import List, Dict, Any, Optional
|
|
|
|
# Initialize Faker
|
|
fake = Faker('en_US')
|
|
|
|
|
|
class SaudiOperatingTheatreDataGenerator:
|
|
"""
|
|
Generates Saudi-influenced data for Operating Theatre models.
|
|
"""
|
|
|
|
# Saudi hospital names (in English)
|
|
SAUDI_HOSPITALS = [
|
|
"King Faisal Specialist Hospital",
|
|
"King Fahad Medical City",
|
|
"King Abdulaziz University Hospital",
|
|
"King Khaled Eye Specialist Hospital",
|
|
"Prince Sultan Military Medical City",
|
|
"King Abdullah Medical Complex",
|
|
"King Saud Medical City",
|
|
"National Guard Hospital",
|
|
"Prince Mohammed Bin Abdulaziz Hospital",
|
|
"King Salman Hospital",
|
|
"Security Forces Hospital",
|
|
"Aramco Medical Center",
|
|
"Dr. Sulaiman Al Habib Medical Group",
|
|
"Saudi German Hospital",
|
|
"International Medical Center"
|
|
]
|
|
|
|
# Saudi cities
|
|
SAUDI_CITIES = [
|
|
"Riyadh", "Jeddah", "Mecca", "Medina", "Dammam",
|
|
"Khobar", "Dhahran", "Taif", "Tabuk", "Buraidah",
|
|
"Khamis Mushait", "Hofuf", "Jubail", "Yanbu", "Abha",
|
|
"Najran", "Jizan", "Hail", "Al-Qassim", "Qatif"
|
|
]
|
|
|
|
# Saudi regions
|
|
SAUDI_REGIONS = [
|
|
"Riyadh Region", "Makkah Region", "Eastern Province",
|
|
"Asir Region", "Madinah Region", "Qassim Region",
|
|
"Tabuk Region", "Hail Region", "Northern Borders",
|
|
"Jazan Region", "Najran Region", "Al-Baha Region"
|
|
]
|
|
|
|
# Common Saudi male first names
|
|
SAUDI_MALE_FIRST_NAMES = [
|
|
"Mohammed", "Abdullah", "Abdulrahman", "Khalid", "Fahad",
|
|
"Sultan", "Salman", "Saud", "Faisal", "Turki", "Ahmed",
|
|
"Omar", "Youssef", "Ibrahim", "Hamad", "Nasser", "Bandar",
|
|
"Mansour", "Majed", "Waleed", "Talal", "Rakan", "Yazeed",
|
|
"Meshal", "Naif", "Abdulaziz", "Saad", "Ali", "Hassan"
|
|
]
|
|
|
|
# Common Saudi female first names
|
|
SAUDI_FEMALE_FIRST_NAMES = [
|
|
"Nora", "Fatima", "Aisha", "Mariam", "Sarah", "Reem",
|
|
"Lama", "Hind", "Mona", "Amal", "Dalal", "Jawaher",
|
|
"Latifa", "Hessa", "Nouf", "Asma", "Khadija", "Layla",
|
|
"Rana", "Dina", "Hala", "Salma", "Yasmin", "Zainab",
|
|
"Lubna", "Hanaa", "Samira", "Najla", "Afaf", "Ghada"
|
|
]
|
|
|
|
# Common Saudi family names
|
|
SAUDI_FAMILY_NAMES = [
|
|
"Al-Saud", "Al-Rasheed", "Al-Qahtani", "Al-Otaibi", "Al-Dossari",
|
|
"Al-Harbi", "Al-Zahrani", "Al-Ghamdi", "Al-Shehri", "Al-Asmari",
|
|
"Al-Mutairi", "Al-Enezi", "Al-Shamari", "Al-Maliki", "Al-Johani",
|
|
"Al-Subaie", "Al-Hajri", "Al-Khaldi", "Al-Turki", "Al-Obaid",
|
|
"Al-Hassan", "Al-Sheikh", "Al-Najjar", "Al-Omari", "Al-Bakri"
|
|
]
|
|
|
|
# Medical specialties common in Saudi Arabia
|
|
MEDICAL_SPECIALTIES = [
|
|
"General Surgery", "Cardiac Surgery", "Neurosurgery",
|
|
"Orthopedic Surgery", "Pediatric Surgery", "Vascular Surgery",
|
|
"Thoracic Surgery", "Plastic Surgery", "Bariatric Surgery",
|
|
"Transplant Surgery", "Ophthalmology", "ENT Surgery",
|
|
"Urology", "Obstetrics", "Maxillofacial Surgery"
|
|
]
|
|
|
|
# Common surgical procedures in Saudi context
|
|
SURGICAL_PROCEDURES = {
|
|
"GENERAL": [
|
|
"Laparoscopic Cholecystectomy",
|
|
"Appendectomy",
|
|
"Hernia Repair",
|
|
"Bowel Resection",
|
|
"Gastric Bypass",
|
|
"Sleeve Gastrectomy",
|
|
"Thyroidectomy",
|
|
"Hemorrhoidectomy"
|
|
],
|
|
"CARDIAC": [
|
|
"Coronary Artery Bypass Grafting",
|
|
"Valve Replacement",
|
|
"Atrial Septal Defect Repair",
|
|
"Pacemaker Insertion",
|
|
"Angioplasty",
|
|
"Heart Transplant"
|
|
],
|
|
"ORTHOPEDIC": [
|
|
"Total Knee Replacement",
|
|
"Total Hip Replacement",
|
|
"Spinal Fusion",
|
|
"ACL Reconstruction",
|
|
"Rotator Cuff Repair",
|
|
"Fracture Fixation",
|
|
"Arthroscopy"
|
|
],
|
|
"NEURO": [
|
|
"Craniotomy",
|
|
"Brain Tumor Resection",
|
|
"Spinal Decompression",
|
|
"VP Shunt Placement",
|
|
"Aneurysm Clipping",
|
|
"Deep Brain Stimulation"
|
|
],
|
|
"OBSTETRIC": [
|
|
"Cesarean Section",
|
|
"Hysterectomy",
|
|
"Myomectomy",
|
|
"Ovarian Cystectomy",
|
|
"Tubal Ligation",
|
|
"D&C Procedure"
|
|
]
|
|
}
|
|
|
|
# Equipment commonly used in Saudi hospitals
|
|
MEDICAL_EQUIPMENT = [
|
|
"Da Vinci Surgical Robot",
|
|
"C-Arm Fluoroscopy",
|
|
"Zeiss Surgical Microscope",
|
|
"Harmonic Scalpel",
|
|
"LigaSure Device",
|
|
"Medtronic Neuromonitoring System",
|
|
"Stryker Navigation System",
|
|
"Karl Storz Laparoscopic Tower",
|
|
"GE Ultrasound Machine",
|
|
"Phillips CT Scanner"
|
|
]
|
|
|
|
def __init__(self, tenant_id: Optional[str] = None):
|
|
"""Initialize the data generator."""
|
|
self.tenant_id = tenant_id or str(uuid.uuid4())
|
|
self.generated_surgeons = []
|
|
self.generated_patients = []
|
|
self.generated_rooms = []
|
|
|
|
def generate_saudi_name(self, gender: str = 'male') -> Dict[str, str]:
|
|
"""Generate a Saudi-style name."""
|
|
if gender.lower() == 'male':
|
|
first_name = random.choice(self.SAUDI_MALE_FIRST_NAMES)
|
|
else:
|
|
first_name = random.choice(self.SAUDI_FEMALE_FIRST_NAMES)
|
|
|
|
family_name = random.choice(self.SAUDI_FAMILY_NAMES)
|
|
|
|
# Sometimes add a middle name (father's name for males)
|
|
if gender.lower() == 'male' and random.random() > 0.5:
|
|
middle_name = random.choice(self.SAUDI_MALE_FIRST_NAMES)
|
|
full_name = f"Dr. {first_name} {middle_name} {family_name}"
|
|
else:
|
|
full_name = f"Dr. {first_name} {family_name}"
|
|
|
|
return {
|
|
'first_name': first_name,
|
|
'last_name': family_name,
|
|
'full_name': full_name
|
|
}
|
|
|
|
def generate_operating_room(self, room_number: int) -> Dict[str, Any]:
|
|
"""Generate operating room data with Saudi context."""
|
|
room_types = ['GENERAL', 'CARDIAC', 'NEURO', 'ORTHOPEDIC', 'OBSTETRIC',
|
|
'PEDIATRIC', 'OPHTHALMOLOGY', 'ENT', 'UROLOGY']
|
|
|
|
room_type = random.choice(room_types)
|
|
hospital = random.choice(self.SAUDI_HOSPITALS)
|
|
|
|
# Advanced equipment more common in Saudi hospitals
|
|
has_advanced_equipment = random.random() > 0.3
|
|
|
|
room_data = {
|
|
'tenant': self.tenant_id,
|
|
'room_id': str(uuid.uuid4()),
|
|
'room_number': f"OR-{room_number:03d}",
|
|
'room_name': f"{hospital} - {room_type.title()} OR {room_number}",
|
|
'room_type': room_type,
|
|
'status': random.choice(['AVAILABLE', 'OCCUPIED', 'CLEANING', 'SETUP']),
|
|
'floor_number': random.randint(1, 5),
|
|
'room_size': round(random.uniform(40, 80), 2),
|
|
'ceiling_height': round(random.uniform(3.0, 4.5), 2),
|
|
'temperature_min': 18.0,
|
|
'temperature_max': 24.0,
|
|
'humidity_min': 30.0,
|
|
'humidity_max': 60.0,
|
|
'air_changes_per_hour': random.randint(20, 25),
|
|
'positive_pressure': True,
|
|
'equipment_list': self._generate_equipment_list(room_type, has_advanced_equipment),
|
|
'special_features': self._generate_special_features(room_type),
|
|
'has_c_arm': room_type in ['ORTHOPEDIC', 'NEURO', 'VASCULAR'] or random.random() > 0.5,
|
|
'has_ct': room_type in ['NEURO', 'CARDIAC'] and has_advanced_equipment,
|
|
'has_mri': room_type == 'NEURO' and has_advanced_equipment and random.random() > 0.7,
|
|
'has_ultrasound': True,
|
|
'has_neuromonitoring': room_type in ['NEURO', 'ORTHOPEDIC'],
|
|
'supports_robotic': has_advanced_equipment and room_type in ['GENERAL', 'UROLOGY', 'CARDIAC'],
|
|
'supports_laparoscopic': room_type in ['GENERAL', 'OBSTETRIC', 'UROLOGY'],
|
|
'supports_microscopy': room_type in ['NEURO', 'OPHTHALMOLOGY', 'ENT'],
|
|
'supports_laser': room_type in ['OPHTHALMOLOGY', 'ENT', 'UROLOGY'],
|
|
'max_case_duration': random.randint(240, 720),
|
|
'turnover_time': random.randint(20, 45),
|
|
'cleaning_time': random.randint(30, 60),
|
|
'required_nurses': random.randint(2, 4),
|
|
'required_techs': random.randint(1, 3),
|
|
'is_active': True,
|
|
'accepts_emergency': room_type != 'OPHTHALMOLOGY',
|
|
'building': f"Building {random.choice(['A', 'B', 'C', 'Main', 'East', 'West'])}",
|
|
'wing': random.choice(['North', 'South', 'East', 'West', 'Central']),
|
|
'created_at': fake.date_time_between(start_date='-2y', end_date='now'),
|
|
'updated_at': fake.date_time_between(start_date='-30d', end_date='now')
|
|
}
|
|
|
|
return room_data
|
|
|
|
def _generate_equipment_list(self, room_type: str, has_advanced: bool) -> List[str]:
|
|
"""Generate equipment list based on room type."""
|
|
basic_equipment = [
|
|
"Anesthesia Machine",
|
|
"Patient Monitor",
|
|
"Surgical Lights",
|
|
"Operating Table",
|
|
"Electrocautery Unit",
|
|
"Suction System",
|
|
"Instrument Tables"
|
|
]
|
|
|
|
specialized_equipment = {
|
|
'CARDIAC': ["Heart-Lung Machine", "IABP", "TEE Machine"],
|
|
'NEURO': ["Surgical Microscope", "Neuronavigation", "CUSA"],
|
|
'ORTHOPEDIC': ["C-Arm", "Power Tools", "Traction Table"],
|
|
'OPHTHALMOLOGY': ["Phaco Machine", "OCT", "YAG Laser"],
|
|
'ENT': ["ENT Microscope", "Shaver System", "Navigation"],
|
|
'UROLOGY': ["Cystoscopy Tower", "Laser System", "Lithotripter"]
|
|
}
|
|
|
|
equipment = basic_equipment.copy()
|
|
|
|
if room_type in specialized_equipment:
|
|
equipment.extend(specialized_equipment[room_type])
|
|
|
|
if has_advanced:
|
|
equipment.extend(random.sample(self.MEDICAL_EQUIPMENT, k=random.randint(2, 4)))
|
|
|
|
return equipment
|
|
|
|
def _generate_special_features(self, room_type: str) -> List[str]:
|
|
"""Generate special features for OR."""
|
|
features = ["Laminar Flow", "HEPA Filtration", "Integrated Video System"]
|
|
|
|
special_features = {
|
|
'CARDIAC': ["Hybrid OR Capability", "Cardiac Catheterization"],
|
|
'NEURO': ["Intraoperative MRI Compatible", "Neuromonitoring"],
|
|
'ORTHOPEDIC': ["Laminar Flow Class 100", "Biplane Imaging"],
|
|
'OPHTHALMOLOGY': ["Laser Safety", "Microscope Integration"],
|
|
'OBSTETRIC': ["Neonatal Resuscitation Area", "Fetal Monitoring"]
|
|
}
|
|
|
|
if room_type in special_features:
|
|
features.extend(special_features[room_type])
|
|
|
|
return features
|
|
|
|
def generate_or_block(self, operating_room_id: str, block_date: date) -> Dict[str, Any]:
|
|
"""Generate OR block schedule."""
|
|
surgeon = self.generate_saudi_name('male')
|
|
services = ['GENERAL', 'CARDIAC', 'NEURO', 'ORTHOPEDIC', 'OBSTETRIC',
|
|
'OPHTHALMOLOGY', 'ENT', 'UROLOGY']
|
|
|
|
# Saudi working hours typically 7 AM to 3 PM or 8 AM to 4 PM
|
|
start_hours = [7, 8, 9, 13, 14]
|
|
start_hour = random.choice(start_hours)
|
|
duration_hours = random.randint(2, 8)
|
|
|
|
block_data = {
|
|
'operating_room': operating_room_id,
|
|
'block_id': str(uuid.uuid4()),
|
|
'date': block_date,
|
|
'start_time': time(start_hour, 0),
|
|
'end_time': time((start_hour + duration_hours) % 24, 0),
|
|
'block_type': random.choice(['SCHEDULED', 'EMERGENCY', 'RESERVED']),
|
|
'primary_surgeon': surgeon['full_name'],
|
|
'service': random.choice(services),
|
|
'status': random.choice(['SCHEDULED', 'ACTIVE', 'COMPLETED']),
|
|
'allocated_minutes': duration_hours * 60,
|
|
'used_minutes': random.randint(duration_hours * 45, duration_hours * 60),
|
|
'special_equipment': random.sample(self.MEDICAL_EQUIPMENT, k=random.randint(0, 3)),
|
|
'special_setup': self._generate_special_setup(),
|
|
'notes': self._generate_block_notes(),
|
|
'created_at': fake.date_time_between(start_date='-30d', end_date='now'),
|
|
'updated_at': fake.date_time_between(start_date='-7d', end_date='now')
|
|
}
|
|
|
|
return block_data
|
|
|
|
def _generate_special_setup(self) -> str:
|
|
"""Generate special setup requirements."""
|
|
setups = [
|
|
"Prone positioning required",
|
|
"Lateral positioning with bean bag",
|
|
"Beach chair position",
|
|
"Microscope setup required",
|
|
"Robot docking from patient left",
|
|
"Fluoroscopy setup",
|
|
"Neuromonitoring setup required",
|
|
"Cell saver required",
|
|
"Warming blanket needed"
|
|
]
|
|
return random.choice(setups) if random.random() > 0.5 else ""
|
|
|
|
def _generate_block_notes(self) -> str:
|
|
"""Generate block notes."""
|
|
notes = [
|
|
"Complex case - allow extra time",
|
|
"Teaching case for residents",
|
|
"VIP patient - special protocols",
|
|
"Latex allergy - latex-free environment",
|
|
"Multi-specialty case",
|
|
"Consultant from King Faisal Hospital attending",
|
|
"Research protocol case",
|
|
"International patient - translator needed"
|
|
]
|
|
return random.choice(notes) if random.random() > 0.3 else ""
|
|
|
|
def generate_surgical_case(self, or_block_id: str, patient_id: str) -> Dict[str, Any]:
|
|
"""Generate surgical case data."""
|
|
case_type = random.choice(['ELECTIVE', 'URGENT', 'EMERGENCY'])
|
|
service_type = random.choice(list(self.SURGICAL_PROCEDURES.keys()))
|
|
procedure = random.choice(self.SURGICAL_PROCEDURES[service_type])
|
|
|
|
# Generate Saudi medical team
|
|
primary_surgeon = self.generate_saudi_name('male')
|
|
anesthesiologist = self.generate_saudi_name(random.choice(['male', 'female']))
|
|
nurse = self.generate_saudi_name('female') # Nursing often female in Saudi
|
|
|
|
scheduled_start = fake.date_time_between(start_date='now', end_date='+30d')
|
|
duration = random.randint(30, 360)
|
|
|
|
case_data = {
|
|
'or_block': or_block_id,
|
|
'case_id': str(uuid.uuid4()),
|
|
'case_number': f"SURG-{datetime.now().strftime('%Y%m%d')}-{random.randint(1, 9999):04d}",
|
|
'patient': patient_id,
|
|
'primary_surgeon': primary_surgeon['full_name'],
|
|
'anesthesiologist': anesthesiologist['full_name'],
|
|
'circulating_nurse': nurse['full_name'],
|
|
'scrub_nurse': self.generate_saudi_name('female')['full_name'],
|
|
'primary_procedure': procedure,
|
|
'secondary_procedures': self._generate_secondary_procedures(service_type),
|
|
'procedure_codes': self._generate_procedure_codes(),
|
|
'case_type': case_type,
|
|
'approach': random.choice(['OPEN', 'LAPAROSCOPIC', 'ROBOTIC', 'ENDOSCOPIC']),
|
|
'anesthesia_type': random.choice(['GENERAL', 'REGIONAL', 'SPINAL', 'LOCAL']),
|
|
'scheduled_start': scheduled_start,
|
|
'estimated_duration': duration,
|
|
'actual_start': scheduled_start + timedelta(minutes=random.randint(-15, 30)) if random.random() > 0.5 else None,
|
|
'actual_end': scheduled_start + timedelta(minutes=duration + random.randint(-30, 60)) if random.random() > 0.5 else None,
|
|
'status': random.choice(['SCHEDULED', 'IN_PROGRESS', 'COMPLETED', 'DELAYED']),
|
|
'diagnosis': self._generate_diagnosis(),
|
|
'diagnosis_codes': self._generate_diagnosis_codes(),
|
|
'clinical_notes': self._generate_clinical_notes(),
|
|
'special_equipment': random.sample(self.MEDICAL_EQUIPMENT, k=random.randint(1, 3)),
|
|
'blood_products': self._generate_blood_products() if random.random() > 0.7 else [],
|
|
'implants': self._generate_implants(service_type) if random.random() > 0.6 else [],
|
|
'patient_position': random.choice(['SUPINE', 'PRONE', 'LATERAL', 'LITHOTOMY']),
|
|
'complications': [] if random.random() > 0.1 else ["Minor bleeding controlled"],
|
|
'estimated_blood_loss': random.randint(10, 500) if random.random() > 0.3 else None,
|
|
'created_at': fake.date_time_between(start_date='-7d', end_date='now'),
|
|
'updated_at': fake.date_time_between(start_date='-1d', end_date='now')
|
|
}
|
|
|
|
return case_data
|
|
|
|
def _generate_secondary_procedures(self, service_type: str) -> List[str]:
|
|
"""Generate secondary procedures."""
|
|
if random.random() > 0.6:
|
|
procedures = self.SURGICAL_PROCEDURES.get(service_type, [])
|
|
return random.sample(procedures, k=min(random.randint(1, 2), len(procedures)))
|
|
return []
|
|
|
|
def _generate_procedure_codes(self) -> List[str]:
|
|
"""Generate CPT procedure codes."""
|
|
# Sample CPT codes
|
|
codes = []
|
|
for _ in range(random.randint(1, 3)):
|
|
codes.append(f"{random.randint(10000, 99999)}")
|
|
return codes
|
|
|
|
def _generate_diagnosis(self) -> str:
|
|
"""Generate diagnosis."""
|
|
diagnoses = [
|
|
"Acute Appendicitis",
|
|
"Cholelithiasis",
|
|
"Inguinal Hernia",
|
|
"Coronary Artery Disease",
|
|
"Brain Tumor",
|
|
"Degenerative Disc Disease",
|
|
"Osteoarthritis",
|
|
"Breast Cancer",
|
|
"Colorectal Cancer",
|
|
"Thyroid Nodule",
|
|
"Uterine Fibroids",
|
|
"Kidney Stones",
|
|
"Prostate Hyperplasia"
|
|
]
|
|
return random.choice(diagnoses)
|
|
|
|
def _generate_diagnosis_codes(self) -> List[str]:
|
|
"""Generate ICD-10 diagnosis codes."""
|
|
# Sample ICD-10 codes
|
|
codes = []
|
|
for _ in range(random.randint(1, 3)):
|
|
letter = random.choice(['K', 'I', 'M', 'C', 'N', 'E'])
|
|
codes.append(f"{letter}{random.randint(10, 99)}.{random.randint(0, 9)}")
|
|
return codes
|
|
|
|
def _generate_clinical_notes(self) -> str:
|
|
"""Generate clinical notes."""
|
|
notes = [
|
|
"Patient with history of diabetes mellitus type 2, controlled on metformin",
|
|
"No known drug allergies. Previous surgery without complications",
|
|
"Hypertensive patient on ACE inhibitors, blood pressure stable",
|
|
"Patient fasting since midnight as per protocol",
|
|
"Preoperative antibiotics administered",
|
|
"Patient counseled about procedure risks and benefits",
|
|
"Informed consent obtained in Arabic and English"
|
|
]
|
|
return random.choice(notes)
|
|
|
|
def _generate_blood_products(self) -> List[str]:
|
|
"""Generate blood product requirements."""
|
|
products = []
|
|
if random.random() > 0.5:
|
|
products.append(f"PRBC {random.randint(1, 4)} units")
|
|
if random.random() > 0.7:
|
|
products.append(f"FFP {random.randint(1, 2)} units")
|
|
if random.random() > 0.8:
|
|
products.append(f"Platelets {random.randint(1, 2)} units")
|
|
return products
|
|
|
|
def _generate_implants(self, service_type: str) -> List[str]:
|
|
"""Generate implant requirements based on service type."""
|
|
implants = {
|
|
'ORTHOPEDIC': [
|
|
"Total Knee Prosthesis - Zimmer",
|
|
"Total Hip Prosthesis - Stryker",
|
|
"Spinal Fusion Cage - Medtronic",
|
|
"ACL Graft",
|
|
"Fracture Plate and Screws"
|
|
],
|
|
'CARDIAC': [
|
|
"Mechanical Valve - St. Jude",
|
|
"Bioprosthetic Valve",
|
|
"Pacemaker - Medtronic",
|
|
"Coronary Stent",
|
|
"Vascular Graft"
|
|
],
|
|
'NEURO': [
|
|
"VP Shunt",
|
|
"Deep Brain Stimulator",
|
|
"Cranial Plate",
|
|
"Aneurysm Clip"
|
|
],
|
|
'GENERAL': [
|
|
"Mesh for Hernia Repair",
|
|
"Gastric Band",
|
|
"Biliary Stent"
|
|
]
|
|
}
|
|
|
|
if service_type in implants:
|
|
return random.sample(implants[service_type], k=1)
|
|
return []
|
|
|
|
def generate_surgical_note(self, surgical_case_id: str, surgeon_id: str) -> Dict[str, Any]:
|
|
"""Generate surgical note."""
|
|
note_data = {
|
|
'surgical_case': surgical_case_id,
|
|
'note_id': str(uuid.uuid4()),
|
|
'surgeon': surgeon_id,
|
|
'preoperative_diagnosis': self._generate_diagnosis(),
|
|
'planned_procedure': random.choice([proc for procs in self.SURGICAL_PROCEDURES.values() for proc in procs]),
|
|
'indication': self._generate_indication(),
|
|
'procedure_performed': self._generate_procedure_details(),
|
|
'surgical_approach': self._generate_surgical_approach(),
|
|
'findings': self._generate_findings(),
|
|
'technique': self._generate_technique(),
|
|
'postoperative_diagnosis': self._generate_diagnosis(),
|
|
'condition': random.choice(['STABLE', 'GOOD', 'FAIR']),
|
|
'disposition': random.choice(['RECOVERY', 'ICU', 'WARD']),
|
|
'complications': "None" if random.random() > 0.1 else "Minor bleeding, controlled",
|
|
'estimated_blood_loss': random.randint(10, 500),
|
|
'blood_transfusion': "None" if random.random() > 0.8 else "1 unit PRBC",
|
|
'specimens': self._generate_specimens() if random.random() > 0.5 else None,
|
|
'implants': self._generate_implant_details() if random.random() > 0.6 else None,
|
|
'drains': "JP drain placed" if random.random() > 0.5 else None,
|
|
'closure': self._generate_closure_details(),
|
|
'postop_instructions': self._generate_postop_instructions(),
|
|
'follow_up': "Follow up in 2 weeks in surgical clinic",
|
|
'status': random.choice(['DRAFT', 'COMPLETED', 'SIGNED']),
|
|
'signed_datetime': fake.date_time_between(start_date='-7d', end_date='now') if random.random() > 0.3 else None,
|
|
'created_at': fake.date_time_between(start_date='-7d', end_date='now'),
|
|
'updated_at': fake.date_time_between(start_date='-1d', end_date='now')
|
|
}
|
|
|
|
return note_data
|
|
|
|
def _generate_indication(self) -> str:
|
|
"""Generate surgical indication."""
|
|
indications = [
|
|
"Symptomatic for 6 months, failed conservative management",
|
|
"Progressive symptoms despite medical therapy",
|
|
"Acute presentation with signs of peritonitis",
|
|
"Elective procedure for quality of life improvement",
|
|
"Urgent intervention to prevent complications",
|
|
"Diagnostic and therapeutic intervention",
|
|
"Staged procedure as per treatment protocol"
|
|
]
|
|
return random.choice(indications)
|
|
|
|
def _generate_procedure_details(self) -> str:
|
|
"""Generate detailed procedure description."""
|
|
return "Procedure performed as planned using standard technique with no intraoperative complications"
|
|
|
|
def _generate_surgical_approach(self) -> str:
|
|
"""Generate surgical approach description."""
|
|
approaches = [
|
|
"Midline laparotomy incision",
|
|
"Laparoscopic approach with 4 ports",
|
|
"Lateral thoracotomy through 5th intercostal space",
|
|
"Posterior approach to spine",
|
|
"Deltopectoral approach",
|
|
"Bikini incision for cesarean section"
|
|
]
|
|
return random.choice(approaches)
|
|
|
|
def _generate_findings(self) -> str:
|
|
"""Generate intraoperative findings."""
|
|
findings = [
|
|
"Findings consistent with preoperative diagnosis",
|
|
"Adhesions from previous surgery noted and lysed",
|
|
"No evidence of metastatic disease",
|
|
"Inflammation noted in surrounding tissues",
|
|
"Anatomy normal, procedure proceeded as planned"
|
|
]
|
|
return random.choice(findings)
|
|
|
|
def _generate_technique(self) -> str:
|
|
"""Generate surgical technique description."""
|
|
return "Standard surgical technique employed with meticulous hemostasis throughout the procedure"
|
|
|
|
def _generate_specimens(self) -> str:
|
|
"""Generate specimen details."""
|
|
specimens = [
|
|
"Appendix sent to pathology",
|
|
"Gallbladder sent to pathology",
|
|
"Lymph nodes sent for frozen section",
|
|
"Tissue biopsy sent to pathology",
|
|
"Tumor specimen sent with margins marked"
|
|
]
|
|
return random.choice(specimens)
|
|
|
|
def _generate_implant_details(self) -> str:
|
|
"""Generate implant details."""
|
|
return "Implant placed per manufacturer protocol, position confirmed with imaging"
|
|
|
|
def _generate_closure_details(self) -> str:
|
|
"""Generate closure details."""
|
|
closures = [
|
|
"Layered closure with absorbable sutures",
|
|
"Skin closed with staples",
|
|
"Subcuticular closure with absorbable sutures",
|
|
"Wound closed in layers, dressing applied",
|
|
"Closure with 3-0 Vicryl and skin adhesive"
|
|
]
|
|
return random.choice(closures)
|
|
|
|
def _generate_postop_instructions(self) -> str:
|
|
"""Generate postoperative instructions."""
|
|
instructions = [
|
|
"NPO until bowel sounds return, advance diet as tolerated",
|
|
"Ambulate within 6 hours, incentive spirometry every hour",
|
|
"Pain control with PCA, transition to oral when tolerating diet",
|
|
"DVT prophylaxis with heparin, sequential compression devices",
|
|
"Monitor vitals every 4 hours, daily labs",
|
|
"Antibiotics for 24 hours postoperatively",
|
|
"Foley catheter to be removed POD 1"
|
|
]
|
|
return "; ".join(random.sample(instructions, k=random.randint(2, 4)))
|
|
|
|
def generate_equipment_usage(self, surgical_case_id: str) -> List[Dict[str, Any]]:
|
|
"""Generate equipment usage records."""
|
|
equipment_records = []
|
|
num_equipment = random.randint(3, 8)
|
|
|
|
for _ in range(num_equipment):
|
|
equipment_type = random.choice([
|
|
'SURGICAL_INSTRUMENT', 'MONITORING_DEVICE', 'ELECTROCAUTERY',
|
|
'LASER', 'MICROSCOPE', 'ULTRASOUND', 'DISPOSABLE'
|
|
])
|
|
|
|
equipment_name = self._get_equipment_name(equipment_type)
|
|
|
|
usage_data = {
|
|
'surgical_case': surgical_case_id,
|
|
'usage_id': str(uuid.uuid4()),
|
|
'equipment_name': equipment_name,
|
|
'equipment_type': equipment_type,
|
|
'manufacturer': self._get_manufacturer(),
|
|
'model': f"Model-{random.randint(100, 999)}",
|
|
'serial_number': f"SN{random.randint(100000, 999999)}",
|
|
'quantity_used': random.randint(1, 5),
|
|
'unit_of_measure': random.choice(['EACH', 'SET', 'PACK', 'BOX']),
|
|
'start_time': fake.date_time_between(start_date='-1d', end_date='now'),
|
|
'end_time': fake.date_time_between(start_date='now', end_date='+4h'),
|
|
'unit_cost': Decimal(str(round(random.uniform(10, 5000), 2))),
|
|
'lot_number': f"LOT{random.randint(10000, 99999)}",
|
|
'expiration_date': fake.date_between(start_date='+30d', end_date='+2y'),
|
|
'sterilization_date': fake.date_between(start_date='-7d', end_date='today'),
|
|
'notes': self._generate_equipment_notes(),
|
|
'created_at': fake.date_time_between(start_date='-1d', end_date='now'),
|
|
'updated_at': fake.date_time_between(start_date='-1h', end_date='now')
|
|
}
|
|
|
|
equipment_records.append(usage_data)
|
|
|
|
return equipment_records
|
|
|
|
def _get_equipment_name(self, equipment_type: str) -> str:
|
|
"""Get equipment name based on type."""
|
|
equipment_names = {
|
|
'SURGICAL_INSTRUMENT': [
|
|
"Harmonic Scalpel", "LigaSure Device", "Surgical Stapler",
|
|
"Laparoscopic Grasper Set", "Retractor Set", "Scalpel Set"
|
|
],
|
|
'MONITORING_DEVICE': [
|
|
"Cardiac Monitor", "Pulse Oximeter", "BIS Monitor",
|
|
"Arterial Line Monitor", "Central Line Kit"
|
|
],
|
|
'ELECTROCAUTERY': [
|
|
"Bovie Electrocautery", "Bipolar Forceps", "Monopolar Cautery"
|
|
],
|
|
'LASER': [
|
|
"CO2 Laser", "YAG Laser", "Holmium Laser", "Argon Laser"
|
|
],
|
|
'MICROSCOPE': [
|
|
"Zeiss OPMI", "Leica Surgical Microscope", "Pentero Microscope"
|
|
],
|
|
'ULTRASOUND': [
|
|
"GE Ultrasound", "Phillips EPIQ", "Sonosite Edge"
|
|
],
|
|
'DISPOSABLE': [
|
|
"Surgical Drape Set", "Gown Pack", "Suture Set",
|
|
"Surgical Gloves", "Sponge Pack"
|
|
]
|
|
}
|
|
|
|
return random.choice(equipment_names.get(equipment_type, ["Generic Equipment"]))
|
|
|
|
def _get_manufacturer(self) -> str:
|
|
"""Get equipment manufacturer."""
|
|
manufacturers = [
|
|
"Medtronic", "Johnson & Johnson", "Stryker", "Boston Scientific",
|
|
"Abbott", "GE Healthcare", "Siemens", "Phillips", "Karl Storz",
|
|
"Olympus", "Zimmer Biomet", "Smith & Nephew", "B. Braun"
|
|
]
|
|
return random.choice(manufacturers)
|
|
|
|
def _generate_equipment_notes(self) -> str:
|
|
"""Generate equipment usage notes."""
|
|
notes = [
|
|
"Equipment functioning properly",
|
|
"Calibrated before use",
|
|
"Backup equipment available",
|
|
"Special settings documented",
|
|
"Used per protocol",
|
|
""
|
|
]
|
|
return random.choice(notes)
|
|
|
|
def generate_surgical_note_template(self) -> Dict[str, Any]:
|
|
"""Generate surgical note template."""
|
|
specialties = ['ALL', 'GENERAL', 'CARDIAC', 'NEURO', 'ORTHOPEDIC',
|
|
'OBSTETRIC', 'OPHTHALMOLOGY', 'ENT', 'UROLOGY']
|
|
|
|
specialty = random.choice(specialties)
|
|
|
|
template_data = {
|
|
'tenant': self.tenant_id,
|
|
'template_id': str(uuid.uuid4()),
|
|
'name': f"{specialty} Surgery Template - {random.choice(['Standard', 'Complex', 'Emergency'])}",
|
|
'description': f"Standard template for {specialty.lower()} surgical procedures",
|
|
'procedure_type': random.choice(self.SURGICAL_PROCEDURES.get(specialty, ["General Procedure"])) if specialty != 'ALL' else None,
|
|
'specialty': specialty,
|
|
'preoperative_diagnosis_template': "Preoperative Diagnosis: [Diagnosis]",
|
|
'planned_procedure_template': "Planned Procedure: [Procedure Name]",
|
|
'indication_template': "Indication: Patient presents with [symptoms] requiring surgical intervention",
|
|
'procedure_performed_template': "Procedure Performed: [Actual procedure]",
|
|
'surgical_approach_template': "Approach: [Describe surgical approach]",
|
|
'findings_template': "Findings: [Describe intraoperative findings]",
|
|
'technique_template': "Technique: [Describe surgical technique in detail]",
|
|
'postoperative_diagnosis_template': "Postoperative Diagnosis: [Final diagnosis]",
|
|
'complications_template': "Complications: [None/Describe if any]",
|
|
'specimens_template': "Specimens: [List specimens sent to pathology]",
|
|
'implants_template': "Implants: [List any implants used]",
|
|
'closure_template': "Closure: [Describe closure technique]",
|
|
'postop_instructions_template': "Postoperative Instructions: [List instructions]",
|
|
'is_active': True,
|
|
'is_default': random.random() > 0.7,
|
|
'usage_count': random.randint(0, 100),
|
|
'created_at': fake.date_time_between(start_date='-1y', end_date='now'),
|
|
'updated_at': fake.date_time_between(start_date='-30d', end_date='now')
|
|
}
|
|
|
|
return template_data
|
|
|
|
def generate_complete_dataset(self,
|
|
num_rooms: int = 10,
|
|
num_blocks_per_room: int = 5,
|
|
num_cases_per_block: int = 3) -> Dict[str, List]:
|
|
"""Generate complete dataset for all models."""
|
|
|
|
dataset = {
|
|
'operating_rooms': [],
|
|
'or_blocks': [],
|
|
'surgical_cases': [],
|
|
'surgical_notes': [],
|
|
'equipment_usage': [],
|
|
'surgical_note_templates': []
|
|
}
|
|
|
|
# Generate operating rooms
|
|
for i in range(1, num_rooms + 1):
|
|
room = self.generate_operating_room(i)
|
|
dataset['operating_rooms'].append(room)
|
|
self.generated_rooms.append(room)
|
|
|
|
# Generate OR blocks for each room
|
|
for j in range(num_blocks_per_room):
|
|
block_date = fake.date_between(start_date='today', end_date='+30d')
|
|
block = self.generate_or_block(room['room_id'], block_date)
|
|
dataset['or_blocks'].append(block)
|
|
|
|
# Generate surgical cases for each block
|
|
for k in range(random.randint(1, num_cases_per_block)):
|
|
# Generate a patient ID (placeholder)
|
|
patient_id = str(uuid.uuid4())
|
|
|
|
case = self.generate_surgical_case(block['block_id'], patient_id)
|
|
dataset['surgical_cases'].append(case)
|
|
|
|
# Generate surgical note for completed cases
|
|
if case['status'] == 'COMPLETED':
|
|
note = self.generate_surgical_note(case['case_id'], block['primary_surgeon'])
|
|
dataset['surgical_notes'].append(note)
|
|
|
|
# Generate equipment usage
|
|
equipment_usage = self.generate_equipment_usage(case['case_id'])
|
|
dataset['equipment_usage'].extend(equipment_usage)
|
|
|
|
# Generate surgical note templates
|
|
for _ in range(15):
|
|
template = self.generate_surgical_note_template()
|
|
dataset['surgical_note_templates'].append(template)
|
|
|
|
return dataset
|
|
|
|
def print_statistics(self, dataset: Dict[str, List]) -> None:
|
|
"""Print statistics about generated data."""
|
|
print("\n" + "="*60)
|
|
print("Saudi Operating Theatre Data Generation Complete")
|
|
print("="*60)
|
|
print(f"Operating Rooms Generated: {len(dataset['operating_rooms'])}")
|
|
print(f"OR Blocks Generated: {len(dataset['or_blocks'])}")
|
|
print(f"Surgical Cases Generated: {len(dataset['surgical_cases'])}")
|
|
print(f"Surgical Notes Generated: {len(dataset['surgical_notes'])}")
|
|
print(f"Equipment Usage Records: {len(dataset['equipment_usage'])}")
|
|
print(f"Surgical Note Templates: {len(dataset['surgical_note_templates'])}")
|
|
print("="*60)
|
|
|
|
# Room type distribution
|
|
room_types = {}
|
|
for room in dataset['operating_rooms']:
|
|
room_type = room['room_type']
|
|
room_types[room_type] = room_types.get(room_type, 0) + 1
|
|
|
|
print("\nOperating Room Types:")
|
|
for room_type, count in sorted(room_types.items()):
|
|
print(f" {room_type}: {count}")
|
|
|
|
# Case type distribution
|
|
case_types = {}
|
|
for case in dataset['surgical_cases']:
|
|
case_type = case['case_type']
|
|
case_types[case_type] = case_types.get(case_type, 0) + 1
|
|
|
|
print("\nSurgical Case Types:")
|
|
for case_type, count in sorted(case_types.items()):
|
|
print(f" {case_type}: {count}")
|
|
|
|
# Status distribution
|
|
case_statuses = {}
|
|
for case in dataset['surgical_cases']:
|
|
status = case['status']
|
|
case_statuses[status] = case_statuses.get(status, 0) + 1
|
|
|
|
print("\nCase Status Distribution:")
|
|
for status, count in sorted(case_statuses.items()):
|
|
print(f" {status}: {count}")
|
|
|
|
|
|
# Example usage
|
|
if __name__ == "__main__":
|
|
# Initialize generator
|
|
generator = SaudiOperatingTheatreDataGenerator()
|
|
|
|
# Generate complete dataset
|
|
dataset = generator.generate_complete_dataset(
|
|
num_rooms=8,
|
|
num_blocks_per_room=4,
|
|
num_cases_per_block=3
|
|
)
|
|
|
|
# Print statistics
|
|
generator.print_statistics(dataset)
|
|
|
|
# Example: Print first operating room
|
|
if dataset['operating_rooms']:
|
|
print("\n" + "="*60)
|
|
print("Sample Operating Room:")
|
|
print("="*60)
|
|
room = dataset['operating_rooms'][0]
|
|
for key, value in room.items():
|
|
if key not in ['equipment_list', 'special_features']:
|
|
print(f"{key}: {value}")
|
|
|
|
# Example: Print first surgical case
|
|
if dataset['surgical_cases']:
|
|
print("\n" + "="*60)
|
|
print("Sample Surgical Case:")
|
|
print("="*60)
|
|
case = dataset['surgical_cases'][0]
|
|
for key, value in case.items():
|
|
if key not in ['special_equipment', 'blood_products', 'implants']:
|
|
print(f"{key}: {value}") |