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
import numpy as np

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)

    pixels = ds.pixel_array.astype('float32')
    pixels = (pixels - pixels.min()) / (pixels.max() - pixels.min() + 1e-8)

    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)

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)

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()

import whisper
import datetime

class AmbientScribe:
    """Real-time ambient clinical documentation pipeline.

    Captures audio from patient encounter, transcribes with Whisper,
    structures the transcript into a clinical note using an LLM,
    and returns HL7 FHIR-compatible structured data.
    """
    def __init__(self):
        self.asr_model = whisper.load_model("large-v3")
        self.phi_patterns = self._load_phi_patterns()

    def process_encounter(self, audio_path: str) -> dict:
        raw_transcript = self.asr_model.transcribe(audio_path)
        clean_text, redactions = self._deidentify(raw_transcript["text"])

        structured_note = self._structure_note(clean_text)

        return {
            "note": structured_note,
            "redactions": redactions,
            "duration_sec": raw_transcript["segments"][-1]["end"],
            "model": "whisper-large-v3"
        }

    def _structure_note(self, text: str) -> dict:
        """Use LLM to extract SOAP sections from transcript."""
        # In production, call a HIPAA-compliant LLM endpoint
        return {
            "subject": "Chest pain follow-up",
            "assessment": "Stable angina, continue current regimen",
            "plan": "Schedule stress test within 2 weeks"
        }

    def _load_phi_patterns(self):
        return {
            "patient_name": r"\b(?:Mr\.|Mrs\.|Ms\.|Dr\.)\s+[A-Z][a-z]+\s+[A-Z][a-z]+\b",
            "mrn": r"(?i)\b(?:MRN|medical record)[:\s]*[A-Z]{0,3}\d{4,10}\b",
        }

    def _deidentify(self, text: str) -> tuple:
        for pattern in self._phi_patterns.values():
            text = re.sub(pattern, "[REDACTED]", text)
        return text, {}

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

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

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)

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_logger.info(
        f"PHI_ACCESS provider={provider['sub']} "
        f"patient={patient_id} resource=Observation "
        f"timestamp={datetime.utcnow().isoformat()}"
    )

    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 requests

SMART_CONFIG = {
    "authorize_url": "https://ehr.example.com/auth/authorize",
    "token_url": "https://ehr.example.com/auth/token",
    "client_id": "your-app-id",
    "redirect_uri": "https://yourapp.com/callback",
    "scope": "patient/Observation.read patient/DiagnosticReport.read"
}

def smart_on_fhir_auth():
    """Initiate SMART on FHIR OAuth2 flow."""
    auth_url = (
        f"{SMART_CONFIG['authorize_url']}?"
        f"response_type=code&"
        f"client_id={SMART_CONFIG['client_id']}&"
        f"redirect_uri={SMART_CONFIG['redirect_uri']}&"
        f"scope={SMART_CONFIG['scope']}&"
        f"aud=https://ehr.example.com/fhir"
    )
    return auth_url

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",
    "address": r"\b\d{1,5}\s+[A-Za-z0-9\s,]+(?:Street|St|Ave|Avenue|Road|Rd|Drive|Dr|Lane|Ln)\b",
    "zip": r"\b\d{5}(?:-\d{4})?\b",
}

def deidentify_text(text: str, replacement: str = "[REDACTED]") -> tuple:
    """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

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."

def strip_dicom_phi(dicom_path: str, output_path: str) -> int:
    """Remove all PHI tags from a DICOM file. Returns count of removed tags."""
    ds = pydicom.dcmread(dicom_path)
    phi_tags = [
        (0x0010, 0x0010),  # Patient Name
        (0x0010, 0x0020),  # Patient ID
        (0x0010, 0x0030),  # Patient Birth Date
        (0x0010, 0x0040),  # Patient Sex
        (0x0008, 0x0080),  # Institution Name
        (0x0008, 0x0090),  # Referring Physician
        (0x0008, 0x1070),  # Operator Name
    ]
    removed = 0
    for tag in phi_tags:
        if tag in ds:
            del ds[tag]
            removed += 1
    ds.save_as(output_path)
    return removed

Hospital Network:
┌─────────────┐    ┌──────────────┐    ┌──────────────┐
│   PACS      │───▶│  AI Inference │───▶│   EMR        │
│  (Images)   │    │  Server (GPU) │    │  (Results)   │
└─────────────┘    └──────────────┘    └──────────────┘
       │                  │                    │
       │        ┌─────────┴─────────┐          │
       │        │  De-identification │         │
       │        │  + Audit Log      │         │
       │        └──────────────────┘          │
       │                                       │
       └─────────── PACS Integration ──────────┘