Skip to content

Clinical ETL Pipelines

ColabBio provides modular, production-ready Nextflow DSL2 pipelines to ingest heterogeneous hospital datasets and securely synchronize them with native clinical microservices.

graph TD
    subgraph Hospital Data Sources
        CSV["Clinical CSV / EHR Extracts"]
        DUMP["Relational DB Dumps"]
        WSI["Histology Slides (.svs, .tiff)"]
    end

    subgraph ColabBio ETL Modules
        F1["FHIR_INGEST (csv_fhir_mapper)"]
        O1["OMOP_INGEST (Vocabulary ETL)"]
        D1["DICOM_ORTHANC_INGEST (WSI to DICOM)"]
        M1["OMERO_IMPORT (CLI Pipeline)"]
    end

    subgraph Clinical Core Services
        FHIR_SRV[("HAPI FHIR R4")]
        OMOP_SRV[("OMOP CDM PostgreSQL")]
        ORTHANC_SRV[("Orthanc PACS")]
        OMERO_SRV[("OMERO Server")]
    end

    CSV --> F1 --> FHIR_SRV
    DUMP --> O1 --> OMOP_SRV
    WSI --> D1 --> ORTHANC_SRV
    WSI --> M1 --> OMERO_SRV

1. FHIR Ingestion Pipeline (`FHIR_INGEST`)

The FHIR loader transforms tabular patient data, laboratory values, and diagnostic reports into standard HL7 FHIR R4 Resources (Patient, Condition, Observation, DocumentReference).

Pipeline Process (Nextflow DSL2)

process FHIR_INGEST {
    tag "FHIR Ingest: ${dataset_id}"
    container 'ghcr.io/colabbio/fhir-tools:latest'

    input:
    tuple val(dataset_id), path(clinical_file)

    output:
    stdout emit: log_output

    script:
    """
    python3 /opt/colabbio/etl/csv_fhir_mapper.py \
        --in ${clinical_file} \
        --out bundle.json \
        --patient-id-col "patient_id" \
        --condition-code-system "http://snomed.info/sct"

    curl -X POST -H "Content-Type: application/fhir+json" \
         -d @bundle.json \
         "${params.fhir_url}/Bundle"
    """
}

Image Linking: The loader creates DocumentReference resources containing pointers to the corresponding WSI slide identifiers in OMERO or Orthanc.

2. OMOP CDM Pipeline (`OMOP_INGEST`)

The OMOP pipeline populates the OMOP Common Data Model (v5.4) running on PostgreSQL, standardizing clinical concepts into SNOMED-CT, RxNorm, and LOINC.

process OMOP_INGEST {
    tag "OMOP Ingest: ${dataset_id}"
    container 'ghcr.io/colabbio/omop-loader:latest'

    input:
    tuple val(dataset_id), path(database_dump)

    output:
    stdout emit: log_output

    script:
    """
    # Direct database staging & vocabulary mapping
    pg_restore --no-owner --role=omop_user -d ${params.omop_dsn} ${database_dump}

    psql ${params.omop_dsn} -f /opt/colabbio/sql/map_vocabularies.sql
    """
}

Populated tables include: person, condition_occurrence, measurement, drug_exposure, and observation_period.

3. DICOM WSI Ingestion (`DICOM_ORTHANC_INGEST`)

Transforms gigapixel Whole Slide Images (SVS, NDPI, TIFF) into standardized DICOM Supplement 145 (Whole Slide Imaging) multi-resolution tile pyramids.

process DICOM_ORTHANC_INGEST {
    tag "Orthanc Ingest: ${slide_name}"
    container 'ghcr.io/colabbio/orthanc-wsi:latest'

    input:
    tuple val(slide_name), path(slide_file)

    output:
    stdout emit: log_output

    script:
    """
    wsimporter --input ${slide_file} \
               --output ./dicom_pyramid \
               --tile-size 512 \
               --compression jpeg

    # STOW-RS bulk upload to PACS
    storescu -xs ${params.orthanc_host} ${params.orthanc_port} ./dicom_pyramid/*.dcm
    """
}

4. OMERO Server Import (`OMERO_IMPORT`)

Imports raw Whole Slide Images into the institutional OMERO repository for multi-channel bio-formats rendering.

process OMERO_IMPORT {
    tag "OMERO Import: ${slide_name}"
    container 'ghcr.io/colabbio/omero-client:latest'

    input:
    tuple val(slide_name), path(slide_file)

    script:
    """
    omero login -s ${params.omero_host} -u ${OMERO_USER} -w ${OMERO_PASS}
    omero import ${slide_file} --target "Dataset:${params.omero_dataset_id}"
    """
}