biolm.protocols package

BioLM Protocols — validation, hosted runs, and local execution.

class biolm.protocols.Protocol(yaml_path: str)

Bases: object

Load and validate BioLM protocol YAML files.

Args:

yaml_path: Path to a protocol YAML file. The file is loaded and

validated on construction; invalid YAML raises ValueError.

Use validate() as a classmethod to validate without instantiating.

execute(inputs: Dict[str, Any] | None = None)

Execute protocol locally with given inputs.

Args:

inputs: Input values for the protocol (optional, uses defaults from protocol if not provided).

Returns:

biolm.protocols.runtime.LocalRunResult with dataframe, records, and metadata.

Requires:

biolm[pipeline] optional dependencies.

classmethod fetch_by_id(protocol_id: str, api_key: str | None = None, base_url: str | None = None) → dict

Fetch a protocol from the platform by ID (synchronous wrapper).

Args:

protocol_id: Protocol ID to fetch api_key: Optional API key for authentication base_url: Optional base URL (defaults to BIOLMAI_BASE_API_URL)

Returns:

Protocol data as dict (same structure as YAML)

Raises:

FileNotFoundError: If protocol not found (404) PermissionError: If not authenticated (401) ValueError: If API request fails

async classmethod fetch_by_id_async(protocol_id: str, api_key: str | None = None, base_url: str | None = None) → dict

Fetch a protocol from the platform by ID.

Args:

protocol_id: Protocol ID to fetch api_key: Optional API key for authentication base_url: Optional base URL (defaults to BIOLMAI_BASE_API_URL)

Returns:

Protocol data as dict (same structure as YAML)

Raises:

FileNotFoundError: If protocol not found (404) PermissionError: If not authenticated (401) ValueError: If API request fails

classmethod init(output_path: str, example: str | None = None, force: bool = False) → str

Initialize a new protocol YAML file.

Args:

output_path: Path where the protocol file should be created example: Optional example template name to use force: If True, overwrite existing file

Returns:

Path to the created file

Raises:

FileExistsError: If file exists and force=False ValueError: If example name is invalid FileNotFoundError: If example file doesn’t exist

static render_report(protocol_data: dict, source: str = 'file', console=None) → None

Render a formatted report of the protocol using Rich.

Args:

protocol_data: Protocol data dictionary source: Source description (e.g., “file”, “platform”) console: Optional Rich Console instance (creates one if not provided)

Raises:

ValueError: If protocol_data is invalid or missing required fields

classmethod validate(yaml_path: str) → ProtocolValidationResult

Validate a protocol YAML file.

class biolm.protocols.ProtocolClient(api_key: str | None = None, base_url: str | None = None)

Bases: object

Submit and monitor BioLM protocol runs from Python.

Wraps the /api/protocols/ REST endpoints. Use submit() to start a run, run_and_wait() for a blocking workflow, or get_run() to poll an existing run ID.

get(slug: str, version: int | None = None) → Dict[str, Any]

get_run(run_id: str) → ProtocolRun

Reconnect to an existing run by ID.

list(search: str | None = None, page: int = 1, page_size: int = 20) → Dict[str, Any]

run_and_wait(slug: str, inputs: Dict[str, Any], run_name: str | None = None, timeout: float = 3600.0, show_progress: bool = True, poll_interval: float = 5.0) → Dict[str, Any]

submit(slug: str, inputs: Dict[str, Any], version: int | None = None, run_name: str | None = None, environment_id: int | None = None, files: Dict[str, Any] | None = None) → ProtocolRun

exception biolm.protocols.ProtocolNotFoundError

Bases: ProtocolRunError

The requested protocol slug/version does not exist or is not accessible.

class biolm.protocols.ProtocolRun(data: Dict[str, Any], client: ProtocolClient)

Bases: object

A submitted protocol run returned by ProtocolClient.submit().

cancel() → Dict[str, Any]

Cancel this run (idempotent; may error if already terminal).

download(output_dir: str | Path = '.', file_type: str = 'csv', overwrite: bool = False) → Path

download_files(*, output_dir: str | Path = '.', file_type: str = 'csv', overwrite: bool = False) → Path

Compatibility alias for download().

progress() → Dict[str, Any]

refresh() → ProtocolRun

results() → Dict[str, Any]

to_dataframe(*, output_dir: str | Path = '.', overwrite: bool = False)

Download CSV zip and return a pandas DataFrame.

wait(timeout: float = 3600.0, show_progress: bool = True, poll_interval: float = 5.0) → ProtocolRun

Block until this run reaches a terminal state.

