UDI = Device Identifier (DI) + Production Identifier (PI)

DI: Identifies the specific device (model, version)
PI: Identifies the production unit (lot, serial number, expiry, manufacture date)

import re
from dataclasses import dataclass
from datetime import datetime
from typing import Optional

@dataclass
class UDI:
    """
    Unique Device Identification structure
    """
    device_identifier: str  # DI - identifies device model
    lot_number: Optional[str] = None
    serial_number: Optional[str] = None
    manufacturing_date: Optional[datetime] = None
    expiration_date: Optional[datetime] = None
    donation_id: Optional[str] = None  # For blood/tissue

    def to_gs1_string(self):
        """Convert to GS1 format"""
        udi_string = f"(01){self.device_identifier}"

        if self.lot_number:
            udi_string += f"(10){self.lot_number}"

        if self.serial_number:
            udi_string += f"(21){self.serial_number}"

        if self.expiration_date:
            exp_date = self.expiration_date.strftime("%y%m%d")
            udi_string += f"(17){exp_date}"

        if self.manufacturing_date:
            mfg_date = self.manufacturing_date.strftime("%y%m%d")
            udi_string += f"(11){mfg_date}"

        return udi_string

    @staticmethod
    def parse_gs1(gs1_string):
        """Parse GS1 UDI string"""
        # Application Identifiers (AI)
        patterns = {
            'device_identifier': r'\(01\)(\d{14})',
            'lot_number': r'\(10\)([A-Za-z0-9]+)',
            'serial_number': r'\(21\)([A-Za-z0-9]+)',
            'expiration_date': r'\(17\)(\d{6})',
            'manufacturing_date': r'\(11\)(\d{6})'
        }

        udi_data = {}

        for field, pattern in patterns.items():
            match = re.search(pattern, gs1_string)
            if match:
                value = match.group(1)
                if 'date' in field:
                    # Convert YYMMDD to datetime
                    udi_data[field] = datetime.strptime(value, "%y%m%d")
                else:
                    udi_data[field] = value

        return UDI(**udi_data)

---

import pandas as pd
from datetime import datetime
from enum import Enum

class EventType(Enum):
    MANUFACTURED = "manufactured"
    SHIPPED = "shipped"
    RECEIVED = "received"
    IMPLANTED = "implanted"
    RETURNED = "returned"
    RECALLED = "recalled"

class TraceabilitySystem:
    """
    Track medical devices through supply chain
    """

    def __init__(self):
        self.events = []

    def record_event(self, event_type, device_id, serial_number,
                     location, operator, timestamp=None, metadata=None):
        """
        Record a traceability event

        Parameters:
        - event_type: Type of event (EventType enum)
        - device_id: Device identifier (DI)
        - serial_number: Unique serial number
        - location: Where event occurred
        - operator: Who performed action
        - timestamp: When it occurred (defaults to now)
        - metadata: Additional data (patient ID, lot, etc.)
        """

        if timestamp is None:
            timestamp = datetime.now()

        event = {
            'event_type': event_type.value,
            'device_id': device_id,
            'serial_number': serial_number,
            'location': location,
            'operator': operator,
            'timestamp': timestamp,
            'metadata': metadata or {}
        }

        self.events.append(event)

        return event

    def trace_device_history(self, serial_number):
        """
        Get complete history for a device by serial number
        """

        device_events = [
            e for e in self.events
            if e['serial_number'] == serial_number
        ]

        # Sort by timestamp
        device_events.sort(key=lambda x: x['timestamp'])

        return pd.DataFrame(device_events)

    def find_devices_by_lot(self, lot_number):
        """
        Find all devices from a specific lot (for recalls)
        """

        devices = [
            e for e in self.events
            if e.get('metadata', {}).get('lot_number') == lot_number
        ]

        # Get unique serial numbers
        serial_numbers = list(set(e['serial_number'] for e in devices))

        return serial_numbers

    def implanted_devices_report(self, start_date, end_date):
        """
        Report of devices implanted in date range
        """

        implanted = [
            e for e in self.events
            if e['event_type'] == EventType.IMPLANTED.value
            and start_date <= e['timestamp'] <= end_date
        ]

        return pd.DataFrame(implanted)

    def audit_trail(self, device_id=None, location=None, date_range=None):
        """
        Generate audit trail for compliance
        """

        filtered_events = self.events

        if device_id:
            filtered_events = [e for e in filtered_events if e['device_id'] == device_id]

        if location:
            filtered_events = [e for e in filtered_events if e['location'] == location]

        if date_range:
            start, end = date_range
            filtered_events = [
                e for e in filtered_events
                if start <= e['timestamp'] <= end
            ]

        return pd.DataFrame(filtered_events)

