Dataset processor

The dataset processor component reads a common OpenProblems dataset and writes one or more task-specific AnnData files. It is the first component in the task pipeline and is responsible for splitting the data into the files your methods, control methods, and metrics expect.

Note

The task template already contains a dataset processor in src/data_processors/process_dataset/. Use it as your starting point and modify the config and script to match your task’s API.

Why use a dataset processor?

The processor is the single place where ground-truth labels are separated from method inputs. By writing a solution.h5ad that is never passed to methods, the pipeline guarantees that no method can access labels it is not supposed to see. Metrics receive the solution file independently, after a method has already produced its output.

How to build the processor

The dataset processor is a standard Viash component: a config.vsh.yaml that declares the interface and a script that implements the logic.

Step 1: Create the Viash config

Python

src/data_processors/process_dataset/config.vsh.yaml

__merge__: ../../api/comp_data_processor.yaml  # 1

name: process_dataset  # 2
namespace: data_processors

info:
  label: Process dataset
  summary: "Filter and split a common dataset into task-specific train, test, and solution files."

arguments:
  - name: "--input"
    __merge__: ../../api/file_common_dataset.yaml
    type: file
    required: true
  - name: "--output_train"
    __merge__: ../../api/file_train.yaml
    type: file
    required: true
    direction: output
  - name: "--output_test"
    __merge__: ../../api/file_test.yaml
    type: file
    required: true
    direction: output
  - name: "--output_solution"
    __merge__: ../../api/file_solution.yaml
    type: file
    required: true
    direction: output

engines:
  - type: docker
    image: openproblems/base_python:1.0.0
    setup:
      - type: python
        github:
          - "openproblems-bio/core#subdirectory=packages/python/openproblems"

runners:
  - type: executable
  - type: nextflow
    directives:
      label: [midtime, midmem, lowcpu]

R

src/data_processors/process_dataset/config.vsh.yaml

__merge__: ../../api/comp_data_processor.yaml  # 1

name: process_dataset  # 2
namespace: data_processors

info:
  label: Process dataset
  summary: "Filter and split a common dataset into task-specific train, test, and solution files."

arguments:
  - name: "--input"
    __merge__: ../../api/file_common_dataset.yaml
    type: file
    required: true
  - name: "--output_train"
    __merge__: ../../api/file_train.yaml
    type: file
    required: true
    direction: output
  - name: "--output_test"
    __merge__: ../../api/file_test.yaml
    type: file
    required: true
    direction: output
  - name: "--output_solution"
    __merge__: ../../api/file_solution.yaml
    type: file
    required: true
    direction: output

engines:
  - type: docker
    image: openproblems/base_python:1.0.0
    setup:
      - type: python
        github:
          - "openproblems-bio/core#subdirectory=packages/python/openproblems"

runners:
  - type: executable
  - type: nextflow
    directives:
      label: [midtime, midmem, lowcpu]

Annotations:

1. __merge__ imports the shared component interface from src/api/comp_data_processor.yaml. This provides the unit test and any shared arguments.
2. name must always be process_dataset for dataset processor components.
3. Add one output argument for each task-specific AnnData file your task requires. Update the __merge__ path for each to reference the matching file_*.yaml in src/api/.
4. The engines block selects the Docker base image. Add setup entries for any additional Python or R packages your script needs.
5. Keep both executable and nextflow runners so the component works in both local testing and the full Nextflow pipeline.

Step 2: Create the script

Python

src/data_processors/process_dataset/script.py

import anndata as ad
from openproblems.data import subset_h5ad_by_format

## VIASH START
par = {
    "input": "resources_test/common/cxg_mouse_pancreas_atlas/dataset.h5ad",
    "output_train": "output_train.h5ad",
    "output_test": "output_test.h5ad",
    "output_solution": "output_solution.h5ad",
}
## VIASH END

print("Reading input dataset", flush=True)
adata = ad.read_h5ad(par["input"])

# Split cells into train and test
is_test = adata.obs["batch"] == adata.obs["batch"].iloc[-1]
adata_train = adata[~is_test].copy()
adata_test = adata[is_test].copy()

print("Writing train output", flush=True)
out_train = subset_h5ad_by_format(adata_train, par, "output_train")
out_train.write_h5ad(par["output_train"], compression="gzip")

print("Writing test output", flush=True)
out_test = subset_h5ad_by_format(adata_test, par, "output_test")
out_test.write_h5ad(par["output_test"], compression="gzip")

print("Writing solution output", flush=True)
out_solution = subset_h5ad_by_format(adata_test, par, "output_solution")
out_solution.write_h5ad(par["output_solution"], compression="gzip")

print("Done!", flush=True)

R

src/data_processors/process_dataset/script.R

library(anndata)
library(openproblems)

## VIASH START
par <- list(
  input = "resources_test/common/cxg_mouse_pancreas_atlas/dataset.h5ad",
  output_train = "output_train.h5ad",
  output_test = "output_test.h5ad",
  output_solution = "output_solution.h5ad"
)
## VIASH END

cat("Reading input dataset\n")
adata <- read_h5ad(par$input)

# Split cells into train and test
is_test <- adata$obs$batch == tail(unique(adata$obs$batch), 1)
adata_train <- adata[!is_test, ]
adata_test <- adata[is_test, ]

cat("Writing train output\n")
out_train <- subset_h5ad_by_format(adata_train, par, "output_train")
out_train$write_h5ad(par$output_train, compression = "gzip")

cat("Writing test output\n")
out_test <- subset_h5ad_by_format(adata_test, par, "output_test")
out_test$write_h5ad(par$output_test, compression = "gzip")

cat("Writing solution output\n")
out_solution <- subset_h5ad_by_format(adata_test, par, "output_solution")
out_solution$write_h5ad(par$output_solution, compression = "gzip")

cat("Done!\n")

Tip

subset_h5ad_by_format is a helper from the openproblems package that automatically keeps only the slots declared in the matching file_*.yaml. This ensures your output files are consistent with the API and that no extra slots leak into downstream components.

Step 3: Create test resources

The test resource script runs the processor on a small dataset and writes the outputs to resources_test/. These files are used by the unit tests of all downstream components.

scripts/create_datasets/test_resources.sh

#!/bin/bash

set -eo pipefail

echo "Running dataset processor"

viash run src/data_processors/process_dataset/config.vsh.yaml -- \
  --input resources_test/common/cxg_mouse_pancreas_atlas/dataset.h5ad \
  --output_train resources_test/task_template/cxg_mouse_pancreas_atlas/train.h5ad \
  --output_test resources_test/task_template/cxg_mouse_pancreas_atlas/test.h5ad \
  --output_solution resources_test/task_template/cxg_mouse_pancreas_atlas/solution.h5ad

echo "Done!"

Run the script from the repository root:

bash scripts/create_datasets/test_resources.sh

Note

Update the --output_* argument names and paths to match the arguments defined in your config.vsh.yaml. The paths under resources_test/ must match the example fields in your file_*.yaml API files so that unit tests can find the generated files.