biomero package

biomero.slurm_client module

class biomero.slurm_client.SlurmClient(host='slurm', user=None, port=None, config=None, gateway=None, forward_agent=None, connect_timeout=None, connect_kwargs=None, inline_ssh_env=True, slurm_data_path: str = 'my-scratch/data', slurm_images_path: str = 'my-scratch/singularity_images/workflows', slurm_converters_path: str = 'my-scratch/singularity_images/converters', slurm_model_paths: dict = None, slurm_model_repos: dict = None, slurm_model_images: dict = None, converter_images: dict = None, slurm_model_jobs: dict = None, slurm_model_jobs_params: dict = None, slurm_model_use_gpu: dict = None, slurm_script_path: str = 'slurm-scripts', slurm_script_repo: str = None, init_slurm: bool = False, track_workflows: bool = True, enable_job_accounting: bool = True, enable_job_progress: bool = True, enable_workflow_analytics: bool = True, sqlalchemy_url: str = None, config_only: bool = False, slurm_data_bind_path: str = None, slurm_conversion_partition: str = None, slurm_default_partition: str = None, sacct_start_time: str = None, sacct_days_ago: int = None, env_file_submission: bool = False, inject_gpu_flag: bool = False, gpu_partition: str = None, gpu_gres: str = None, gpu_gpus: str = None, slurm_global_job_params: list = None, slurm_image_pull_via_sbatch: bool = False, image_pull_cpus: str = None, image_pull_mem: str = None, apptainer_tmpdir: str = None, apptainer_cachedir: str = None, slurm_zip_cmd: str = None, analytics_rebuild_start_time: str = None, analytics_rebuild_days_ago: int = None, image_pull_time: str = None, image_pull_concurrency: int = 1, image_pull_partition: str = None, remote_shallow_zarr: bool = False, remote_shallower_image: str = 'cellularimagingcf/biomero-shallower:latest', remote_shallower_version: str = None, remote_shallower_workers: int = 1, remote_shallower_partition: str = None, remote_shallower_mem: str = None, remote_shallower_time: str = None, job_status_attempts: int = 5, job_status_retry_seconds: float = 3, ssh_recovery_timeout: int = 900, slurm_lock_cmd: str = 'flock -w {wait_seconds} -E {busy_exit_code} {lock_file}')

Bases: Connection

A client for connecting to and interacting with a Slurm cluster over SSH.

This class extends the Connection class, adding methods and attributes specific to working with Slurm.

SlurmClient accepts the same arguments as Connection. Below only mentions the added ones:

The easiest way to set this client up is by using a slurm-config.ini and the from_config() method.

slurm_data_path

The path to the directory containing the data files for Slurm jobs.

Type:

str

slurm_images_path

The path to the directory containing the Singularity images for Slurm jobs.

Type:

str

slurm_converters_path

The path to the directory containing the Singularity images for file converters.

Type:

str

slurm_model_paths

A dictionary containing the paths to the Singularity images for specific Slurm job models.

Type:

dict

slurm_model_repos

A dictionary containing the git repositories of Singularity images for specific Slurm job models.

Type:

dict

slurm_model_images

A dictionary containing the dockerhub of the Singularity images for specific Slurm job models. Will fill automatically from the data in the git repository, if you set init_slurm.

Type:

dict

slurm_script_path

The path to the directory containing the Slurm job submission scripts on Slurm.

Type:

str

slurm_script_repo

The git https URL for cloning the repo containing the Slurm job submission scripts. Optional.

Type:

str

Example

# Create a SlurmClient object as contextmanager

with SlurmClient.from_config() as client:

# Run a command on the remote host

result = client.run(‘sbatch myjob.sh’)

# Check whether the command succeeded

if result.ok:

print(‘Job submitted successfully!’)

# Print the output of the command

print(result.stdout)

Example 2:

# Create a SlurmClient and setup Slurm (download containers etc.)

with SlurmClient.from_config(init_slurm=True) as client:

client.run_workflow(…)

bring_listener_uptodate(listener, start=1)

build_sbatch_wrap_command(wrapped_command: str, *, job_name: str, output_path: str, job_params: List[str] | None = None) → str

Build an sbatch --wrap command that honours the site defaults.

Parameters:

• wrapped_command (str) – The shell command the job runs.

• job_name (str) – Slurm job name.

• output_path (str) – Remote path of the job log; may contain Slurm filename patterns such as %j.

• job_params (Optional[List[str]]) – Resource flags; defaults to get_job_params(), which applies the configured default partition and global sbatch_* parameters, except the flags this method sets itself (SBATCH_WRAP_OWNED_FLAGS).

Returns:

The sbatch command, printing the job ID on submission.

Return type:

str

cancel_slurm_jobs(slurm_job_ids: List[int]) → Result | None

Cancel Slurm jobs, for example when the workflow they belong to is cancelled.

Cancelling a job that already ended is not an error for the caller: the job is gone either way, so this logs what Slurm answered and returns the result instead of raising.

The command restricts scancel to jobs owned by the cluster account BIOMERO connects with (--user, which Slurm documents as “restrict the scancel operation to jobs owned by the given user”). An account with Slurm operator, admin or coordinator rights may otherwise cancel other users’ jobs, so a wrong or stale job ID can never hit a job that is not BIOMERO’s. It uses --user="$USER" rather than --me, which older Slurm versions lack.

Parameters:

slurm_job_ids (List[int]) – The job IDs of the jobs to cancel.

Returns:

The result of the scancel command, or None when none

of the IDs is a valid job ID.

Return type:

Result

check_job_status(slurm_job_ids: List[int], env: Dict[str, str] | None = None) → Tuple[Dict[int, str], Result | None]

Check the status of Slurm jobs with the given job IDs.

Note: This doesn’t return job arrays individually. It takes the last value returned for those sub ids (generally the one still PENDING).

Parameters:

• slurm_job_ids (List[int]) – The job IDs of the Slurm jobs to check.

• env (Dict[str, str], optional) – Optional environment variables to set when running the command. Defaults to None.

Returns:

A tuple containing the status per input ID and the result of the command execution. Without any valid (positive integer) job ID nothing is run: it returns ({}, None).

Return type:

Tuple[Dict[int, str], Optional[Result]]

Raises:

SSHException – If the command fails, or sacct returns no job rows after job_status_attempts attempts.

check_remote_commands_available() → List[str]

Check that the cluster has the commands that recovery relies on.

Tracked submissions, result archives and input extraction repeat safely after a lost connection only with a lock command (slurm_lock_cmd, flock by default) and with Slurm accounting (sacct), which finds a job whose submission reply was lost. With Info-ZIP as slurm_zip_cmd, extraction also needs mktemp and a find with -cnewer (GNU findutils; only the presence of find is checked). A missing command is logged as a WARNING and does not stop the setup.

Returns:

The executables that were not found.

Return type:

List[str]

cleanup_tmp_files(slurm_job_id: str, filename: str = None, data_location: str = None, logfile: str = None) → Result

Cleanup zip and unzipped files/folders associated with a Slurm job.

Parameters:

• slurm_job_id (str) – The job ID of the Slurm script.

• filename (str) – The zip filename on Slurm.

• data_location (str, optional) – The location of data files on Slurm. If not provided, it will be extracted from the log file.

• logfile (str, optional) – The log file of the Slurm job. If not provided, a default log file will be used.

Returns:

The result of the cleanup operation.

Return type:

Result

Note

The cleanup process involves removing the specified zip file, the log file, and associated data files and folders.

Example

# Cleanup temporary files for a Slurm job client.cleanup_tmp_files(“12345”, “output.zip”)

convert_param_type_to_omtype(param_type: str, _default, *args, **kwargs) → Any

Convert a generic type to an OMERO type and instantiates it with args/kwargs.

