biolm.protocols.runtime package

Subpackages

biolm.protocols.runtime.backends package

Submodules

biolm.protocols.runtime.compile module

Protocol dict → ExecutionPlan → DataPipeline.

biolm.protocols.runtime.compile.build_execution_plan(protocol: dict, inputs: dict[str, Any], *, run_id: str | None = None) → ExecutionPlan: Compile protocol + inputs to an ExecutionPlan (no pipeline construction).

biolm.protocols.runtime.compile.compile_protocol(protocol: dict, inputs: dict[str, Any], *, output_dir: str | None = None, run_id: str | None = None, verbose: bool = False, **pipeline_kwargs: Any) → tuple[biolm.protocols.runtime.spec.ExecutionPlan, biolm.pipeline.data.DataPipeline]: Compile protocol and build a configured DataPipeline.

biolm.protocols.runtime.compile.compile_to_pipeline(protocol: dict, inputs: dict[str, Any], *, output_dir: str | None = None, run_id: str | None = None, verbose: bool = False, **pipeline_kwargs: Any) → tuple[biolm.protocols.runtime.spec.ExecutionPlan, biolm.pipeline.data.DataPipeline]: Compile protocol and build a configured DataPipeline.

biolm.protocols.runtime.context module

Input context for protocol expression evaluation.

class biolm.protocols.runtime.context.InputContext(inputs: dict[str, Any], protocol: dict | None = None)

Bases: object

Wraps user inputs for ${{ }} evaluation (inputs-only in v1).

as_eval_context() → dict[str, Any]: Context dict passed to expression_evaluator.

resolve_sequences() → list[str]: Resolve primary sequence list from inputs per Local Profile v1.

biolm.protocols.runtime.execute module

Orchestration entrypoint for local protocol execution.

class biolm.protocols.runtime.execute.LocalRunResult(dataframe: ~pandas.core.frame.DataFrame, records: list[dict[str, typing.Any]], plan: ~biolm.protocols.runtime.spec.ExecutionPlan, pipeline: ~biolm.pipeline.data.DataPipeline | None = None, run_id: str | None = None, output_selections: list[biolm.protocols.outputs.OutputSelection] = <factory>, selected_records: list[dict[str, typing.Any]] = <factory>)

Bases: object

Result of a local protocol run.

dataframe: DataFrame

output_selections: list[biolm.protocols.outputs.OutputSelection]

pipeline: DataPipeline | None = None

plan: ExecutionPlan

records: list[dict[str, Any]]

run_id: str | None = None

selected_records: list[dict[str, Any]]

to_seqframe(*, molecule_type: str | None = None, path: str | Path | None = None): Materialize results as a SeqFrame (requires seqframe extra).

biolm.protocols.runtime.execute.compile_protocol(protocol: dict, inputs: dict[str, Any], *, output_dir: str | None = None, run_id: str | None = None, verbose: bool = False, **pipeline_kwargs: Any) → tuple[biolm.protocols.runtime.spec.ExecutionPlan, biolm.pipeline.data.DataPipeline]: Compile protocol and build a configured DataPipeline.

biolm.protocols.runtime.execute.compile_to_pipeline(protocol: dict, inputs: dict[str, Any], *, output_dir: str | None = None, run_id: str | None = None, verbose: bool = False, **pipeline_kwargs: Any) → tuple[biolm.protocols.runtime.spec.ExecutionPlan, biolm.pipeline.data.DataPipeline]: Compile protocol and build a configured DataPipeline.

biolm.protocols.runtime.execute.run_local_protocol(protocol: dict, inputs: dict[str, Any] | None = None, *, output_dir: str | Path | None = None, verbose: bool = False, **kwargs: Any) → LocalRunResult: Run a protocol dict locally (main public API).

biolm.protocols.runtime.expressions module

Recursive evaluation of protocol template expressions.

biolm.protocols.runtime.expressions.evaluate_tree(obj: Any, context: dict[str, Any]) → Any: Recursively evaluate template expressions in dicts/lists.

biolm.protocols.runtime.expressions.evaluate_value(value: Any, context: dict[str, Any]) → Any: Evaluate ${{ }} templates in a scalar value.

biolm.protocols.runtime.mapping module

response_mapping → PredictionStage extraction specs.

class biolm.protocols.runtime.mapping.MappingSpec(column: str, kind: str, response_key: str, reduction: str | None = None, jmespath: str | None = None)

Bases: object

Parsed response_mapping entry.

column: str

jmespath: str | None = None

kind: str

reduction: str | None = None

response_key: str

