Workflow I/O Contracts#

Helpers for attaching standardised inputnode / outputnode IdentityInterface boundary nodes to a Nipype workflow, so a meta-workflow can wire stage-to-stage through a stable named contract instead of reaching into a sub-workflow’s internal node names.

Pattern#

A sub-workflow’s builder:

  1. calls attach_inputnode() with its runtime-overridable input fields (seeding build-time defaults so standalone runs resolve statically), and fans each field out to the consuming nodes with fan_out();

  2. calls attach_outputnode() for its stable outputs and connects the producing nodes into it.

A meta-workflow (e.g. full_pipeline) then connects only upstream.outputnode.<field> → downstream.inputnode.<field> — no node-name introspection of nested sub-workflows. An incoming edge into inputnode.<field> overrides the build-time default (standard Nipype behaviour), which is what lets one builder serve both standalone and embedded runs.

from thesis.core.contracts import attach_inputnode, attach_outputnode, fan_out

inputnode = attach_inputnode(wf, ["t1_brain", "dwi_mask"])
fan_out(wf, inputnode, "t1_brain", [(validator, "reference_image")])

outputnode = attach_outputnode(wf, ["fdt_paths"])
wf.connect(density_writer, "out_file", outputnode, "fdt_paths")

Module reference#

Workflow I/O contract helpers.

Small factory helpers that attach standardized inputnode / outputnode IdentityInterface boundary nodes to a Nipype workflow. They let a meta-workflow wire to a stable named contract instead of reaching into a sub-workflow’s internal node names.

The pattern: a sub-workflow’s builder calls attach_inputnode() with the runtime-overridable input fields (defaulted from config so standalone runs resolve statically), then connects inputnode.<field> to each internal consumer via fan_out(). It calls attach_outputnode() for stable outputs and connects producers into it. A meta-workflow then connects only upstream.outputnode.<f> -> downstream.inputnode.<f>.

See docs/superpowers/specs/2026-05-26-full-pipeline-contracts-design.md.

thesis.core.contracts.ROI_OUTPUT_FIELDS: tuple[str, str, str, str] = ('roi_seed', 'roi_stop', 'roi_avoid', 'roi_target')#

ROI terminus output fields published on a tractography workflow outputnode.

Both backends re-expose their final ROI source (seed / stop / avoid / target masks) under these stable contract names so a meta-workflow (e.g. the *_synthseg ROI validation) can wire to them directly instead of scanning wf._graph for a (possibly renamed) internal node name.

thesis.core.contracts.MRTRIX3_ROI_OUTPUT_FIELDS: tuple[str, str, str, str] = ('roi_seed', 'roi_stop', 'roi_avoid', 'roi_target')#

Backend-specific aliases (identical fields; kept for call-site clarity).

thesis.core.contracts.attach_inputnode(wf, fields, defaults=None, *, name='inputnode')[source]#

Create an IdentityInterface input contract node on wf.

Parameters:

  • wf (Workflow) – Workflow to attach the node to.

  • fields (Sequence[str]) – Input field names exposed by the contract.

  • defaults (Optional[Mapping[str, Any]]) – Optional field -> value map applied as build-time node.inputs.<field>. None values are skipped so the trait stays Undefined (a meta-workflow edge supplies it at run time). Keys not declared in fields raise ValueError.

  • name (str) – Node name (default "inputnode").

Return type:

Node

Returns:

The created nipype.Node.

thesis.core.contracts.attach_outputnode(wf, fields, *, name='outputnode')[source]#

Create an IdentityInterface output contract node on wf.

Parameters:

  • wf (Workflow) – Workflow to attach the node to.

  • fields (Sequence[str]) – Output field names exposed by the contract.

  • name (str) – Node name (default "outputnode").

Return type:

Node

Returns:

The created nipype.Node.

thesis.core.contracts.fan_out(wf, inputnode, field, targets)[source]#

Connect inputnode.<field> to each (node, port) in targets.

Captures the per-hemisphere / per-source fan-out that used to live in the meta-workflow as startswith() loops. The builder owns the target node handles, so no introspection is needed.

Parameters:

  • wf (Workflow) – Workflow owning both ends.

  • inputnode (Node) – The input contract node.

  • field (str) – Source field on inputnode.

  • targets (Iterable[tuple[Node, str]]) – Iterable of (consumer_node, consumer_port) pairs.

Return type:

None