The Public Health AI Handbook: Evaluating AI Tools for Public Health Practice

Tegomoh, Bryan

doi:10.5281/zenodo.18263442

System Integration and Regulatory Compliance

Integration architecture, interoperability, security, and regulatory controls for public health AI deployment. The material is maintained separately so each operational question has a stable, focused reference.

Learning Objectives

Identify the evidence and controls relevant to this decision area
Distinguish technical performance from operational and population impact
Apply the included framework without extending claims beyond the cited evidence

Chapter Summary (TL;DR)

Use explicit targets, populations, thresholds, and decision consequences. Require external evidence and local monitoring where deployment can affect people or programs. Preserve uncertainty and document limits.

Introduction

This focused reference is part of the broader System Integration and Regulatory Compliance overview. It preserves the detailed methods, examples, and exercises while reducing page size and improving direct navigation.

System Integration

EHR Integration Patterns

Healthcare systems require integration with Electronic Health Records to be clinically useful.

Mandel et al., 2016, Journal of the American Medical Informatics Association - “SMART on FHIR: A standards-based, interoperable apps platform for electronic health records”

FHIR API Integration

FHIR (Fast Healthcare Interoperability Resources) is the modern standard for health data exchange.

HL7 FHIR Specification

from fhirclient import client
from fhirclient.models.observation import Observation
from fhirclient.models.patient import Patient
from fhirclient.models.condition import Condition
import requests
from typing import Dict, List, Optional
from datetime import datetime, timedelta

class FHIRIntegration:
 """
 Integrate ML model with FHIR-compatible EHR systems

 Supports:
 - Reading patient data from FHIR server
 - Extracting features for prediction
 - Writing predictions back as FHIR Observations
 """

 def __init__(self, fhir_base_url: str, auth_token: str):
  """
  Initialize FHIR client

  Args:
   fhir_base_url: Base URL of FHIR server (e.g., https://fhir.hospital.org/api)
   auth_token: OAuth2 access token
  """
  self.settings = {
   'app_id': 'sepsis_predictor_app',
   'api_base': fhir_base_url
  }
  self.client = client.FHIRClient(settings=self.settings)
  self.auth_token = auth_token
  self.headers = {
   'Authorization': f'Bearer {auth_token}',
   'Accept': 'application/fhir+json',
   'Content-Type': 'application/fhir+json'
  }

 def get_patient_data(self, patient_id: str, lookback_hours: int = 24) -> Dict:
  """
  Fetch patient data from FHIR server and extract features

  Args:
   patient_id: FHIR patient ID
   lookback_hours: How far back to look for observations

  Returns:
   Dict with extracted features for ML model
  """
  try:
   # Fetch patient demographics
   patient = Patient.read(patient_id, self.client.server)

   # Calculate age
   from dateutil.parser import parse
   birth_date = parse(patient.birthDate.isoString)
   age = (datetime.now() - birth_date).days // 365

   # Fetch recent observations
   lookback_date = datetime.now() - timedelta(hours=lookback_hours)

   observations = self._fetch_observations(
    patient_id,
    lookback_date.isoformat()
   )

   # Extract features from observations
   features = self._extract_features_from_observations(observations)
   features['age'] = age
   features['patient_id'] = patient_id

   return features

  except Exception as e:
   raise Exception(f"Failed to fetch patient data: {e}")

 def _fetch_observations(self, patient_id: str, date_gte: str) -> List:
  """Fetch observations for patient"""
  url = f"{self.settings['api_base']}/Observation"
  params = {
   'patient': patient_id,
   'date': f'gt{date_gte}',
   '_sort': '-date',
   '_count': 100
  }

  response = requests.get(url, params=params, headers=self.headers)
  response.raise_for_status()

  bundle = response.json()

  if bundle.get('entry'):
   return [entry['resource'] for entry in bundle['entry']]
  return []

 def _extract_features_from_observations(self, observations: List) -> Dict:
  """
  Extract ML features from FHIR Observations

  Maps LOINC codes to feature names
  """
  features = {}

  # LOINC code mapping to feature names
  loinc_mapping = {
   '8867-4': 'heart_rate',   # Heart rate
   '9279-1': 'respiratory_rate',  # Respiratory rate
   '8310-5': 'temperature',   # Body temperature
   '8480-6': 'systolic_bp',   # Systolic BP
   '8462-4': 'diastolic_bp',   # Diastolic BP
   '6690-2': 'white_blood_cell',  # WBC count
   '2524-7': 'lactate',    # Lactate
   '2345-7': 'glucose',    # Glucose
   '20570-8': 'creatinine'   # Creatinine
  }

  # Extract most recent value for each LOINC code
  for obs in observations:
   if not obs.get('code') or not obs['code'].get('coding'):
    continue

   loinc_code = obs['code']['coding'][0].get('code')

   if loinc_code in loinc_mapping:
    feature_name = loinc_mapping[loinc_code]

    # Only take most recent value if not already set
    if feature_name not in features:
     if obs.get('valueQuantity'):
      features[feature_name] = obs['valueQuantity']['value']
     elif obs.get('valueCodeableConcept'):
      # Handle coded values if needed
      pass

  return features

 def write_prediction(self, patient_id: str, prediction: Dict) -> str:
  """
  Write prediction back to EHR as FHIR Observation

  Args:
   patient_id: FHIR patient ID
   prediction: Dict with prediction results

  Returns:
   ID of created Observation resource
  """
  # Create FHIR Observation resource
  observation = {
   'resourceType': 'Observation',
   'status': 'final',
   'category': [{
    'coding': [{
     'system': 'http://terminology.hl7.org/CodeSystem/observation-category',
     'code': 'survey',
     'display': 'Survey'
    }]
   }],
   'code': {
    'coding': [{
     'system': 'http://loinc.org',
     'code': '82810-3', # LOINC code for sepsis prediction (example)
     'display': 'Sepsis Risk Score'
    }],
    'text': 'AI-predicted sepsis risk score'
   },
   'subject': {
    'reference': f'Patient/{patient_id}'
   },
   'effectiveDateTime': datetime.utcnow().isoformat() + 'Z',
   'issued': datetime.utcnow().isoformat() + 'Z',
   'valueQuantity': {
    'value': prediction['sepsis_risk'],
    'unit': 'probability',
    'system': 'http://unitsofmeasure.org',
    'code': '1'
   },
   'interpretation': [{
    'coding': [{
     'system': 'http://terminology.hl7.org/CodeSystem/v3-ObservationInterpretation',
     'code': 'H' if prediction['sepsis_risk'] > 0.7 else 'N' if prediction['sepsis_risk'] < 0.3 else 'I',
     'display': 'High' if prediction['sepsis_risk'] > 0.7 else 'Normal' if prediction['sepsis_risk'] < 0.3 else 'Intermediate'
    }]
   }],
   'note': [{
    'text': f"AI model prediction (version {prediction.get('model_version', '1.0')}). "
      f"Risk category: {prediction['risk_category']}. "
      f"Confidence: {prediction.get('confidence', 0):.2f}."
   }],
   'device': {
    'display': 'Sepsis Prediction ML Model v1.2.0'
   }
  }

  # POST to FHIR server
  url = f"{self.settings['api_base']}/Observation"
  response = requests.post(url, json=observation, headers=self.headers)

  if response.status_code in [200, 201]:
   created_observation = response.json()
   observation_id = created_observation['id']
   print(f"[OK] Prediction written to EHR: Observation/{observation_id}")
   return observation_id
  else:
   raise Exception(f"Failed to write prediction: {response.status_code} - {response.text}")

