neversight/learn-skills.dev/clinical-trial-logistics

from enum import Enum
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import List, Optional, Dict
import random
class TreatmentArm(Enum):
    INVESTIGATIONAL = "investigational"
    COMPARATOR = "comparator"
    PLACEBO = "placebo"
class PatientStatus(Enum):
    SCREENED = "screened"
    RANDOMIZED = "randomized"
    ON_TREATMENT = "on_treatment"
    COMPLETED = "completed"
    DISCONTINUED = "discontinued"
@dataclass
class StratificationFactor:
    """Stratification criteria for randomization"""
    factor_name: str
    value: str
@dataclass
class Patient:
    """Clinical trial patient"""
    patient_id: str
    site_id: str
    screening_date: datetime
    status: PatientStatus
    stratification_factors: List[StratificationFactor] = None
    treatment_arm: Optional[TreatmentArm] = None
    randomization_date: Optional[datetime] = None
    allocated_kits: List[str] = None
@dataclass
class DrugKit:
    """IMP drug kit"""
    kit_number: str
    treatment_arm: TreatmentArm
    lot_number: str
    expiry_date: datetime
    site_id: str
    status: str  # available, allocated, dispensed, returned, destroyed
    patient_id: Optional[str] = None
    dispensed_date: Optional[datetime] = None