---

import numpy as np
import pandas as pd

class ConsignmentInventoryManager:
    """
    Manage consignment inventory for medical devices
    """

    def __init__(self, location_name, par_level_policy=None):
        self.location_name = location_name
        self.inventory = {}  # {serial_number: device_info}
        self.par_levels = par_level_policy or {}
        self.usage_history = []

    def add_device(self, device_id, serial_number, cost, vendor, metadata=None):
        """Add device to consignment inventory"""

        self.inventory[serial_number] = {
            'device_id': device_id,
            'serial_number': serial_number,
            'cost': cost,
            'vendor': vendor,
            'status': 'available',
            'received_date': datetime.now(),
            'metadata': metadata or {}
        }

    def consume_device(self, serial_number, patient_id, procedure_info):
        """
        Record device consumption (trigger payment to vendor)
        """

        if serial_number not in self.inventory:
            raise ValueError(f"Device {serial_number} not in consignment inventory")

        device = self.inventory[serial_number]

        if device['status'] != 'available':
            raise ValueError(f"Device {serial_number} is not available (status: {device['status']})")

        # Record usage
        usage_record = {
            'serial_number': serial_number,
            'device_id': device['device_id'],
            'vendor': device['vendor'],
            'cost': device['cost'],
            'consumption_date': datetime.now(),
            'patient_id': patient_id,
            'procedure_info': procedure_info,
            'days_in_consignment': (datetime.now() - device['received_date']).days
        }

        self.usage_history.append(usage_record)

        # Update device status
        device['status'] = 'consumed'
        device['consumption_date'] = datetime.now()

        return usage_record

    def check_par_levels(self):
        """
        Check if consignment inventory meets PAR levels
        Trigger vendor replenishment if needed
        """

        replenishment_needed = []

        for device_id, par_level in self.par_levels.items():
            # Count available devices
            available_count = sum(
                1 for device in self.inventory.values()
                if device['device_id'] == device_id and device['status'] == 'available'
            )

            if available_count < par_level:
                shortage = par_level - available_count

                replenishment_needed.append({
                    'device_id': device_id,
                    'current_qty': available_count,
                    'par_level': par_level,
                    'replenishment_qty': shortage
                })

        return replenishment_needed

    def vendor_reconciliation_report(self, vendor, month):
        """
        Generate monthly reconciliation report for vendor billing
        """

        usage_df = pd.DataFrame(self.usage_history)

        if len(usage_df) == 0:
            return pd.DataFrame()

        # Filter by vendor and month
        vendor_usage = usage_df[
            (usage_df['vendor'] == vendor) &
            (usage_df['consumption_date'].dt.to_period('M') == month)
        ]

        # Aggregate
        summary = vendor_usage.groupby('device_id').agg({
            'serial_number': 'count',
            'cost': 'sum'
        }).rename(columns={'serial_number': 'quantity', 'cost': 'total_cost'})

        summary['average_cost'] = summary['total_cost'] / summary['quantity']

        return summary

    def aging_report(self):
        """
        Report devices sitting in consignment for extended periods
        (potential obsolescence risk)
        """

        aging_data = []

        for serial, device in self.inventory.items():
            if device['status'] == 'available':
                days_on_hand = (datetime.now() - device['received_date']).days

                aging_data.append({
                    'serial_number': serial,
                    'device_id': device['device_id'],
                    'vendor': device['vendor'],
                    'cost': device['cost'],
                    'days_on_hand': days_on_hand,
                    'risk_level': self._aging_risk(days_on_hand)
                })

        aging_df = pd.DataFrame(aging_data)

        if len(aging_df) > 0:
            aging_df = aging_df.sort_values('days_on_hand', ascending=False)

        return aging_df

    @staticmethod
    def _aging_risk(days):
        """Categorize aging risk"""
        if days > 365:
            return 'Critical'
        elif days > 180:
            return 'High'
        elif days > 90:
            return 'Medium'
        else:
            return 'Low'

---

from datetime import datetime, timedelta