# Example usage
fhir = FHIRIntegration(
 fhir_base_url='https://fhir.hospital.org/api',
 auth_token='your_oauth2_token'
)

# Fetch patient data
patient_data = fhir.get_patient_data('patient-12345', lookback_hours=24)

# Make prediction (assuming model loaded)
prediction = model.predict(patient_data)

# Write prediction back to EHR
observation_id = fhir.write_prediction('patient-12345', prediction)

HL7 v2 Integration

HL7 v2 remains widely used despite being older than FHIR.

HL7 Version 2 Product Suite

from hl7apy.parser import parse_message
from hl7apy.core import Message, Segment
from datetime import datetime

class HL7Integration:
 """
 Integrate ML model with HL7 v2 messaging systems

 Common message types:
 - ADT (Admission/Discharge/Transfer)
 - ORU (Observation Results)
 - ORM (Orders)
 """

 def parse_adt_message(self, hl7_message: str) -> Dict:
  """
  Parse HL7 ADT message and extract patient data

  Example ADT^A01 (Patient Admission):
  MSH|^~\&|HIS|HOSPITAL|AI|PREDICTOR|20240315120000||ADT^A01|MSG001|P|2.5
  EVN|A01|20240315120000
  PID|1||MRN12345||DOE^JOHN^A||19700101|M|||123 MAIN ST^^CITY^ST^12345
  PV1|1|I|ICU^101^A||||^SMITH^JANE^MD^^^DR|||||||||||V123456
  OBX|1|NM|8867-4^Heart Rate^LN||110|/min|||||F
  OBX|2|NM|9279-1^Respiratory Rate^LN||24|/min|||||F
  OBX|3|NM|8310-5^Body Temperature^LN||38.5|Cel|||||F
  """
  try:
   msg = parse_message(hl7_message)

   features = {}

   # Extract patient demographics from PID segment
   if hasattr(msg, 'pid'):
    pid = msg.pid

    # Birth date
    if hasattr(pid, 'date_time_of_birth') and pid.date_time_of_birth.value:
     birth_date = pid.date_time_of_birth.value
     features['age'] = self._calculate_age(birth_date)

    # Patient ID
    if hasattr(pid, 'patient_identifier_list'):
     features['patient_id'] = pid.patient_identifier_list.id_number.value

    # Gender
    if hasattr(pid, 'administrative_sex') and pid.administrative_sex.value:
     features['gender'] = pid.administrative_sex.value

   # Extract observations from OBX segments
   if hasattr(msg, 'obx'):
    obx_segments = msg.obx if isinstance(msg.obx, list) else [msg.obx]

    for obx in obx_segments:
     # Get LOINC code
     if hasattr(obx, 'observation_identifier'):
      loinc_code = obx.observation_identifier.identifier.value

      # Get value
      if hasattr(obx, 'observation_value') and obx.observation_value.value:
       value = float(obx.observation_value.value)

       # Map LOINC to feature
       if loinc_code == '8867-4':
        features['heart_rate'] = value
       elif loinc_code == '9279-1':
        features['respiratory_rate'] = value
       elif loinc_code == '8310-5':
        features['temperature'] = value
       elif loinc_code == '8480-6':
        features['systolic_bp'] = value
       elif loinc_code == '6690-2':
        features['white_blood_cell'] = value
       elif loinc_code == '2524-7':
        features['lactate'] = value

   return features

  except Exception as e:
   raise Exception(f"Failed to parse HL7 message: {e}")

 def create_oru_message(self, patient_id: str, prediction: Dict) -> str:
  """
  Create HL7 ORU (Observation Result) message with prediction

  Returns:
   HL7 v2.5 message string
  """
  # Create message
  msg = Message("ORU_R01", version="2.5")

  # MSH segment (Message Header)
  msg.msh.msh_3 = "AI_PREDICTOR"
  msg.msh.msh_4 = "HOSPITAL"
  msg.msh.msh_5 = "HIS"
  msg.msh.msh_6 = "HOSPITAL"
  msg.msh.msh_7 = datetime.now().strftime("%Y%m%d%H%M%S")
  msg.msh.msh_9 = "ORU^R01^ORU_R01"
  msg.msh.msh_10 = f"MSG{int(datetime.now().timestamp())}"
  msg.msh.msh_11 = "P" # Production
  msg.msh.msh_12 = "2.5"

  # PID segment (Patient Identification)
  msg.pid.pid_1 = "1"
  msg.pid.pid_3 = patient_id

  # OBR segment (Observation Request)
  msg.oru_r01_patient_result.oru_r01_order_observation.obr.obr_1 = "1"
  msg.oru_r01_patient_result.oru_r01_order_observation.obr.obr_4 = "SEPSIS_RISK^Sepsis Risk Prediction^L"
  msg.oru_r01_patient_result.oru_r01_order_observation.obr.obr_7 = datetime.now().strftime("%Y%m%d%H%M%S")
  msg.oru_r01_patient_result.oru_r01_order_observation.obr.obr_25 = "F" # Final result

  # OBX segment (Observation Result)
  obx = msg.oru_r01_patient_result.oru_r01_order_observation.oru_r01_observation.obx
  obx.obx_1 = "1"
  obx.obx_2 = "NM" # Numeric
  obx.obx_3 = "82810-3^Sepsis Risk Score^LN"
  obx.obx_5 = str(prediction['sepsis_risk'])
  obx.obx_6 = "probability^probability^UCUM"
  obx.obx_8 = "H" if prediction['sepsis_risk'] > 0.7 else "L" # Abnormal flags
  obx.obx_11 = "F" # Final result
  obx.obx_14 = datetime.now().strftime("%Y%m%d%H%M%S")

  # NTE segment (Notes and Comments)
  nte = msg.oru_r01_patient_result.oru_r01_order_observation.oru_r01_observation.nte
  nte.nte_1 = "1"
  nte.nte_3 = f"AI model prediction. Risk category: {prediction['risk_category']}. Model version: {prediction.get('model_version', '1.0')}"

  # Convert to ER7 (pipe-delimited) format
  return msg.to_er7()

 def _calculate_age(self, birth_date_str: str) -> int:
  """Calculate age from HL7 date format (YYYYMMDD)"""
  try:
   birth_date = datetime.strptime(birth_date_str[:8], "%Y%m%d")
   return (datetime.now() - birth_date).days // 365
  except:
   return None

