Data Flow

End-to-end data pipeline documentation covering the complete flow from raw data sources through transformations to API responses.

Overview
Reference Data Pipeline
Runtime Data Flow
SIR Interpretation Flow
Antibiogram Generation Flow
Run Persistence Flow
Data Refresh Flow
Error and Dead-Letter Flow
Data Transformations

Overview

Python AMR processes two distinct types of data:

Reference Data - Static datasets (microorganisms, antimicrobials, breakpoints)
Runtime Data - User inputs and analytical results

High-Level Architecture

┌────────────────────────────────────────────────────────────┐
│                   Reference Data Pipeline                  │
│  External → Import → NDJSON → Transforms → Parquet        │
│  Sources                                                   │
└────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌────────────────────────────────────────────────────────────┐
│                    Application Layer                       │
│  API/CLI → Core Logic → Reference Store → Results         │
└────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌────────────────────────────────────────────────────────────┐
│                   Persistence Pipeline                     │
│  Results → Queue → Repository → Database → Storage        │
└────────────────────────────────────────────────────────────┘

Reference Data Pipeline

Pipeline Stages

External Sources (TSV/CSV/XLSX/RDA)
    │
    ├─► data-raw/external/
    │       • Loinc.csv (200 MB)
    │       • SNOMED_*.txt (5 MB)
    │       • GBIF Taxon.tsv (1 GB)
    │       • EUCAST breakpoints
    │
    ▼
Import Scripts (scripts/import_raw_sources.py)
    │
    ├─► Format detection and parsing
    ├─► Schema validation
    ├─► Deduplication
    │
    ▼
Canonical NDJSON (data-raw/sources/)
    │
    ├─► microorganisms.ndjson (75k records)
    ├─► antimicrobials.ndjson (465 records)
    ├─► antivirals.ndjson (85 records)
    ├─► clinical_breakpoints.ndjson (50k records)
    ├─► interpretive_rules.ndjson (1.3k records)
    ├─► intrinsic_resistant.ndjson (10k records)
    │
    ▼
Transform Pipeline (scripts/run_data_pipeline.py)
    │
    ├─► Load NDJSON with Polars
    ├─► Apply dataset-specific transforms
    ├─► Add computed fields
    ├─► Fix data quality issues
    │
    ▼
Parquet Snapshots (data/snapshots/)
    │
    ├─► microorganisms.parquet
    ├─► antimicrobials.parquet
    ├─► antivirals.parquet
    ├─► clinical_breakpoints.parquet
    ├─► interpretive_rules.parquet
    ├─► intrinsic_resistant.parquet
    │
    ▼
Snapshot Manifest (data/manifests/snapshot_manifest.json)
    │
    └─► Metadata, checksums, schemas

Detailed Flow

1. External Sources → Import

# scripts/import_raw_sources.py

def import_external_sources():
    """
    Import external formats to canonical NDJSON.

    Supported formats:
    - CSV/TSV (Polars)
    - XLSX (pandas + openpyxl)
    - Feather/IPC (Polars)
    - Parquet (Polars)
    - RDA (via feather intermediate)
    """
    mapping = load_import_mapping()  # data-raw/import_mapping.json

    for dataset_name, source_file in mapping.items():
        # Detect format
        format = detect_format(source_file)

        # Parse with appropriate loader
        df = parse_file(source_file, format)

        # Validate schema
        validate_schema(df, dataset_name)

        # Write NDJSON
        output_path = f"data-raw/sources/{dataset_name}.ndjson"
        df.write_ndjson(output_path)

2. NDJSON → Transforms

# scripts/run_data_pipeline.py

from amr.data.pipeline import DataPipeline
from amr.data.transforms import TRANSFORM_REGISTRY

def run_pipeline():
    """
    Execute transformation pipeline.

    Order defined in data-raw/pipeline.json
    """
    pipeline = DataPipeline(
        raw_dir="data-raw/sources",
        snapshot_dir="data/snapshots"
    )

    # Load execution order
    execution_order = load_pipeline_config()

    for dataset_name in execution_order:
        # Load NDJSON
        df = pipeline.load_raw(dataset_name)

        # Apply transforms
        transform = TRANSFORM_REGISTRY.get(dataset_name)
        if transform:
            df = transform.apply(df)

        # Write snapshot
        pipeline.write_snapshot(dataset_name, df)

    # Build manifest
    pipeline.build_manifest()

3. Transform Examples

Microorganisms Transform:

# src/amr/data/transforms/microorganisms.py

