tomviz-pipeline

A pure-Python node-graph pipeline engine for processing volumetric / tomographic data. It originated as the external-execution runtime of tomviz and is being developed as a standalone library so that other applications can build, run, and persist data processing pipelines without depending on the tomviz application, Qt, or VTK.

Layout

The package is layered:

• tomviz_pipeline.core — a generic, dependency-free (stdlib only) pipeline engine: Pipeline, Node (SourceNode / TransformNode / SinkNode / SinkGroupNode), typed InputPort / OutputPort (PassthroughOutputPort), Link, PortData, dirty tracking (New / Stale / Current), execution planning, blocking (DefaultExecutor) and threaded (ThreadedExecutor) pipeline executors, ExecutionFuture, thread-safe Signal / EventQueue observers, and a per-node NodeExecutor abstraction (in-process vs. out-of-process node execution).
• The rest of tomviz_pipeline — the tomviz-flavored layer: the numpy Dataset model (plus pure-Python Table and Molecule payloads for the corresponding port types), EMD/HDF5 I/O, schema-v2 state files (.tvsm JSON and .tvh5 HDF5 containers), the built-in source/transform/sink nodes, kernel authoring APIs, progress reporting channels, the ExternalNodeExecutor (run a node under a different Python environment), the batch runner, and the tomviz-pipeline CLI.

Applications that just need a pipeline engine can depend on tomviz_pipeline.core alone; tomviz uses the whole package.

Installation

pip install tomviz-pipeline

Requires Python 3.9+. The runtime dependencies are numpy, h5py, click and tqdm. The optional tomviz-pipeline[vtk] extra lets state files carry tables built directly with VTK by kernel scripts; kernel scripts bring their own scientific dependencies (scipy, tomopy, ...).

Quick start