# Example usage
hl7 = HL7Integration()

# Incoming HL7 message from EHR
incoming_msg = """MSH|^~\&|HIS|HOSPITAL|AI|PREDICTOR|20240315120000||ADT^A01|MSG001|P|2.5
PID|1||MRN12345||DOE^JOHN^A||19700101|M
OBX|1|NM|8867-4^Heart Rate^LN||110|/min|||||F
OBX|2|NM|9279-1^Respiratory Rate^LN||24|/min|||||F
OBX|3|NM|8310-5^Body Temperature^LN||38.5|Cel|||||F
OBX|4|NM|8480-6^Systolic BP^LN||95|mm[Hg]|||||F
OBX|5|NM|6690-2^WBC^LN||15.2|10*3/uL|||||F
OBX|6|NM|2524-7^Lactate^LN||2.8|mmol/L|||||F"""

# Parse and extract features
patient_data = hl7.parse_adt_message(incoming_msg)
print("Extracted features:", patient_data)

# Make prediction
prediction = {
 'sepsis_risk': 0.73,
 'risk_category': 'High',
 'model_version': '1.2.0'
}

# Create response message
response_msg = hl7.create_oru_message('MRN12345', prediction)
print("\nHL7 ORU message:")
print(response_msg)

Database Integration for Audit Logging

All predictions must be logged for regulatory compliance and monitoring.

import psycopg2
from psycopg2.extras import execute_values
from contextlib import contextmanager
import json
from typing import Dict, List
from datetime import datetime