class MicroorganismsTransform:
    """Transform microorganisms dataset."""

    def apply(self, df: pl.DataFrame) -> pl.DataFrame:
        """
        Apply transformations:
        1. Add fullname column (genus + species)
        2. Normalize prevalence scores
        3. Add rank_index for sorting
        4. Validate taxonomy hierarchy
        """
        return (
            df
            .with_columns([
                # Combine genus and species
                (pl.col("genus") + " " + pl.col("species")).alias("fullname"),

                # Normalize prevalence
                (pl.col("prevalence") / pl.col("prevalence").max()).alias("prevalence_norm"),

                # Add rank index
                pl.col("rank").map_dict({
                    "kingdom": 1,
                    "phylum": 2,
                    "class": 3,
                    "order": 4,
                    "family": 5,
                    "genus": 6,
                    "species": 7,
                    "subspecies": 8
                }).alias("rank_index")
            ])
            .filter(pl.col("genus").is_not_null())  # Remove invalid entries
        )

Clinical Breakpoints Transform:

# src/amr/data/transforms/clinical_breakpoints.py

class ClinicalBreakpointsTransform:
    """Transform clinical breakpoints dataset."""

    def apply(self, df: pl.DataFrame) -> pl.DataFrame:
        """
        Apply transformations:
        1. Parse MIC/disk breakpoint ranges
        2. Add host_type (human/animal)
        3. Validate guideline consistency
        4. Add effective_date
        """
        return (
            df
            .with_columns([
                # Parse MIC breakpoints
                pl.col("breakpoint_S").cast(pl.Float64, strict=False),
                pl.col("breakpoint_R").cast(pl.Float64, strict=False),

                # Infer host type
                pl.when(pl.col("host").is_not_null())
                  .then(pl.lit("animal"))
                  .otherwise(pl.lit("human"))
                  .alias("host_type"),

                # Extract year from guideline
                pl.col("guideline")
                  .str.extract(r"(\d{4})", 1)
                  .cast(pl.Int32, strict=False)
                  .alias("guideline_year")
            ])
            .filter(
                # Remove invalid breakpoints
                (pl.col("breakpoint_S").is_not_null()) |
                (pl.col("breakpoint_R").is_not_null())
            )
        )

Runtime Data Flow

API Request → Response Flow

┌─────────────────────────────────────────────────────┐
│ 1. HTTP Request                                     │
│    POST /v1/sir/interpret                           │
│    {values: [0.5, 2, 16], mo: "B_ESCHR_COLI", ...} │
└──────────────────┬──────────────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────────────┐
│ 2. FastAPI Handler                                  │
│    • Pydantic validation                            │
│    • Request parsing                                │
└──────────────────┬──────────────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────────────┐
│ 3. Core Logic (amr.core.interpretation)             │
│    • Load breakpoints from reference store          │
│    • Apply interpretation rules                     │
│    • Return SIR results                             │
└──────────────────┬──────────────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────────────┐
│ 4. Persistence Queue (optional)                     │
│    • Enqueue run data                               │
│    • Return optimistic run_id                       │
└──────────────────┬──────────────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────────────┐
│ 5. HTTP Response                                    │
│    {results: ["S", "S", "R"], run_id: 123}         │
└─────────────────────────────────────────────────────┘

SIR Interpretation Flow

Detailed Sequence

User Input: values=[0.5, 2, 16], mo="B_ESCHR_COLI", ab="CIP"
    │
    ▼
┌─────────────────────────────────────────┐
│ 1. Normalize Inputs                     │
│    • mo → "B_ESCHR_COLI" (if needed)   │
│    • ab → "CIP" (if needed)            │
│    • Detect method (MIC/disk)          │
└────────────┬────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────┐
│ 2. Load Reference Data                  │
│    ReferenceStore.get_breakpoints()     │
│    └─► Returns Polars DataFrame        │
└────────────┬────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────┐
│ 3. Filter Breakpoints                   │
│    df.filter(                           │
│        (pl.col("mo") == mo) &           │
│        (pl.col("ab") == ab) &           │
│        (pl.col("guideline") == guide)   │
│    )                                    │
└────────────┬────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────┐
│ 4. Apply Breakpoint Logic               │
│    For each value:                      │
│      if value <= S_breakpoint: "S"      │
│      elif value >= R_breakpoint: "R"    │
│      else: "I"                          │
└────────────┬────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────┐
│ 5. Apply Interpretive Rules (optional)  │
│    • EUCAST expert rules                │
│    • Intrinsic resistance               │
│    • Disk/MIC discrepancy resolution    │
└────────────┬────────────────────────────┘
             │
             ▼
Output: ["S", "S", "R"]

Code Flow

# src/amr/core/interpretation.py