Waiting is websocket-first: it connects to the run’s telemetry channel for live progress. If the websockets dependency is missing, the connection fails, or the stream closes while the run is still nonterminal, it transparently falls back to REST polling of the run detail endpoint until the run succeeds, fails, or is cancelled.

Parameters:

• timeout – Total deadline in seconds shared across the websocket and REST polling phases. The deadline is not reset after a websocket failure.

• show_progress – Print status changes to stdout.

• poll_interval – Seconds between REST polls during fallback. Each sleep is clamped to the remaining deadline and must be greater than zero.

Returns:

self for chaining.

Raises:

• ProtocolRunError – If the run fails or is cancelled.

• TimeoutError – If the total deadline elapses before completion.

• ValueError – If poll_interval is not greater than zero.

exception biolm.protocols.ProtocolRunError

Bases: Exception

A protocol run failed, was cancelled, or the API returned an error.

class biolm.protocols.ProtocolValidationResult(is_valid: bool, errors: ~typing.List[~biolm.protocols.validation.ValidationError] = <factory>, warnings: ~typing.List[str] = <factory>, statistics: ~typing.Dict[str, ~typing.Any] = <factory>)

Bases: object

Result of protocol validation.

add_error(message: str, path: str = '', error_type: str = 'unknown')

add_warning(message: str)

errors: List[ValidationError]

is_valid: bool

statistics: Dict[str, Any]

warnings: List[str]

class biolm.protocols.ValidationError(message: str, path: str = '', error_type: str = 'unknown')

Bases: object

Represents a single validation error.

error_type: str = 'unknown'

message: str

path: str = ''

biolm.protocols.validate_protocol_file(yaml_path: str) → ProtocolValidationResult

Validate a protocol YAML file.

Protocol model — load, validate, execute, and inspect protocol YAML.

class biolm.protocols.model.Protocol(yaml_path: str)

Bases: object

Load and validate BioLM protocol YAML files.

Args:

yaml_path: Path to a protocol YAML file. The file is loaded and

validated on construction; invalid YAML raises ValueError.

Use validate() as a classmethod to validate without instantiating.

execute(inputs: Dict[str, Any] | None = None)

Execute protocol locally with given inputs.

Args:

inputs: Input values for the protocol (optional, uses defaults from protocol if not provided).

Returns:

biolm.protocols.runtime.LocalRunResult with dataframe, records, and metadata.

Requires:

biolm[pipeline] optional dependencies.

classmethod fetch_by_id(protocol_id: str, api_key: str | None = None, base_url: str | None = None) → dict

Fetch a protocol from the platform by ID (synchronous wrapper).

Args:

protocol_id: Protocol ID to fetch api_key: Optional API key for authentication base_url: Optional base URL (defaults to BIOLMAI_BASE_API_URL)

Returns:

Protocol data as dict (same structure as YAML)

Raises:

FileNotFoundError: If protocol not found (404) PermissionError: If not authenticated (401) ValueError: If API request fails

async classmethod fetch_by_id_async(protocol_id: str, api_key: str | None = None, base_url: str | None = None) → dict

Fetch a protocol from the platform by ID.

Args:

protocol_id: Protocol ID to fetch api_key: Optional API key for authentication base_url: Optional base URL (defaults to BIOLMAI_BASE_API_URL)

Returns:

Protocol data as dict (same structure as YAML)

Raises:

FileNotFoundError: If protocol not found (404) PermissionError: If not authenticated (401) ValueError: If API request fails

classmethod init(output_path: str, example: str | None = None, force: bool = False) → str

Initialize a new protocol YAML file.

Args:

output_path: Path where the protocol file should be created example: Optional example template name to use force: If True, overwrite existing file

Returns:

Path to the created file

Raises:

FileExistsError: If file exists and force=False ValueError: If example name is invalid FileNotFoundError: If example file doesn’t exist

static render_report(protocol_data: dict, source: str = 'file', console=None) → None

Render a formatted report of the protocol using Rich.

Args:

protocol_data: Protocol data dictionary source: Source description (e.g., “file”, “platform”) console: Optional Rich Console instance (creates one if not provided)

Raises:

ValueError: If protocol_data is invalid or missing required fields

classmethod validate(yaml_path: str) → ProtocolValidationResult

Validate a protocol YAML file.

class biolm.protocols.model.ProtocolValidationResult(is_valid: bool, errors: ~typing.List[~biolm.protocols.validation.ValidationError] = <factory>, warnings: ~typing.List[str] = <factory>, statistics: ~typing.Dict[str, ~typing.Any] = <factory>)

Bases: object

Result of protocol validation.