class DatabaseLogger:
 """
 Log all predictions to database for:
 - Audit trail
 - Performance monitoring
 - Model retraining
 - Regulatory compliance
 """

 def __init__(self, db_config: Dict):
  """
  Initialize database connection

  Args:
   db_config: Dict with database connection parameters
  """
  self.config = db_config

 @contextmanager
 def get_connection(self):
  """Context manager for database connections"""
  conn = psycopg2.connect(**self.config)
  try:
   yield conn
   conn.commit()
  except Exception as e:
   conn.rollback()
   raise e
  finally:
   conn.close()

 def initialize_schema(self):
  """Create database schema if doesn't exist"""
  schema = """
  -- Predictions table
  CREATE TABLE IF NOT EXISTS predictions (
   id SERIAL PRIMARY KEY,
   prediction_id VARCHAR(50) UNIQUE NOT NULL,
   patient_id VARCHAR(50) NOT NULL,
   timestamp TIMESTAMP NOT NULL DEFAULT NOW(),
   model_name VARCHAR(100) NOT NULL,
   model_version VARCHAR(20) NOT NULL,
   sepsis_risk FLOAT NOT NULL,
   risk_category VARCHAR(10) NOT NULL,
   confidence FLOAT,
   features JSONB NOT NULL,
   prediction_time_ms FLOAT,
   api_version VARCHAR(20),
   INDEX idx_patient_timestamp (patient_id, timestamp),
   INDEX idx_timestamp (timestamp),
   INDEX idx_model_version (model_version)
  );

  -- Actual outcomes table (populated later with ground truth)
  CREATE TABLE IF NOT EXISTS outcomes (
   id SERIAL PRIMARY KEY,
   prediction_id VARCHAR(50) REFERENCES predictions(prediction_id),
   actual_sepsis BOOLEAN NOT NULL,
   outcome_timestamp TIMESTAMP NOT NULL DEFAULT NOW(),
   confirmed_by VARCHAR(100),
   notes TEXT,
   INDEX idx_prediction_id (prediction_id)
  );

  -- Model performance metrics table
  CREATE TABLE IF NOT EXISTS performance_metrics (
   id SERIAL PRIMARY KEY,
   model_version VARCHAR(20) NOT NULL,
   evaluation_date DATE NOT NULL,
   metric_name VARCHAR(50) NOT NULL,
   metric_value FLOAT NOT NULL,
   subgroup VARCHAR(50),
   n_samples INTEGER,
   INDEX idx_model_date (model_version, evaluation_date)
  );

  -- Drift detection results table
  CREATE TABLE IF NOT EXISTS drift_detections (
   id SERIAL PRIMARY KEY,
   detection_date DATE NOT NULL,
   feature_name VARCHAR(50) NOT NULL,
   ks_statistic FLOAT,
   p_value FLOAT,
   psi_score FLOAT,
   drift_detected BOOLEAN,
   severity VARCHAR(20),
   INDEX idx_detection_date (detection_date)
  );
  """

  with self.get_connection() as conn:
   cursor = conn.cursor()
   cursor.execute(schema)
   print("[OK] Database schema initialized")

 def log_prediction(self, prediction_data: Dict) -> int:
  """
  Log prediction to database

  Args:
   prediction_data: Dict with prediction details

  Returns:
   Database ID of logged prediction
  """
  with self.get_connection() as conn:
   cursor = conn.cursor()

   cursor.execute("""
    INSERT INTO predictions (
     prediction_id,
     patient_id,
     timestamp,
     model_name,
     model_version,
     sepsis_risk,
     risk_category,
     confidence,
     features,
     prediction_time_ms,
     api_version
    ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
    RETURNING id
   """, (
    prediction_data['prediction_id'],
    prediction_data['patient_id'],
    prediction_data.get('timestamp', datetime.utcnow()),
    prediction_data.get('model_name', 'sepsis_predictor'),
    prediction_data['model_version'],
    prediction_data['sepsis_risk'],
    prediction_data['risk_category'],
    prediction_data.get('confidence'),
    json.dumps(prediction_data.get('features', {})),
    prediction_data.get('prediction_time_ms'),
    prediction_data.get('api_version', '1.0')
   ))

   prediction_id = cursor.fetchone()[0]

   return prediction_id

 def log_outcome(self, prediction_id: str, actual_sepsis: bool,
     confirmed_by: str, notes: Optional[str] = None):
  """
  Log actual outcome (ground truth) for model evaluation

  Called after clinical confirmation of sepsis status
  """
  with self.get_connection() as conn:
   cursor = conn.cursor()

   cursor.execute("""
    INSERT INTO outcomes (
     prediction_id,
     actual_sepsis,
     confirmed_by,
     notes
    ) VALUES (%s, %s, %s, %s)
   """, (prediction_id, actual_sepsis, confirmed_by, notes))

 def get_recent_predictions(self, hours: int = 24) -> pd.DataFrame:
  """Retrieve recent predictions for monitoring"""
  with self.get_connection() as conn:
   query = f"""
    SELECT
     p.*,
     o.actual_sepsis,
     o.outcome_timestamp
    FROM predictions p
    LEFT JOIN outcomes o ON p.prediction_id = o.prediction_id
    WHERE p.timestamp > NOW() - INTERVAL '{hours} hours'
    ORDER BY p.timestamp DESC
   """

   return pd.read_sql(query, conn)

 def calculate_performance_metrics(self, days: int = 7) -> Dict:
  """
  Calculate model performance metrics from predictions with outcomes

  Args:
   days: Number of days to look back

  Returns:
   Dict with performance metrics
  """
  with self.get_connection() as conn:
   query = f"""
    SELECT
     p.sepsis_risk,
     p.risk_category,
     o.actual_sepsis,
     p.model_version
    FROM predictions p
    INNER JOIN outcomes o ON p.prediction_id = o.prediction_id
    WHERE p.timestamp > NOW() - INTERVAL '{days} days'
   """

   df = pd.read_sql(query, conn)

   if len(df) == 0:
    return {'error': 'No predictions with outcomes in time period'}

   from sklearn.metrics import roc_auc_score, accuracy_score, recall_score, precision_score

   y_true = df['actual_sepsis'].values
   y_pred_proba = df['sepsis_risk'].values
   y_pred = (y_pred_proba >= 0.5).astype(int)

   metrics = {
    'n_samples': len(df),
    'prevalence': y_true.mean(),
    'auc': roc_auc_score(y_true, y_pred_proba),
    'accuracy': accuracy_score(y_true, y_pred),
    'sensitivity': recall_score(y_true, y_pred),
    'specificity': recall_score(1-y_true, 1-y_pred),
    'ppv': precision_score(y_true, y_pred),
    'npv': precision_score(1-y_true, 1-y_pred)
   }

   return metrics