Note that Cytomine has a Python Client, and some conversion methods to python types, but nothing particularly worth depending on that library for yet. Might be useful in the future perhaps. (e.g. https://github.com/Cytomine-ULiege/Cytomine-python-client/ blob/master/cytomine/cytomine_job.py)

Parameters:

• param_type (str) – The param type to convert.

• _default – The default value. Required to distinguish between float and int.

• *args – Additional positional arguments.

• **kwargs – Additional keyword arguments.

Returns:

The converted OMERO type class instance or None if errors occured.

Return type:

Any

convert_url(input_url: str, ext: str = '.json') → str

Convert the input GitHub URL to an output URL that retrieves the ‘descriptor.json’ file in raw format.

Parameters:

• input_url (str) – The input GitHub URL.

• ext (str) – (Optional) The input file extension.

Returns:

The output URL to the ‘descriptor.json’ file.

Return type:

str

Raises:

ValueError – If the input URL is not a valid GitHub URL.

copy_zip_locally(local_tmp_storage: str, filename: str) → Result

Copy a zip file from Slurm to the local server.

Note about (Transfer)Result:

Unlike similar classes such as invoke.runners.Result or fabric.runners.Result (which have a concept of “warn and return anyways on failure”) this class has no useful truthiness behavior. If a file transfer fails, some exception will be raised, either an OSError or an error from within Paramiko.

Parameters:

• local_tmp_storage (String) – Path to store the zip file locally.

• filename (String) – Zip filename on Slurm.

Returns:

The result of the scp attempt.

Return type:

TransferResult

extract_data_location_from_log(slurm_job_id: str = None, logfile: str = None) → str

Read SLURM job logfile to find location of the data.

One of the parameters is required, either id or file.

Parameters:

• slurm_job_id (str) – Id of the slurm job

• logfile (str) – Path to the logfile

Returns:

Data location according to the log

Return type:

str

Raises:

SSHException – If there is an issue with the command execution.

extract_job_id(result: Result) → int

Extract the Slurm job ID from the result of a command.

Parameters:

result (Result) – The result of a command execution.

Returns:

The Slurm job ID extracted from the result, or -1 if not found.

Return type:

int

extract_parts_from_url(input_url: str) → Tuple[List[str], str]

Extract the repository and branch information from the input URL.

Parameters:

input_url (str) – The input GitHub URL.

Returns:

The list of url parts and the branch/version. If no branch is found, it will return “master”

Return type:

Tuple[List[str], str]

Raises:

ValueError – If the input URL is not a valid GitHub URL.

classmethod from_config(configfile: str = '', init_slurm: bool = False, config_only: bool = False) → SlurmClient

Creates a new SlurmClient object using the parameters read from a configuration file (.ini).

Defaults paths to look for config files are:

• /etc/slurm-config.ini

• /OMERO/slurm-config.ini

• ~/slurm-config.ini

Set BIOMERO_SLURM_CONFIG_FILE to enable authoritative-file mode. In that mode only the configured file is read; default paths and an explicit configfile argument are intentionally ignored.

Note that this is only for the SLURM-specific values that we added. Most configuration values are set via configuration mechanisms from Fabric library, like SSH settings being loaded from SSH config, /etc/fabric.yml or environment variables. See Fabric’s documentation for more info on configuration if needed.

Parameters:

• configfile (str) – The path to your configuration file. Optional.

• init_slurm (bool) – Initiate / validate slurm setup. Optional Might take some time the first time with downloading, etc.

Returns:

A new SlurmClient object.

Return type:

SlurmClient

generate_slurm_job_for_workflow(workflow: str, substitutes: Dict[str, str], template: str = 'job_template.sh') → str

Generate a Slurm job script for a specific workflow.

Parameters:

• workflow (str) – The name of the workflow.

• substitutes (Dict[str, str]) – A dictionary containing key-value pairs for substituting placeholders in the job template.

• template (str, optional) – The filename of the job template. Defaults to “job_template.sh”.

Returns:

The generated Slurm job script as a string.

Return type:

str

generic_descriptor_from_github(workflow: str) → Dict

Pull the workflow descriptor from GitHub and convert to generic format.

Parameters:

workflow (str) – Workflow name (looked up in slurm_model_repos) or a direct GitHub repository URL.

Returns:

The descriptor in biomero-schema format.

Return type:

Dict

Raises:

ValueError – If no descriptor file is found or the URL is invalid.

get_active_job_progress(slurm_job_id: str, pattern: str = '\\d+%', env: Dict[str, str] | None = None) → Any

Get the progress of an active Slurm job from its logfiles.

Parameters:

• slurm_job_id (str) – The ID of the Slurm job.

• pattern (str) – The pattern to match in the job log to extract the progress (default: r”d+%”).

• env (Dict[str, str], optional) – Optional environment variables to set when running the command. Defaults to None.

Returns:

The progress of the Slurm job according to the pattern, or None.

Return type:

Any

get_all_image_versions_and_data_files() → Tuple[Dict[str, List[str]], List[str]]

Retrieve all available image versions and data files from the Slurm cluster.

Returns:

A tuple containing: - A dictionary mapping models to available versions. - A list of available input data folders.

Return type:

Tuple[Dict[str, List[str]], List[str]]

classmethod get_authoritative_config_path()

Return the configured authoritative ini path, if enabled.

get_cellpose_command(image_version: str, input_data: str, cp_model: str, nuc_channel: int, prob_threshold: float, cell_diameter: float, email: str | None = None, time: str | None = None, use_gpu: bool = True, model: str = 'cellpose') → Tuple[str, Dict]

Return the command and environment dictionary to run a CellPose job on the Slurm workload manager. A specific example of using the generic ‘get_workflow_command’.

Parameters:

• image_version (str) – The version of the Singularity image to use.

• input_data (str) – The name of the input data folder on the shared file system.

• cp_model (str) – The name of the CellPose model to use.

• nuc_channel (int) – The index of the nuclear channel.

• prob_threshold (float) – The probability threshold for nuclei detection.

• cell_diameter (float) – The expected cell diameter in pixels.

• email (Optional[str]) – The email address to send notifications to. Defaults to None.

• time (Optional[str]) – The maximum time for the job to run. Defaults to None.

• use_gpu (bool) – Whether to use GPU for the CellPose job. Defaults to True.

• model (str) – The name of the folder of the Docker image to use. Defaults to “cellpose”.

Returns:

A tuple containing the Slurm sbatch command and the environment dictionary.

Return type:

Tuple[str, dict]

classmethod get_config_paths(configfile: str = '') → list[str]

Resolve config paths in effective read order.

BIOMERO_SLURM_CONFIG_FILE enables authoritative-file mode and replaces all default and caller-supplied config layers.

classmethod get_config_write_path() → str

Resolve the writable admin overlay or authoritative ini path.

Used by the OMERO.biomero admin UI to know which file to save its settings to.

get_conversion_command(data_path: str, config_file: str, source_format: str = 'zarr', target_format: str = 'tiff', submission_job_name: str | None = None) → Tuple[str, Dict, str, str]

Generate Slurm conversion command and environment variables for data conversion.

Parameters:

• data_path (str) – Path to the data folder.

• config_file (str) – Path to the configuration file.

• source_format (str) – Source data format (default is ‘zarr’).

• target_format (str) – Target data format (default is ‘tiff’).

• submission_job_name (str, optional) – Slurm job name that identifies the submission of a tracked task (see get_submission_job_name). Defaults to conversion.

Returns:

A tuple containing the Slurm conversion command and the environment variables, followed by the converter image name and version.

Return type:

Tuple[str, Dict, str, str]

Warning

The default implementation only supports conversion from ‘zarr’ to ‘tiff’. If using other source or target formats, users must implement and configure additional converters themselves.

get_conversion_logfile_from_slurm(slurm_job_id: str, local_tmp_storage: str = '/tmp/', logfile: str = None) → Tuple[str, str, Any]

Copy a conversion job’s log(s) to local storage as one combined file.

A conversion is submitted as a Slurm array job with --output=omero-%A_%a.log, so it produces one log per array task (omero-<jobid>_1.log, omero-<jobid>_2.log, …). This reads all matching logs on the Slurm side and writes a single combined log file locally, so it can be attached to OMERO exactly like a workflow log – including when the conversion failed and the logs are the only clue.

Unlike get_logfile_from_slurm() (which sftp-copies a single named file) this resolves the array glob via cat and tolerates the case where no log files exist (writes an empty file rather than erroring).

Parameters:

• slurm_job_id (str) – The ID of the conversion (array) Slurm job.

• local_tmp_storage (str, optional) – Local directory to write the combined log to. Defaults to “/tmp/”.

• logfile (str, optional) – Explicit log glob to read (e.g. SlurmJob.log_file). Defaults to the converter glob omero-<jobid>_*.log.

Returns:

local directory, combined log file path, and

the raw cat Result.

Return type:

Tuple[str, str, Any]

get_file_attachment_params(workflow: str) → Dict[str, Dict[str, Any]]

Return only the file-attachment params for a workflow.

Thin filter over get_workflow_parameters().

get_image_pull_status() → Dict[str, Any]

Return structured status for the most recent image submission.

get_image_versions_and_data_files(model: str) → Tuple[List[str], List[str]]

Retrieve the available image versions and input data files for a given model.

Parameters:

model (str) – The name of the model to query for.

Returns:

A tuple containing two lists: - The first list includes the available image versions,

sorted in descending order.

• The second list includes the available data files.

Return type:

Tuple[List[str], List[str]]

Raises:

ValueError – If the provided model is not found in the SlurmClient’s known model paths.

get_job_params(overrides: Dict[str, Any] | None = None, *, excluded: Iterable[str] = (), base_params: Iterable[str] = ()) → List[str]

Merge job resources, the default partition and global sbatch flags.

The one place that combines these sources, for workflow, conversion, image-pull and helper jobs. For each flag the first source that sets it wins: base_params, overrides, then the default partition (--partition only, when overrides has no partition), then the global sbatch_* settings. The source of every flag is logged at DEBUG.

Parameters:

• overrides (Optional[Dict[str, Any]]) – Job-specific resources by sbatch flag name without the dashes, e.g. {'cpus-per-task': 4}. None and empty values are skipped.

• excluded (Iterable[str]) – Flag names (without --) that global sbatch_* settings must not add, because the caller owns them.

• base_params (Iterable[str]) – --flag=value parameters already chosen by a more specific source, such as per-workflow ones. Surrounding whitespace is stripped; they come first.

Returns:

--flag=value parameters, in the shell syntax the

administrator configured.

Return type:

List[str]

get_job_status_command(slurm_job_ids: List[int]) → str

Return the Slurm command to get the status of jobs with the given job IDs.

Parameters:

slurm_job_ids (List[int]) – The job IDs of the jobs to check.

Returns:

The Slurm command to get the status of the jobs.

Return type:

str

get_jobs_info_command(start_time: str = None, end_time: str = 'now', columns: str = 'JobId', states: str = 'r,cd,f,to,rs,dl,nf') → str

Return the Slurm command to retrieve information about old jobs.

The command will be formatted with the specified start time, which is expected to be in the ISO format “YYYY-MM-DD”. The command will use the “sacct” tool to query the Slurm accounting database for jobs that started on or after the specified start time, and will output only the job IDs (-o JobId) without header or trailer lines (-n -X).

Parameters:

• start_time (str) – The start time from which to retrieve job information. Defaults to None, which resolves in priority order: class default (2023-01-01), then sacct_start_time config, then sacct_days_ago config (relative), then env var BIOMERO_SACCT_START_TIME, then env var BIOMERO_SACCT_START_DAYS_AGO (relative, highest priority).

• end_time (str) – The end time until which to retrieve job information. Defaults to “now”.

• columns (str) – The columns to retrieve from the job information. Defaults to “JobId”. It is comma separated, e.g. “JobId,State”.

• states (str) – The job states to include in the query. Defaults to “r,cd,f,to,rs,dl,nf”.

Returns:

A string representing the Slurm command to retrieve information about old jobs.

Return type:

str

get_listeners(runner)

get_logfile_from_slurm(slurm_job_id: str, local_tmp_storage: str = '/tmp/', logfile: str = None) → Tuple[str, str, Result]

Copy the logfile of the given SLURM job to the local server.

Note about (Transfer)Result:

Unlike similar classes such as invoke.runners.Result or fabric.runners.Result (which have a concept of “warn and return anyways on failure”) this class has no useful truthiness behavior. If a file transfer fails, some exception will be raised, either an OSError or an error from within Paramiko.

Parameters:

• slurm_job_id (str) – The ID of the SLURM job.

• local_tmp_storage (str, optional) – Path to store the logfile locally. Defaults to “/tmp/”.

• logfile (str, optional) – Path to the logfile on the SLURM server. Defaults to None.

Returns:

directory, full path of the logfile, and TransferResult

Return type:

Tuple

get_or_create_github_session()

get_recent_log_command(log_file: str, n: int = 10) → str

Get the command to retrieve the recent log entries from a specified log file.

Parameters:

• log_file (str) – The path to the log file.

• n (int, optional) – The number of recent log entries to retrieve. Defaults to 10.

Returns:

The command to retrieve the recent log entries.

Return type:

str

get_remote_shallower_receipts(workflow_id, canonical_inputs)

Read trusted completed receipts from event-sourced helper tasks.

get_shallower_job_params()

Single-process CPU helper resources, also used for recovery jobs.

Global sbatch_* flags that would break a single-process CPU job (SHALLOWER_EXCLUDED_FLAGS, GPU and per-task flags) are not added.

static get_submission_job_name(task_id: Any) → str

Return the Slurm job name that identifies a task’s submission.

A job whose submission reply was lost is found again by this name.

Parameters:

task_id (Any) – ID of the task.

Returns:

The job name, for example biomero-<task id>.

Return type:

str

get_submission_state_prefix(task_id: Any) → str

Return the remote path prefix of a task’s submission receipts.

Parameters:

task_id (Any) – ID of the task.

Returns:

<slurm_data_path>/.biomero-recovery/<task id>.

Return type:

str

get_task(task_id: Any) → Task

Read a task from the workflow tracker.

Parameters:

task_id (Any) – ID of the task.

Returns:

The task aggregate.

Return type:

Task

get_unzip_command(zipfile: str, filter_filetypes: str = '*.zarr *.ome.zarr *.tiff *.tif') → str

Generate a command string for unzipping a data archive and creating required directories for Slurm jobs.

Parameters:

• zipfile (str) – The name of the zip archive file to extract. Without extension.

• filter_filetypes (str, optional) – A space-separated string containing the file extensions to extract from the zip file. E.g. defaults to “*.zarr *.ome.zarr *.tiff *.tif”. Setting this argument to None will omit the file filter and extract all files.

Returns:

The command to extract the specified filetypes from the zip file.

Return type:

str

get_update_slurm_scripts_command() → str

Generate the command to update the Git repository containing the Slurm scripts, if necessary.

Returns:

A string containing the Git command to update the Slurm scripts.

Return type:

str

get_workflow(workflow_id: Any) → WorkflowRun

Read a workflow from the workflow tracker.

Parameters:

workflow_id (Any) – ID of the workflow.

Returns:

The workflow aggregate.

Return type:

WorkflowRun

get_workflow_command(workflow: str, workflow_version: str, input_data: str, email: str | None = None, time: str | None = None, submission_job_name: str | None = None, **kwargs) → Tuple[str, Dict]

Generate the Slurm workflow command and environment variables.

Parameters:

• workflow (str) – The name of the workflow.

• workflow_version (str) – The version of the workflow.

• input_data (str) – The name of the input data folder containing the input image files.

• email (str, optional) – The email address for job notifications. Defaults to None, which defaults to what is in the job script.

• time (str, optional) – The time limit for the job in the format HH:MM:SS. Defaults to None, which defaults to what is in the job script.

• submission_job_name (str, optional) – Slurm job name that identifies the submission of a tracked task (see get_submission_job_name). Defaults to None, which keeps the job name of the job script.

• **kwargs – Additional keyword arguments for the workflow.

Returns:

A tuple containing the Slurm workflow command and the environment variables.

Return type:

Tuple[str, Dict]

get_workflow_parameters(workflow: str) → Dict[str, Dict[str, Any]]

Retrieve the parameters of a workflow.

Parameters:

workflow (str) – The workflow for which to retrieve the parameters.

Returns:

A dictionary containing the workflow parameters.

Return type:

Dict[str, Dict[str, Any]]

Raises:

ValueError – If an error occurs while retrieving the workflow parameters.

get_workflow_tasks(workflow_id: Any, *, task_name: str | None = None, input_data: str | None = None) → List[Task]

Read the tasks of a workflow from the workflow tracker.

Parameters:

• workflow_id (Any) – ID of the workflow, as UUID or string.

• task_name (Optional[str]) – Only return tasks with this name.

• input_data (Optional[str]) – Only return tasks with this input data.

Returns:

The matching task aggregates, in workflow order;

empty when workflow tracking is disabled.

Return type:

List[Task]

get_zip_command(data_location: str, filename: str) → str

Generate a command string for zipping the data on Slurm.

Parameters:

• data_location (str) – The folder to be zipped.

• filename (str) – The name of the zip archive file to extract. Without extension.

Returns:

The command to create the zip file.

Return type:

str

init_workflows(force_update: bool = False)

Retrieves the required info for the configured workflows from github. It will fill slurm_model_images with dockerhub links.

Parameters:

force_update (bool) – Will overwrite already given paths in slurm_model_images

initialize_analytics_system(reset_tables=False)

Initialize the analytics system based on the analytics configuration passed to the constructor.

Parameters:

reset_tables (bool) – If True, drops and recreates all views.

list_active_jobs(env: Dict[str, str] | None = None) → List[str]

Get a list of active jobs from SLURM.

Parameters:

env (Dict[str, str], optional) – Optional environment variables to set when running the command. Defaults to None.

Returns:

A list of job IDs.

Return type:

List[str]

list_all_jobs(env: Dict[str, str] | None = None) → List[str]

Get a list of all jobs from SLURM.

Parameters:

env (Dict[str, str], optional) – Optional environment variables to set when running the command. Defaults to None.

Returns:

A list of job IDs.

Return type:

List[str]

list_available_converter_versions() → Dict

Note, assumes you use versioned converters. Will return a dict with a version of each converter on your Slurm. However, doesn’t work properly with unversioned sif.

list_completed_jobs(env: Dict[str, str] | None = None) → List[str]

Get a list of completed jobs from SLURM.

Parameters:

env (Dict[str, str], optional) – Optional environment variables to set when running the command. Defaults to None.

Returns:

A list of job IDs.

Return type:

List[str]

classmethod load_config(configfile: str = '') → ConfigParser

Load BIOMERO configuration using the effective path policy.

Logs which files were read, and warns when authoritative-file mode ignores a caller-supplied configfile or its file is missing.

open()

Open the SSH connection and make a quiet link detectable.

parse_docker_image_version(image: str) → Tuple[str, str]

Parses the Docker image string to extract the image name and version tag.

Parameters:

image (str) – The Docker image string in the format ‘image_name:version’.

Returns:

A tuple containing:

• version (str or None): The version tag if present, otherwise None.

• image_name (str): The image name if matched, otherwise the original image string.

Return type:

tuple

static positive_job_ids(slurm_job_ids: List) → List[int]

Keep the entries that are valid Slurm job IDs: positive integers.

Parameters:

slurm_job_ids (List) – Job IDs as numbers or strings; None and values that are not numbers (such as the -1 that extract_job_id returns for “not found”) are dropped.

Returns:

The valid job IDs, in the original order.

Return type:

List[int]

prepare_converters() → List[Dict[str, str]]

Stage converter runtime files and return their image specifications.

remove_submission_receipts(state_prefix: str) → None

Remove the receipt files of a submission whose outcome is recorded.

Call this once the event store holds the job ID (or the failure): the files are then no longer needed. A failure is only logged, as the files are harmless leftovers.

Parameters:

state_prefix (str) – Remote path prefix of the receipt files; see get_submission_state_prefix.

run_commands(cmdlist: List[str], env: Dict[str, str] | None = None, sep: str = ' && ', log_stdout: bool = True, log_level: int = 20, **kwargs) → Result

Run a list of shell commands consecutively on the Slurm cluster, ensuring the success of each before proceeding to the next.

The environment variables can be set using the env argument. These commands retain the same session (environment variables etc.), unlike running them separately.

Parameters:

• cmdlist (List[str]) – A list of shell commands to run on Slurm.

• env (Dict[str, str], optional) – Optional environment variables to set when running the command. Defaults to None.

• sep (str, optional) – The separator used to concatenate the commands. Defaults to ‘ && ‘.

• log_stdout (bool, optional) – Whether to log the stdout output. Defaults to True.

• log_level (int, optional) – Log level of the command line, for long generated scripts. Defaults to INFO.

• **kwargs – Additional keyword arguments.

Returns:

The result of the last command in the list.

Return type:

Result

run_commands_read_only(commands, env=None, **kwargs)

Read remote state while preserving the last known job status.

run_commands_split_out(cmdlist: List[str], env: Dict[str, str] | None = None) → List[str]

Run a list of shell commands consecutively and split the output of each command.

Each command in the list is executed with a separator in between that is unique and can be used to split the output of each command later. The separator used is specified by the _OUT_SEP attribute of the SlurmClient instance.

Parameters:

• cmdlist (List[str]) – A list of shell commands to run.

• env (Dict[str, str], optional) – Optional environment variables to set when running the command. Defaults to None.

Returns:

A list of strings, where each string corresponds to the output of a single command in cmdlist split by the separator _OUT_SEP.

Return type:

List[str]

Raises:

SSHException – If any of the commands fail to execute successfully.

run_conversion_workflow_job(folder_name: str, source_format: str = 'zarr', target_format: str = 'tiff', wf_id: UUID = None) → SlurmJob

Run the data conversion workflow on Slurm using the given data folder.

Parameters:

• folder_name (str) – The name of the data folder containing source format files.

• source_format (str) – Source data format for conversion (default is ‘zarr’).

• target_format (str) – Target data format after conversion (default is ‘tiff’).

Returns:

the conversion job

Return type:

SlurmJob

Warning

The default implementation only supports conversion from ‘zarr’ to ‘tiff’. If using other source or target formats, users must implement and configure additional converters themselves.

run_read_only(command, **kwargs)

Bound, reconnect and repeat a caller-declared short read.

run_workflow(workflow_name: str, workflow_version: str, input_data: str, email: str | None = None, time: str | None = None, wf_id: UUID | None = None, output_settings: Dict | None = None, **kwargs) → Tuple[Result, int, UUID, UUID]

Run a specified workflow on Slurm using the given parameters.

Parameters:

• workflow_name (str) – Name of the workflow to execute.

• workflow_version (str) – Version of the workflow (image version on Slurm).

• input_data (str) – Name of the input data folder containing input image files.

• email (str, optional) – Email address for Slurm job notifications.

• time (str, optional) – Time limit for the Slurm job in the format HH:MM:SS.

• wf_id (UUID, optional) – Workflow ID for tracking purposes. If not provided, a new one is created.

• **kwargs – Additional keyword arguments for the workflow.

Returns:

A tuple containing the result of starting the workflow job, the Slurm job ID, the workflow ID, and the task ID. If the Slurm job ID could not be extracted, it returns -1 for the job ID.

Return type:

Tuple[Result, int, UUID, UUID]

Note

The Slurm job ID is extracted from the result of the run_commands method. If track_workflows is enabled, workflow and task tracking is performed.

run_workflow_job(workflow_name: str, workflow_version: str, input_data: str, email: str | None = None, time: str | None = None, wf_id: UUID | None = None, output_settings: Dict | None = None, **kwargs) → SlurmJob

Run a specified workflow on Slurm using the given parameters and return a SlurmJob instance.

Parameters:

• workflow_name (str) – Name of the workflow to execute.

• workflow_version (str) – Version of the workflow (image version on Slurm).

• input_data (str) – Name of the input data folder containing input image files.

• email (str, optional) – Email address for Slurm job notifications.

• time (str, optional) – Time limit for the Slurm job in the format HH:MM:SS.

• wf_id (UUID, optional) – Workflow ID for tracking purposes. If not provided, a new one is created.

• output_settings (dict, optional) – Selected output settings for results import.

• **kwargs – Additional keyword arguments for the workflow.

Returns:

A SlurmJob instance representing the started workflow job.

Return type:

SlurmJob

setup_container_images(extra_image_specs: List[Dict[str, str]] = None) → int | None

Create workflow image directories and submit one combined array.

This is a submission of its own: get_image_pull_status reports only the most recent one, so pass the converter images as extra_image_specs (see prepare_converters) to track them together, as setup_slurm does.

Parameters:

extra_image_specs (List[Dict[str, str]], optional) – More images to make available next to the workflow images.

Returns:

The Slurm array job ID, or None if nothing was

submitted to Slurm.

Return type:

Optional[int]

setup_converters() → int | None

Stage converters and submit them through the shared image runner.

This is a submission of its own: get_image_pull_status reports only the most recent one. setup_slurm submits converters and workflow images together.

Returns:

The Slurm array job ID, or None if nothing was

submitted to Slurm.

Return type:

Optional[int]

setup_directories()

Creates necessary directories for Slurm operations.

This function creates directories for data storage, scripts, and workflows as specified in the SlurmClient object.

Raises:

SSHException – If there is an issue executing directory creation commands.

setup_job_scripts()

Sets up job scripts for Slurm operations.

This function either clones a Git repository containing job scripts into the specified script path or generates scripts locally if no repository is provided.

Raises:

SSHException – If there is an issue executing Git commands or generating scripts.

setup_listeners(runner, reset_tables)

setup_slurm() → int | None

Create the cluster setup and return the image array job ID.

shallow_results_on_slurm(data_path, workflow_id, canonical_inputs, *, heartbeat=None)

Run/adopt optional CPU shallowing before result archiving.

The whole run is repeated within the SSH recovery window after a lost connection, so every step inside it must be safe to repeat: it adopts the recorded job or submits through submit_job_once.

heartbeat is a caller-owned no-argument callback invoked during job waits.

str_to_class(module_name: str, class_name: str, *args, **kwargs)

Return a class instance from a string reference.

Parameters:

• module_name (str) – The name of the module.

• class_name (str) – The name of the class.

• *args – Additional positional arguments for the class constructor.

• **kwargs – Additional keyword arguments for the class constructor.

Returns:

An instance of the specified class, or None if the class or

module does not exist.

Return type:

object

submit_job_once(sbatch_command: str, state_prefix: str, *, job_name: str, env: Dict[str, str] | None = None, raise_on_rejection: bool = True, verify_command_unchanged: bool = True) → Result

Submit a Slurm job exactly once, even when the SSH reply is lost.

This is the one submit-exactly-once implementation, for workflows, conversions and helper jobs. It runs the submission under a remote lock with receipt files (see remote_scripts.submit_once_script): asked again, it adopts the job already submitted, finds a job whose reply was lost by its name in Slurm, and never runs sbatch again while an earlier attempt may have created a job. The lock command is slurm_lock_cmd; sacct must be available.

Parameters:

• sbatch_command (str) – The submission command (any sbatch …). It must use job_name as its Slurm job name.

• state_prefix (str) – Remote path prefix of the receipt files; see get_submission_state_prefix.

• job_name (str) – The Slurm job name of this submission.

• env (Optional[Dict[str, str]]) – Environment variables for the command.

• raise_on_rejection (bool) – Raise UnexpectedExit when sbatch refuses the job (default), and keep a receipt that repeats that answer. When False, return the failed Result and forget the attempt, so that a later call submits again.

• verify_command_unchanged (bool) – Stop when a recorded submission used a different command (default).

Returns:

stdout is Submitted batch job <id>; see

extract_job_id. A refused job gives a failed Result when raise_on_rejection is False.

Return type:

Result

Raises:

• UnexpectedExit – If sbatch refused the job, or the script failed.

• ssh_recovery.SSHRecoveryTimeout – If the connection did not recover in time.

• ssh_recovery.RemoteOperationStillPending – If the submission is still locked or its job is not visible in Slurm after the waiting time. Nothing is submitted again.

transfer_data(local_path: str) → Result

Transfers a file or directory from the local machine to the remote Slurm cluster.

Parameters:

local_path (str) – The local path to the file or directory to transfer.

Returns:

The result of the file transfer operation.

Return type:

Result

unpack_data(zipfile: str, env: Dict[str, str] | None = None) → Result

Unpacks a zipped file on the remote Slurm cluster.

Parameters:

• zipfile (str) – The name of the zipped file to be unpacked.

• env (Dict[str, str], optional) – Optional environment variables to set when running the command. Defaults to None.

Returns:

The result of the command.

Return type:

Result

update_slurm_scripts(generate_jobs: bool = False, env: Dict[str, str] | None = None) → Result

Updates the local copy of the Slurm job submission scripts.

This function pulls the latest version of the scripts from the Git repository and copies them to the slurm_script_path directory. Alternatively, it can generate scripts from a template. This is the default behavior if no Git repo is provided or can be forced via the generate_jobs parameter.

Parameters:

• generate_jobs (bool) – Whether to generate new slurm job scripts INSTEAD (of pulling from git). Defaults to False, except if no slurm_script_repo is configured.

• env (Dict[str, str], optional) – Optional environment variables to set when running the command. Defaults to None.

Returns:

The result of the command.

Return type:

Result

validate(validate_slurm_setup: bool = False)

Validate the connection to the Slurm cluster by running a simple command.

Parameters:

validate_slurm_setup (bool) – Whether to also check and fix the Slurm setup (folders, images, etc.)

Returns:

True if the validation is successful, False otherwise.

Return type:

bool

workflow_bilayers_folder_params_to_subs(descriptor: Dict) → Dict[str, str]

Build INPARAMS and OUTPARAMS substitution strings for bilayers job templates.

Folder inputs (image/file/array/measurement/executable) are mapped to $DATA_PATH/data/in. Outputs with a cli_tag and any parameter with output_dir_set=True (marked set-by-server after schema parsing) are mapped to $DATA_PATH/data/out.

Parameters:

descriptor (Dict) – The parsed workflow descriptor.

Returns:

Dictionary with keys INPARAMS and OUTPARAMS.

Return type:

Dict[str, str]

workflow_params_to_envvars(**kwargs) → Dict

Convert workflow parameters to environment variables.

Parameters:

**kwargs – Additional keyword arguments for the workflow.

Returns:

A dictionary containing the environment variables.

Return type:

Dict

workflow_params_to_subs(params) → Dict[str, str]

Convert workflow parameters to substitution dictionary for job script.

Parameters:

params – A dictionary containing workflow parameters.

Returns:

A dictionary with parameter names as keys and

corresponding flags with placeholders as values.

Return type:

Dict[str, str]

workflow_template_subs(descriptor: Dict) → Dict[str, str]

Build shared-template substitutions for workflow family differences.

zip_data_on_slurm_server(data_location: str, filename: str, env: Dict[str, str] | None = None) → Result

Zip the output folder of a job on Slurm

Parameters:

• data_location (String) – Folder on SLURM with the “data/out” subfolder.

• filename (String) – Name to give to the zipfile.

• env (Dict[str, str], optional) – Optional environment variables to set when running the command. Defaults to None.

Returns:

The result of the zip attempt.

Return type:

Result

class biomero.slurm_client.SlurmJob(submit_result: Result | None, job_id: int, wf_id: UUID, task_id: UUID, slurm_polling_interval: int = 10, log_file: str = None)

Bases: object

Represents a job submitted to a Slurm cluster.

This class encapsulates information and methods related to managing a job submitted to a Slurm cluster. It provides functionality to monitor the job’s state, wait for completion, and perform cleanup operations.

job_id

The ID of the Slurm job.

Type:

int

submit_result

The result of submitting the job.

Type:

Result

ok

Indicates whether the job submission was successful.

Type:

bool

job_state

The current state of the Slurm job.

Type:

str

progress

The progress of the Slurm job.

Type:

str

error_message

The error message, if any, encountered during job submission.

Type:

str

wf_id

The workflow ID associated with the job.

Type:

UUID

task_id

The task ID within the workflow.

Type:

UUID

slurm_polling_interval

The polling interval (in seconds) for checking the job status.

Type:

int

Example

# Submit some job with the SlurmClient submit_result, job_id, wf_id, task_id = slurmClient.run_workflow(

workflow_name, workflow_version, input_data, email, time, wf_id, **kwargs)