class LoanerSetManager:
    """
    Track loaner instrument sets for surgeries
    """

    def __init__(self):
        self.loaner_sets = {}
        self.bookings = []
        self.movements = []

    def register_loaner_set(self, set_id, vendor, description, value, contents):
        """
        Register a loaner set in the system
        """

        self.loaner_sets[set_id] = {
            'set_id': set_id,
            'vendor': vendor,
            'description': description,
            'value': value,
            'contents': contents,
            'status': 'available',
            'location': 'Vendor',
            'registered_date': datetime.now()
        }

    def book_loaner_set(self, set_id, procedure_date, surgeon, procedure_type, patient_id):
        """
        Book loaner set for upcoming surgery
        """

        if set_id not in self.loaner_sets:
            raise ValueError(f"Loaner set {set_id} not registered")

        booking = {
            'booking_id': f"BOOK-{len(self.bookings)+1}",
            'set_id': set_id,
            'procedure_date': procedure_date,
            'surgeon': surgeon,
            'procedure_type': procedure_type,
            'patient_id': patient_id,
            'booking_date': datetime.now(),
            'status': 'booked'
        }

        self.bookings.append(booking)

        # Update set status
        self.loaner_sets[set_id]['status'] = 'booked'

        return booking

    def receive_loaner_set(self, set_id, received_by, condition='good'):
        """
        Record receipt of loaner set at hospital
        """

        if set_id not in self.loaner_sets:
            raise ValueError(f"Loaner set {set_id} not registered")

        movement = {
            'set_id': set_id,
            'event_type': 'received',
            'timestamp': datetime.now(),
            'location': 'Hospital Central Sterile',
            'operator': received_by,
            'condition': condition
        }

        self.movements.append(movement)

        # Update set status and location
        self.loaner_sets[set_id]['status'] = 'in_house'
        self.loaner_sets[set_id]['location'] = 'Hospital Central Sterile'
        self.loaner_sets[set_id]['received_date'] = datetime.now()

    def send_to_sterile_processing(self, set_id, processor):
        """
        Send loaner set for sterilization
        """

        movement = {
            'set_id': set_id,
            'event_type': 'sent_to_sterilization',
            'timestamp': datetime.now(),
            'location': 'Sterile Processing',
            'operator': processor
        }

        self.movements.append(movement)

        self.loaner_sets[set_id]['location'] = 'Sterile Processing'
        self.loaner_sets[set_id]['status'] = 'processing'

    def ready_for_surgery(self, set_id, sterilization_lot):
        """
        Mark loaner set as sterile and ready
        """

        movement = {
            'set_id': set_id,
            'event_type': 'sterilization_complete',
            'timestamp': datetime.now(),
            'location': 'Sterile Storage',
            'sterilization_lot': sterilization_lot
        }

        self.movements.append(movement)

        self.loaner_sets[set_id]['location'] = 'Sterile Storage'
        self.loaner_sets[set_id]['status'] = 'ready'

    def use_in_surgery(self, set_id, procedure_info):
        """
        Record use in surgery
        """

        movement = {
            'set_id': set_id,
            'event_type': 'used_in_surgery',
            'timestamp': datetime.now(),
            'location': f"OR-{procedure_info.get('or_room', 'Unknown')}",
            'procedure_info': procedure_info
        }

        self.movements.append(movement)

        self.loaner_sets[set_id]['status'] = 'used'
        self.loaner_sets[set_id]['last_used_date'] = datetime.now()

    def return_to_vendor(self, set_id, carrier, tracking_number):
        """
        Ship loaner set back to vendor
        """

        movement = {
            'set_id': set_id,
            'event_type': 'returned_to_vendor',
            'timestamp': datetime.now(),
            'location': 'In Transit to Vendor',
            'carrier': carrier,
            'tracking_number': tracking_number
        }

        self.movements.append(movement)

        self.loaner_sets[set_id]['status'] = 'returned'
        self.loaner_sets[set_id]['location'] = 'In Transit to Vendor'
        self.loaner_sets[set_id]['return_date'] = datetime.now()

    def overdue_loaners_report(self, days_threshold=14):
        """
        Identify loaner sets held beyond expected return date
        """

        overdue = []

        for set_id, loaner in self.loaner_sets.items():
            if loaner['status'] in ['available', 'returned']:
                continue

            if 'received_date' in loaner:
                days_on_hand = (datetime.now() - loaner['received_date']).days

                if days_on_hand > days_threshold:
                    overdue.append({
                        'set_id': set_id,
                        'vendor': loaner['vendor'],
                        'description': loaner['description'],
                        'value': loaner['value'],
                        'days_on_hand': days_on_hand,
                        'status': loaner['status'],
                        'location': loaner['location']
                    })

        return pd.DataFrame(overdue)

    def loaner_utilization_report(self):
        """
        Analyze loaner set utilization and costs
        """

        utilization_data = []

        for set_id, loaner in self.loaner_sets.items():
            # Count uses
            uses = [m for m in self.movements if m['set_id'] == set_id and m['event_type'] == 'used_in_surgery']

            # Calculate average time on-site
            received_events = [m for m in self.movements if m['set_id'] == set_id and m['event_type'] == 'received']
            returned_events = [m for m in self.movements if m['set_id'] == set_id and m['event_type'] == 'returned_to_vendor']

            cycles = min(len(received_events), len(returned_events))
            if cycles > 0:
                avg_days_on_site = sum([
                    (returned_events[i]['timestamp'] - received_events[i]['timestamp']).days
                    for i in range(cycles)
                ]) / cycles
            else:
                avg_days_on_site = 0

            utilization_data.append({
                'set_id': set_id,
                'vendor': loaner['vendor'],
                'description': loaner['description'],
                'value': loaner['value'],
                'times_used': len(uses),
                'avg_days_on_site': round(avg_days_on_site, 1),
                'cycles': cycles
            })

        return pd.DataFrame(utilization_data)