# Example configuration
db_config = {
 'host': 'localhost',
 'database': 'ml_monitoring',
 'user': 'mlops',
 'password': 'secure_password', # Use secrets management
 'port': 5432
}

db_logger = DatabaseLogger(db_config)

# Initialize schema (run once)
db_logger.initialize_schema()

# Log prediction
prediction_data = {
 'prediction_id': 'pred_abc123',
 'patient_id': 'MRN12345',
 'model_version': '1.2.0',
 'sepsis_risk': 0.73,
 'risk_category': 'High',
 'confidence': 0.89,
 'features': {
  'heart_rate': 110,
  'respiratory_rate': 24,
  'temperature': 38.5,
  'systolic_bp': 95,
  'white_blood_cell': 15.2,
  'lactate': 2.8,
  'age': 67
 },
 'prediction_time_ms': 45.2
}

db_id = db_logger.log_prediction(prediction_data)
print(f"[OK] Prediction logged with ID: {db_id}")

# Later: Log actual outcome
db_logger.log_outcome(
 prediction_id='pred_abc123',
 actual_sepsis=True,
 confirmed_by='Dr. Smith',
 notes='Patient developed sepsis 4 hours after prediction'
)

# Calculate recent performance
metrics = db_logger.calculate_performance_metrics(days=7)
print(f"Past 7 days performance: AUC={metrics['auc']:.3f}, Sensitivity={metrics['sensitivity']:.3f}")

Regulatory Compliance

FDA Pathways for AI/ML Medical Devices

FDA Software as a Medical Device (SaMD)

AI/ML systems that diagnose, treat, prevent, or mitigate disease are considered medical devices and require FDA clearance.

Three main pathways:

510(k) Premarket Notification - Most common for AI (Moderate risk, Class II)
De Novo Classification - Novel low-to-moderate risk devices
PMA (Premarket Approval) - High-risk devices (Class III)

Understanding 510(k) for AI

510(k) requires demonstration of “substantial equivalence” to a predicate device.

Benjamens et al., 2020, npj Digital Medicine - “The state of artificial intelligence-based FDA-approved medical devices and algorithms”

Key components:

Device Description

Intended use
Indications for use
Contraindications
Target population

Predicate Device

Identification of cleared predicate
Comparison of technological characteristics
Comparison of performance

Performance Testing

Clinical validation study
Performance metrics
Subgroup analysis

Risk Analysis

Failure modes and effects analysis (FMEA)
Risk mitigation strategies

Software Description

Level of concern (minor, moderate, major)
Software development lifecycle
Verification and validation

Example: Documenting for 510(k) submission:

from dataclasses import dataclass
from typing import List, Dict, Optional
from datetime import datetime

