phenotypic.sdk_.slurm

Shared SLURM utilities for array job chunking, sbatch submission, and dispatching.

This subpackage consolidates SLURM logic used by the main CLI (phenotypic._cli).

Functions

calculate_optimal_array_chunks	Split images into array job chunks based on SLURM array size limits.
estimate_concurrent_capacity	Estimate max concurrent tasks from partition resources via sinfo.
format_sbatch_directives	Generate #SBATCH directive lines for a SLURM script.
generate_dispatcher_chain	Generate dispatcher scripts for a chain of chunk scripts.
generate_dispatcher_script	Generate a dispatcher script that submits the next chunk and dispatcher.
generation_script_key	Map any lifecycle generation to one collision-resistant path component.
get_slurm_array_limit	Query SLURM for MaxArraySize configuration.
get_slurm_max_submit_jobs	Query a conservative SLURM MaxSubmitJobs limit per user.
parse_job_id	Extract the SLURM job ID from sbatch output.
parse_slurm_time	Validate and canonicalize a SLURM time limit.
sbatch_submission_environment	Snapshot the caller's Python path under a SLURM-safe variable.
submit_drip_feed_start	Submit the first chunk and first dispatcher to start a drip-feed chain.
submit_script	Submit a script to SLURM via sbatch and return the job ID.
validate_array_chunk	Validate that an array chunk is within acceptable bounds.
write_slurm_array_script	Write spec to path and mark it executable.

Classes

SlurmArrayScriptSpec

Specification for a bash SLURM array script.

class phenotypic.sdk_.slurm.SlurmArrayScriptSpec(job_name: str, slurm_args: ~collections.abc.Mapping[str, ~typing.Any], log_path: ~pathlib.Path, task_indices: ~collections.abc.Sequence[int | str], body: str, error_log_path: ~pathlib.Path | None = None, prelude: str = '', comments: ~collections.abc.Sequence[str] = <factory>, array_name: str = 'TASK_INDICES', current_var: str = 'CURRENT_TASK_INDEX', missing_task_id_message: str = 'ERROR: SLURM_ARRAY_TASK_ID not set', bounds_error_message: str | None = None, signal_grace: int | None = None, requeue: bool = False)

Bases: object

Specification for a bash SLURM array script.

Parameters:

• job_name (str) – SLURM job name for #SBATCH --job-name.

• slurm_args (Mapping[str, Any]) – CLI-style SLURM arguments passed to the shared directive formatter.

• log_path (Path) – Path used for both stdout and stderr logs.

• error_log_path (Path | None) – Optional stderr path. Defaults to log_path.

• task_indices (Sequence[int | str]) – Values mapped from SLURM_ARRAY_TASK_ID into the current task variable. Strings are shell-quoted.

• body (str) – Bash body executed once the current task variable is available.

• prelude (str) – Optional bash block inserted after strict mode and before the task array.

• comments (Sequence[str]) – Optional comment lines inserted after SBATCH directives.

• array_name (str) – Bash array variable name.

• current_var (str) – Bash variable that receives the current array entry.

• missing_task_id_message (str) – Error printed when not running as an array job.

• bounds_error_message (str | None) – Error printed when the array task id is out of bounds. When omitted, a message is derived from array_name.

• signal_grace (int | None) – Optional seconds for #SBATCH --signal=B:TERM@N.

• requeue (bool) – Whether to include #SBATCH --requeue.

render() → str

Render the script content.

Return type:

str

array_name: str = 'TASK_INDICES'

body: str

bounds_error_message: str | None = None

comments: Sequence[str]

current_var: str = 'CURRENT_TASK_INDEX'

error_log_path: Path | None = None

job_name: str

log_path: Path

missing_task_id_message: str = 'ERROR: SLURM_ARRAY_TASK_ID not set'

prelude: str = ''

requeue: bool = False

signal_grace: int | None = None

slurm_args: Mapping[str, Any]

task_indices: Sequence[int | str]

phenotypic.sdk_.slurm.calculate_optimal_array_chunks(num_images: int, array_limit: int) → List[Tuple[int, int]]