# Create a SlurmJob instance slurmJob = SlurmJob(submit_result, job_id, wf_id, task_id)

if not slurmJob.ok:

logger.warning(f”Error with job: {slurmJob.get_error()}”)

else:

try:

# heartbeat is a no-argument callback supplied by the caller. slurmJob.wait_for_completion(slurmClient, heartbeat=heartbeat) if not slurmJob.completed():

raise Exception(f”Job is not completed: {slurmJob}”)

else:

slurmJob.cleanup(slurmClient)

except Exception as e:

logger.error(f” ERROR WITH JOB: {e}”) raise e

SLURM_POLLING_INTERVAL = 10

cleanup(slurmClient) → Result

Cleanup remaining log files.

Parameters:

slurmClient – The Slurm client.

Returns:

The result of the cleanup operation.

Return type:

Result

completed()

Check if the Slurm job has completed successfully.

Returns:

True if the job has completed; False otherwise.

Return type:

bool

static ended_without_success(state: str | None) → bool

Tell whether a (raw) job state is final and not COMPLETED.

Failed, cancelled, timed-out, out-of-memory … jobs are never re-attached to on resume; a COMPLETED job may be, so that its outputs are processed instead of running the analysis again.

classmethod from_job_id(job_id, wf_id=None, task_id=None, **kwargs)

