AFL.double_agent.Pipeline module#

Pipeline module for the double_agent package.

This module provides the Pipeline class, which serves as a container for defining and executing computational workflows. Pipelines are composed of PipelineOp objects that perform specific operations on data.

PipelineOps in this system follow these conventions: - Each operator takes input_variable and output_variable parameters in its constructor - Each operator implements a .calculate method that:

  • Maintains the same signature as the base class (PipelineOp)

  • Writes results to an xarray Dataset or DataArray

  • Returns self for method chaining

class AFL.double_agent.Pipeline.Pipeline(name: str | None = None, ops: List | None = None, description: str | None = None)#

Bases: PipelineContext

Container class for defining and executing computational workflows.

The Pipeline class serves as a framework for organizing and running sequences of operations (PipelineOps) on data. Each operation in the pipeline takes input data, performs a specific transformation, and produces output data that can be used by subsequent operations.

Parameters:

  • name (Optional[str], default=None) – Name of the pipeline. If None, defaults to “Pipeline”.

  • ops (Optional[List], default=None) – List of PipelineOp objects to initialize the pipeline with.

result#

Stores the final result after pipeline execution

Type:

Any

ops#

List of PipelineOp objects in the pipeline

Type:

List

graph#

NetworkX directed graph representation of the pipeline

Type:

nx.DiGraph

graph_edge_labels#

Edge labels for the pipeline graph visualization

Type:

Dict

append(op: PipelineOp) → Self#

Mirrors the behavior of python lists

calculate(dataset: Dataset, tiled_data=None, disable_progress_bar: bool = False) → Dataset#

Execute all operations in pipeline on provided dataset

clear_outputs()#

Clears the output dict of all PipelineOps in this pipeline

copy() → Self#
draw(figsize=(8, 8), edge_labels=True)#

Draw the pipeline as a graph

draw_plotly()#
extend(op: List[PipelineOp]) → Self#

Mirrors the behavior of python lists

input_variables() → List[str]#

Get the input variables needed for the pipeline to fully execute

Warning

This list will currently include “Generators” that aren’t required to be in the pipeline. These will hopefully be distinguishable by having Generator in the name, but this isn’t enforced at the moment.

make_graph()#

Build a networkx graph representation of this pipeline

output_variables() → List[str]#

Get the outputs variables of the pipeline

print() → None#

Print a summary of the pipeline

print_code() → None#

String representation of approximate code to generate this pipeline

Run this method to produce a string of Python code that should recreate this Pipeline.

static read_json(filename: str)#

Read pipeline from json file on disk

Usage#

`python from AFL.double_agent.Pipeline import Pipeline pipeline1 = Pipeline.read_json('pickled_pipeline.pkl') ``

static read_pkl(filename: str)#

Read pipeline from pickle file on disk

Warning

Please use the read_json and write_json methods. The pickle methods are insecure and prone to errors.

Usage#

`python from AFL.double_agent.Pipeline import Pipeline pipeline1 = Pipeline.read('pickled_pipeline.pkl') ``

search(name: str, contains: bool = False) → PipelineOp | None#
write_json(filename: str, overwrite=False, description: str | None = None)#

Write pipeline to disk as a JSON

Parameters:

  • filename (str) – Filename or filepath to be written

  • overwrite (bool, default=False) – Whether to overwrite an existing file

  • description (str, optional) – A descriptive text about the pipeline’s purpose and functionality

write_pkl(filename: str)#

Write pipeline to disk as a pkl

Warning

Please use the read_json and write_json methods. The pickle methods are insecure and prone to errors.

Parameters:

filename (str) – Filename or filepath to be written