Split images into array job chunks based on SLURM array size limits.

Calculates the minimum number of array jobs needed to process all images while respecting the cluster’s MaxArraySize limit. Each chunk is represented as a (start, end) tuple of array indices.

Parameters:

• num_images (int) – Total number of images to process.

• array_limit (int) – Maximum array size allowed by SLURM.

Returns:

List of (start_idx, end_idx) tuples defining each array job chunk. Indices are 0-based and end is exclusive (Python slice convention).

Raises:

ValueError – If array_limit is not positive.

Return type:

List[Tuple[int, int]]

Examples

>>> calculate_optimal_array_chunks(500, 1000)
[(0, 500)]

>>> calculate_optimal_array_chunks(2500, 1000)
[(0, 1000), (1000, 2000), (2000, 2500)]

>>> calculate_optimal_array_chunks(1000, 1000)
[(0, 1000)]

>>> calculate_optimal_array_chunks(1001, 1000)
[(0, 1000), (1000, 1001)]

phenotypic.sdk_.slurm.estimate_concurrent_capacity(partition: str, cpus_per_task: int = 1, mem_gb_per_task: float = 4.0) → int

Estimate max concurrent tasks from partition resources via sinfo.

Queries SLURM’s sinfo for the given partition to determine total CPUs, memory, and node count, then estimates how many tasks can run concurrently given per-task resource requirements.

Parameters:

• partition (str) – SLURM partition name.

• cpus_per_task (int) – CPUs requested per task.

• mem_gb_per_task (float) – Memory in GB per task.

Returns:

Estimated number of concurrent tasks. Falls back to 100 if sinfo is unavailable.

Return type:

int

Examples

>>> capacity = estimate_concurrent_capacity("compute")
>>> capacity >= 1  # Always at least 1
True

phenotypic.sdk_.slurm.format_sbatch_directives(job_name: str, slurm_args: Dict[str, Any], output_log: Path, error_log: Path) → str

Generate #SBATCH directive lines for a SLURM script.

Converts CLI SLURM parameters to #SBATCH directives with proper formatting. Reserved keys (array, output, error, job-name) are silently skipped because they are managed by the script generators.

Parameters:

• job_name (str) – Job name for --job-name.

• slurm_args (Dict[str, Any]) – SLURM parameters dict (CLI-style keys like slurm_partition, mem_gb, time).

• output_log (Path) – Path for stdout log.

• error_log (Path) – Path for stderr log.

Returns:

String with all #SBATCH directives joined by newlines.

Return type:

str

Notes

• Time parameters (time, slurm_time) as integers are treated as minutes and converted to HH:MM:SS.

• mem_gb is converted to --mem=<N>G.

phenotypic.sdk_.slurm.generate_dispatcher_chain(chunk_scripts: List[Path], output_dir: Path, slurm_args: Dict[str, Any], log_dir: Path, finalizer_script: Path | None = None, continuation_dependency_kinds: Sequence[Literal['afterany', 'afterok']] | None = None, generation: str | None = None, lifecycle_output_dir: Path | None = None) → List[Path]

Generate dispatcher scripts for a chain of chunk scripts.

For N chunk scripts, generates N-1 dispatcher scripts. Each dispatcher submits the next chunk and (if not last) the next dispatcher with the corresponding dependency kind on that chunk.

Parameters:

• chunk_scripts (List[Path]) – Ordered list of array job chunk script paths.

• output_dir (Path) – Directory to write dispatcher scripts into.

• slurm_args (Dict[str, Any]) – SLURM parameters dict (partition, etc.).

• log_dir (Path) – Directory for dispatcher log files.

• finalizer_script (Path | None) – Terminal finalizer passed only to the last dispatcher in the chain.

• continuation_dependency_kinds (Sequence[Literal['afterany', 'afterok']] | None) – Dependency kind for every chunk-to-continuation edge. The sequence has N - 1 entries without a finalizer and N entries with one. Defaults to afterany for every edge.

• generation (str | None) – Optional explicit lifecycle generation. When supplied, dispatcher scripts are isolated under that generation.