biolm.protocols.runtime.mapping.mapping_to_stage_kwargs(action: str, response_mapping: dict[str, Any]) → dict[str, Any]: Convert protocol response_mapping to PredictionStage constructor kwargs.

biolm.protocols.runtime.mapping.parse_mapping_entry(column: str, value: Any) → MappingSpec: Parse one response_mapping key/value pair.

biolm.protocols.runtime.profile module

Local Protocol Profile v1 — supported-feature checks.

exception biolm.protocols.runtime.profile.UnsupportedProtocolFeature(message: str, *, feature: str | None = None)

Bases: ValueError

Raised when a protocol uses features outside Local Protocol Profile v1.

biolm.protocols.runtime.profile.check_supported(protocol: dict) → None

Validate protocol dict against Local Protocol Profile v1.

Raises:: UnsupportedProtocolFeature: if any unsupported construct is present.

biolm.protocols.runtime.results module

DataFrame → normalized record list.

biolm.protocols.runtime.results.dataframe_to_records(df: DataFrame) → list[dict[str, Any]]: Convert pipeline final DataFrame to list[dict] with sequence field.

biolm.protocols.runtime.spec module

Compiled protocol execution plan dataclasses.

class biolm.protocols.runtime.spec.CompiledTask(task_id: str, slug: str, action: str, depends_on: list[str] = <factory>, params: dict[str, typing.Any] = <factory>, response_mapping: dict[str, typing.Any] = <factory>, stage_kwargs: dict[str, typing.Any] = <factory>)

Bases: object

One ApiTask after compile-time resolution.

action: str

depends_on: list[str]

params: dict[str, Any]

response_mapping: dict[str, Any]

slug: str

stage_kwargs: dict[str, Any]

task_id: str

class biolm.protocols.runtime.spec.ExecutionPlan(protocol_name: str, sequences: list[str], tasks: list[biolm.protocols.runtime.spec.CompiledTask], inputs: dict[str, typing.Any] = <factory>, run_id: str | None = None)

Bases: object

Compile output: resolved sequences + ordered tasks.

inputs: dict[str, Any]

protocol_name: str

run_id: str | None = None

sequences: list[str]

tasks: list[biolm.protocols.runtime.spec.CompiledTask]

Module contents

Local Protocol Runtime — compile and execute Protocol YAML via biolm.pipeline.

Requires optional dependencies:

default

pip install "biolm[pipeline]"

class biolm.protocols.runtime.LocalRunResult(dataframe: ~pandas.core.frame.DataFrame, records: list[dict[str, typing.Any]], plan: ~biolm.protocols.runtime.spec.ExecutionPlan, pipeline: ~biolm.pipeline.data.DataPipeline | None = None, run_id: str | None = None, output_selections: list[biolm.protocols.outputs.OutputSelection] = <factory>, selected_records: list[dict[str, typing.Any]] = <factory>)

Bases: object

Result of a local protocol run.

dataframe: DataFrame

output_selections: list[biolm.protocols.outputs.OutputSelection]

pipeline: DataPipeline | None = None

plan: ExecutionPlan

records: list[dict[str, Any]]

run_id: str | None = None

selected_records: list[dict[str, Any]]

to_seqframe(*, molecule_type: str | None = None, path: str | Path | None = None): Materialize results as a SeqFrame (requires seqframe extra).

exception biolm.protocols.runtime.UnsupportedProtocolFeature(message: str, *, feature: str | None = None)

Bases: ValueError

Raised when a protocol uses features outside Local Protocol Profile v1.

biolm.protocols.runtime.compile_protocol(protocol: dict, inputs: dict[str, Any], *, output_dir: str | None = None, run_id: str | None = None, verbose: bool = False, **pipeline_kwargs: Any) → tuple[biolm.protocols.runtime.spec.ExecutionPlan, biolm.pipeline.data.DataPipeline]: Compile protocol and build a configured DataPipeline.

biolm.protocols.runtime.compile_to_pipeline(protocol: dict, inputs: dict[str, Any], *, output_dir: str | None = None, run_id: str | None = None, verbose: bool = False, **pipeline_kwargs: Any) → tuple[biolm.protocols.runtime.spec.ExecutionPlan, biolm.pipeline.data.DataPipeline]: Compile protocol and build a configured DataPipeline.

biolm.protocols.runtime.run_local_protocol(protocol: dict, inputs: dict[str, Any] | None = None, *, output_dir: str | Path | None = None, verbose: bool = False, **kwargs: Any) → LocalRunResult: Run a protocol dict locally (main public API).