def as_sir(
    values: list,
    mo: str | None,
    ab: str | None,
    guideline: str = "EUCAST 2025",
    method: str | None = None,
    add_intrinsic_resistance: bool = True,
    interpretive_rules: str | None = None
) -> list[str]:
    """
    Interpret MIC/disk values to SIR categories.

    Data flow:
    1. Input validation
    2. Reference data lookup
    3. Breakpoint application
    4. Rule application
    5. Result return
    """
    # 1. Validate inputs
    if not values:
        raise ValueError("values cannot be empty")

    # 2. Load breakpoints
    from amr.data.reference_store import get_breakpoints
    breakpoints_df = get_breakpoints()

    # 3. Filter to relevant breakpoints
    filtered = breakpoints_df.filter(
        (pl.col("mo") == mo) &
        (pl.col("ab") == ab) &
        (pl.col("guideline") == guideline) &
        (pl.col("method") == (method or "MIC"))
    )

    if filtered.height == 0:
        raise ValueError(f"No breakpoints found for {mo}/{ab}/{guideline}")

    # 4. Extract breakpoint values
    s_breakpoint = filtered["breakpoint_S"][0]
    r_breakpoint = filtered["breakpoint_R"][0]

    # 5. Apply breakpoints
    results = []
    for value in values:
        if value <= s_breakpoint:
            sir = "S"
        elif value >= r_breakpoint:
            sir = "R"
        else:
            sir = "I"
        results.append(sir)

    # 6. Apply interpretive rules
    if interpretive_rules == "EUCAST":
        results = apply_eucast_rules(results, mo, ab)

    if add_intrinsic_resistance:
        results = apply_intrinsic_resistance(results, mo, ab)

    return results

Antibiogram Generation Flow

Input DataFrame:
    patient_id | organism    | AMX | CIP | GEN
    1          | E. coli     | R   | S   | S
    2          | E. coli     | S   | S   | S
    3          | K. pneumo   | R   | R   | S
    │
    ▼
┌─────────────────────────────────────────┐
│ 1. Normalize Organisms                  │
│    organism → mo_code via as_mo()       │
└────────────┬────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────┐
│ 2. Group by Pathogen                    │
│    df.group_by("mo_code")               │
└────────────┬────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────┐
│ 3. Calculate Resistance %                │
│    For each (pathogen, antimicrobial):  │
│      n_resistant = count(SIR == "R")    │
│      n_total = count(*)                 │
│      pct = 100 * n_resistant / n_total  │
└────────────┬────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────┐
│ 4. Filter by Minimum                    │
│    Keep only pathogens with n >= min    │
└────────────┬────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────┐
│ 5. Compute Confidence Intervals          │
│    Wilson score interval for each %     │
└────────────┬────────────────────────────┘
             │
             ▼
Output DataFrame:
    microorganism    | AMX  | CIP  | GEN
    Escherichia coli | 50.0 | 0.0  | 0.0
    K. pneumoniae    | 100  | 100  | 0.0

Run Persistence Flow

Async Queue Architecture

API Handler
    │
    │ enqueue_persist(run_data)
    ▼
┌─────────────────────────────────┐
│     AsyncQueue (4096 max)       │
│  ┌───┬───┬───┬───┬───┬───┐     │
│  │ 1 │ 2 │ 3 │ 4 │ 5 │...│     │
│  └───┴───┴───┴───┴───┴───┘     │
└─────────┬───────────────────────┘
          │
          │ queue.get()
          ▼
┌─────────────────────────────────┐
│   Worker (async coroutine)      │
│   • Retry loop (max 3)          │
│   • Exponential backoff         │
│   • Error handling              │
└─────────┬───────────────────────┘
          │
          ▼
┌─────────────────────────────────┐
│      RunRepository              │
│  • save_run_async()             │
└─────────┬───────────────────────┘
          │
          ├──► SQLite/Postgres (metadata)
          │    INSERT INTO runs (...)
          │
          └──► DuckDB (output rows)
               INSERT INTO run_outputs (...)

Data Transformation

# Input (API request)
{
  "values": [0.5, 2, 16],
  "microorganism": "B_ESCHR_COLI",
  "antimicrobial": "CIP",
  "guideline": "EUCAST 2025"
}

# Transformed for persistence
{
  "run_type": "sir",
  "input_data": {
    "values": [0.5, 2, 16],
    "microorganism": "B_ESCHR_COLI",
    "antimicrobial": "CIP",
    "guideline": "EUCAST 2025"
  },
  "output_data": {
    "results": ["S", "S", "R"]
  },
  "row_count": 3,
  "created_at": "2026-02-15T12:00:00Z",
  "metadata": {
    "source": "lab_system",
    "batch_id": "2026-02-15-001"
  }
}

