Metadata-Version: 2.4
Name: tomviz-pipeline
Version: 3.2.1
Summary: Node-graph pipeline engine for tomographic data processing, usable standalone or embedded in applications such as tomviz.
Author-email: Alessandro Genova <alessandro.genova@kitware.com>
License-Expression: BSD-3-Clause
Project-URL: Homepage, https://www.tomviz.org/
Project-URL: Repository, https://github.com/OpenChemistry/tomviz-pipeline
Project-URL: Issues, https://github.com/OpenChemistry/tomviz-pipeline/issues
Keywords: tomography,tomviz,pipeline,volumetric,image-processing
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Science/Research
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Scientific/Engineering :: Image Processing
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: numpy
Requires-Dist: h5py
Requires-Dist: click
Requires-Dist: tqdm
Provides-Extra: vtk
Requires-Dist: vtk; extra == "vtk"
Provides-Extra: test
Requires-Dist: pytest; extra == "test"
Provides-Extra: dev
Requires-Dist: pytest; extra == "dev"
Requires-Dist: flake8; extra == "dev"
Dynamic: license-file

# tomviz-pipeline

A pure-Python node-graph pipeline engine for processing volumetric /
tomographic data. It originated as the external-execution runtime of
[tomviz](https://tomviz.org) 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

```bash
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):

```python
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:

```python
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`:

```python
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.

## 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 `Signal`s. 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.)

```python
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:

```python
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):

```python
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:

```python
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:

```python
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
```

## 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:

```python
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:

```python
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:

```python
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

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

## Development

```bash
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`.