Monitor an already submitted job without submitting it again.

This process did not verify the submission, so the job is not ok: monitoring it with wait_for_completion is the way to find out. Submission reconciliation remains the caller’s responsibility.

get_error() → str

Get the error message associated with the Slurm job submission.

Returns:

The error message, or an empty string if no error occurred.

Return type:

str

static is_terminal(state: str | None) → bool

Tell whether a (raw) job state is final.

static state_name(state: str | None) → str

Return the bare Slurm state from a raw sacct/squeue state.

sacct may add a suffix (CANCELLED+) or details (CANCELLED by 1000).

Parameters:

state (Optional[str]) – The raw state, or None.

Returns:

The state name, for example CANCELLED; empty if unknown.

Return type:

str

wait_for_completion(slurmClient, omeroConn=None, *, heartbeat=None, track_progress=True, update_task=True, strict_status=False) → str

Wait for the Slurm job to reach completion, cancellation, failure, or timeout.

Parameters:

• slurmClient – The Slurm client.

• omeroConn – Deprecated, kept for callers of BIOMERO 2.8.0 that pass the connection positionally. Its keepAlive() is used as the heartbeat; pass heartbeat instead. Will be removed in a future major release.

• heartbeat – Optional no-argument callback, invoked before each poll. Return values are ignored; raise to abort monitoring. Exceptions propagate unchanged without marking the Slurm job failed. Takes precedence over omeroConn.

• track_progress – Read analysis progress from job logs (default True).

• update_task – Publish task status/progress (default True). Helpers whose success also requires report validation can disable this.

• strict_status – Raise on unavailable status instead of recording a failure. This lets resumable callers preserve an existing job.

Returns:

The final state of the Slurm job.

Return type:

str

biomero.eventsourcing module

class biomero.eventsourcing.NoOpWorkflowTracker

Bases: object

A no-operation workflow tracker that makes all method calls no-op.

All method calls to this class will return None and log the function name and its parameters for debugging purposes.

class biomero.eventsourcing.ResultDict(result: Result)

Bases: dict

A dictionary subclass that stores details from a Fabric Result object.

Parameters:

result (Result) – The Fabric result object.

command

The command that was executed.

Type:

str

env

The environment variables during the command execution.

Type:

dict

stdout

The standard output from the command.

Type:

str

stderr

The standard error from the command.

Type:

str

class biomero.eventsourcing.Task(*args: Any, **kwargs: Any)

Bases: Aggregate

Represents a task in a workflow as an aggregate in the domain model.

workflow_id

The UUID of the associated workflow.

Type:

UUID

task_name

The name of the task.

Type:

str

task_version

The version of the task.

Type:

str

input_data

Input data for the task.

Type:

dict

params

Additional parameters for the task.

Type:

dict

job_ids

List of job IDs associated with the task.

Type:

list

results

List of results from the task execution.

Type:

list

result_message

Message related to the result of the task.

Type:

str

status

The current status of the task.

Type:

str

completed

Whether the task was completed.

Type:

bool

failed

Whether the task failed.

Type:

bool

claimed_by

The supervisor that holds the claim, for a launcher.

Type:

str

claimed_at

When that claim was made or last renewed.

Type:

datetime

