luminarycloud.pipelines¶

Submodules¶

luminarycloud.pipelines.stages

Attributes¶

`ArgNamedVariableSet`	Deprecated alias for PP_NAMED_VARIABLE_SET_ID.
`PP_NAMED_VARIABLE_SET_ID`	A constant PipelineParameter that can be used in a PipelineArgs params list to add a Named Variable
`Stage`

Exceptions¶

StopRun

Raised by RunScript code to indicate that the pipeline run should stop intentionally.

Classes¶

`BoolPipelineParameter`	A Bool Pipeline Parameter can replace a hard-coded bool in Pipeline operator arguments to
`CreateGeometry`	Creates a new geometry by copying a base geometry and optionally mutating it by applying a named
`CreateGeometryOutputs`	The outputs of the CreateGeometry stage.
`CreateMesh`	Generates a Mesh from a Geometry.
`CreateMeshOutputs`	The outputs of the CreateMesh stage.
`CreateSimulation`	Creates and runs a Simulation.
`CreateSimulationOutputs`	The outputs of the CreateSimulation stage.
`FloatPipelineParameter`	A Float Pipeline Parameter can replace a hard-coded float in Pipeline operator arguments to
`FlowableIOSchema`	Typed representation of RunScript input/output schema.
`FlowableType`	Canonical flowable type identifiers.
`IntPipelineParameter`	An Int Pipeline Parameter can replace a hard-coded int in Pipeline operator arguments to
`Mesh`	DEPRECATED: Use CreateMesh instead.
`MeshOutputs`	The outputs of the Mesh stage.
`Pipeline`	A Pipeline is a reusable, parameterized workflow that automates large batches of work on the
`PipelineArgs`	A table of arguments with which one invokes a Pipeline to create a PipelineJob.
`PipelineOutputGeometry`	A representation of a Geometry in a Pipeline.
`PipelineOutputMesh`	A representation of a Mesh in a Pipeline.
`PipelineOutputSimulation`	A representation of a Simulation in a Pipeline.
`PipelineParameter`	Base class for all concrete PipelineParameters.
`ReadGeometry`	Reads a Geometry into the Pipeline and runs a geometry check.
`ReadGeometryOutputs`	The outputs of the ReadGeometry stage.
`ReadGeometryV1`	DEPRECATED: Use CreateGeometry or ReadGeometry instead.
`ReadMesh`	Reads a Mesh into the Pipeline.
`ReadMeshOutputs`	The outputs of the ReadMesh stage.
`RunScript`	RunScript is a stage that runs a user-provided Python function.
`RunScriptContext`	Context object passed to the user's function at runtime.
`Simulate`	DEPRECATED: Use CreateSimulation instead.
`SimulateOutputs`	The outputs of the Simulate stage.
`StringPipelineParameter`	A String Pipeline Parameter can replace a hard-coded string in Pipeline operator arguments to
`WaitForMesh`	Waits for a Mesh to be ready for simulation.
`WaitForMeshOutputs`	The outputs of the WaitForMesh stage.

Functions¶