class IRTSystem:
    """
    Interactive Response Technology for clinical trials
    """
    def __init__(self, trial_id, randomization_ratio, blinded=True):
        self.trial_id = trial_id
        self.randomization_ratio = randomization_ratio  # e.g., {'investigational': 2, 'comparator': 1}
        self.blinded = blinded
        self.patients = {}
        self.drug_kits = {}
        self.randomization_list = []
        self.site_inventory = {}
    def generate_randomization_list(self, total_patients, block_size=6,
                                    stratification_factors=None):
        """
        Generate randomization list with blocking and stratification
        Parameters:
        - total_patients: Total randomization codes to generate
        - block_size: Block size for randomization
        - stratification_factors: List of stratification combinations
        """
        if stratification_factors is None:
            stratification_factors = [None]  # No stratification
        randomization_list = []
        randomization_number = 1
        for strata in stratification_factors:
            num_patients_per_strata = total_patients // len(stratification_factors)
            # Generate treatment sequence based on ratio
            sequence = []
            for treatment, count in self.randomization_ratio.items():
                sequence.extend([TreatmentArm[treatment.upper()]] * count)
            # Generate blocks
            num_blocks = (num_patients_per_strata // block_size) + 1
            for block in range(num_blocks):
                # Shuffle within block
                random.shuffle(sequence)
                for treatment in sequence:
                    if len(randomization_list) >= total_patients:
                        break
                    randomization_list.append({
                        'randomization_number': f"RND-{randomization_number:05d}",
                        'treatment_arm': treatment,
                        'stratification': strata,
                        'block': block + 1
                    })
                    randomization_number += 1
        self.randomization_list = randomization_list[:total_patients]
        return self.randomization_list
    def randomize_patient(self, patient_id, site_id, stratification_values=None):
        """
        Randomize patient and allocate treatment
        Parameters:
        - patient_id: Patient identifier
        - site_id: Study site
        - stratification_values: Dict of stratification factor values
        """
        if patient_id in self.patients:
            raise ValueError(f"Patient {patient_id} already randomized")
        # Find next available randomization code for stratification
        # In real system, this would be from pre-generated randomization list
        available_codes = [
            code for code in self.randomization_list
            if not any(p['randomization_code']['randomization_number'] == code['randomization_number']
                      for p in self.patients.values() if 'randomization_code' in p)
        ]
        if not available_codes:
            raise ValueError("No randomization codes available")
        # Assign next code
        randomization_code = available_codes[0]
        # Create patient record
        patient = {
            'patient_id': patient_id,
            'site_id': site_id,
            'randomization_date': datetime.now(),
            'status': PatientStatus.RANDOMIZED,
            'randomization_code': randomization_code,
            'treatment_arm': randomization_code['treatment_arm'],
            'stratification_values': stratification_values,
            'allocated_kits': []
        }
        self.patients[patient_id] = patient
        # Allocate drug kit
        kit = self._allocate_kit(patient_id, site_id, randomization_code['treatment_arm'])
        if kit:
            patient['allocated_kits'].append(kit['kit_number'])
        return {
            'patient_id': patient_id,
            'randomization_number': randomization_code['randomization_number'],
            'kit_number': kit['kit_number'] if kit else None,
            'dispensing_instructions': self._get_dispensing_instructions()
        }
    def _allocate_kit(self, patient_id, site_id, treatment_arm):
        """
        Allocate drug kit to patient from site inventory
        """
        # Find available kits at site for treatment arm
        site_kits = [
            kit for kit_num, kit in self.drug_kits.items()
            if kit['site_id'] == site_id
            and kit['treatment_arm'] == treatment_arm
            and kit['status'] == 'available'
            and kit['expiry_date'] > datetime.now()
        ]
        if not site_kits:
            # Trigger resupply
            self._trigger_resupply(site_id, treatment_arm)
            return None
        # Allocate kit with earliest expiry (FEFO)
        site_kits.sort(key=lambda x: x['expiry_date'])
        kit = site_kits[0]
        kit['status'] = 'allocated'
        kit['patient_id'] = patient_id
        kit['allocated_date'] = datetime.now()
        return kit
    def dispense_kit(self, kit_number, patient_id, dispensed_by):
        """
        Record kit dispensing to patient
        """
        if kit_number not in self.drug_kits:
            raise ValueError(f"Kit {kit_number} not found")
        kit = self.drug_kits[kit_number]
        if kit['status'] != 'allocated':
            raise ValueError(f"Kit {kit_number} is not allocated (status: {kit['status']})")
        if kit['patient_id'] != patient_id:
            raise ValueError(f"Kit {kit_number} is allocated to different patient")
        kit['status'] = 'dispensed'
        kit['dispensed_date'] = datetime.now()
        kit['dispensed_by'] = dispensed_by
        # Update patient status
        if patient_id in self.patients:
            self.patients[patient_id]['status'] = PatientStatus.ON_TREATMENT
        return {
            'kit_number': kit_number,
            'patient_id': patient_id,
            'dispensed_date': kit['dispensed_date'],
            'accountability_required': True
        }
    def return_kit(self, kit_number, return_reason, returned_by):
        """
        Record kit return (unused or partially used)
        """
        if kit_number not in self.drug_kits:
            raise ValueError(f"Kit {kit_number} not found")
        kit = self.drug_kits[kit_number]
        kit['status'] = 'returned'
        kit['return_date'] = datetime.now()
        kit['return_reason'] = return_reason
        kit['returned_by'] = returned_by
        return kit
    def check_site_inventory(self, site_id):
        """
        Check site inventory levels and trigger resupply if needed
        """
        site_kits = [
            kit for kit in self.drug_kits.values()
            if kit['site_id'] == site_id and kit['status'] == 'available'
        ]
        # Group by treatment arm
        inventory_by_arm = {}
        for arm in TreatmentArm:
            arm_kits = [k for k in site_kits if k['treatment_arm'] == arm]
            inventory_by_arm[arm.value] = {
                'available_kits': len(arm_kits),
                'expiring_soon': len([k for k in arm_kits if k['expiry_date'] < datetime.now() + timedelta(days=90)])
            }
        return inventory_by_arm
    def _trigger_resupply(self, site_id, treatment_arm):
        """
        Trigger site resupply when inventory low
        """
        resupply_request = {
            'site_id': site_id,
            'treatment_arm': treatment_arm,
            'request_date': datetime.now(),
            'priority': 'high',
            'requested_quantity': 20  # Standard resupply quantity
        }
        # In real system, this would integrate with depot management
        print(f"RESUPPLY TRIGGERED: Site {site_id} needs {treatment_arm.value} kits")
        return resupply_request
    def _get_dispensing_instructions(self):
        """
        Generate dispensing instructions for site staff
        """
        if self.blinded:
            return "Dispense assigned kit to patient. DO NOT OPEN OR INSPECT CONTENTS."
        else:
            return "Dispense assigned kit to patient. Verify drug name and strength."
# Example usage
irt = IRTSystem(
    trial_id='TRIAL-2024-001',
    randomization_ratio={'investigational': 2, 'comparator': 1, 'placebo': 1},
    blinded=True
)