phenotypic.sdk_.slurm package#

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

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

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

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[source]#

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[source]#

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) → List[Path][source]#

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 --dependency=afterany 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.

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) → Path[source]#

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

The dispatcher requests minimal resources (1 CPU, 100M, 5 min) and runs two sbatch commands: one for the next processing chunk, and one for the next dispatcher (with --dependency on the chunk).

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.

Returns:

Path to the generated dispatcher script.

Return type:

Path

phenotypic.sdk_.slurm.get_slurm_array_limit() → int[source]#

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[source]#

Query SLURM for 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:

Integer limit or None if not configured/available.

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)

  • This limit is typically much higher than array limits (e.g., 10000+)

phenotypic.sdk_.slurm.parse_job_id(sbatch_stdout: str) → str[source]#

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.submit_drip_feed_start(chunk_scripts: List[Path], dispatcher_scripts: List[Path]) → Tuple[List[str], str | None][source]#

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

Parameters:
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[source]#

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[source]#

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