A scriptable node is built from exactly two artifacts: a definition (the interface — ports, parameters, label; the same JSON vocabulary as tomviz's .json sidecar files) and a kernel (the compute — the same class a tomviz kernel script contains, so anyone who has written one already knows how):

import numpy as np

from tomviz_pipeline import Pipeline, ScriptableNode
from tomviz_pipeline.dataset import Dataset
from tomviz_pipeline.kernels import SourceKernel, TransformKernel

class ConstantVolume(SourceKernel):
    def produce(self, value=1.0, side=32):
        arr = np.full((side, side, side), value, dtype=np.float32)
        return {'volume': Dataset({'Scalars': arr}, active='Scalars')}

class Multiply(TransformKernel):
    def transform(self, inputs, factor=2.0):
        ds = inputs['volume']
        return {'volume': ds.apply_to_each_scalar_array(
            lambda a: a * factor)}

CONSTANT_VOLUME = {
    'schemaVersion': 2,
    'name': 'ConstantVolume',
    'outputs': [{'name': 'volume', 'type': 'ImageData'}],
    'parameters': [{'name': 'value', 'type': 'double', 'default': 1.0},
                   {'name': 'side', 'type': 'int', 'default': 32}],
}

MULTIPLY = {
    'schemaVersion': 2,
    'name': 'Multiply',
    'inputs':  [{'name': 'volume', 'type': 'ImageData'}],
    'outputs': [{'name': 'volume', 'type': 'ImageData'}],
    'parameters': [{'name': 'factor', 'type': 'double', 'default': 2.0}],
}

pipeline = Pipeline()
source = pipeline.add_node(ScriptableNode(CONSTANT_VOLUME,
                                          kernel=ConstantVolume))
scale = pipeline.add_node(ScriptableNode(MULTIPLY, kernel=Multiply))
pipeline.create_link(source.output_port('volume'),
                     scale.input_port('volume'))
source.set_parameters(value=21.0)

pipeline.execute()
print(scale.output_port('volume').data().payload.active_scalars.max())
# 42.0

Both constructor arguments are polymorphic, with one rule: strings are content, paths are files.

• definition: a dict, a JSON string, or a pathlib.Path to a .json file. The definition decides the node class: with "schemaVersion": 2, no inputs means a ScriptableSourceNode (source.python), otherwise a ScriptableTransformNode (transform.python). A definition without schemaVersion is a v1 definition, run by a LegacyScriptableTransformNode (transform.legacyPython): the transform(dataset, **params) scripts of the tomviz kernel catalog. Until 4.0, one that declares inputs or outputs without schemaVersion is still read as schema 2, with a DeprecationWarning.
• kernel: a kernel class you hold (schema 2 only), the source text of a script, or a pathlib.Path to a .py file.
• parameters (optional): initial values in place of the definition's defaults, e.g. parameters={'factor': 3.0}. Only parameters the definition declares are accepted. Unlike set_parameters() later on, this marks nothing stale: the node is not in a graph yet.

ScriptableNode(...) returns the right class, and isinstance(node, ScriptableNode) holds for all three. PythonNode is the former name of ScriptableNode and still works.

That makes the classic sidecar pair — and kernels whose imports only resolve in another Python environment — work without ever importing the kernel here:

from pathlib import Path
from tomviz_pipeline import ExternalNodeExecutor

recon = pipeline.add_node(ScriptableNode(Path('kernels/Reconstruct.json'),
                                         kernel=Path('kernels/Reconstruct.py')))
recon.set_parameters(iterations=100)
recon.node_executor = ExternalNodeExecutor('/envs/tomopy')

Notes on the kernel side:

• Kernels get self.progress, self.canceled, and self.completed for long-running work; a kernel sees deep-copied datasets and keyword parameters — no graph machinery — which is what lets the same class run unchanged inside the tomviz application, in this standalone runtime, and in an external environment.
• A class-bound kernel executes in-process directly. Whenever the node must be serialized (state files, external execution) the class is re-expressed as a script by capturing its source; for that the class body must be self-contained apart from numpy / tomviz_pipeline imports. Kernels given as script text or a .py path serialize as-is.
• Kernels also get self.state: a dict the engine preserves across executions of the node (each run gets a fresh instance, so plain attributes don't survive). It round-trips through external execution but is never written to state files. Keep its values JSON-friendly.
• A kernel may implement should_auto_execute(self, **parameters) -> bool for periodic execution: when node.auto_execute_enabled is set an application polls executor.should_auto_execute(node) every node.auto_execute_interval_seconds — through the node's executor, so an external node is asked inside its own environment — and re-executes the pipeline on True. The hook receives the node's parameters and shares self.state with produce / transform.
• A kernel can write back to its own parameters with self.set_parameter(name, value) (the name must be declared in the description; the value is coerced to the declared type). Updates are installed on the node when the method returns — quietly: nothing is marked stale and no re-execution is triggered, because the run that made the change is deemed to have consumed it — and surface as the node's parameters_updated(node, changed) signal so an application can refresh its parameter UI. The next run receives the new values; inside should_auto_execute, return True to run with them right away. Updates round-trip through external execution like self.state does.
• Older scripts that import tomviz.nodes (the historical spelling, tomviz.nodes.TransformNode) keep working: those names resolve to the kernel classes through a compatibility alias.

Every scriptable node exposes its definition and script: node.json_description and node.script read them. To edit a node in use, set node.script and call node.reconfigure_description(json_text). Both mark the node stale. A new definition keeps the parameter values it still declares with the same type and returns the names it reset to their defaults; one that would change the node's schema or ports raises ValueError, since ports are fixed once a node exists.

Parameters

Node configuration lives in a private parameters store (so names can never collide with node internals like label or state) and is changed through set_parameters(), which marks the node and everything downstream stale and emits parameters_applied:

scale.set_parameters(factor=3)      # scale (and downstream) now Stale
pipeline.execute()                  # re-runs just what's needed

pipeline.auto_execute = True        # optional: C++-style behavior where
scale.set_parameters(factor=4)      # applying parameters re-executes

Values a kernel writes back with self.set_parameter (see above) take the other door, apply_parameter_updates(): the store changes, nothing is marked stale, and only parameters_updated is emitted.

A parameter the definition declares without a default holds no value, and the kernel's own Python default applies. One with a data-default instead takes its value from the data on the node's primary input whenever the node runs: num-voxels-x, num-voxels-y and num-voxels-z are the input's size along that axis. The resolved value is passed to the kernel but never stored, so it follows the input from run to run and is not saved in state files; setting a value (anything but None) overrides it. resolve_data_default(spec, shape) computes the same value, for a UI that wants to show it:

# {"name": "Nrecon", "type": "int", "data-default": "num-voxels-y"}
recon.parameter('Nrecon')                     # None: the data decides
resolve_data_default(spec, (512, 256, 61))    # 256, what a run passes

Using it from an application

A pipeline embedded in an application must not block the UI thread. Install a ThreadedExecutor once and every execute() variant returns immediately with an ExecutionFuture; the plan runs on a worker thread.

Notifications are Signals. A plain connect(handler) is a direct connection — the handler runs on whichever thread emits, which during execution is the worker, so it must be thread-safe and must not touch UI. Connecting through an EventQueue defers delivery: emission only enqueues, and the handler runs when your UI thread drains the queue — the moral equivalent of Qt's queued connections, without Qt. (The optional second argument to connect() is a dispatcher — EventQueue here, AsyncioDispatcher for event-loop apps below.)

from tomviz_pipeline import EventQueue, ThreadedExecutor

# Fresh pipeline with the ConstantVolume / Multiply kernels from the
# quick start.
pipeline = Pipeline()
source = pipeline.add_node(ScriptableNode(CONSTANT_VOLUME,
                                          kernel=ConstantVolume))
scale = pipeline.add_node(ScriptableNode(MULTIPLY, kernel=Multiply))
pipeline.create_link(source.output_port('volume'),
                     scale.input_port('volume'))

pipeline.set_executor(ThreadedExecutor())

# Everything below runs on the thread that drains `events` — never on
# the worker.
events = EventQueue()
pipeline.executor.node_execution_started.connect(
    lambda node: print(f'started:  {node.label}'), events)
pipeline.executor.node_execution_finished.connect(
    lambda node, ok: print(f'finished: {node.label} ok={ok}'), events)
scale.progress_step_changed.connect(
    lambda node, step: print(f'progress: {step}'), events)
pipeline.execution_finished.connect(
    lambda future: print(f'done, succeeded={future.succeeded()}'), events)

future = pipeline.execute()          # returns immediately

# A real application drains the queue from its main loop (see the Qt
# sketch below). A minimal stand-in loop:
while not future.is_finished():
    events.process(block=True, timeout=0.05)
events.process()                     # deliver anything queued after finish

Interactive control, all cooperative and non-blocking:

future = pipeline.execute()          # user starts a run...
pipeline.cancel_execution()          # ...and changes their mind: stops at
future.wait(10)                      # the next node boundary; running
                                     # kernels observe self.canceled

pipeline.set_paused(True)            # batch edits without re-running
scale.set_parameters(factor=3.0)
pipeline.set_paused(False)           # un-pausing re-executes what's stale

pipeline.auto_execute = True         # or: every set_parameters() re-runs,
scale.set_parameters(factor=4.0)     # as in the tomviz application

pipeline.executor.cancel_and_wait()  # teardown: join the worker.
                                     # pipeline.clear() does this for you.

Calling execute() while a run is in flight never queues behind it: the new plan becomes pending, the in-flight run is canceled at its next node boundary, and the pending plan starts when the worker exits — the right semantics for "the user dragged the slider again".

In a Qt application the drain is a timer on the GUI thread (illustrative sketch):

class PipelinePanel(QWidget):
    def __init__(self, pipeline):
        ...
        self.events = EventQueue()
        pipeline.set_executor(ThreadedExecutor(sync_queue=self.events))
        pipeline.execution_finished.connect(self._on_finished,
                                            self.events)
        self._timer = QTimer(self, interval=16)   # ~60 Hz
        self._timer.timeout.connect(self.events.process)
        self._timer.start()

The optional sync_queue makes the worker wait, after each node, until the application thread has drained the queue — use it when per-node UI updates (e.g. re-rendering) must complete before the next node starts, mirroring the C++ tomviz executor's barrier.

In an asyncio application there is no drain timer to write: an AsyncioDispatcher plays the EventQueue's role, delivering handlers onto the event loop. Create it once, on the loop, and pass it to any number of connect() calls; plain callbacks run as loop callbacks, and coroutine-function handlers are scheduled as tasks. Await a run's completion by parking future.wait() on a thread-pool thread:

import asyncio

from tomviz_pipeline import AsyncioDispatcher

async def run_pipeline(pipeline):
    dispatcher = AsyncioDispatcher()        # bound to this event loop

    connection = pipeline.executor.node_execution_finished.connect(
        lambda node, ok: print(f'finished: {node.label} ok={ok}'),
        dispatcher)

    future = pipeline.execute()             # returns immediately
    await asyncio.to_thread(future.wait)    # yields until the run ends
    connection.disconnect()
    return future.succeeded()

scale.set_parameters(factor=5.0)
assert asyncio.run(run_pipeline(pipeline)) is True

Cancelling an asyncio task that is awaiting a run does not cancel the pipeline — awaiting is observation, not ownership. Couple them explicitly when you want to:

async def run_and_own(pipeline):
    future = pipeline.execute()
    try:
        await asyncio.to_thread(future.wait)
    except asyncio.CancelledError:
        pipeline.cancel_execution()   # propagate deliberately
        raise

What kernels print

An executor built with capture_output=True hands what each kernel prints (sys.stdout and sys.stderr) to its node's output_written(node, text) signal, in batches about a second apart and once more when the kernel returns, before the node's finished signals. The text still reaches the terminal as before.

pipeline.set_executor(ThreadedExecutor(capture_output=True))
scale.output_written.connect(
    lambda node, text: print(f'{node.label} printed {text!r}'),
    dispatcher)

An in-process kernel is captured by thread: sys.stdout and sys.stderr get stand-ins that route the executing thread's writes to its node and pass every other write straight through. A node run by ExternalNodeExecutor sends its output over the progress channel as output messages (the child CLI captures whenever it reports to a parent), and the parent emits the same signal. Not seen: output of C code writing to the file descriptors directly, and of threads or processes a kernel starts itself.

Output persistence

Each output port is either transient (port.persistent = False — its payload lives only while some consumer holds a handle; the planner re-runs the producer when the data is needed again) or persistent, in which case port.persistence_mode picks the medium: InMemory pins the payload on the port, OnDisk spills it to a temp cache file ($TOMVIZ_PORT_CACHE_DIR or the system temp dir) once the last handle drops and reloads it lazily via port.materialize(). port.data() is a non-loading peek. Sources default to persistent-InMemory; transform outputs follow the pipeline-wide default:

from tomviz_pipeline import PipelineSettings, TransformPersistenceDefault

PipelineSettings.instance().transform_persistence_default = \
    TransformPersistenceDefault.OnDisk   # keep intermediates out of RAM

Modes can be switched at runtime and round-trip through schema-v2 state files (persistent / persistenceMode keys, identical to the C++ implementation).

Graph nodes (application integration)

Kernels cover custom computation. Applications embedding the engine that need custom graph machinery — their own port layouts, payload types, serialization, or state handling — subclass the core graph classes directly and register them under their own type strings:

from tomviz_pipeline import (
    DefaultExecutor, NodeFactory, Pipeline, PortData, SourceNode,
    TransformNode,
)

class Constant(SourceNode):
    type_name = 'example.constant'

    def __init__(self, value=0):
        super().__init__()
        self._parameters['value'] = value
        self.add_output('output', 'ImageData')

    def execute(self):
        self.output_port('output').set_data(
            PortData(self.parameter('value'), 'ImageData'))
        return True

class Scale(TransformNode):
    type_name = 'example.scale'

    def __init__(self, factor=1):
        super().__init__()
        self._parameters['factor'] = factor
        self.add_input('input', 'ImageData')
        self.add_output('output', 'ImageData')

    def transform(self, inputs):
        value = inputs['input'].payload * self.parameter('factor')
        return {'output': PortData(value, 'ImageData')}

NodeFactory.register('example.constant', Constant)
NodeFactory.register('example.scale', Scale)

pipeline = Pipeline()
source = pipeline.add_node(Constant(21))
scale = pipeline.add_node(Scale(2))
pipeline.create_link(source.output_port('output'),
                     scale.input_port('input'))
DefaultExecutor(pipeline).execute()
print(scale.output_port('output').data().payload)  # 42

This is the layer the built-in tomviz node types (readers, crops, reconstructions, ...) are made of. Note the trade-off versus kernels: graph nodes speak PortData and manage their own ports, but only deserialize in processes where the application's classes are importable — a scriptable node round-trips anywhere because it travels with its script.

Saving and loading state files

Applications save their session with save_state, the counterpart of load_state. A .tvh5 file also embeds the data of the persistent output ports, so loading it does not recompute them; .tvsm (or .json) is the graph alone:

from tomviz_pipeline.state import load_state, read_state_json, save_state

pipeline = load_state('session.tvh5')
extra = {k: v for k, v in read_state_json('session.tvh5').items()
         if k in ('views', 'layouts')}
save_state('copy.tvh5', pipeline, extra)

extra holds the application's own top-level sections (views, layouts, ...), which the library writes without interpreting, as it does with each output port's metadata dict (color maps, active scalars) and the settings of sinks it has no class for. Relative reader paths are rewritten so they still name the same files from the new location. To build the document on one thread and write it on another, call build_state then write_state.

Running a state file

tomviz-pipeline -s pipeline.tvsm -o output/
tomviz-pipeline -s pipeline.tvsm -o output/ --input 'data/*.emd'

Development

git clone https://github.com/OpenChemistry/tomviz-pipeline
cd tomviz-pipeline
pip install -e .[dev]
flake8 --config setup.cfg src tests
pytest

Releases are built and uploaded to PyPI by the Publish to PyPI workflow when a GitHub release is published; bump __version__ in src/tomviz_pipeline/__init__.py (the single source of the package version) and tag v<version>.

License

BSD 3-Clause. See LICENSE.