• lifecycle_output_dir (Path | None) – Optional lifecycle fence root when dispatcher scripts themselves are written beneath a separate control root.

Returns:

List of dispatcher script paths (one fewer than chunk_scripts, since the last chunk does not need a dispatcher). Empty if only one chunk exists.

Return type:

List[Path]

phenotypic.sdk_.slurm.generate_dispatcher_script(next_chunk_script: Path, next_dispatcher_script: Path | None, output_path: Path, slurm_args: Dict[str, Any], log_dir: Path, *, output_dir: Path | None = None, generation: str | None = None, chunk_index: int = 1, finalizer_script: Path | None = None, dependency_kind: Literal['afterany', 'afterok'] = 'afterany') → Path

Generate a dispatcher script that submits the next chunk and dispatcher.

The dispatcher requests control-plane resources (1 CPU, 512M, 5 min) and invokes the Python lifecycle entry point. That entry point durably submits the next processing chunk and optional dependent dispatcher.

Parameters:

• next_chunk_script (Path) – Path to the next array job chunk script.

• next_dispatcher_script (Path | None) – Path to the next dispatcher script, or None for the last chunk (no further dispatcher needed).

• output_path (Path) – Where to write the generated dispatcher script.

• slurm_args (Dict[str, Any]) – SLURM parameters dict (used to extract partition).

• log_dir (Path) – Directory for dispatcher log files.

• output_dir (Path | None) – Base output directory containing the lifecycle state.

• generation (str | None) – Exact scheduler generation for lifecycle submissions.

• chunk_index (int) – Zero-based index of the chunk this dispatcher submits.

• finalizer_script (Path | None) – Terminal finalizer submitted after the last chunk becomes terminal.

• dependency_kind (Literal['afterany', 'afterok']) – Dependency condition for the continuation submitted after next_chunk_script.

Returns:

Path to the generated dispatcher script.

Return type:

Path

phenotypic.sdk_.slurm.generation_script_key(generation: str) → str

Map any lifecycle generation to one collision-resistant path component.

Parameters:

generation (str)

Return type:

str

phenotypic.sdk_.slurm.get_slurm_array_limit() → int

Query SLURM for MaxArraySize configuration.

Uses scontrol show config to retrieve the maximum number of array tasks allowed per job. Falls back to a conservative default if the query fails or SLURM is not available.

Returns:

1000).

Return type:

Integer limit for array job size (default

Examples

>>> limit = get_slurm_array_limit()
>>> limit >= 1000  # At least the default
True

Notes

• Result is cached for the session (lru_cache)

• Default fallback is 1000 (conservative for most clusters)

• Common SLURM values: 1001, 10000, 100000

phenotypic.sdk_.slurm.get_slurm_max_submit_jobs() → int | None

Query a conservative SLURM MaxSubmitJobs limit per user.

Uses sacctmgr to retrieve the maximum number of jobs a user can have in the queue simultaneously. This is typically set by QoS (Quality of Service) policies.

Returns:

Smallest configured positive QoS or user-association limit, or None if no limit is configured or available. Taking the smallest value is conservative when the cluster exposes several QoS or association records and the caller cannot reliably infer which one the submitted job will use.

Return type:

int | None

Examples

>>> limit = get_slurm_max_submit_jobs()
>>> limit is None or limit > 0
True

Notes

• Result is cached for the session

• Returns None if sacctmgr is not available

• Returns None if no limit is configured (unlimited)

• QoS and association records can carry different limits; returning the largest value can oversize an array for the active policy.

phenotypic.sdk_.slurm.parse_job_id(sbatch_stdout: str) → str

Extract the SLURM job ID from sbatch output.

Parameters:

sbatch_stdout (str) – Standard output from an sbatch command, typically "Submitted batch job 12345\n".

Returns:

The job ID as a string.

Raises:

RuntimeError – If the job ID cannot be parsed from the output.

Return type:

str

Examples

>>> parse_job_id("Submitted batch job 12345\n")
'12345'

phenotypic.sdk_.slurm.parse_slurm_time(value: object) → str | None

Validate and canonicalize a SLURM time limit.

Parameters:

value (object) – Empty input, positive integer minutes, or a SLURM duration in HH:MM:SS or D-HH:MM:SS form.

Returns:

Canonical SLURM duration, or None when value is empty.

Raises:

ValueError – If value is not one of the supported forms, contains an invalid clock field, or represents a nonpositive duration.

Return type:

str | None

Examples

>>> parse_slurm_time(90)
'01:30:00'
>>> parse_slurm_time("00:10:00")
'00:10:00'
>>> parse_slurm_time("1-04:00:00")
'1-04:00:00'

phenotypic.sdk_.slurm.sbatch_submission_environment(environment: Mapping[str, str] | None = None) → dict[str, str]

Snapshot the caller’s Python path under a SLURM-safe variable.

Some clusters explicitly filter PYTHONPATH even when sbatch uses --export=ALL. PhenoTypic batch scripts restore this namespaced copy before invoking Python.

Parameters:

environment (Mapping[str, str] | None) – Environment to copy. Defaults to os.environ.

Returns:

A detached environment mapping suitable for subprocess.run.

Return type:

dict[str, str]

phenotypic.sdk_.slurm.submit_drip_feed_start(chunk_scripts: List[Path], dispatcher_scripts: List[Path], *, finalizer_script: Path | None = None, continuation_dependency_kind: Literal['afterany', 'afterok'] = 'afterany', output_dir: Path | None = None, generation: str | None = None) → Tuple[List[str], str | None]

Submit the first chunk and first dispatcher to start a drip-feed chain.

Parameters:

• chunk_scripts (List[Path]) – Ordered list of chunk script paths (must be non-empty).

• dispatcher_scripts (List[Path]) – Dispatcher scripts from generate_dispatcher_chain() (may be empty for single-chunk).

• finalizer_script (Path | None) – Terminal finalizer submitted after the only chunk when no dispatcher is required.

• continuation_dependency_kind (Literal['afterany', 'afterok']) – Dependency condition for the initial dispatcher or single-chunk finalizer.

• output_dir (Path | None) – Explicit lifecycle output root for attempt-scoped chains.

• generation (str | None) – Explicit lifecycle generation for attempt-scoped chains.

Returns:

Tuple of (job_ids, warning_message). job_ids contains the submitted job IDs (1 or 2). warning_message is None on success, or a string with recovery instructions if the dispatcher submission failed (chunk 0 was still submitted).

Raises:

RuntimeError – If the first chunk submission fails.

Return type:

Tuple[List[str], str | None]

phenotypic.sdk_.slurm.submit_script(script_path: Path, dependency_job_id: str | None = None, array_index: int | None = None) → str

Submit a script to SLURM via sbatch and return the job ID.

Parameters:

• script_path (Path) – Path to the SLURM batch script.

• dependency_job_id (str | None) – When set, adds --dependency=afterany:<id> so this job starts only after the dependency finishes.

• array_index (int | None) – When set, overrides any script array directive and submits only this array index.

Returns:

SLURM job ID string.

Raises:

RuntimeError – If sbatch is not available, the submission fails, or the job ID cannot be parsed.

Return type:

str

phenotypic.sdk_.slurm.validate_array_chunk(chunk: Tuple[int, int], num_images: int, array_limit: int) → bool

Validate that an array chunk is within acceptable bounds.

Parameters:

• chunk (Tuple[int, int]) – (start, end) tuple to validate.

• num_images (int) – Total number of images.

• array_limit (int) – Maximum array size.

Returns:

True if chunk is valid, False otherwise.

Return type:

bool

Examples

>>> validate_array_chunk((0, 500), 1000, 1000)
True

>>> validate_array_chunk((0, 1500), 1000, 1000)
False

>>> validate_array_chunk((-1, 100), 1000, 1000)
False

phenotypic.sdk_.slurm.write_slurm_array_script(path: Path, spec: SlurmArrayScriptSpec) → Path

Write spec to path and mark it executable.

Parameters:

• path (Path) – Destination script path.

• spec (SlurmArrayScriptSpec) – Script specification to render.

Returns:

The destination path.

Return type:

Path