AI in Healthcare: Technical Implementation Guide — Medical Imaging, HIPAA APIs, and ML Pipelines

import pydicom
import torch
import torch.nn as nn
from torchvision import transforms
from PIL import Image
import io

# --- DICOM Loading and PHI Stripping ---

def load_dicom_image(dicom_path: str) -> torch.Tensor:
    """Load a DICOM file, strip PHI metadata, return normalized tensor.

    The DICOM file contains both pixel data and protected health information
    (patient name, ID, DOB). We extract only the pixel array and discard
    all metadata for privacy.
    """
    ds = pydicom.dcmread(dicom_path)

    # Extract pixel array and rescale to 0-1
    pixels = ds.pixel_array.astype('float32')
    pixels = (pixels - pixels.min()) / (pixels.max() - pixels.min() + 1e-8)

    # Convert single-channel grayscale to 3-channel RGB (for pretrained models)
    if len(pixels.shape) == 2:
        pixels = np.stack([pixels] * 3, axis=0)
    elif len(pixels.shape) == 3 and pixels.shape[0] == 1:
        pixels = np.repeat(pixels, 3, axis=0)

    return torch.from_numpy(pixels).unsqueeze(0)  # Add batch dim

# --- Classification Model ---

class ChestXRayClassifier(nn.Module):
    """Binary classifier for chest X-rays (normal vs abnormal).

    Uses a pretrained ResNet-18 backbone fine-tuned on medical images.
    Medical imaging models typically use transfer learning from ImageNet
    weights due to limited labeled medical datasets.
    """
    def __init__(self, num_classes=2, pretrained=True):
        super().__init__()
        from torchvision.models import resnet18
        self.backbone = resnet18(weights='IMAGENET1K_V1' if pretrained else None)
        in_features = self.backbone.fc.in_features
        self.backbone.fc = nn.Sequential(
            nn.Dropout(0.3),
            nn.Linear(in_features, 512),
            nn.ReLU(),
            nn.Dropout(0.2),
            nn.Linear(512, num_classes)
        )

    def forward(self, x):
        return self.backbone(x)

# --- Inference ---

model = ChestXRayClassifier()
model.load_state_dict(torch.load("chest-xray-v1.pt"))
model.eval()

image_tensor = load_dicom_image("study_12345.dcm")
with torch.no_grad():
    logits = model(image_tensor)
    probs = torch.softmax(logits, dim=1)
    prediction = "abnormal" if probs[0][1] > 0.5 else "normal"
    confidence = probs[0][1].item() if prediction == "abnormal" else probs[0][0].item()
    print(f"Prediction: {prediction} (confidence: {confidence:.3f})")

from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
import logging

app = FastAPI(title="Clinical AI Inference API")
security = HTTPBearer()

# Audit logger — all PHI access is logged
audit_logger = logging.getLogger("phi_audit")
audit_handler = logging.FileHandler("/var/log/phi_access.log")
audit_logger.addHandler(audit_handler)
audit_logger.setLevel(logging.INFO)

# JWT verification (HIPAA requires authentication)
def verify_token(credentials: HTTPAuthorizationCredentials = Security(security)):
    """Verify the access token and return the requesting provider's identity."""
    try:
        payload = jwt.decode(
            credentials.credentials,
            algorithms=["RS256"],
            options={"verify_aud": True, "aud": "clinical-api"}
        )
        return payload
    except jwt.PyJWTError as e:
        raise HTTPException(status_code=401, detail=f"Invalid token: {e}")

@app.get("/fhir/Observation/{patient_id}")
async def get_predictions(
    patient_id: str,
    provider=Depends(verify_token)
):
    """Return AI predictions for a patient. Logs all PHI access."""
    # Audit log: who accessed whose data
    audit_logger.info(
        f"PHI_ACCESS provider={provider['sub']} "
        f"patient={patient_id} resource=Observation "
        f"timestamp={datetime.utcnow().isoformat()}"
    )

    # Fetch de-identified results from the inference DB
    results = await db.fetch_predictions(patient_id)
    if not results:
        raise HTTPException(status_code=404, detail="No predictions found")

    return {
        "resourceType": "Bundle",
        "type": "searchset",
        "entry": [{"resource": r} for r in results]
    }

import re

PHI_PATTERNS = {
    "patient_name": r"\b(?:Mr\.|Mrs\.|Ms\.|Dr\.)\s+[A-Z][a-z]+\s+[A-Z][a-z]+\b",
    "date_of_birth": r"\b\d{2}/\d{2}/\d{4}\b",
    "ssn": r"\b\d{3}-\d{2}-\d{4}\b",
    "phone": r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b",
    "email": r"\b[\w.]+@[\w.]+\.\w+\b",
    "mrn": r"(?i)\b(?:MRN|medical record)[:\s]*[A-Z]{0,3}\d{4,10}\b",
}

def deidentify_text(text: str, replacement: str = "[REDACTED]") -> str:
    """Remove PHI from clinical notes by replacing matches with [REDACTED].

    Returns both the de-identified text and a count of redactions per category.
    """
    redactions = {}
    for label, pattern in PHI_PATTERNS.items():
        matches = re.findall(pattern, text)
        if matches:
            redactions[label] = len(matches)
            text = re.sub(pattern, replacement, text)
    return text, redactions

# Example
note = "Patient John Doe (MRN: 12345, DOB: 04/15/1985) presents with chest pain."
clean_note, counts = deidentify_text(note)
# clean_note: "Patient [REDACTED] ([REDACTED]: [REDACTED], [REDACTED]: [REDACTED]) presents with chest pain."
# counts: {'patient_name': 1, 'date_of_birth': 1, 'mrn': 1}