@dataclass
class RegulatoryDocumentation:
 """
 Structure regulatory documentation for FDA submission

 Based on FDA guidance for AI/ML SaMD
 """
 device_name: str
 manufacturer: str

 # Intended Use
 intended_use: str
 indications: List[str]
 contraindications: List[str]
 target_population: str

 # Predicate Device
 predicate_510k_number: Optional[str] = None
 predicate_device_name: Optional[str] = None

 # Training Data
 training_dataset: Dict = None

 # Performance
 validation_study: Dict = None
 performance_metrics: Dict = None

 # Risk Analysis
 risk_analysis: List[Dict] = None

 # Software
 software_level_of_concern: str = "moderate" # minor, moderate, major
 development_lifecycle: str = None

 def generate_510k_summary(self) -> str:
  """Generate 510(k) summary document"""

  summary = f"""
510(k) SUMMARY

Submitter Information:
 Manufacturer: {self.manufacturer}
 Date Prepared: {datetime.now().strftime('%B %d, %Y')}

Device Trade Name: {self.device_name}

Device Classification:
 Product Code: DQK (Clinical Decision Support Software)
 Regulation Number: 21 CFR 870.1310
 Device Class: II
 Review Panel: Cardiovascular

Predicate Device:
 510(k) Number: {self.predicate_510k_number or 'N/A'}
 Device Name: {self.predicate_device_name or 'N/A'}

Intended Use:
{self.intended_use}

Indications for Use:
{chr(10).join(' - ' + indication for indication in self.indications)}

Contraindications:
{chr(10).join(' - ' + contra for contra in self.contraindications)}

Target Population:
{self.target_population}

DEVICE DESCRIPTION:

The {self.device_name} is a software-only medical device that uses machine learning
algorithms to predict the risk of sepsis in adult intensive care unit (ICU) patients.
The device analyzes patient vital signs and laboratory results to generate a risk
score between 0 and 1, indicating the probability of sepsis development within the
next 6 hours.

The software receives input data from the hospital's electronic health record (EHR)
system via HL7/FHIR interface. The device outputs:
1. Sepsis risk score (0-1 scale)
2. Risk category (Low, Medium, High)
3. Confidence estimate

The device is intended to be used as an adjunct to clinical decision-making and
does not automate any clinical decisions without physician oversight.

TRAINING DATASET:

Dataset Size: {self.training_dataset.get('size', 'N/A')} patients
Time Period: {self.training_dataset.get('date_range', 'N/A')}
Number of Sites: {self.training_dataset.get('n_sites', 'N/A')}
Geographic Distribution: {self.training_dataset.get('geographic_distribution', 'N/A')}

Inclusion Criteria:
{chr(10).join(' - ' + criteria for criteria in self.training_dataset.get('inclusion_criteria', []))}

Exclusion Criteria:
{chr(10).join(' - ' + criteria for criteria in self.training_dataset.get('exclusion_criteria', []))}

Demographics:
 Age Range: {self.training_dataset.get('age_range', 'N/A')}
 Gender Distribution: {self.training_dataset.get('gender_distribution', 'N/A')}
 Race/Ethnicity Distribution: {self.training_dataset.get('race_distribution', 'N/A')}

Label Source: {self.training_dataset.get('label_source', 'N/A')}
Data Quality Measures: {self.training_dataset.get('quality_measures', 'N/A')}

CLINICAL VALIDATION STUDY:

Study Design: {self.validation_study.get('design', 'N/A')}
Number of Patients: {self.validation_study.get('n_patients', 'N/A')}
Number of Sites: {len(self.validation_study.get('sites', []))}
Study Period: {self.validation_study.get('date_range', 'N/A')}

Primary Endpoint: {self.validation_study.get('primary_endpoint', 'N/A')}

PERFORMANCE METRICS:

Overall Performance:
 AUC-ROC: {self.performance_metrics.get('auc', 'N/A'):.3f}
 Sensitivity: {self.performance_metrics.get('sensitivity', 'N/A'):.3f}
 Specificity: {self.performance_metrics.get('specificity', 'N/A'):.3f}
 PPV: {self.performance_metrics.get('ppv', 'N/A'):.3f}
 NPV: {self.performance_metrics.get('npv', 'N/A'):.3f}

Subgroup Analysis:
{self._format_subgroup_performance()}

RISK ANALYSIS:

{self._format_risk_analysis()}

SOFTWARE DESCRIPTION:

Level of Concern: {self.software_level_of_concern.title()}
Development Methodology: {self.development_lifecycle}

The device software has been developed following FDA guidance on Software Development
Activities and incorporates:
 - Requirements traceability
 - Design verification and validation
 - Version control
 - Cybersecurity measures
 - Post-market monitoring capabilities

CONCLUSION:

The {self.device_name} has been demonstrated to be substantially equivalent to the
predicate device. Performance testing shows comparable safety and effectiveness.
The device meets applicable FDA requirements for software as a medical device.
  """

  return summary

 def _format_subgroup_performance(self) -> str:
  """Format subgroup analysis results"""
  if not self.performance_metrics.get('subgroups'):
   return " No subgroup analysis performed"

  output = []
  for subgroup, metrics in self.performance_metrics['subgroups'].items():
   output.append(f" {subgroup}:")
   output.append(f" AUC: {metrics.get('auc', 'N/A'):.3f}")
   output.append(f" Sensitivity: {metrics.get('sensitivity', 'N/A'):.3f}")
   output.append(f" Specificity: {metrics.get('specificity', 'N/A'):.3f}")

  return '\n'.join(output)

 def _format_risk_analysis(self) -> str:
  """Format risk analysis (FMEA)"""
  if not self.risk_analysis:
   return "See attached risk analysis document"

  output = ["Identified Risks and Mitigations:\n"]
  for i, risk in enumerate(self.risk_analysis, 1):
   output.append(f"Risk {i}: {risk['hazard']}")
   output.append(f" Severity: {risk['severity']}")
   output.append(f" Probability: {risk['probability']}")
   output.append(f" Mitigation: {risk['mitigation']}\n")

  return '\n'.join(output)

