nirs4all.visualization.pipeline_diagram module
Pipeline Diagram - DAG visualization for pipeline execution structure.
This module provides visualization tools for displaying the complete pipeline structure as a directed acyclic graph (DAG).
The diagram shows: - All pipeline steps with operator names - Dataset shape at each step (samples × processings × features) - Branching and merging points - Model training steps - Cross-validation splitters
Shape notation: S×P×F - S = samples - P = processings (preprocessing views) - F = features (wavelengths/columns)
Example
>>> from nirs4all.visualization.pipeline_diagram import PipelineDiagram
>>> diagram = PipelineDiagram(pipeline_steps, predictions)
>>> fig = diagram.render()
>>> fig.savefig('pipeline_diagram.png')
Note
Maintenance status (debt campaign 2026-06-05)
The internal DAG model in this module is slated to be replaced by the
dag-ml engine (separate repository). Do not invest in decomposing or
deep-testing this file; keep changes to bug fixes until the replacement lands.
- class nirs4all.visualization.pipeline_diagram.PipelineDiagram(pipeline_steps: list[Any] | None = None, predictions: Any = None, execution_trace: Any = None, config: dict[str, Any] | None = None)[source]
Bases:
objectCreate DAG visualization for pipeline execution structure.
Renders a visual diagram showing the complete pipeline topology, including all steps, shapes, branches, and models.
- Variables:
pipeline_steps – List of pipeline step definitions
predictions – Optional Predictions object with execution data
execution_trace – Optional ExecutionTrace with actual runtime shapes
config – Optional dict for customization
- NODE_STYLES = {'branch': ('#E0F2F1', '#00796B'), 'concat_transform': ('#F3E5F5', '#7B1FA2'), 'default': ('#ECEFF1', '#455A64'), 'feature_augmentation': ('#E0F2F1', '#00796B'), 'input': ('#FAFAFA', '#616161'), 'merge': ('#E0F2F1', '#00796B'), 'merge_sources': ('#E0F2F1', '#00796B'), 'model': ('#FFEBEE', '#D32F2F'), 'output': ('#FAFAFA', '#616161'), 'preprocessing': ('#E3F2FD', '#1976D2'), 'sample_augmentation': ('#E8F5E9', '#388E3C'), 'source_branch': ('#E0F2F1', '#00796B'), 'splitter': ('#F3E5F5', '#7B1FA2'), 'y_processing': ('#FFF8E1', '#FFA000')}
- classmethod from_trace(execution_trace: Any, predictions: Any = None, config: dict[str, Any] | None = None) PipelineDiagram[source]
Create a PipelineDiagram from an ExecutionTrace.
This builds the diagram using actual runtime data including measured shapes at each step.
- Parameters:
execution_trace – ExecutionTrace object from pipeline execution
predictions – Optional Predictions object to enrich nodes with scores
config – Optional configuration dict
- Returns:
PipelineDiagram instance ready for rendering
Example
>>> from nirs4all.visualization import PipelineDiagram >>> diagram = PipelineDiagram.from_trace(trace) >>> fig = diagram.render(title="Execution Trace")
- render(show_shapes: bool | None = None, figsize: tuple[float, float] | None = None, title: str | None = None, initial_shape: tuple[int, int, int] | None = None) Figure[source]
Render the pipeline diagram.
- Parameters:
show_shapes – Override config’s show_shapes setting
figsize – Override figure size
title – Optional title for the diagram
initial_shape – Initial dataset shape (samples, processings, features)
- Returns:
matplotlib Figure object
- class nirs4all.visualization.pipeline_diagram.PipelineNode(id: str, step_index: int, label: str, node_type: str = 'preprocessing', shape_before: tuple[int, int, int] | None=None, shape_after: tuple[int, int, int] | None=None, input_layout_shape: tuple[int, int] | None=None, output_layout_shape: tuple[int, int] | None=None, features_shape: list[tuple[int, int, int]] | None=None, branch_id: int | None = None, branch_name: str = '', substep_index: int | None = None, parent_ids: list[str] = <factory>, children_ids: list[str] = <factory>, duration_ms: float = 0.0, metadata: dict[str, ~typing.Any]=<factory>)[source]
Bases:
objectRepresents a node in the pipeline DAG.
- Variables:
id (str) – Unique node identifier
step_index (int) – Pipeline step index (1-based)
label (str) – Display label for the node
node_type (str) – Type of node (preprocessing, model, splitter, branch, merge, etc.)
shape_before (tuple[int, int, int] | None) – Dataset shape before this step (samples, processings, features)
shape_after (tuple[int, int, int] | None) – Dataset shape after this step
input_layout_shape (tuple[int, int] | None) – 2D layout shape before step (samples, features)
output_layout_shape (tuple[int, int] | None) – 2D layout shape after step (samples, features)
features_shape (list[tuple[int, int, int]] | None) – List of 3D per-source shapes (samples, processings, features)
branch_id (int | None) – Branch ID if inside a branch (None if not)
branch_name (str) – Branch name if inside a branch
substep_index (int | None) – Index within a branch’s substeps
duration_ms (float) – Execution duration in milliseconds (from trace)
- nirs4all.visualization.pipeline_diagram.plot_pipeline_diagram(pipeline_steps: list[Any] | None = None, predictions: Any = None, show_shapes: bool = True, figsize: tuple[float, float] | None = None, title: str | None = None, initial_shape: tuple[int, int, int] | None = None, config: dict[str, Any] | None = None, execution_trace: Any = None) Figure[source]
Convenience function to create a pipeline diagram.
- Parameters:
pipeline_steps – List of pipeline step definitions
predictions – Optional Predictions object with execution data
show_shapes – Whether to show shape info in nodes
figsize – Figure size tuple
title – Optional title for the diagram
initial_shape – Initial dataset shape (samples, processings, features)
config – Additional configuration dict
execution_trace – Optional ExecutionTrace object
- Returns:
matplotlib Figure object
Example
>>> from nirs4all.visualization.pipeline_diagram import plot_pipeline_diagram >>> fig = plot_pipeline_diagram(pipeline, initial_shape=(189, 1, 2151)) >>> fig.savefig('pipeline_diagram.png')