pipeline.runner

PipelineRunner: compose an application from several pipelines.

A complicated application can be split into several smaller pipelines. Each
pipeline has its own attributes and results (one frame in the frontend), and the
runner controls when each pipeline may run and passes data between them.

- Every pipeline has an order level. Pipelines on the same level are independent
  and can run in any order; a pipeline can only run after all pipelines on lower
  levels have finished.
- Output ports of modules in one pipeline can be connected to input ports of
  modules in a pipeline on a higher level. The data is delivered as soon as the
  source pipeline finishes, so the UI schemas of the target modules are up to
  date before the target pipeline's attributes are requested.
- Re-running a pipeline marks every finished pipeline on a higher level as
  ``stale``; they must be run again.
- The runner definition is saved to a ``.runner`` file that refers to one
  ``.pipe`` file per pipeline, relative to the ``.runner`` file.

Flow control (loops, conditions) is only supported inside a pipeline, see
:mod:``flowControl``.

Classes

class pipeline.runner.RunnerConnection

Connection from an output port of a module in one pipeline to an input port of a module in another.

Inherits from:: BaseModel

Attributes:

source_pipeline: str = <ast.Call object at 0x0000021FE192AAA0>

source_module: str = <ast.Call object at 0x0000021FE192A950>

source_port: str = <ast.Call object at 0x0000021FE192A800>

target_pipeline: str = <ast.Call object at 0x0000021FE192A6B0>

target_module: str = <ast.Call object at 0x0000021FE192A560>

target_port: str = <ast.Call object at 0x0000021FE192A410>

class pipeline.runner.RunnerPipelineEntry

A pipeline entry of the .runner file.

Inherits from:: BaseModel

Attributes:

name: str = <ast.Call object at 0x0000021FE1929C60>

file: str = <ast.Call object at 0x0000021FE1929B10>

order: int = <ast.Call object at 0x0000021FE19299C0>

local_functions_path: str | None = <ast.Call object at 0x0000021FE1929810>

class pipeline.runner.RunnerData

Content of a .runner file.

Inherits from:: BaseModel

Attributes:

name: str = <ast.Call object at 0x0000021FE19295A0>

title: str | None = <ast.Call object at 0x0000021FE19293F0>

description: str | None = <ast.Call object at 0x0000021FE19291E0>

version: str | None = <ast.Call object at 0x0000021FE1928FD0>

pipelines: list[RunnerPipelineEntry] = <ast.Call object at 0x0000021FE1928DC0>

connections: list[RunnerConnection] = <ast.Call object at 0x0000021FE1928BB0>

class pipeline.runner.PipelineRunner

Run several pipelines as one application and pass data between them.

Examples

>>> runner = PipelineRunner(name="PileCapacity", title="桩基承载力计算")
>>> runner.add_pipeline(pile_pipe)       # order 1; the runner name is pile_pipe.app_name
>>> runner.add_pipeline(bores_pipe)      # order 2
>>> runner.add_pipeline(capacity_pipe)   # order 3
>>> runner.connect(create_pile.OutputPile >> cal_capacity.InputPile)
>>> runner.connect_ports("bores", "TablesToMultiProfile1D", "OutputMultiProfile1D",
...                      "capacity", "CalPileCapacity", "InputProfiles")
>>> runner.update_gdim_state(token=token, proj_id=proj_id)
>>> runner.run("pile", length=45)
>>> runner["bores"].get_attribute_info()                  # options come from "pile" outputs
>>> runner.run("bores", selected_bores=["ZK1"])
>>> runner.run("capacity")
>>> runner.save("pileCapacity.runner")                     # also writes pile.pipe, bores.pipe, ...
>>> runner = PipelineRunner.load("pileCapacity.runner")

Methods:

Initialize the PipelineRunner.

Parameters

namestr, default: “runner”: Name of the runner application.
titlestr | None, default: None: Title of the runner application.
descriptionstr | None, default: None: Description of the runner application.
versionstr | None, default: None: Version of the runner application.
workspacestr | Path | None, default: None: Working directory of the runner. Each pipeline uses the sub folder workspace/<pipeline name>. If None, the pipelines keep their own workspace.
verbosebool | None, default: None: Verbose setting forwarded to all pipelines. If None, the pipelines keep their own setting.