# Example usage
reg_doc = RegulatoryDocumentation(
 device_name="AI Sepsis Predictor",
 manufacturer="Hospital AI Systems Inc.",

 intended_use=(
  "The AI Sepsis Predictor is intended to predict the risk of sepsis in adult "
  "patients admitted to intensive care units. The device is intended for use by "
  "qualified healthcare professionals as an aid in clinical decision-making."
 ),

 indications=[
  "Prediction of sepsis risk in adult ICU patients (age ≥18 years)",
  "Identification of patients who may benefit from enhanced monitoring or early intervention",
  "Risk stratification for sepsis development within 6 hours"
 ],

 contraindications=[
  "Not intended for use in pediatric patients (age <18 years)",
  "Not intended for use as sole basis for clinical decision-making",
  "Not intended for use in patients with incomplete vital signs or laboratory data"
 ],

 target_population="Adult patients admitted to intensive care units",

 predicate_510k_number="K201234",
 predicate_device_name="Clinical Decision Support System for Sepsis Detection",

 training_dataset={
  'size': 50000,
  'date_range': '2018-2022',
  'n_sites': 12,
  'geographic_distribution': 'Multi-state (CA, NY, TX, FL, IL)',
  'inclusion_criteria': [
   'Adult patients (≥18 years)',
   'ICU admission',
   'Complete vital signs within 24 hours',
   'Laboratory results available'
  ],
  'exclusion_criteria': [
   'Pediatric patients (<18 years)',
   'Missing >20% of required data elements',
   'Pre-existing sepsis diagnosis on ICU admission'
  ],
  'age_range': '18-95 years (median 64)',
  'gender_distribution': '52% male, 48% female',
  'race_distribution': '60% White, 15% Black, 15% Hispanic, 10% Other',
  'label_source': 'Sepsis-3 criteria applied by trained physician reviewers',
  'quality_measures': 'Inter-rater reliability kappa = 0.87'
 },

 validation_study={
  'design': 'Prospective, multi-center observational study',
  'n_patients': 10000,
  'sites': ['Academic Medical Center A', 'Community Hospital B', 'Tertiary Care Center C'],
  'date_range': 'January 2023 - December 2023',
  'primary_endpoint': 'Development of sepsis (Sepsis-3 criteria) within 6 hours of prediction'
 },

 performance_metrics={
  'auc': 0.876,
  'sensitivity': 0.851,
  'specificity': 0.823,
  'ppv': 0.612,
  'npv': 0.947,
  'subgroups': {
   'Age ≥65 years': {'auc': 0.869, 'sensitivity': 0.843, 'specificity': 0.817},
   'Age <65 years': {'auc': 0.883, 'sensitivity': 0.859, 'specificity': 0.829},
   'Male': {'auc': 0.874, 'sensitivity': 0.847, 'specificity': 0.821},
   'Female': {'auc': 0.878, 'sensitivity': 0.855, 'specificity': 0.825},
   'White': {'auc': 0.881, 'sensitivity': 0.857, 'specificity': 0.828},
   'Black': {'auc': 0.867, 'sensitivity': 0.841, 'specificity': 0.815},
   'Hispanic': {'auc': 0.872, 'sensitivity': 0.848, 'specificity': 0.819}
  }
 },

 risk_analysis=[
  {
   'hazard': 'False negative prediction (missed sepsis case)',
   'severity': 'Major',
   'probability': 'Medium',
   'mitigation': 'Device intended as adjunct to clinical judgment. Clinicians maintain '
       'responsibility for diagnosis. Device includes confidence estimate. '
       'Training emphasizes limitations.'
  },
  {
   'hazard': 'False positive prediction (unnecessary intervention)',
   'severity': 'Moderate',
   'probability': 'Medium',
   'mitigation': 'Risk category thresholds set conservatively. Clinicians make final '
       'treatment decisions. Device does not automate interventions.'
  },
  {
   'hazard': 'Software malfunction or incorrect prediction',
   'severity': 'Major',
   'probability': 'Low',
   'mitigation': 'Extensive verification and validation testing. Automated monitoring '
       'of predictions. Human oversight required. Fallback to standard care.'
  }
 ],

 software_level_of_concern="moderate",
 development_lifecycle="Agile development with regulatory compliance checkpoints"
)

# Generate 510(k) summary
summary = reg_doc.generate_510k_summary()

# Save to file
with open('510k_summary.txt', 'w') as f:
 f.write(summary)

print("[OK] 510(k) summary generated")

Post-Market Surveillance

FDA Post-Market Surveillance guidance

FDA requires ongoing monitoring after device clearance.

from enum import Enum
from dataclasses import dataclass
from typing import List, Optional
import logging

class EventSeverity(Enum):
 MINOR = "minor"
 MODERATE = "moderate"
 SEVERE = "severe"
 DEATH = "death"

@dataclass
class AdverseEvent:
 """Structure for adverse event reporting"""
 event_id: str
 timestamp: datetime
 patient_id: str
 event_type: str # e.g., 'false_negative', 'system_error', 'incorrect_prediction'
 severity: EventSeverity
 description: str
 clinical_outcome: Optional[str] = None
 corrective_action: Optional[str] = None
 root_cause: Optional[str] = None
 preventable: Optional[bool] = None

