luminarycloud.pipelines.stages¶

Classes¶

`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.
`Mesh`	DEPRECATED: Use CreateMesh instead.
`MeshOutputs`	The outputs of the Mesh stage.
`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.
`Simulate`	DEPRECATED: Use CreateSimulation instead.
`SimulateOutputs`	The outputs of the Simulate stage.
`WaitForMesh`	Waits for a Mesh to be ready for simulation.
`WaitForMeshOutputs`	The outputs of the WaitForMesh stage.

Package Contents¶

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 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 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.

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 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.