# Stored in database
runs table:
  id | run_type | row_count | created_at | metadata_json

run_outputs table (DuckDB):
  run_id | row_index | input_value | sir_result
  123    | 0         | 0.5         | S
  123    | 1         | 2.0         | S
  123    | 2         | 16.0        | R

Data Refresh Flow

LOINC Refresh Example

External Source
    │
    ├─► Download Loinc.csv (200 MB)
    │   from loinc.org
    ▼
┌─────────────────────────────────┐
│ scripts/refresh_loinc.py         │
│  • Parse Loinc.csv              │
│  • Filter DRUG/TOX classes      │
│  • Normalize names              │
└─────────┬───────────────────────┘
          │
          ▼
┌─────────────────────────────────┐
│ Load antimicrobials.ndjson       │
│  • Read existing data           │
└─────────┬───────────────────────┘
          │
          ▼
┌─────────────────────────────────┐
│ Fuzzy Match                      │
│  • Normalize both sides         │
│  • Match by name similarity     │
│  • Add LOINC codes              │
└─────────┬───────────────────────┘
          │
          ▼
┌─────────────────────────────────┐
│ Write antimicrobials.ndjson      │
│  • Preserve existing data       │
│  • Add 'loinc' field            │
└─────────┬───────────────────────┘
          │
          ▼
Run Data Pipeline
    └─► Rebuild snapshots

Error and Dead-Letter Flow

Queue Worker
    │
    │ save_run_async(run_data)
    ▼
┌─────────────────────────────────┐
│  Try: Save to database          │
└─────────┬───────────────────────┘
          │
          ├─► Success
          │   └─► Increment counter
          │
          ├─► Retriable Error (transient)
          │   └─► Retry with backoff
          │       ├─► Retry 1 (50ms delay)
          │       ├─► Retry 2 (100ms delay)
          │       └─► Retry 3 (200ms delay)
          │
          └─► Permanent Failure
              └─► Save to Dead Letter Queue
                  │
                  ▼
┌─────────────────────────────────┐
│  dead_letters table              │
│  • Redact sensitive data        │
│  • Truncate large payloads      │
│  • Store error message          │
│  • Add timestamp                │
└─────────────────────────────────┘

Data Transformations

Common Transform Patterns

1. Type Coercion

df.with_columns([
    pl.col("numeric_field").cast(pl.Float64, strict=False),
    pl.col("date_field").str.strptime(pl.Date, "%Y-%m-%d"),
    pl.col("boolean_field").cast(pl.Boolean)
])

2. Null Handling

df.with_columns([
    pl.col("field").fill_null("DEFAULT_VALUE"),
    pl.when(pl.col("field").is_null())
      .then(pl.lit("N/A"))
      .otherwise(pl.col("field"))
      .alias("field_safe")
])

3. String Normalization

df.with_columns([
    pl.col("name")
      .str.to_uppercase()
      .str.strip_chars()
      .str.replace_all(r"\s+", " ")
      .alias("name_normalized")
])

4. Computed Fields

df.with_columns([
    (pl.col("genus") + " " + pl.col("species")).alias("fullname"),
    (pl.col("resistance_count") / pl.col("total_count") * 100).alias("resistance_pct")
])

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Data Flow

Table of Contents

Overview

High-Level Architecture

Reference Data Pipeline

Pipeline Stages

Detailed Flow

1. External Sources → Import

2. NDJSON → Transforms

3. Transform Examples

Runtime Data Flow

API Request → Response Flow

SIR Interpretation Flow

Detailed Sequence

Code Flow

Antibiogram Generation Flow

Run Persistence Flow

Async Queue Architecture

Data Transformation

Data Refresh Flow

LOINC Refresh Example

Error and Dead-Letter Flow

Data Transformations

Common Transform Patterns

1. Type Coercion

2. Null Handling

3. String Normalization

4. Computed Fields

Related Documentation

FilesExpand file tree

data_flow.md

Latest commit

History

data_flow.md

File metadata and controls

Data Flow

Table of Contents

Overview

High-Level Architecture

Reference Data Pipeline

Pipeline Stages

Detailed Flow

1. External Sources → Import

2. NDJSON → Transforms

3. Transform Examples

Runtime Data Flow

API Request → Response Flow

SIR Interpretation Flow

Detailed Sequence

Code Flow

Antibiogram Generation Flow

Run Persistence Flow

Async Queue Architecture

Data Transformation

Data Refresh Flow

LOINC Refresh Example

Error and Dead-Letter Flow

Data Transformations

Common Transform Patterns

1. Type Coercion

2. Null Handling

3. String Normalization

4. Computed Fields

Related Documentation