class PostMarketSurveillance:
 """
 Monitor deployed model for FDA post-market surveillance requirements

 Implements:
 - Adverse event tracking
 - MDR (Medical Device Report) filing
 - Quarterly performance reports
 - Safety signal detection
 """

 def __init__(self):
  self.adverse_events: List[AdverseEvent] = []
  self.logger = logging.getLogger(__name__)

 def log_adverse_event(self, event: AdverseEvent):
  """
  Log adverse event for FDA reporting

  FDA requires reporting of:
  - Deaths
  - Serious injuries
  - Malfunctions that could cause serious injury/death
  """
  self.adverse_events.append(event)

  self.logger.warning(
   f"Adverse event logged: {event.event_id} - "
   f"Severity: {event.severity.value}, Type: {event.event_type}"
  )

  # Check if Medical Device Report (MDR) required
  if self._requires_mdr(event):
   self._initiate_mdr_filing(event)

 def _requires_mdr(self, event: AdverseEvent) -> bool:
  """
  Determine if event requires MDR filing to FDA

  MDR required for:
  - Death
  - Serious injury (hospitalization, disability, intervention to prevent harm)
  - Malfunction that would be likely to cause death/serious injury if it recurred
  """
  # Death always requires MDR
  if event.severity == EventSeverity.DEATH:
   return True

  # Severe injuries require MDR
  if event.severity == EventSeverity.SEVERE:
   return True

  # Malfunction assessment
  malfunction_keywords = ['system failure', 'crash', 'incorrect prediction leading to harm']
  if any(keyword in event.description.lower() for keyword in malfunction_keywords):
   return True

  return False

 def _initiate_mdr_filing(self, event: AdverseEvent):
  """
  Initiate Medical Device Report filing with FDA

  Timeline: Within 30 days of becoming aware
  """
  self.logger.critical(
   f"MDR FILING REQUIRED for event {event.event_id}\n"
   f"Severity: {event.severity.value}\n"
   f"Description: {event.description}\n"
   f"Clinical Outcome: {event.clinical_outcome}"
  )

  # In production: Integrate with FDA MAUDE database
  # MedWatch Form 3500A for mandatory reporting

  # Alert responsible parties
  self._send_urgent_notification(
   title=f'MDR Filing Required: {event.event_id}',
   message=f'Adverse event requires FDA Medical Device Report within 30 days',
   event_details=event
  )

 def _send_urgent_notification(self, title: str, message: str, event_details: AdverseEvent):
  """Send urgent notification to compliance team"""
  # Implementation depends on organization's alerting system
  # Could be PagerDuty, email, Slack, etc.
  print(f"\n{'='*60}")
  print(f"🚨 URGENT: {title}")
  print(f"{'='*60}")
  print(f"{message}\n")
  print(f"Event ID: {event_details.event_id}")
  print(f"Severity: {event_details.severity.value}")
  print(f"Description: {event_details.description}")
  print(f"{'='*60}\n")

 def generate_quarterly_report(self, quarter: str, year: int) -> Dict:
  """
  Generate quarterly post-market surveillance report for FDA

  Required reporting includes:
  - Total device uses
  - Adverse events summary
  - Performance metrics
  - Corrective actions taken
  """
  # Filter events for quarter
  quarter_start, quarter_end = self._get_quarter_dates(quarter, year)

  quarter_events = [
   e for e in self.adverse_events
   if quarter_start <= e.timestamp <= quarter_end
  ]

  # Categorize events
  events_by_severity = {
   'minor': sum(1 for e in quarter_events if e.severity == EventSeverity.MINOR),
   'moderate': sum(1 for e in quarter_events if e.severity == EventSeverity.MODERATE),
   'severe': sum(1 for e in quarter_events if e.severity == EventSeverity.SEVERE),
   'death': sum(1 for e in quarter_events if e.severity == EventSeverity.DEATH)
  }

  events_by_type = {}
  for event in quarter_events:
   events_by_type[event.event_type] = events_by_type.get(event.event_type, 0) + 1

  # MDR filings
  mdr_filed = sum(1 for e in quarter_events if self._requires_mdr(e))

  # Get performance metrics from database
  performance = self._get_quarter_performance(quarter_start, quarter_end)

  report = {
   'quarter': f'Q{quarter} {year}',
   'reporting_period': f'{quarter_start.date()} to {quarter_end.date()}',
   'total_predictions': performance.get('total_predictions', 0),
   'total_adverse_events': len(quarter_events),
   'adverse_events_by_severity': events_by_severity,
   'adverse_events_by_type': events_by_type,
   'mdr_filed': mdr_filed,
   'performance_metrics': {
    'auc': performance.get('auc'),
    'sensitivity': performance.get('sensitivity'),
    'specificity': performance.get('specificity'),
    'ppv': performance.get('ppv')
   },
   'corrective_actions': self._summarize_corrective_actions(quarter_events),
   'preventable_events': sum(1 for e in quarter_events if e.preventable)
  }

  self.logger.info(f"Generated quarterly report for Q{quarter} {year}")

  return report

 def _get_quarter_dates(self, quarter: str, year: int) -> tuple:
  """Get start and end dates for quarter"""
  quarter_starts = {
   '1': (1, 1),
   '2': (4, 1),
   '3': (7, 1),
   '4': (10, 1)
  }

  month, day = quarter_starts[quarter]
  start = datetime(year, month, day)

  # Calculate end date
  if quarter == '4':
   end = datetime(year, 12, 31, 23, 59, 59)
  else:
   next_month = month + 3
   end = datetime(year, next_month, 1) - timedelta(seconds=1)

  return start, end

 def _get_quarter_performance(self, start_date: datetime, end_date: datetime) -> Dict:
  """Get performance metrics for quarter from database"""
  # In production: Query from database
  # Placeholder implementation
  return {
   'total_predictions': 125000,
   'auc': 0.867,
   'sensitivity': 0.843,
   'specificity': 0.815,
   'ppv': 0.598
  }

 def _summarize_corrective_actions(self, events: List[AdverseEvent]) -> List[str]:
  """Summarize corrective actions taken"""
  actions = set()
  for event in events:
   if event.corrective_action:
    actions.add(event.corrective_action)
  return list(actions)

# Example usage
surveillance = PostMarketSurveillance()

# Log adverse event
event = AdverseEvent(
 event_id='AE-2024-001',
 timestamp=datetime.now(),
 patient_id='MRN12345',
 event_type='false_negative',
 severity=EventSeverity.MODERATE,
 description='Model predicted low risk (0.18) but patient developed sepsis within 3 hours. '
    'Vital signs were within normal ranges at time of prediction.',
 clinical_outcome='Patient recovered after appropriate treatment. Delay in recognition '
      'led to 2-hour delay in antibiotic administration.',
 corrective_action='Case reviewed by clinical team. Added to model monitoring for similar cases.',
 root_cause='Patient presented with atypical sepsis (low-grade fever, minimal tachycardia)',
 preventable=False
)

surveillance.log_adverse_event(event)

# Generate quarterly report
report = surveillance.generate_quarterly_report(quarter='1', year=2024)

print("\n" + "="*60)
print("POST-MARKET SURVEILLANCE QUARTERLY REPORT")
print("="*60)
print(f"\nReporting Period: {report['reporting_period']}")
print(f"Total Predictions: {report['total_predictions']:,}")
print(f"\nAdverse Events: {report['total_adverse_events']}")
print(f" By Severity: {report['adverse_events_by_severity']}")
print(f" MDRs Filed: {report['mdr_filed']}")
print(f"\nPerformance Metrics:")
print(f" AUC: {report['performance_metrics']['auc']:.3f}")
print(f" Sensitivity: {report['performance_metrics']['sensitivity']:.3f}")
print(f" Specificity: {report['performance_metrics']['specificity']:.3f}")