stage(→ Callable[[Callable[Ellipsis, dict[str, ...)

Decorator for building a RunScript stage from a Python function.

Package Contents¶

exception StopRun¶: Raised by RunScript code to indicate that the pipeline run should stop intentionally.

class BoolPipelineParameter(name: str)¶

A Bool Pipeline Parameter can replace a hard-coded bool in Pipeline operator arguments to allow its value to be set when the Pipeline is invoked.

name¶

property type: str¶

class CreateGeometry(*, stage_name: str | None = None, base_geometry_id: str | luminarycloud.pipelines.parameters.StringPipelineParameter, geo_name: str | luminarycloud.pipelines.parameters.StringPipelineParameter | None = None)¶

Creates a new geometry by copying a base geometry and optionally mutating it by applying a named variable set (NVS). Also runs a geometry check and fails if the check produces errors.

Parameters:

base_geometry_idstr | StringPipelineParameter: The ID of the geometry to copy as a basis of the newly created one.
geo_namestr | StringPipelineParameter | None: The name to assign to the new geometry. If None, a default name will be used.

downstream_stages() → list[Stage]¶

inputs_dict() → dict[str, tuple[Stage, str]]¶

is_source() → bool¶

outputs¶

class CreateGeometryOutputs¶

The outputs of the CreateGeometry stage.

downstream_inputs() → list[luminarycloud.pipelines.flowables.PipelineInput]¶

geometry: luminarycloud.pipelines.flowables.PipelineOutputGeometry¶: The resultant Geometry.

class CreateMesh(*, stage_name: str | None = None, geometry: luminarycloud.pipelines.flowables.PipelineOutputGeometry, mesh_name: str | luminarycloud.pipelines.parameters.StringPipelineParameter | None = None, target_cv_count: int | luminarycloud.pipelines.parameters.IntPipelineParameter | None = None, mesh_gen_params: luminarycloud.meshing.MeshGenerationParams | None = None)¶

Generates a Mesh from a Geometry.

This is the basic stage for generating a minimal Mesh, a Mesh with a target number of control volumes, or a Mesh from arbitrary MeshGenerationParams.

Parameters:

geometryPipelineOutputGeometry: The Geometry to mesh.
mesh_namestr | StringPipelineParameter | None: The name to assign to the Mesh. If None, a default name will be used.
target_cv_countint | IntPipelineParameter | None: The target number of control volumes to generate. If None, a minimal mesh will be generated. Default: None
mesh_gen_paramsMeshGenerationParams | None: Mesh generation parameters. If provided, target_cv_count must be None.

downstream_stages() → list[Stage]¶

inputs_dict() → dict[str, tuple[Stage, str]]¶

is_source() → bool¶

outputs¶

class CreateMeshOutputs¶

The outputs of the CreateMesh stage.

downstream_inputs() → list[luminarycloud.pipelines.flowables.PipelineInput]¶

mesh: luminarycloud.pipelines.flowables.PipelineOutputMesh¶: The Mesh generated from the given Geometry.

class CreateSimulation(*, stage_name: str | None = None, mesh: luminarycloud.pipelines.flowables.PipelineOutputMesh, sim_name: str | luminarycloud.pipelines.parameters.StringPipelineParameter | None = None, sim_template_id: str | luminarycloud.pipelines.parameters.StringPipelineParameter, batch_processing: bool | luminarycloud.pipelines.parameters.BoolPipelineParameter = True, gpu_type: luminarycloud.enum.GPUType | str | luminarycloud.pipelines.parameters.StringPipelineParameter | None = None, gpu_count: int | luminarycloud.pipelines.parameters.IntPipelineParameter | None = None)¶

Creates and runs a Simulation.

Parameters:

meshPipelineOutputMesh: The Mesh to use for the Simulation.
sim_template_idstr | StringPipelineParameter: The ID of the SimulationTemplate to use for the Simulation.
sim_namestr | StringPipelineParameter | None: The name to assign to the Simulation. If None, a default name will be used.
batch_processingbool | BoolPipelineParameter: If True, the Simulation will run as a standard job. If False, the Simulation will run as a priority job. Default: True
gpu_typeGPUType | str | StringPipelineParameter | None: GPU type to use for the Simulation (e.g. GPUType.H100, "h100"). If None, the default GPU type will be used.
gpu_countint | IntPipelineParameter | None: Number of GPUs to use for the Simulation. Only relevant if gpu_type is specified. If set to 0 or omitted and gpu_type is specified, the number of GPUs will be automatically determined.

downstream_stages() → list[Stage]¶

inputs_dict() → dict[str, tuple[Stage, str]]¶

is_source() → bool¶

outputs¶

class CreateSimulationOutputs¶

The outputs of the CreateSimulation stage.

downstream_inputs() → list[luminarycloud.pipelines.flowables.PipelineInput]¶

simulation: luminarycloud.pipelines.flowables.PipelineOutputSimulation¶: The Simulation.

class FloatPipelineParameter(name: str)¶

A Float Pipeline Parameter can replace a hard-coded float in Pipeline operator arguments to allow its value to be set when the Pipeline is invoked.

name¶

property type: str¶

class FlowableIOSchema¶

Typed representation of RunScript input/output schema.

classmethod from_dict(data: Mapping[str, Mapping[str, FlowableType | str]]) → FlowableIOSchema¶

to_dict() → dict[str, dict[str, str]]¶

inputs: dict[str, FlowableType]¶

outputs: dict[str, FlowableType]¶

class FlowableType¶

Canonical flowable type identifiers.

GEOMETRY = 'Geometry'¶

MESH = 'Mesh'¶

SIMULATION = 'Simulation'¶

class IntPipelineParameter(name: str)¶

An Int Pipeline Parameter can replace a hard-coded int in Pipeline operator arguments to allow its value to be set when the Pipeline is invoked.

name¶

property type: str¶

class Mesh(*, stage_name: str | None = None, geometry: luminarycloud.pipelines.flowables.PipelineOutputGeometry, mesh_name: str | luminarycloud.pipelines.parameters.StringPipelineParameter | None = None, target_cv_count: int | luminarycloud.pipelines.parameters.IntPipelineParameter | None = None, mesh_gen_params: luminarycloud.meshing.MeshGenerationParams | None = None)¶

DEPRECATED: Use CreateMesh instead.

Generates a Mesh from a Geometry.

This is the basic stage for generating a minimal Mesh, a Mesh with a target number of control volumes, or a Mesh from arbitrary MeshGenerationParams.

Parameters:

geometryPipelineOutputGeometry: The Geometry to mesh.
mesh_namestr | StringPipelineParameter | None: The name to assign to the Mesh. If None, a default name will be used.
target_cv_countint | IntPipelineParameter | None: The target number of control volumes to generate. If None, a minimal mesh will be generated. Default: None
mesh_gen_paramsMeshGenerationParams | None: Mesh generation parameters. If provided, target_cv_count must be None.

downstream_stages() → list[Stage]¶

inputs_dict() → dict[str, tuple[Stage, str]]¶

is_source() → bool¶

outputs¶

class MeshOutputs¶

The outputs of the Mesh stage.

downstream_inputs() → list[luminarycloud.pipelines.flowables.PipelineInput]¶

mesh: luminarycloud.pipelines.flowables.PipelineOutputMesh¶: The Mesh generated from the given Geometry.

class Pipeline(stages: list[Stage])¶

A Pipeline is a reusable, parameterized workflow that automates large batches of work on the Luminary platform.

get_stage_id(stage: Stage) → str¶

pipeline_params() → set[PipelineParameter]¶

to_yaml() → str¶

stages¶

class PipelineArgs(params: list[luminarycloud.pipelines.core.PipelineParameter], args: list[list[PipelineArgValueType]])¶

A table of arguments with which one invokes a Pipeline to create a PipelineJob.

Each column of the PipelineArgs table is a PipelineParameter, and each row is an ordered list of values to substitute in for those parameters. The resulting PipelineJob will have one run per row. Every PipelineParameter of the Pipeline must be included in the PipelineArgs columns. The PP_NAMED_VARIABLE_SET_ID PipelineParameter may also be included.

column_for(param_name: str) → int¶

has_column_for(param_name: str) → bool¶

print_as_table() → None¶

params¶

rows¶

class PipelineOutputGeometry(owner: luminarycloud.pipelines.core.Stage, name: str)¶

A representation of a Geometry in a Pipeline.

downstream_inputs: list[PipelineInput] = []¶

name¶

owner¶

class PipelineOutputMesh(owner: luminarycloud.pipelines.core.Stage, name: str)¶

A representation of a Mesh in a Pipeline.

downstream_inputs: list[PipelineInput] = []¶

name¶

owner¶

class PipelineOutputSimulation(owner: luminarycloud.pipelines.core.Stage, name: str)¶

A representation of a Simulation in a Pipeline.

downstream_inputs: list[PipelineInput] = []¶

name¶

owner¶

class PipelineParameter(name: str)¶

Base class for all concrete PipelineParameters.

name¶

property type: str¶

class ReadGeometry(*, stage_name: str | None = None, geometry_id: str | luminarycloud.pipelines.parameters.StringPipelineParameter)¶

Reads a Geometry into the Pipeline and runs a geometry check.

Reads the geometry by ID without copying or modifying it.

Parameters:

geometry_idstr | StringPipelineParameter: The ID of the Geometry to retrieve.

downstream_stages() → list[Stage]¶

inputs_dict() → dict[str, tuple[Stage, str]]¶

is_source() → bool¶

outputs¶

class ReadGeometryOutputs¶

The outputs of the ReadGeometry stage.

downstream_inputs() → list[luminarycloud.pipelines.flowables.PipelineInput]¶

geometry: luminarycloud.pipelines.flowables.PipelineOutputGeometry¶: The resultant Geometry.

class ReadGeometryV1(*, stage_name: str | None = None, geometry_id: str | luminarycloud.pipelines.parameters.StringPipelineParameter, geo_name: str | luminarycloud.pipelines.parameters.StringPipelineParameter | None = None, use_geo_without_copying: bool | luminarycloud.pipelines.parameters.BoolPipelineParameter = False)¶

DEPRECATED: Use CreateGeometry or ReadGeometry instead.

Reads a Geometry into the Pipeline, optionally makes a copy of it, and optionally mutates that copy by applying a named variable set (NVS). Also runs a geometry check and fails if the check produces errors.

Parameters:

geometry_idstr | StringPipelineParameter

The ID of the Geometry to retrieve (and copy).

geo_namestr | StringPipelineParameter | None

The name to assign to the Geometry copy. If None, a default name will be used.

use_geo_without_copyingbool | BoolPipelineParameter

By default, this is False, meaning that each Geometry this stage references will be copied and the PipelineJob will actually operate on the copied Geometry. This is done for multiple reasons, one of which is so that a PipelineJob can be based on a single parametric Geometry which each run of the job modifies by applying a NamedVariableSet. That modification mutates the Geometry, so those runs can only happen in parallel without interfering with each other if they each operate on a different copy of the Geometry. (The second reason is discussed in the “Details” section below.)

However, if you’ve already prepared your Geometry in advance and you don’t want the PipelineJob to create copies, you can set this to True. In that case, the referenced Geometry will be used directly without being copied.

IMPORTANT: If you set this to True, you must ensure no two PipelineJobRuns operate on the same Geometry, i.e. no two PipelineArgs rows contain the same Geometry ID.

downstream_stages() → list[Stage]¶

inputs_dict() → dict[str, tuple[Stage, str]]¶

is_source() → bool¶

outputs¶

class ReadMesh(*, stage_name: str | None = None, mesh_id: str | luminarycloud.pipelines.parameters.StringPipelineParameter)¶

Reads a Mesh into the Pipeline.

Does not complete until the Mesh is ready for simulation. Fails if the meshing job completes with errors.

Parameters:

mesh_idstr | StringPipelineParameter: The ID of the Mesh to retrieve.

downstream_stages() → list[Stage]¶

inputs_dict() → dict[str, tuple[Stage, str]]¶

is_source() → bool¶

outputs¶

class ReadMeshOutputs¶

The outputs of the ReadMesh stage.

downstream_inputs() → list[luminarycloud.pipelines.flowables.PipelineInput]¶

mesh: luminarycloud.pipelines.flowables.PipelineOutputMesh¶: The Mesh read from the given mesh_id.

RunScript is a stage that runs a user-provided Python function.

All inputs, outputs, and params must be declared. Inputs and params get passed to the function at runtime, and output types are validated at runtime.

If the function specifies a parameter named context, it will be passed a RunScriptContext.

RunScript stages have a very limited runtime. They are not suitable for long-running operations.

While you can instantiate a RunScript stage directly, the usual way to construct one is to decorate a function with the @stage decorator.

Examples

If you’re using a pipeline to generate geometry variants, you can codify some validations on those variants to flag problems early.

pp_geo_id = lc.pipelines.StringPipelineParameter("geo_id")
pp_sim_template_id = lc.pipelines.StringPipelineParameter("sim_template_id")
pp_expected_num_volumes = lc.pipelines.IntPipelineParameter("expected_num_volumes")

create_geo = lc.pipelines.CreateGeometry(base_geometry_id=pp_geo_id)

@lc.pipelines.stage(
    inputs={"geometry": create_geo.outputs.geometry},
    outputs={"geometry": lc.pipelines.PipelineOutputGeometry},
    params={"expected_num_volumes": pp_expected_num_volumes},
)
def validate_geo(geometry: lc.Geometry, expected_num_volumes: int):
    _surfaces, volumes = geometry.list_entities()
    if len(volumes) != expected_num_volumes:
        raise lc.pipelines.StopRun(
            f"Geometry {geometry.id} has {len(volumes)} volumes, expected {expected_num_volumes}"
        )

    return {"geometry": geometry}

create_mesh = lc.pipelines.CreateMesh(geometry=validate_geo.outputs.geometry)

create_sim = lc.pipelines.CreateSimulation(
    sim_template_id=pp_sim_template_id,
    mesh=create_mesh.outputs.mesh,
)

pipeline = lc.pipelines.create_pipeline(
    name="Custom geo validations demo",
    stages=[create_geo, validate_geo, create_mesh, create_sim],
)

downstream_stages() → list[Stage]¶

inputs_dict() → dict[str, tuple[Stage, str]]¶

is_source() → bool¶

outputs¶

class RunScriptContext¶

Context object passed to the user’s function at runtime.

idempotency_key: str¶: A key that is guaranteed to be unique to the current RunScript task. Pipelines provide at-least-once execution, meaning that a task may be executed multiple times under certain failure conditions. This is useful to pass as a request_id for SDK calls that support that parameter for idempotency.

class Simulate(*, stage_name: str | None = None, mesh: luminarycloud.pipelines.flowables.PipelineOutputMesh, sim_name: str | luminarycloud.pipelines.parameters.StringPipelineParameter | None = None, sim_template_id: str | luminarycloud.pipelines.parameters.StringPipelineParameter, batch_processing: bool | luminarycloud.pipelines.parameters.BoolPipelineParameter = True, gpu_type: luminarycloud.enum.GPUType | str | luminarycloud.pipelines.parameters.StringPipelineParameter | None = None, gpu_count: int | luminarycloud.pipelines.parameters.IntPipelineParameter | None = None)¶

DEPRECATED: Use CreateSimulation instead.

Runs a Simulation.

Parameters:

meshPipelineOutputMesh: The Mesh to use for the Simulation.
sim_template_idstr | StringPipelineParameter: The ID of the SimulationTemplate to use for the Simulation.
sim_namestr | StringPipelineParameter | None: The name to assign to the Simulation. If None, a default name will be used.
batch_processingbool | BoolPipelineParameter: If True, the Simulation will run as a standard job. If False, the Simulation will run as a priority job. Default: True
gpu_typeGPUType | str | StringPipelineParameter | None: GPU type to use for the Simulation (e.g. GPUType.H100, "h100"). If None, the default GPU type will be used.
gpu_countint | IntPipelineParameter | None: Number of GPUs to use for the Simulation. Only relevant if gpu_type is specified. If set to 0 or omitted and gpu_type is specified, the number of GPUs will be automatically determined.

downstream_stages() → list[Stage]¶

inputs_dict() → dict[str, tuple[Stage, str]]¶

is_source() → bool¶

outputs¶

class SimulateOutputs¶

The outputs of the Simulate stage.

downstream_inputs() → list[luminarycloud.pipelines.flowables.PipelineInput]¶

simulation: luminarycloud.pipelines.flowables.PipelineOutputSimulation¶: The Simulation.

class StringPipelineParameter(name: str)¶

A String Pipeline Parameter can replace a hard-coded string in Pipeline operator arguments to allow its value to be set when the Pipeline is invoked.

name¶

property type: str¶

class WaitForMesh(*, stage_name: str | None = None, mesh: luminarycloud.pipelines.flowables.PipelineOutputMesh)¶

Waits for a Mesh to be ready for simulation.

This is useful if you have a more complicated meshing setup than the Mesh stage can handle.

For example, you can use a RunScript stage to call lc.create_mesh() with arbitrary meshing parameters, then pass the resulting mesh to this stage to wait for it to be ready. Mesh creation can take minutes to hours, and RunScript has a short timeout, so you can’t wait for it to complete in the RunScript stage, but lc.create_mesh() returns immediately and mesh creation happens asynchronously, so you can wait for it to be ready in this stage.

Parameters:

meshPipelineOutputMesh: The Mesh to wait for.

downstream_stages() → list[Stage]¶

inputs_dict() → dict[str, tuple[Stage, str]]¶

is_source() → bool¶

outputs¶

class WaitForMeshOutputs¶

The outputs of the WaitForMesh stage.

downstream_inputs() → list[luminarycloud.pipelines.flowables.PipelineInput]¶

mesh: luminarycloud.pipelines.flowables.PipelineOutputMesh¶: The Mesh that will be waited for.

stage(*, inputs: dict[str, luminarycloud.pipelines.flowables.PipelineOutput] | None = None, outputs: dict[str, type[luminarycloud.pipelines.flowables.PipelineOutput]] | None = None, stage_name: str | None = None, params: dict[str, PipelineParameter | luminarycloud.pipelines.arguments.PipelineArgValueType] | None = None) → Callable[[Callable[Ellipsis, dict[str, Any] | None]], RunScript]¶

Decorator for building a RunScript stage from a Python function.

Examples

>>> @pipelines.stage(
...     inputs={"geometry": create_geo.outputs.geometry},
...     outputs={"geometry": pipelines.PipelineOutputGeometry},
... )
... def ensure_single_volume(geometry: lc.Geometry):
...     _, volumes = geometry.list_entities()
...     if len(volumes) != 1:
...         raise pipelines.StopRun("expected exactly one volume")
...     return {"geometry": geometry}

ArgNamedVariableSet¶: Deprecated alias for PP_NAMED_VARIABLE_SET_ID.

PP_NAMED_VARIABLE_SET_ID¶: A constant PipelineParameter that can be used in a PipelineArgs params list to add a Named Variable Set ID column to the args table. There must be zero or one of these in a PipelineArgs params list.

Stage¶