Pipeline and runner API#
.. py:class:: Pipeline(graph: dict[~typing.Any, set[~typing.Any]] = NOTHING, name: str = ‘Unnamed Pipeline’, trace_events: dict[str, list[~modacor.dataclasses.trace_event.TraceEvent]] = NOTHING, authored_yaml: str | None = None, authored_spec: dict[str, ~typing.Any] | None = None, step_origins: dict[str, ~modacor.runner.pipeline_schema.StepOrigin] = NOTHING) :module: modacor.runner.pipeline
Pipeline nodes are assumed to be of type ProcessStep.
The underlying graph maps each node to the set of prerequisite nodes
that must complete before it can run.
.. py:method:: Pipeline.add(node: ~typing.Any, *predecessors: ~typing.Any) -> None :module: modacor.runner.pipeline
Add a node and its prerequisites to the pipeline graph.
This mirrors the commonly used ``TopologicalSorter.add`` shape while
keeping graph ownership inside ``Pipeline``.
.. py:method:: Pipeline.add_incoming_branch(branch: ~typing.Self, branching_node: ~modacor.dataclasses.process_step.ProcessStep) -> ~typing.Self :module: modacor.runner.pipeline
Add a pipeline as a branch whose outcome shall be combined with
the existing pipeline at `branching_node`.
This assumes that the branch to be added has a single exit point.
.. py:method:: Pipeline.add_outgoing_branch(branch: ~typing.Self, branching_node: ~modacor.dataclasses.process_step.ProcessStep) -> ~typing.Self :module: modacor.runner.pipeline
Add a pipeline as a branch whose input is based on the existing pipeline.
This assumes that the branch to be added has a single entry point.
.. py:method:: Pipeline.attach_tracer_event(node: ~modacor.dataclasses.process_step.ProcessStep, tracer: ~typing.Any | None, *, include_rendered_trace: bool = False, include_rendered_config: bool = False, rendered_format: str = ‘text/html’, chunk_identity: dict[str, ~typing.Any] | None = None) -> ~modacor.dataclasses.trace_event.TraceEvent :module: modacor.runner.pipeline
Create & attach a TraceEvent for `node`, using `tracer.events` if available.
- Always attaches a TraceEvent so the graph UI can show config/module info.
- Adds datasets diffs only if a matching tracer event for this step_id exists.
.. py:method:: Pipeline.create_scheduler() -> ~graphlib.TopologicalSorter :module: modacor.runner.pipeline
Create a fresh topological scheduler for the current graph.
.. py:method:: Pipeline.done(*nodes: ~typing.Any) -> None :module: modacor.runner.pipeline
Mark nodes as complete on the compatibility scheduler.
.. py:method:: Pipeline.from_dict(graph_dict: ~typing.Mapping[~modacor.dataclasses.process_step.ProcessStep, ~typing.Iterable[~modacor.dataclasses.process_step.ProcessStep]], name: str = ‘’) -> ~modacor.runner.pipeline.Pipeline :module: modacor.runner.pipeline :classmethod:
Instantiate a Pipeline from a mapping.
:Parameters: **graph_dict** -- Mapping of node -> iterable of prerequisite nodes.
.. rubric:: Notes
This is a low-level constructor mainly intended for internal use or
tests. Normal users should prefer `from_yaml_file` or `from_yaml`.
.. py:method:: Pipeline.from_spec(spec: dict, registry: ~modacor.runner.process_step_registry.ProcessStepRegistry | None = None) -> ~modacor.runner.pipeline.Pipeline :module: modacor.runner.pipeline :classmethod:
Build a Pipeline from a graph spec of the shape produced by `to_spec`.
Expected shape:
{
"name": "...",
"nodes": [
{"id": "...", "module": "...", "config": {...}},
...
],
"edges": [
{"from": "...", "to": "..."},
...
],
}
.. py:method:: Pipeline.from_yaml(yaml_string: str, registry: ~modacor.runner.process_step_registry.ProcessStepRegistry | None = None, *, max_expanded_steps: int | None = None) -> ~modacor.runner.pipeline.Pipeline :module: modacor.runner.pipeline :classmethod:
Instantiate a Pipeline from a YAML configuration string.
Expected YAML schema (keyed by step_id):
```yaml
name: my_pipeline
steps:
1:
module: PoissonUncertainties
requires_steps: []
configuration: {...}
"pu":
module: PoissonUncertainties
requires_steps: [1] # may be int or string
configuration: {...}
```
.. rubric:: Notes
* The keys under `steps` (`1`, `"pu"`, etc.) are treated as the
canonical `step_id`s and are normalized to `str`.
* `requires_steps` entries can be ints or strings; they are also
normalized to `str`.
* If a `step_id` field is present inside a step, it must match
the outer key (after string conversion), otherwise an error
is raised to avoid silent mismatches.
.. py:method:: Pipeline.from_yaml_file(yaml_file: ~pathlib.Path | str, registry: ~modacor.runner.process_step_registry.ProcessStepRegistry | None = None, *, max_expanded_steps: int | None = None) -> ~modacor.runner.pipeline.Pipeline :module: modacor.runner.pipeline :classmethod:
Instantiate a Pipeline from a YAML configuration file.
:Parameters: * **yaml_file** -- Path to the YAML file.
* **registry** -- Optional ProcessStepRegistry. If omitted, the global
DEFAULT_PROCESS_STEP_REGISTRY is used.
.. py:method:: Pipeline.get_ready() -> tuple[~typing.Any, …] :module: modacor.runner.pipeline
Return nodes currently ready from the compatibility scheduler.
.. py:method:: Pipeline.is_active() -> bool :module: modacor.runner.pipeline
Return whether the compatibility scheduler still has active work.
.. py:method:: Pipeline.prepare() -> None :module: modacor.runner.pipeline
Prepare a compatibility scheduler for manual step iteration.
.. py:method:: Pipeline.provenance(*, include_trace_events: bool = False) -> ~modacor.dataclasses.pipeline_provenance.PipelineProvenance :module: modacor.runner.pipeline
Return authored and expanded representations for persistent provenance.
.. py:method:: Pipeline.run(**kwargs) -> None :module: modacor.runner.pipeline
Run pipeline with simple topological scheduling.
Any keyword arguments are passed through to `ProcessStep.execute`.
.. py:method:: Pipeline.static_order() -> tuple[~typing.Any, …] :module: modacor.runner.pipeline
Return a static topological ordering using a fresh scheduler.
.. py:method:: Pipeline.to_dot(direction: str = ‘LR’, *, group_step_blocks: bool = True) -> str :module: modacor.runner.pipeline
Export the pipeline as a Graphviz DOT string for visualization.
Nodes are labeled with "<step_id>: <calling_name/module_name>".
:Parameters: * **direction** -- Graphviz rank direction, e.g. "LR" for left-to-right or "TB" for
top-to-bottom.
* **group_step_blocks** -- Group expanded nodes by their authored block and item. Pipelines
without expanded blocks render identically either way.
.. py:method:: Pipeline.to_drawio(direction: str = ‘LR’, *, group_step_blocks: bool = True, dot_executable: str = ‘dot’, pixels_per_inch: float = 96.0) -> str :module: modacor.runner.pipeline
Export the pipeline as editable, uncompressed draw.io XML.
Graphviz supplies the initial node and cluster layout. Expanded
``step_blocks`` become nested block and item containers, and every
pipeline node and dependency remains individually editable.
:Parameters: * **direction** -- Graphviz rank direction, e.g. ``LR`` or ``TB``.
* **group_step_blocks** -- Preserve expanded block and item grouping as draw.io containers.
* **dot_executable** -- Graphviz ``dot`` executable name or path.
* **pixels_per_inch** -- Conversion scale from Graphviz points to draw.io canvas units.
.. py:method:: Pipeline.to_mermaid(direction: str = ‘LR’, *, group_step_blocks: bool = True) -> str :module: modacor.runner.pipeline
Export the pipeline as a Mermaid flowchart definition.
:Parameters: * **direction** -- Mermaid direction: "LR" (left-right), "TB" (top-bottom), etc.
* **group_step_blocks** -- Group expanded nodes by their authored block and item. Pipelines
without expanded blocks render identically either way.
.. py:method:: Pipeline.to_spec() -> dict[str, ~typing.Any] :module: modacor.runner.pipeline
Export the pipeline to a JSON-serializable graph spec.
:returns: * *dict with structure*
* *{* -- "name": "<pipeline_name>",
"nodes": [
{
"id": "<step_id>",
"label": "<human readable label>",
"module": "<ProcessStep class name>",
"module_path": "<path to module>" or "",
"version": "<module version>" or "",
"config": {...} # current configuration dict
},
...
],
"edges": [
{"from": "<source_step_id>", "to": "<target_step_id>"},
...
],
* *}*
* *e.g.*
* *{*
* **"id"** (*"FL",*)
* **"label"** (*"Divide by relative flux",*)
* **"module"** (*"Divide",*)
* **"requires_steps"** (*["DC"],*)
* **"config"** (*{...},*)
* **"trace_events"** (*[*) -- {
"step_id": "FL",
"config_hash": "...",
"datasets": {
"sample.signal": {"diff": ["units","nan_signal"], "prev": {...}, "now": {...}}
}
}
* *]*
* *}*
* *Adds*
* *- requires_steps per node (derived from graph prereqs)*
* *- optional trace events (if Pipeline.trace_events is populated)*
* *- config_hash per node (stable)*
.. py:method:: Pipeline.to_yaml() -> str :module: modacor.runner.pipeline
Export the pipeline to a YAML string using the same schema
that `from_yaml` expects (keyed by step_id).
The result looks like:
```yaml
name: my_pipeline
steps:
1:
module: SomeStep
requires_steps: []
configuration: {...}
"pu":
module: OtherStep
requires_steps: [1]
configuration: {...}
```
.. py:function:: run_pipeline_job(pipeline: ~modacor.runner.pipeline.Pipeline | ~pathlib.Path | str, *, sources: ~modacor.io.io_sources.IoSources | None = None, sinks: ~modacor.io.io_sinks.IoSinks | None = None, processing_data: ~modacor.dataclasses.processing_data.ProcessingData | None = None, trace: bool = False, trace_watch: dict[str, list[str]] | None = None, tracer_kwargs: dict[str, ~typing.Any] | None = None, stop_after: str | None = None, selected_step_ids: set[str] | list[str] | tuple[str, …] | None = None, capture_partial_on_error: bool = False, chunk_spec: ~modacor.io.chunking.ChunkSpec | ~modacor.io.chunking.ProvisionalChunkSpec | None = None) -> ~modacor.runner.pipeline_runner.RunResult :module: modacor.runner
Execute a pipeline end-to-end.
This helper is intentionally shared between notebooks and CLI usage so both execution modes follow the exact same scheduler path.
.. py:class:: RunResult(processing_data: ~modacor.dataclasses.processing_data.ProcessingData, pipeline: ~modacor.runner.pipeline.Pipeline, tracer: ~modacor.debug.pipeline_tracer.PipelineTracer | None, step_durations: dict[str, float], executed_steps: list[str], stopped_after_step: str | None, chunk_spec: ~modacor.io.chunking.ChunkSpec | ~modacor.io.chunking.ProvisionalChunkSpec | None = None) :module: modacor.runner :canonical: modacor.runner.pipeline_runner.RunResult
Container returned by :func:run_pipeline_job.
.. py:class:: PipelineRunError(message: str, *, result: ~modacor.runner.pipeline_runner.RunResult, original_exception: Exception, failed_step_id: str | None) :module: modacor.runner :canonical: modacor.runner.pipeline_runner.PipelineRunError
Raised when a pipeline run fails after collecting partial run context.