add_error(message: str, path: str = '', error_type: str = 'unknown')

add_warning(message: str)

errors: List[ValidationError]

is_valid: bool

statistics: Dict[str, Any]

warnings: List[str]

class biolm.protocols.model.ValidationError(message: str, path: str = '', error_type: str = 'unknown')

Bases: object

Represents a single validation error.

error_type: str = 'unknown'

message: str

path: str = ''

Protocol YAML validation (schema + semantic checks).

class biolm.protocols.validation.ProtocolValidationResult(is_valid: bool, errors: ~typing.List[~biolm.protocols.validation.ValidationError] = <factory>, warnings: ~typing.List[str] = <factory>, statistics: ~typing.Dict[str, ~typing.Any] = <factory>)

Bases: object

Result of protocol validation.

add_error(message: str, path: str = '', error_type: str = 'unknown')

add_warning(message: str)

errors: List[ValidationError]

is_valid: bool

statistics: Dict[str, Any]

warnings: List[str]

class biolm.protocols.validation.ValidationError(message: str, path: str = '', error_type: str = 'unknown')

Bases: object

Represents a single validation error.

error_type: str = 'unknown'

message: str

path: str = ''

biolm.protocols.validation.load_yaml(yaml_path: str) → dict

Load a protocol YAML file.

biolm.protocols.validation.validate_protocol_file(yaml_path: str) → ProtocolValidationResult

Validate a protocol YAML file.

Protocol Submission API client — programmatic run submission, progress tracking, and results retrieval.

This file is a renamed copy of biolmai/protocol_runs.py from py-biolm, kept for backwards compatibility when migrating to the biolm namespace.

class biolm.protocols.runs.ProtocolClient(api_key: str | None = None, base_url: str | None = None)

Bases: object

Submit and monitor BioLM protocol runs from Python.

Wraps the /api/protocols/ REST endpoints. Use submit() to start a run, run_and_wait() for a blocking workflow, or get_run() to poll an existing run ID.

get(slug: str, version: int | None = None) → Dict[str, Any]

get_run(run_id: str) → ProtocolRun

Reconnect to an existing run by ID.

list(search: str | None = None, page: int = 1, page_size: int = 20) → Dict[str, Any]

run_and_wait(slug: str, inputs: Dict[str, Any], run_name: str | None = None, timeout: float = 3600.0, show_progress: bool = True, poll_interval: float = 5.0) → Dict[str, Any]

submit(slug: str, inputs: Dict[str, Any], version: int | None = None, run_name: str | None = None, environment_id: int | None = None, files: Dict[str, Any] | None = None) → ProtocolRun

exception biolm.protocols.runs.ProtocolNotFoundError

Bases: ProtocolRunError

The requested protocol slug/version does not exist or is not accessible.

class biolm.protocols.runs.ProtocolRun(data: Dict[str, Any], client: ProtocolClient)

Bases: object

A submitted protocol run returned by ProtocolClient.submit().

cancel() → Dict[str, Any]

Cancel this run (idempotent; may error if already terminal).

download(output_dir: str | Path = '.', file_type: str = 'csv', overwrite: bool = False) → Path

download_files(*, output_dir: str | Path = '.', file_type: str = 'csv', overwrite: bool = False) → Path

Compatibility alias for download().

progress() → Dict[str, Any]

refresh() → ProtocolRun

results() → Dict[str, Any]

to_dataframe(*, output_dir: str | Path = '.', overwrite: bool = False)

Download CSV zip and return a pandas DataFrame.

wait(timeout: float = 3600.0, show_progress: bool = True, poll_interval: float = 5.0) → ProtocolRun

Block until this run reaches a terminal state.

Waiting is websocket-first: it connects to the run’s telemetry channel for live progress. If the websockets dependency is missing, the connection fails, or the stream closes while the run is still nonterminal, it transparently falls back to REST polling of the run detail endpoint until the run succeeds, fails, or is cancelled.

Parameters:

• timeout – Total deadline in seconds shared across the websocket and REST polling phases. The deadline is not reset after a websocket failure.

• show_progress – Print status changes to stdout.

• poll_interval – Seconds between REST polls during fallback. Each sleep is clamped to the remaining deadline and must be greater than zero.

Returns:

self for chaining.

Raises:

• ProtocolRunError – If the run fails or is cancelled.

• TimeoutError – If the total deadline elapses before completion.

• ValueError – If poll_interval is not greater than zero.

exception biolm.protocols.runs.ProtocolRunError

Bases: Exception

A protocol run failed, was cancelled, or the API returned an error.

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

Requires optional dependencies:

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