class Created(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime')

Bases: Event, Created

originator_id_type

alias of UUID

class Event(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime')

Bases: Event

originator_id_type

alias of UUID

INITIAL_VERSION: int = 0

class JobIdAdded(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime', job_id: str)

Bases: DecoratedFuncCaller, JobIdAdded

class LauncherClaimed(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime', owner: str, claimed_at: datetime.datetime)

Bases: DecoratedFuncCaller, LauncherClaimed

class LauncherReleased(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime')

Bases: DecoratedFuncCaller, LauncherReleased

class ProgressUpdated(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime', progress: str)

Bases: DecoratedFuncCaller, ProgressUpdated

class ResultAdded(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime', result: biomero.eventsourcing.ResultDict)

Bases: DecoratedFuncCaller, ResultAdded

class StatusUpdated(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime', status: str)

Bases: DecoratedFuncCaller, StatusUpdated

class StorageProvenanceRecorded(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime')

Bases: DecoratedFuncCaller, Event

evidence: dict

target_key: str

class TaskCompleted(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime', result: str)

Bases: DecoratedFuncCaller, TaskCompleted

class TaskCreated(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime', originator_topic: 'str', workflow_id: uuid.UUID, task_name: str, task_version: str, input_data: Dict[str, Any], params: Dict[str, Any])

Bases: TaskCreated, Event

originator_id_type

alias of UUID

class TaskFailed(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime', error_message: str)

Bases: DecoratedFuncCaller, TaskFailed

class TaskStarted(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime')

Bases: DecoratedFuncCaller, TaskStarted

add_job_id = <eventsourcing.domain.UnboundCommandMethodDecorator object>

add_result(result: Result)

claim_launcher = <eventsourcing.domain.UnboundCommandMethodDecorator object>

complete_task = <eventsourcing.domain.UnboundCommandMethodDecorator object>

fail_task = <eventsourcing.domain.UnboundCommandMethodDecorator object>

is_claimed_by_other(owner: str, now: datetime, lease_seconds: float) → bool

Whether someone else holds a claim that has not expired yet.

Parameters:

• owner (str) – The supervisor that wants to claim the task.

• now (datetime) – The current time (timezone-aware).

• lease_seconds (float) – How long a claim stays valid.

Returns:

True when another owner’s claim is still within its lease.

Return type:

bool

record_storage_provenance = <eventsourcing.domain.UnboundCommandMethodDecorator object>

release_launcher = <eventsourcing.domain.UnboundCommandMethodDecorator object>

start_task = <eventsourcing.domain.UnboundCommandMethodDecorator object>

update_task_progress = <eventsourcing.domain.UnboundCommandMethodDecorator object>

update_task_status = <eventsourcing.domain.UnboundCommandMethodDecorator object>

exception biomero.eventsourcing.WorkflowCancelled

Bases: WorkflowInterrupted

A workflow was asked to cancel: stop, cancel its jobs, fail it.

exception biomero.eventsourcing.WorkflowInterrupted

Bases: Exception

Whoever runs a workflow must stop now, between its steps.

Deliberately not an InterruptedError: that is an OSError, which except OSError handlers and the SSH recovery classifier would catch. Stage handlers let it pass; subclasses say why the run stopped.

class biomero.eventsourcing.WorkflowRun(*args: Any, **kwargs: Any)

Bases: Aggregate

Represents a workflow run as an aggregate in the domain model.

name

The name of the workflow.

Type:

str

description

A description of the workflow.

Type:

str

user

The ID of the user who initiated the workflow.

Type:

int

group

The ID of the group associated with the workflow.

Type:

int

tasks

A list of task UUIDs associated with the workflow.

Type:

list

class CancelRequested(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime', reason: str)

Bases: DecoratedFuncCaller, CancelRequested

class CanonicalInputsRecorded(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime', export_task_id: uuid.UUID, canonical_inputs: List[Dict[str, Any]])

Bases: DecoratedFuncCaller, CanonicalInputsRecorded

class Created(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime')

Bases: Event, Created

originator_id_type

alias of UUID

class Event(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime')

Bases: Event

originator_id_type

alias of UUID

INITIAL_VERSION: int = 0

class TaskAdded(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime', task_id: uuid.UUID)

Bases: DecoratedFuncCaller, TaskAdded

class WorkflowCompleted(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime')

Bases: DecoratedFuncCaller, WorkflowCompleted

class WorkflowFailed(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime', error_message: str)

Bases: DecoratedFuncCaller, WorkflowFailed

class WorkflowInitiated(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime', originator_topic: 'str', name: str, description: str, user: int, group: int)

Bases: WorkflowInitiated, Event

originator_id_type

alias of UUID

class WorkflowStarted(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime')

Bases: DecoratedFuncCaller, WorkflowStarted

add_task = <eventsourcing.domain.UnboundCommandMethodDecorator object>

complete_workflow() → bool

Complete the workflow, unless it already reached a final state.

The refusal is not part of the events, so streams that already contain two final events still replay; only new transitions are refused.

Returns:

False when the workflow was already final and nothing

was recorded.

Return type:

bool

fail_workflow(error_message: str) → bool

Fail the workflow, unless it already reached a final state.

Returns:

False when the workflow was already final and nothing

was recorded.

Return type:

bool

record_canonical_inputs = <eventsourcing.domain.UnboundCommandMethodDecorator object>

request_cancel = <eventsourcing.domain.UnboundCommandMethodDecorator object>

start_workflow = <eventsourcing.domain.UnboundCommandMethodDecorator object>

class biomero.eventsourcing.WorkflowTracker(env: Mapping[str, str] | None = None)

Bases: Application

Application service class for managing workflow and task lifecycle operations.

initiate_workflow()

Creates a new workflow.

add_task_to_workflow()

Adds a new task to an existing workflow.

start_workflow()

Starts a workflow by its UUID.

complete_workflow()

Marks a workflow as completed.

fail_workflow()

Marks a workflow as failed with an error message.

start_task()

Starts a task by its UUID.

complete_task()

Marks a task as completed with a result message.

fail_task()

Marks a task as failed with an error message.

add_job_id()

Adds a job ID to a task.

add_result()

Adds a result to a task.

update_task_status()

Updates the status of a task.

update_task_progress()

Updates the progress of a task.

add_job_id(task_id, slurm_job_id)

Adds a Slurm job ID to the task with the given UUID.

Parameters:

• task_id (UUID) – The UUID of the task.

• slurm_job_id (str) – The Slurm job ID to associate with the task.

add_launcher_task(workflow_id: UUID, task_name: str, task_version: str, input_data: Dict[str, Any], kwargs: Dict[str, Any]) → UUID

Adds the launcher task that queues a workflow for a detached supervisor: an ordinary task of a launching script, marked as launcher.

What goes into kwargs is up to the caller (the script knows what its run needs); this only adds the marker, with the script version as its value, to a copy of it.

Parameters:

• workflow_id (UUID) – The UUID of the workflow.

• task_name (str) – The launching script, one of constants.LAUNCHER_TASK_NAMES.

• task_version (str) – The version of the launching script.

• input_data (dict) – Input data for the task.

• kwargs (dict) – Everything the supervisor needs to run the workflow.

Returns:

The UUID of the launcher task.

Return type:

UUID

Raises:

ValueError – If the task name is not a launching script.

add_result(task_id, result)

Adds a result to the task with the given UUID.

Parameters:

• task_id (UUID) – The UUID of the task.

• result (Result) – The Fabric result object to add to the task.

add_task_to_workflow(workflow_id: UUID, task_name: str, task_version: str, input_data: Dict[str, Any], kwargs: Dict[str, Any]) → UUID

Adds a task to the specified workflow.

Parameters:

• workflow_id (UUID) – The UUID of the workflow.

• task_name (str) – The name of the task.

• task_version (str) – The version of the task.

• input_data (dict) – Input data for the task.

• kwargs (dict) – Additional parameters for the task.

Returns:

The UUID of the newly added task.

Return type:

UUID

claim_launcher_task(task_id: UUID, owner: str, lease_seconds: float = 300) → bool

Claims a launcher task for a supervisor, atomically.

This is a compare-and-set on the task’s event stream: the claim is appended as the next event of the task, and the event store accepts only one event per position. When two supervisors read the same task and both try to claim it, one of them loses and gets False.

A claim is valid for lease_seconds. The owner renews it by claiming again; a claim that was not renewed (for example because the supervisor crashed) can be taken over by any other owner. Renew well within the lease (the processor renews three times per lease). The lease compares the claim’s timestamp with this process’s clock, so supervisors must have clocks that agree to well within the lease; the event-store position, not the clock, is what keeps two claims apart.

Parameters:

• task_id (UUID) – The launcher task to claim.

• owner (str) – Identity of the claiming supervisor.

• lease_seconds (float, optional) – How long a claim stays valid. Defaults to constants.LAUNCHER_CLAIM_LEASE_SECONDS.

Returns:

True when the caller now holds the claim.

Return type:

bool

Raises:

ValueError – If the task is not a launcher task.

commit_open_transaction() → None

Commits what was saved on a datastore that leaves the transaction to its caller (the scoped-session store); other stores commit on save.

Use this after save() for aggregates that are not written through one of the tracker’s own methods.

complete_task(task_id: UUID, message: str)

Marks the task with the given UUID as completed with a result message.

Parameters:

• task_id (UUID) – The UUID of the task to complete.

• message (str) – The result message of the task.

complete_workflow(workflow_id: UUID)

Marks the workflow with the given UUID as completed.

Parameters:

workflow_id (UUID) – The UUID of the workflow to complete.

fail_task(task_id: UUID, error_message: str)

Marks the task with the given UUID as failed with an error message.

Parameters:

• task_id (UUID) – The UUID of the task to fail.

• error_message (str) – The error message describing the failure.

fail_workflow(workflow_id: UUID, error_message: str)

Marks the workflow with the given UUID as failed with an error message.

Parameters:

• workflow_id (UUID) – The UUID of the workflow to fail.

• error_message (str) – The error message describing the failure.

find_launcher_task(workflow_id: UUID) → Task | None

Finds the launcher task that queued a workflow for a supervisor.

Parameters:

workflow_id (UUID) – The UUID of the workflow.

Returns:

The launcher task, or None when the workflow was not

queued for a supervisor.

Return type:

Task

Raises:

Exception – When the workflow or one of its tasks cannot be read.

get_canonical_input_manifest(workflow_id: UUID) → CanonicalInputManifest

Returns the validated, JSON-serializable recovery manifest.

get_canonical_inputs(workflow_id: UUID) → Dict[str, Any]

Returns the canonical input snapshot recorded for a workflow.

initiate_workflow(name: str, description: str, user: int, group: int) → UUID

Initiates a new workflow.

Parameters:

• name (str) – The name of the workflow.

• description (str) – A description of the workflow.

• user (int) – The ID of the user initiating the workflow.

• group (int) – The group ID associated with the workflow.

Returns:

The UUID of the newly initiated workflow.

Return type:

UUID

is_workflow_cancel_requested(workflow_id: UUID) → bool

Whether a cancel was requested for the workflow.

Parameters:

workflow_id (UUID) – The workflow to check.

Returns:

True once request_workflow_cancel() was called for it.

Return type:

bool

is_workflow_finished(workflow_id: UUID) → bool

Whether the workflow already completed or failed.

Parameters:

workflow_id (UUID) – The workflow to check.

Returns:

True once the workflow reached a final state.

Return type:

bool

iter_workflow_tasks(workflow_id: UUID) → Iterator[Task]

Yields the tasks of a workflow, in the order they were added.

Read errors are raised, not hidden: an unreadable history is not an empty one.

Parameters:

workflow_id (UUID) – The UUID of the workflow.

Yields:

Task – One task aggregate at a time.

name = 'WorkflowTracker'

raise_if_cancel_requested(workflow_id: UUID) → None

Stops the caller when a cancel was requested for the workflow. Call it between the steps of a workflow run.

Parameters:

workflow_id (UUID) – The workflow that is being run.

Raises:

WorkflowCancelled – When request_workflow_cancel() was called.

record_canonical_inputs(workflow_id: UUID, export_task_id: UUID, canonical_inputs: Sequence[CanonicalInput | Mapping[str, Any]]) → None

Records the exact canonical input generations used by a workflow.

record_storage_provenance(task_id, target_key, evidence)

Persist observed storage provenance for an opaque result target.

release_database_connections() → None

Gives this tracker’s database connections back, for a thread that is done with its own tracker.

release_launcher_task(task_id: UUID, owner: str) → bool

Gives a claim up, so another supervisor can take the run over at once instead of after the lease has expired.

Parameters:

• task_id (UUID) – The launcher task.

• owner (str) – Identity of the supervisor that holds the claim.

Returns:

True when the claim was released; False when the owner did

not hold it.

Return type:

bool

request_workflow_cancel(workflow_id: UUID, reason: str) → None

Records that a workflow must be cancelled. Whoever executes the workflow (the inline script or the detached supervisor) reads this and stops; asking again changes nothing.

Parameters:

• workflow_id (UUID) – The workflow to cancel.

• reason (str) – Why, for the record.

start_task(task_id: UUID)

Starts the task with the given UUID.

Parameters:

task_id (UUID) – The UUID of the task to start.

start_workflow(workflow_id: UUID)

Starts the workflow with the given UUID.

Parameters:

workflow_id (UUID) – The UUID of the workflow to start.

update_task_progress(task_id, progress)

Updates the progress of the task with the given UUID.

Parameters:

• task_id (UUID) – The UUID of the task.

• progress (str) – The updated progress of the task.

update_task_status(task_id, status)

Updates the status of the task with the given UUID.

Parameters:

• task_id (UUID) – The UUID of the task.

• status (str) – The new status of the task.

biomero.eventsourcing.is_launcher_task(task) → bool

Whether a task is a launcher queued for a detached supervisor.

Works on a Task aggregate and on a Task.TaskCreated event alike, so every reader (supervisor, views, provenance) applies the same rule.

Parameters:

task – Anything with task_name and params.

Returns:

True for a launching script’s task that carries the marker.

Return type:

bool

biomero.views module

class biomero.views.JobAccounting(*args, **kwargs)

Bases: ProcessApplication

get_jobs(user=None, group=None)

Retrieve jobs for a specific user and/or group.

Parameters: - user (int, optional): The user ID to filter by. - group (int, optional): The group ID to filter by.

Returns: - Dictionary of user IDs to lists of job IDs if no user is specified. - Dictionary with a single user ID key and a list of job IDs if user is specified.

Raises: - ValueError: If neither user nor group is provided.

get_task_id(job_id)

Retrieve the task ID associated with a given job ID.

Parameters: - job_id (int): The job ID (slurm_job_id) to look up.

Returns: - UUID: The task ID associated with the job ID, or None if not found.

name = 'JobAccounting'

policy(domain_event, process_event)

policy(domain_event: WorkflowInitiated, process_event)

policy(domain_event: TaskAdded, process_event)

policy(domain_event: JobIdAdded, process_event)

Default policy

resolve_task(task_id)

Look a task up when we did not see it created ourselves.

resolve_workflow(wf_id)

Look a workflow up when we did not see it started ourselves.

topics: Sequence[str] = ()

update_view_table(job_id, user, group, task_id)

Update the view table with new job information.

Uses merge (upsert) so that if SLURM recycles a job ID the new task_id/user/group overwrites the stale row instead of silently failing and leaving the old mapping in place.

class biomero.views.JobProgress(*args, **kwargs)

Bases: ProcessApplication

name = 'JobProgress'

policy(domain_event, process_event)

policy(domain_event: JobIdAdded, process_event)

policy(domain_event: StatusUpdated, process_event)

policy(domain_event: ProgressUpdated, process_event)

Default policy

topics: Sequence[str] = ()

update_view_table(job_id)

Update the view table with new job status and progress information.

class biomero.views.WorkflowAnalytics(*args, **kwargs)

Bases: ProcessApplication

get_average_task_duration(user=None, group=None)

Retrieve the average task duration grouped by task name and version.

Parameters: - user (int, optional): The user ID to filter by. - group (int, optional): The group ID to filter by.

Returns: - Dictionary of task names and versions to average duration (in seconds).

get_task_counts(user=None, group=None)

Retrieve task execution counts grouped by task name and version.

Parameters: - user (int, optional): The user ID to filter by. - group (int, optional): The group ID to filter by.

Returns: - Dictionary of task names and versions to counts.

get_task_failures(user=None, group=None)

Retrieve tasks that failed, grouped by task name and version.

Parameters: - user (int, optional): The user ID to filter by. - group (int, optional): The group ID to filter by.

Returns: - Dictionary of task names and versions to lists of failure reasons.

get_task_usage_over_time(task_name, user=None, group=None)

Retrieve task usage over time for a specific task.

Parameters: - task_name (str): The name of the task to filter by. - user (int, optional): The user ID to filter by. - group (int, optional): The group ID to filter by.

Returns: - Dictionary mapping date to the count of task executions on that date.

name = 'WorkflowAnalytics'

policy(domain_event, process_event)

policy(domain_event: WorkflowInitiated, process_event)

policy(domain_event: TaskAdded, process_event)

policy(domain_event: TaskCreated, process_event)

policy(domain_event: TaskStarted, process_event)

policy(domain_event: StatusUpdated, process_event)

policy(domain_event: TaskCompleted, process_event)

policy(domain_event: TaskFailed, process_event)

Default policy

resolve_task(task_id)

Look a task up when we did not see it created ourselves.

resolve_workflow(wf_id)

topics: Sequence[str] = ()

update_view_table(task_id)

Update the view table with new task execution information.

class biomero.views.WorkflowProgress(*args, **kwargs)

Bases: ProcessApplication

name = 'WorkflowProgress'

policy(domain_event, process_event)

policy(domain_event: WorkflowInitiated, process_event)

policy(domain_event: WorkflowCompleted, process_event)

policy(domain_event: WorkflowFailed, process_event)

policy(domain_event: TaskAdded, process_event)

policy(domain_event: TaskCreated, process_event)

policy(domain_event: TaskStarted, process_event)

policy(domain_event: StatusUpdated, process_event)

policy(domain_event: ProgressUpdated, process_event)

Default policy

resolve_task(task_id)

Look a task up when we did not see it created ourselves.

resolve_workflow(wf_id)

topics: Sequence[str] = ()

update_view_table(wf_id)

Update the view table with new workflow status, progress, user, and group.

biomero.views.tracker_repository()

Read-only access to the workflow tracker’s aggregates.

biomero.provenance module

Versioned OMERO key/value views of immutable workflow history.

This module neither writes events nor changes OMERO objects. Full CSV export is deliberately independent of this searchable, selectively reduced view.

class biomero.provenance.MetadataAnnotation(namespace: str, values: dict)

Bases: object

class biomero.provenance.MetadataChange(before: biomero.provenance.MetadataAnnotation, after: biomero.provenance.MetadataAnnotation | None)

Bases: object

biomero.provenance.plan_metadata_refresh(tracker, workflow_id, annotations, *, view_version='v0', target_key=None, refresh_storage=False)

Plan a snapshot-preserving update. None means unlink, not global delete.

Unknown namespaces and extra keys are preserved. Missing task snapshots, duplicate identities and changed namespaces fail closed. Large fields that were already replaced by CSV references remain reduced.

Explicit storage repair may advance the import-task snapshot to a version carrying newly recorded evidence for target_key. Other snapshots stay pinned. Existing, different storage evidence is never silently replaced.

biomero.provenance.render_workflow_metadata(tracker, workflow_id, *, view_version='v0', aggregate_versions=None, revision_fields=True, target_key=None)

Return ordered namespace/value pairs without writing to any datastore.

v0 is the only supported view and retains legacy task/job fields, including job command, environment and result-message fields. It omits detached coordination tasks and duplicated output settings. Scientific false/zero values are retained. Version still means the software version.

An explicit aggregate_versions mapping pins the workflow and every retained task independently. Missing versions fail instead of mixing a historical workflow with current task state.

biomero.provenance.resolve_metadata_versions(tracker, workflow_id, annotations)

Recover exact aggregate versions from existing KV snapshots.

Legacy maps have no revision number: require a unique Modified_On match. Job maps inherit the paired task snapshot, never the current task version. Missing/inconsistent snapshots are errors; callers must not guess.

biomero.provenance.storage_metadata_view(evidence, target_key)

Compact per-result storage facts; feature flags are not evidence.

biomero.detached module

The detached-mode setting, and aliases for names that moved.

A BIOMERO workflow run outlives the OMERO session that asks for it, so a script can queue the run in a launcher task and a supervisor (the OMERO processor in NL-BIOMERO) executes it. Core only provides the mechanisms:

• the launcher marker and rule: constants.LAUNCHER_MARKER, WorkflowTracker.add_launcher_task and eventsourcing.is_launcher_task;

• task lookup and the atomic claim: WorkflowTracker.find_launcher_task, WorkflowTracker.claim_launcher_task and WorkflowTracker.release_launcher_task;

• cancellation: WorkflowTracker.request_workflow_cancel and SlurmClient.cancel_slurm_jobs.

The scripts decide what to record in a launcher task and how to resume a run; the supervisor decides which runs to execute. This module holds the one reader of the setting that switches the feature on, which the scripts and the supervisor all need, plus aliases that keep the names of the 2.9 betas working. The aliases are deprecated: use the names above.

biomero.detached.detached_mode_enabled() → bool

Whether workflow scripts should queue runs instead of executing them.

Opt-in through the BIOMERO_DETACHED_WORKFLOWS environment variable of the OMERO processor running the script, because it requires a supervisor to be present: without one, a queued run would never be executed. The variable is declared in constants.slurm_env so that the processor forwards it to the script subprocesses it starts, which is where this is read. This is the only reader of the variable: callers use this function.

Returns:

True when the variable has one of constants.TRUTHY_VALUES.

Return type:

bool

biomero.maintenance module

Durable metadata-maintenance requests, separate from analysis workflows.

Only plain request options and compact outcomes are persisted. The processor owns execution; the scripts layer owns all external annotation operations.

Why this is in core and not in the processor or the scripts: a request is an aggregate in the WorkflowTracker event store, next to the workflow aggregates. Every projection of that store (biomero.views) resolves the class of each event it reads from the event’s stored topic, and the topic is the import path of the class: biomero.maintenance:MetadataRefresh.Started and so on. A class that only the processor or a script can import would make every other process that replays the store fail on these events, and moving the class would orphan the requests already stored. The request lifecycle (this aggregate and its notification scan) therefore stays here; running a request does not.

class biomero.maintenance.MetadataRefresh(*args: Any, **kwargs: Any)

Bases: Aggregate

A resumable maintenance request in the WorkflowTracker event store.

The public methods guard the lifecycle and raise ValueError for a transition that cannot happen. The guards are not part of the events, so a stream written before they existed still replays.

class Created(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime', originator_topic: 'str')

Bases: Created, Event

originator_id_type

alias of UUID

class Event(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime')

Bases: Event

originator_id_type

alias of UUID

class Finished(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime')

Bases: DecoratedFuncCaller, Event

class Progress(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime')

Bases: DecoratedFuncCaller, Event

class Started(self, originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime')

Bases: DecoratedFuncCaller, Event

finished(report: dict, error: str = '') → None

Record the outcome: FAILED on an error or failed targets, else DONE.

Parameters:

• report (dict) – The outcome of the refresh, with the failure count under report["counts"]["failed"].

• error (str, optional) – Why the request could not run at all.

is_finished() → bool

Whether the request reached a terminal state.

progressed(report: dict) → None

Record the progress report of a running request.

started() → None

Mark the request as running; also done when a restart resumes it.

biomero.maintenance.metadata_refresh_statuses(tracker, recent: int = 5) → List[Dict]

Plain snapshots of all active and the most recent finished requests.

This replays the request notifications from the start on every call, which is meant for an occasional administrator status check.

Parameters:

• tracker (WorkflowTracker) – Tracker whose event store holds requests.

• recent (int, optional) – How many finished requests to include.

Returns:

One dict per request, newest first.

Return type:

list

biomero.maintenance.pending_metadata_refreshes(tracker, start: int = 1, pending: Set[UUID] | None = None) → Tuple[int, Set[UUID]]

Catch up on request notifications; a RUNNING request stays pending.

A request that was RUNNING when its supervisor stopped has no Finished event, so it is pending again after a restart and gets resumed.

Parameters:

• tracker (WorkflowTracker) – Tracker whose notification log to read.

• start (int, optional) – Notification cursor from a previous call.

• pending (set, optional) – Pending request IDs from a previous call.

Returns:

The cursor to continue from, and the IDs of the requests that

are queued or running.

Return type:

tuple

biomero.maintenance.queue_metadata_refresh(tracker, user: int, group: int, options: dict) → UUID

Persist the handoff before the requesting script closes its session.

Parameters:

• tracker (WorkflowTracker) – Tracker whose event store holds requests.

• user (int) – OMERO user ID of the administrator who asked.

• group (int) – OMERO group ID to run in.

• options (dict) – Plain request options for the refresh.

Returns:

The request ID.

Return type:

UUID

biomero.remote_shallower module

Optional CPU result shallowing before result archiving.

SlurmClient owns remote execution, image acquisition, job submission and scheduling policy; SlurmJob owns monitoring. This module coordinates the shallower’s manifest, durable submission records, recovery command and receipt validation.

class biomero.remote_shallower.HelperJob(task_id: Any, task: Task, image: str, sif_path: str, output_path: str, manifest_path: str, state_dir: str)

The task, image and remote paths of one helper (shallower) job.

task is the snapshot read when the job was started or resumed; read the task again with task_id where recorded job IDs may have changed.

image: str

Alias for field number 2

manifest_path: str

Alias for field number 5

output_path: str

Alias for field number 4

sif_path: str

Alias for field number 3

state_dir: str

Alias for field number 6

task: Task

Alias for field number 1

task_id: Any

Alias for field number 0

class biomero.remote_shallower.HelperSettings(contract_version: int = 1, manifest_schema: int = 2, polling_interval: int = 15, lock_wait_seconds: int = 30)

Values the entry point can change, threaded to the steps that use them.

Each value defaults to the module constant of the same meaning.

contract_version: int

Alias for field number 0

lock_wait_seconds: int

Alias for field number 3

manifest_schema: int

Alias for field number 1

polling_interval: int

Alias for field number 2

biomero.remote_shallower.build_helper_command(slurm_client: SlurmClient, output_path: str, sif_path: str, manifest_path: str, task_id: str, *, recovery: bool = False, image: str | None = None, contract_version: int = 1) → str

Build the shallowing or recovery job using one helper resource policy.

Paths refer to the remote filesystem. The job is submitted through the SlurmClient, so the configured site defaults apply.

Parameters:

• slurm_client (SlurmClient) – The Slurm client holding the configuration.

• output_path (str) – Result directory to shallow.

• sif_path (str) – Path of the helper image.

• manifest_path (str) – Path of the canonical input manifest.

• task_id (str) – ID of the helper task.

• recovery (bool) – Build the recovery command instead of the shallowing command (default False).

• image (Optional[str]) – Image reference recorded for the task; defaults to the configured remote_shallower_image.

• contract_version (int) – Runtime contract the helper must follow.

Returns:

The sbatch command.

Return type:

str

Raises:

ValueError – If the configured worker count is not positive.

biomero.remote_shallower.completed_receipts(slurm_client: SlurmClient, workflow_id: Any, canonical: CanonicalInputManifest) → Tuple[RemoteShallowReceipt, ...]

Read persisted helper receipts without contacting the cluster.

Parameters:

• slurm_client (SlurmClient) – The Slurm client with the workflow tracker.

• workflow_id (Any) – ID of the workflow.

• canonical (CanonicalInputManifest) – The canonical input manifest the receipts must match.

Returns:

The receipts of completed helper

tasks; empty when remote shallowing or workflow tracking is disabled.

Return type:

Tuple[RemoteShallowReceipt, …]

biomero.remote_shallower.helper_job_name(task_id: Any, *, recovery: bool = False) → str

Keep submission reconciliation specific to one task and operation.

Parameters:

• task_id (Any) – ID of the helper task.

• recovery (bool) – Name the recovery job instead of the shallowing job (default False).

Returns:

The Slurm job name.

Return type:

str

biomero.remote_shallower.image_spec(slurm_client: SlurmClient, *, image: str | None = None) → Dict[str, str]

Describe the configured helper image for SlurmClient image acquisition.

Parameters:

• slurm_client (SlurmClient) – The Slurm client holding the configuration.

• image (Optional[str]) – Image reference to describe instead of the configured remote_shallower_image.

Returns:

The image specification for image acquisition.

Return type:

Dict[str, str]

Raises:

ValueError – If no image is configured, or its digest or version is missing or invalid.

biomero.remote_shallower.installed_tool_version(slurm_client: SlurmClient, sif_path: str) → str

Read the installed helper’s version without executing its payload.

A floating image tag is not a tool version. Persist the OCI version label before submission so receipt validation and recovery use a concrete value.

Parameters:

• slurm_client (SlurmClient) – The Slurm client used to run commands.

• sif_path (str) – Remote path of the installed helper image.

Returns:

The installed tool version.

Return type:

str

biomero.remote_shallower.parse_batch_report(slurm_client: SlurmClient, report_json: str, canonical: CanonicalInputManifest, *, task: Task | None = None) → ShallowBatchReport

Parse a helper batch report and check its input, image and tool version.

When a task is supplied, its recorded image/version and submission own the report; current deployment defaults must not invalidate historical results. Import the optional schema dependency only when shallowing is used.

Parameters:

• slurm_client (SlurmClient) – The Slurm client holding the configuration.

• report_json (str) – The JSON batch report written by the helper.

• canonical (CanonicalInputManifest) – The canonical input manifest the report must match.

• task (Optional[Task]) – The recorded helper task owning the report.

Returns:

The parsed batch report.

Return type:

ShallowBatchReport

Raises:

ValueError – If the report does not match the configuration, input, task or job.

biomero.remote_shallower.publish_or_verify_manifest(slurm_client: SlurmClient, manifest_path: str, canonical: CanonicalInputManifest, *, submitted: bool, lock_wait_seconds: int = 30) → None

Verify an existing manifest or publish a new one without overwriting.

A sibling temporary file is fully uploaded before an atomic hard-link publishes it under a lock. Readers never see a partly uploaded manifest. Existing content is verified, not replaced on resume.

Parameters:

• slurm_client (SlurmClient) – The Slurm client used to run commands.

• manifest_path (str) – Remote path of the canonical input manifest.

• canonical (CanonicalInputManifest) – The canonical input manifest to publish or verify.

• submitted (bool) – Whether the helper job was already submitted, which makes a missing manifest an error.

• lock_wait_seconds (int) – How long to wait for the publication lock.

Raises:

• RuntimeError – If the manifest cannot be read or published, or is missing for a submitted job.

• ValueError – If an existing manifest differs from the workflow input.

biomero.remote_shallower.sha256_of_json(payload: Any) → str

Fingerprint JSON content independently of key order and whitespace.

Parameters:

payload (Any) – A JSON-serializable value.

Returns:

The hex SHA-256 of the canonical JSON encoding.

Return type:

str

biomero.remote_shallower.shallow_workflow_results(slurm_client: SlurmClient, data_path: str, workflow_id: Any, canonical: CanonicalInputManifest, heartbeat: Callable[[], None] | None = None, *, contract_version: int = 1, manifest_schema: int = 2, polling_interval: int = 15, lock_wait_seconds: int = 30) → ShallowBatchReport | None

Shallow the workflow results on the cluster, or resume that work.

Only initialization acquires container images. A rejected shallowing submission returns an empty-receipt report so the caller can retain full results. Callers must not archive potentially incomplete output.

Parameters:

• slurm_client (SlurmClient) – The Slurm client with the workflow tracker.

• data_path (str) – Remote data folder of the workflow results.

• workflow_id (Any) – ID of the workflow.

• canonical (CanonicalInputManifest) – The canonical input manifest of the workflow.

• heartbeat (Optional[Callable[[], None]]) – Callback invoked during job waits.

• contract_version (int) – Runtime contract the helper must follow.

• manifest_schema (int) – Shallow manifest schema the helper must emit.

• polling_interval (int) – Seconds between job status polls.

• lock_wait_seconds (int) – How long to wait for the manifest lock.

Returns:

The validated report, or None when

shallowing is disabled or inapplicable.

Return type:

Optional[ShallowBatchReport]

Raises:

• RuntimeError – If the image is missing, the submission or recovery is unresolved or failed, or the report is missing.

• ValueError – If the installed tool or the report is invalid.

biomero.remote_shallower.validate_installed_tool(slurm_client: SlurmClient, sif_path: str, *, expected_version: str | None = None, runtime_contract: int = 1, manifest_schema: int = 2) → str

Fail closed unless the installed SIF satisfies this runtime contract.

Parameters:

• slurm_client (SlurmClient) – The Slurm client used to run commands.

• sif_path (str) – Remote path of the installed helper image.

• expected_version (Optional[str]) – Version the image must have; defaults to the configured remote_shallower_version.

• runtime_contract (int) – Runtime contract the image must support.

• manifest_schema (int) – Shallow manifest schema the image must emit.

Returns:

The installed tool version.

Return type:

str

Raises:

• ValueError – If the labels or the configured version are invalid.

• RuntimeError – If the version, capability schema, runtime contract or manifest schema is not supported.

biomero.remote_shallower.verify_task_canonical_inputs(task: Task, canonical: CanonicalInputManifest) → None

Raise unless the canonical inputs equal those recorded on the task.

Parameters:

• task (Any) – The recorded helper task.

• canonical (Any) – The canonical input manifest of the workflow.

Raises:

ValueError – If the task recorded a different canonical manifest.

biomero.remote_shallower.wait_for_helper_job(slurm_client: SlurmClient, job_id: int, heartbeat: Callable[[], None] | None = None, polling_interval: int = 15) → str

Monitor an adopted job without analysis-log or task-status updates.

The helper task is completed after report validation, not merely because its Slurm process exited successfully.

Parameters:

• slurm_client (SlurmClient) – The Slurm client used to poll the job.

• job_id (int) – The Slurm job ID.

• heartbeat (Optional[Callable[[], None]]) – Callback invoked before each poll.

• polling_interval (int) – Seconds between status polls.

Returns:

The final state of the Slurm job.

Return type:

str

biomero.ssh_recovery module

Bounded recovery from interrupted SSH connections.

An SSH connection to the cluster can drop while a Slurm job keeps running. This module retries an operation that is safe to repeat, within a time window (ssh_recovery_timeout), and stops with a clear error when the window ends. A window of 0 means no retries: the same code runs once and raises the original error.

Only transport loss is retried: never authentication, never a command that failed on the cluster, and never an arbitrary command. The caller decides which operations are safe to repeat and wraps them in retry. The Slurm side (never submitting twice) lives in SlurmClient and remote_scripts.

exception biomero.ssh_recovery.IncompleteTransfer

Bases: ConnectionError

A file transfer ended with a different size than its source.

class biomero.ssh_recovery.OutageWindow(timeout: float, outage_deadline: float | None = None, pending_deadline: float | None = None, delay: float = 1)

Bases: object

The time budget of one retry call.

A deadline starts at the first failure of its kind and is never moved by anything else, so that repeated failures cannot extend a call forever. A new retry call (the next poll, the next transfer) gets a new window.

timeout

Seconds a kind of failure may last.

Type:

float

outage_deadline

End of the SSH outage window.

Type:

Optional[float]

pending_deadline

End of the window in which a healthy remote operation is awaited.

Type:

Optional[float]

delay

The next pause between attempts, in seconds.

Type:

float

next_pause(seconds_left: float) → float

Return the pause before the next attempt and grow the delay.

seconds_left(error: BaseException, now: float) → float

Seconds left in the window of this kind of failure.

start_if_new(error: BaseException, now: float) → None

Open the window of this kind of failure at its first occurrence.

exception biomero.ssh_recovery.ReadOnlyCommandTimeout

Bases: TimeoutError

A short read-only command did not answer within its bound.

exception biomero.ssh_recovery.RecoveryPending

Bases: Exception

A durable remote operation is still locked or not yet visible.

The connection works; the operation may still be running remotely. It is asked again after a pause, and never resubmitted.

class biomero.ssh_recovery.RecoveryState(timeout: float, windows: ~typing.List[~biomero.ssh_recovery.OutageWindow] = <factory>, heartbeats: ~typing.List[~typing.Callable[[], None]] = <factory>, heartbeat_error: Exception | None = None)

Bases: object

What nested operations of one client share while recovering.

timeout

The recovery window in seconds; 0 means no retries.

Type:

float

windows

The windows of the active retry calls, outermost first.

Type:

List[OutageWindow]

heartbeats

Caller callbacks that run during waits (for example an OMERO session keepalive).

Type:

List[Callable[[], None]]

heartbeat_error

An error raised by a callback; it is the caller’s, so it is never retried.

Type:

Optional[Exception]

pulse() → None

Run the caller callbacks once each.

Raises:

Exception – Whatever a callback raised; it is remembered so that it is not mistaken for an SSH error.

static raise_if_cancelled(cancel: Any | None, description: str) → None

Raise SSHRecoveryCancelled when the caller’s event is set.

seconds_left_in_outage() → float | None

Seconds left in the tightest active outage window, if any.

wait(seconds: float, cancel: Any | None, description: str) → None

Sleep, staying responsive to cancellation and heartbeats.

exception biomero.ssh_recovery.RemoteOperationStillPending

Bases: RuntimeError

A remote operation was still running when its waiting time ended.

The SSH connection is healthy and nothing was repeated: the operation continues on the cluster and can be asked again later.

exception biomero.ssh_recovery.SSHRecoveryCancelled

Bases: InterruptedError

The caller cancelled while waiting for connectivity.

exception biomero.ssh_recovery.SSHRecoveryTimeout

Bases: ConnectionError

Connectivity did not recover within the configured time window.

biomero.ssh_recovery.atomic_get(connection: Connection, remote: str, directory: str) → Result

Download a file so that the local name never shows a partial file.

The data goes to a temporary local file, is checked against the remote file (unchanged and complete), and then replaces the local file. An earlier complete local file stays available until then.

Parameters:

• connection (Connection) – The SlurmClient.

• remote (str) – The remote path.

• directory (str) – The local directory to download into.

Returns:

The result, with local set to the final path.

Return type:

TransferResult

Raises:

IncompleteTransfer – If the remote file changed or the download is incomplete (retried).

biomero.ssh_recovery.atomic_put(connection: Connection, local: str | StringIO | BytesIO, remote: str, partial_suffix: str = '.biomero-partial') → Result

Upload a file so that the remote name never shows a partial file.

The data goes to <remote><partial_suffix> first, is checked for its size, and is renamed to remote. An interrupted upload restarts from the beginning. A previous remote file keeps its permission bits. Concurrent uploads to one remote name are not coordinated: the last rename wins.

Parameters:

• connection (Connection) – The SlurmClient.

• local (Union[str, io.StringIO, io.BytesIO]) – A local path or an in-memory source.

• remote (str) – The final remote path.

• partial_suffix (str) – Suffix of the temporary remote name.

Returns:

The result, with remote set to the final path.

Return type:

TransferResult

Raises:

IncompleteTransfer – If the uploaded size is wrong (retried).

biomero.ssh_recovery.bounded_read_timeout(client: Any, requested: float | None) → float | None

Return the timeout for a short read-only command.

Parameters:

• client (Any) – The SlurmClient, inside a scope.

• requested (Optional[float]) – A longer timeout the caller asked for because the read is legitimately slow.

Returns:

The requested timeout (capped by what is left of an

outage), or the detection limit when none was requested. The request is returned unchanged when recovery is off.

Return type:

Optional[float]

biomero.ssh_recovery.download(connection: Connection, remote: str, directory: str) → Result

Download a file so that the local name never shows a partial file.

The data goes to a temporary local file, is checked against the remote file (unchanged and complete), and then replaces the local file. An earlier complete local file stays available until then.

Parameters:

• connection (Connection) – The SlurmClient.

• remote (str) – The remote path.

• directory (str) – The local directory to download into.

Returns:

The result, with local set to the final path.

Return type:

TransferResult

Raises:

IncompleteTransfer – If the remote file changed or the download is incomplete (retried).

biomero.ssh_recovery.io_timeout(client: Any, configured: float | None = None) → float

Return the native limit for connection stages and short reads.

This is a detection limit, not a limit on healthy long commands. It is never longer than the recovery window, and during an outage never longer than what is left of it.

Parameters:

• client (Any) – The SlurmClient, inside a scope.

• configured (Optional[float]) – A stricter limit set by the caller, which stays in force.

Returns:

The limit in seconds.

Return type:

float

biomero.ssh_recovery.is_transient(error: BaseException) → bool

Tell whether an error is transport loss (see transient_reason).

Parameters:

error (BaseException) – The error raised by an SSH operation.

Returns:

True if retrying may help.

Return type:

bool

biomero.ssh_recovery.limit_channel_open(client: Any) → None

Limit channel opening on the current transport, within native_timeouts.

Remembers the transport’s own setting so that native_timeouts can put it back. Does nothing outside native_timeouts.

Parameters:

client (Any) – The SlurmClient.

biomero.ssh_recovery.native_timeouts(client: Any) → Iterator[None]

Bound connection stages with the supported Fabric/Paramiko settings.

Within an attempt the banner, authentication, connect and channel-open timeouts are limited by io_timeout; the previous values are restored afterwards, also on a reconnected transport. Nothing happens when recovery is off. Paramiko’s exec/subsystem acknowledgement waits and name resolution are not covered by these settings.

Parameters:

client (Any) – The SlurmClient, inside a scope.

Yields:

None – Control, while the temporary limits are in force.

biomero.ssh_recovery.read_only(function: Callable[[...], Any]) → Callable[[...], Any]

Declare a short remote read that is safe to repeat.

The decorated method keeps its signature. It runs through retry with a bounded timeout, so that a silent stall returns control; a timeout counts as a lost connection (the Slurm job has not failed).

Parameters:

function (Callable[..., Any]) – A method taking client first.

Returns:

The wrapped method.

Return type:

Callable[…, Any]

biomero.ssh_recovery.retry(client: Any, operation: Callable[[], Any], description: str, *, heartbeat: Callable[[], None] | None = None) → Any

Run an operation that is safe to repeat, recovering from SSH loss.

On a lost connection the stale transport is closed and the operation is tried again with growing pauses, until the window set by client.ssh_recovery_timeout ends. The window starts at the first failure of this call and is not extended by anything else. With a window of 0 the operation runs once and its own error is raised.

Parameters:

• client (Any) – The SlurmClient.

• operation (Callable[[], Any]) – The repeatable operation.

• description (str) – What the operation does, for logs and errors.

• heartbeat (Optional[Callable[[], None]]) – A callback that runs during waits.

Returns:

The result of the first successful attempt.

Return type:

Any

Raises:

• SSHRecoveryTimeout – If the connection did not come back in time.

• RemoteOperationStillPending – If a healthy remote operation was still pending when the window ended.

• SSHRecoveryCancelled – If the caller cancelled during a wait.

biomero.ssh_recovery.scope(client: Any, heartbeat: Callable[[], None] | None = None) → Iterator[RecoveryState]

Share recovery state and caller callbacks with nested operations.

The outermost scope creates the state from the client’s ssh_recovery_timeout; nested scopes join it.

Parameters:

• client (Any) – The SlurmClient.

• heartbeat (Optional[Callable[[], None]]) – A caller callback that runs during waits, such as a session keepalive.

Yields:

RecoveryState – The shared state.

Raises:

TypeError – If heartbeat is not callable.

biomero.ssh_recovery.sftp_timeout(client: Any) → Iterator[Any]

Bound each SFTP read, not the whole transfer.

Parameters:

client (Any) – The SlurmClient, inside a scope.

Yields:

paramiko.SFTPClient – The SFTP client.

biomero.ssh_recovery.transient_reason(error: BaseException) → str | None

Explain why an error counts as a lost connection, if it does.

Credentials, host keys and remote command failures never count.

Parameters:

error (BaseException) – The error raised by an SSH operation.

Returns:

A short reason when the error is transport loss, else

None.

Return type:

Optional[str]

biomero.remote_scripts module

Shell scripts that BIOMERO runs durably on the cluster.

After a lost SSH connection a caller cannot know whether its remote command ran. These builders make a command safe to ask again: it runs under a lock and leaves a marker or receipt file, so that a second request finds the first result instead of repeating the work (for example a second sbatch).

The builders only create strings and know nothing about connections; the caller (SlurmClient) runs them. Each script is built from small named parts, so you can read what it does without tracing nested quotes.

Receipt protocol of submit_once_script. All files share one prefix, the state prefix (<slurm_data_path>/.biomero-recovery/<task id>):

• .lock: the lock file; one submission per prefix runs at a time.

• .digest: hash of the command, to notice a changed command on resume.

• .intent: written BEFORE sbatch runs. It holds the submission time. An intent without a .job means the outcome of sbatch is unknown.

• .job: the Slurm job id, published atomically once known.

• .failed: sbatch refused the job (no job exists); keeps the output.

• .stdout / .stderr: the output of the sbatch call.

Exit codes with a meaning to the caller: REMOTE_BUSY_EXIT_CODE (the lock is held elsewhere, or the job is not visible in Slurm yet; ask again later) and SUBMISSION_REJECTED_EXIT_CODE (sbatch refused the job).

biomero.remote_scripts.create_then_publish_script(create_command: str, partial_file: str, final_file: str) → str

Create a file under a temporary name and rename it when complete.

Parameters:

• create_command (str) – The command that creates partial_file.

• partial_file (str) – The temporary remote path.

• final_file (str) – The remote path to publish.

Returns:

A script that never shows final_file half written.

Return type:

str

biomero.remote_scripts.locked_script(lock_template: str, lock_file: str, script: str, wait_seconds: int = 5) → str

Run a script while holding a lock on the cluster.

The lock command (flock by default) takes an exclusive lock on the lock file and then runs the script, so concurrent callers run one at a time. A caller that waits longer than wait_seconds gets REMOTE_BUSY_EXIT_CODE instead of the script’s own result.

Parameters:

• lock_template (str) – The lock command template.

• lock_file (str) – Remote path of the lock file.

• script (str) – The shell script to run under the lock.

• wait_seconds (int) – How long to wait for the lock.

Returns:

The command that runs the script under the lock.

Return type:

str

biomero.remote_scripts.remove_state_files_command(state_prefix: str) → str

Build the command that removes the receipt files of one submission.

Parameters:

state_prefix (str) – The state prefix of the submission.

Returns:

The rm -f command.

Return type:

str

biomero.remote_scripts.run_once_script(work_command: str, done_marker: str, *, stale_if_newer: Iterable[str] = (), published_file: str | None = None, reuse_message: str = 'Reusing the completed result') → str

Run a command once and skip it when its marker shows it completed.

The marker records a finished run. A finished run is reused only while it is still current: not when an input is newer than the marker (a zip that was uploaded again, results that changed), nor when the published file no longer has the recorded size.

Parameters:

• work_command (str) – The command to run when there is no current marker.

• done_marker (str) – Remote path of the marker file.

• stale_if_newer (Iterable[str]) – Remote paths; if any of them (or anything below a directory) is newer than the marker, the work runs again.

• published_file (Optional[str]) – A file the work produces. Its size is stored in the marker and compared on reuse.

• reuse_message (str) – Printed when the marker is reused.

Returns:

The shell script.

Return type:

str

biomero.remote_scripts.submit_once_script(sbatch_command: str, state_prefix: str, job_name: str, command_digest: str, *, verify_command_unchanged: bool = True, remember_rejection: bool = True) → str

Submit a Slurm job exactly once, however often the script is run.

Run again after a lost connection, the script adopts the job recorded by the first run; when the first run died between sbatch and the record, it finds the job by its name in Slurm. It never runs sbatch a second time while an earlier attempt may have created a job.

Parameters:

• sbatch_command (str) – The submission command. Its job name must be job_name, because that name is how a lost job is found again.

• state_prefix (str) – Remote path prefix of the receipt files.

• job_name (str) – The Slurm job name of this submission.

• command_digest (str) – Hash of the command and its environment.

• verify_command_unchanged (bool) – Stop when a recorded submission used a different command (default). Without it, no digest is kept.

• remember_rejection (bool) – Keep a .failed receipt when sbatch refuses the job, and repeat that answer on later calls (default). Without it the attempt is forgotten, so a later call submits again.

Returns:

The shell script. It prints Submitted batch job <id>.

Return type:

str