---

class RecallManager:
    """
    Manage medical device recalls
    """

    def __init__(self, traceability_system):
        self.traceability_system = traceability_system
        self.recalls = []

    def initiate_recall(self, recall_id, device_id, lot_numbers, reason,
                        recall_class, recall_date, corrective_action):
        """
        Initiate a device recall
        """

        recall = {
            'recall_id': recall_id,
            'device_id': device_id,
            'lot_numbers': lot_numbers,
            'reason': reason,
            'recall_class': recall_class,  # I, II, or III
            'recall_date': recall_date,
            'corrective_action': corrective_action,
            'status': 'active'
        }

        self.recalls.append(recall)

        return recall

    def identify_affected_devices(self, recall_id):
        """
        Identify all devices affected by recall using traceability
        """

        recall = next((r for r in self.recalls if r['recall_id'] == recall_id), None)

        if not recall:
            raise ValueError(f"Recall {recall_id} not found")

        affected_devices = []

        for lot_number in recall['lot_numbers']:
            # Find devices from affected lots
            serial_numbers = self.traceability_system.find_devices_by_lot(lot_number)

            for serial in serial_numbers:
                device_history = self.traceability_system.trace_device_history(serial)

                # Get current location/status
                latest_event = device_history.iloc[-1] if len(device_history) > 0 else None

                affected_devices.append({
                    'serial_number': serial,
                    'lot_number': lot_number,
                    'current_location': latest_event['location'] if latest_event is not None else 'Unknown',
                    'status': latest_event['event_type'] if latest_event is not None else 'Unknown',
                    'last_event_date': latest_event['timestamp'] if latest_event is not None else None
                })

        return pd.DataFrame(affected_devices)

    def identify_implanted_devices(self, recall_id):
        """
        Identify devices from recall that have been implanted (most critical)
        """

        affected = self.identify_affected_devices(recall_id)

        # Filter to implanted devices
        implanted = affected[affected['status'] == 'implanted']

        return implanted

    def customer_notification_list(self, recall_id):
        """
        Generate list of customers to notify about recall
        """

        affected = self.identify_affected_devices(recall_id)

        # Group by current location (customer)
        customers = affected.groupby('current_location').agg({
            'serial_number': 'count',
            'status': lambda x: ', '.join(x.unique())
        }).rename(columns={'serial_number': 'affected_devices'})

        return customers

    def recall_effectiveness_check(self, recall_id):
        """
        Track recall effectiveness (how many devices recovered)
        """

        affected = self.identify_affected_devices(recall_id)
        total_affected = len(affected)

        # Devices that have been returned/recovered
        recovered = affected[affected['status'] == 'returned']
        recovered_count = len(recovered)

        # Devices still implanted
        implanted = affected[affected['status'] == 'implanted']
        implanted_count = len(implanted)

        # Devices unaccounted for
        unaccounted = affected[~affected['status'].isin(['returned', 'implanted'])]
        unaccounted_count = len(unaccounted)

        effectiveness = {
            'recall_id': recall_id,
            'total_affected': total_affected,
            'recovered': recovered_count,
            'recovery_rate': (recovered_count / total_affected * 100) if total_affected > 0 else 0,
            'still_implanted': implanted_count,
            'unaccounted': unaccounted_count
        }

        return effectiveness