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')
nirs4all.visualization.pipeline_diagram.BranchDiagram

alias of PipelineDiagram

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

Create DAG visualization for pipeline execution structure.

Renders a visual diagram showing the complete pipeline topology, including all steps, shapes, branches, and models.

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')}
edges: List[Tuple[str, str]]
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")
nodes: Dict[str, PipelineNode]
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: 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: object

Represents a node in the pipeline DAG.

id

Unique node identifier

Type:

str

step_index

Pipeline step index (1-based)

Type:

int

label

Display label for the node

Type:

str

node_type

Type of node (preprocessing, model, splitter, branch, merge, etc.)

Type:

str

shape_before

Dataset shape before this step (samples, processings, features)

Type:

Tuple[int, int, int] | None

shape_after

Dataset shape after this step

Type:

Tuple[int, int, int] | None

input_layout_shape

2D layout shape before step (samples, features)

Type:

Tuple[int, int] | None

output_layout_shape

2D layout shape after step (samples, features)

Type:

Tuple[int, int] | None

features_shape

List of 3D per-source shapes (samples, processings, features)

Type:

List[Tuple[int, int, int]] | None

branch_id

Branch ID if inside a branch (None if not)

Type:

int | None

branch_name

Branch name if inside a branch

Type:

str

substep_index

Index within a branch’s substeps

Type:

int | None

parent_ids

List of parent node IDs

Type:

List[str]

children_ids

List of child node IDs

Type:

List[str]

duration_ms

Execution duration in milliseconds (from trace)

Type:

float

metadata

Additional node metadata

Type:

Dict[str, Any]

branch_id: int | None = None
branch_name: str = ''
children_ids: List[str]
duration_ms: float = 0.0
features_shape: List[Tuple[int, int, int]] | None = None
id: str
input_layout_shape: Tuple[int, int] | None = None
label: str
metadata: Dict[str, Any]
node_type: str = 'preprocessing'
output_layout_shape: Tuple[int, int] | None = None
parent_ids: List[str]
shape_after: Tuple[int, int, int] | None = None
shape_before: Tuple[int, int, int] | None = None
step_index: int
substep_index: int | None = None
nirs4all.visualization.pipeline_diagram.plot_branch_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

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')
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')