update_gdim_state() → None

Update the GDIM state of all pipelines.

The keyword arguments are the same as :meth:PipeLine.update_gdim_state, e.g. token, proj_id, proj_name, personal_space, space_name, tpl_id and host. They are also applied to pipelines added later.

add_pipeline(pipeline: PipeLine, order: int | None = None, pipe_file: str | None = None) → None

Add a pipeline to the runner.

The pipeline’s app_name is its name in the runner, so it must be unique in the runner and a legal file name on Windows and Linux. A module object that is already in another pipeline of the runner is rejected. Modules in different pipelines may share a module name.

Parameters

pipelinePipeLine: The pipeline to add. Pipelines with execution steps are not supported.
orderint | None, default: None: Order level of the pipeline. Pipelines on the same level are independent of each other. If None, one level higher than the highest existing level is used (1 for the first pipeline).
pipe_filestr | None, default: None: Path of the .pipe file relative to the .runner file, used by :meth:save. If None, <app_name>.pipe is used. The path must end with .pipe.

get_order(name: str) → int: Get the order level of a pipeline.

connect(links: Link | list[Link]) → None

Connect pipelines with links created by the >> operator between modules of different pipelines.

The pipelines of the modules are found automatically. If the modules of the two pipelines have the same module name, the >> operator cannot be used; use :meth:connect_ports instead.

Examples

>>> runner.connect(create_pile.OutputPile >> cal_capacity.InputPile)
>>> runner.connect(read.OutputTables >> to_profiles.InputTables | read.OutputTables >> merge.InputTables)

connect_ports(source_pipeline: str, source_module: str, source_port: str, target_pipeline: str, target_module: str, target_port: str) → None: Connect an output port of a module in one pipeline to an input port of a module in another pipeline.

status(name: str) → PipelineStatus: Get the run status of a pipeline: not_run, done, failed or stale.

can_run(name: str) → bool: Whether all pipelines on lower order levels have finished, so the pipeline can run.

run(name: str, verbose: bool | None = None, run_trace_file: str | Path | None = _UNSET) → PipelineResult

Run one pipeline and deliver its output data to the connected pipelines.

Parameters

namestr: Name of the pipeline to run.
verbosebool | None, default: None: Per-run verbose override passed to :meth:PipeLine.run.
run_trace_filestr | Path | None, default: unset: Per-run trace file passed to :meth:PipeLine.run. When omitted, the pipeline keeps its own run_trace_file. A trace is written only when verbose is True for this run. A bare filename is placed in the pipeline’s workspace. **attributes Pipeline attribute values to set before running.

Returns

Any

PipelineResult
    The result of the pipeline, according to its ``return_results_config``.

run_all(attributes: dict[str, dict[str, Any]] | None = None, verbose: bool | None = None, run_trace_file: str | Path | None = _UNSET) → dict[str, PipelineResult]

Run all pipelines by order level.

Returns

Any

dict[str, PipelineResult]
    Results keyed by pipeline name.

Attributes

verbosebool | None: Per-run verbose override passed to :meth:PipeLine.run.
run_trace_filestr | Path | None: Per-run trace file passed to every pipeline. When omitted, each pipeline keeps its own run_trace_file. A trace is written only when verbose is True. A bare filename is placed in each pipeline’s workspace. A path that includes a directory is one shared file, so each pipeline overwrites it and the last pipeline’s trace remains.

save(file: str | Path) → None

Save the runner to a .runner file and each pipeline to its .pipe file.

The .pipe files are written relative to the folder of the .runner file.

Parameters

filestr | Path: Path of the .runner file.

load(cls, file: str | Path, silent: bool = False) → PipelineRunner

Load a runner from a .runner file and the .pipe files it refers to.

Parameters

filestr | Path: Path of the .runner file.
silentbool, default: False: If True, suppresses messages while restoring the pipelines.

Properties:

workspace

verbose

llm_key

pipeline_names: Names of the pipelines sorted by order level, then by the order they were added.