cape.cfdx.queue: Manage PBS and Slurm interfaces

This module contains direct interface for functions like qsub and qstat. These methods provide an easy interface to command-line PBS utilities and also provide some access to the PBS information. For example, the method pqsub() writes a file jobID.dat with the PBS job number of the submitted job.

class cape.cfdx.queue.QStat(active: bool = True, scheduler: str = 'pbs', timeout: float | int = 180.0)

Collection of PBS/Slurm job statuses

Call:
>>> q = QStat(active=True, scheduler="pbs", timeout=180.0)
Inputs:
active: {True} | False

Whether or not new qstat calls should be made

scheduler: {"pbs"} | "slurm"

Name of scheduling software to use

timeout: {180.0} | float | int

Time [s] until qstat results are considered stale

Keys:
q[jobid]: dict

Status for job with name PBS/Slurm job ID jobid

Attributes:
active

bool Whether new qstat queries should be made

calltimes

dict[float] Time at which each user/server combo was last updated

check_job(j: str | int, u: str | None = None) → dict | None

Check status of given job (owned by specific user)

Call:
>>> stats = q.check_job(j, u=None)
Inputs:
q: QStat

PBS/Slurm job stats collection

j: str | int

PBS/Slurm job ID

u: {None} | str

User name

Versions:

  • 2025-05-03 @ddalle: v1.0

clear_queue(u: str, server: str | None = None)

Clear old qstat results for a given user and server

Call:
>>> q.clear_queue(u, server=None)
Inputs:
q: QStat

PBS/Slurm job stats collection

u: str

User name

server: {None} | str

Name of PBS/Slurm server

Versions:

  • 2025-05-03 @ddalle: v1.0

defaultserver

str Default server, filled in after first empty qstat call

qstat(u: str, server: str | None = None)

Call qstat and add results to collection

Call:
>>> q.qstat(u, server=None)
Inputs:
q: QStat

PBS/Slurm job stats collection

u: str

User name

server: {None} | str

Name of PBS/Slurm server

Versions:

  • 2025-05-03 @ddalle: v1.0

scheduler

str Name of scheduler, "pbs" | "slurm"

squeue(u: str, server: str | None = None)

Call squeue and add results to collection

Call:
>>> q.squeue(u, server=None)
Inputs:
q: QStat

PBS/Slurm job stats collection

u: str

User name

server: {None} | str

Name of PBS/Slurm server

Versions:

  • 2025-05-03 @ddalle: v1.0

timeout

float Time until qstat results are considered stale [s]

cape.cfdx.queue.get_job_id() → str | None

Get job ID of currently running job (if any)

Call:
>>> job_id = get_job_id()
Outputs:
job_id: str | None

Current PBS/Slurm job ID, if found

Versions:

  • 2024-06-12 @ddalle: v1.0

cape.cfdx.queue.open_jobfile_r() → IOBase | None

Open the best available PBS/Slurm job ID file for reading

Call:
>>> fp = open_jobfile_r()
Outputs:
fp: IOBase | None

File handle, open for reading, if available

Versions:

  • 2024-06-13 @ddalle: v1.0

cape.cfdx.queue.openfile_w(fname: str, mode: str = 'w') → IOBase

Open a file for writing, creating folder if needed

Call:
>>> fp = openfile_w(fname, mode='w')
Inputs:
fname: str

Name of file to write

mode: {'w'} | 'wb'

Mode for resulting file

Outputs:
fp: IOBase

File handle, open for writing

Versions:

  • 2024-06-12 @ddalle: v1.0

cape.cfdx.queue.pqsub(fname: str, fout: str = 'jobID.dat') → str

Submit a PBS script and save the job number in an fout file

Call:
>>> pbs = pqsub(fname, fout="jobID.dat")
Inputs:
fname: str

Name of PBS script to submit

fout: str

Name of output file to contain PBS job number

Outputs:
pbs: int | None

PBS job ID number if submission was successful

Versions:

  • 2014-10-06 @ddalle: v1.0

  • 2021-08-09 @ddalle: v1.1; allow non-int PBS IDs

cape.cfdx.queue.psbatch(fname: str, fout: str = 'jobID.dat') → str

Submit a PBS script and save the job number in an fout file

Call:
>>> pbs = psbatch(fname, fout="jobID.dat")
Inputs:
fname: str

Name of PBS script to submit

fout: str

Name of output file to contain PBS job number

Outputs:
pbs: int | None

Slurm job ID number if submission was successful

Versions:

  • 2022-10-05 @ddalle: v1.0

cape.cfdx.queue.qdel(jobID: str | int, force: bool = False)

Delete a PBS job by ID

Call:
>>> qdel(jobID, force=False)
Inputs:
jobID: str | int

PBS job ID number if submission was successful

force: True | {False}

Option to force-kill troublesome job

Versions:

  • 2014-12-26 @ddalle: v1.0

  • 2024-06-12 @ddalle: v2.0; eliminate loop

  • 2024-06-16 @ddalle: v2.1; add force

cape.cfdx.queue.qstat(u: str | None = None, j: str | int | None = None, server: str | None = None, timeout: float | str = 10.0) → dict

Call qstat and process information

Call:
>>> jobs = qstat(u=None, j=None, server=None)
Inputs:
u: {None} | str

User name, defaults to current user’s name

j: {None} | str | int

Specific job ID for which to check

server: {None} | str

Name of PBS server

Outputs:
jobs: dict

Information on each job, jobs[jobID] for each submitted job

Versions:

  • 2014-10-06 @ddalle: v1.0

  • 2015-06-19 @ddalle: v1.1; add qstat -J option

  • 2025-05-03 @ddalle: v1.2; add server

cape.cfdx.queue.qsub(fname: str) → str

Submit a PBS script and return the job number

Call:
>>> jobid = qsub(fname)
Inputs:
fname: str

Name of PBS script to submit

Outputs:
jobid: str | None

PBS job ID if submission was successful

Versions:

  • 2014-10-05 @ddalle: v1.0

  • 2021-08-09 @ddalle: v2.0

    • Support Python 3

    • Use full job ID (not just int) as a backup

  • 2023-11-07 @ddalle: v2.1; test for qsub

  • 2024-06-18 @ddalle: v3.0; string output

cape.cfdx.queue.read_job_ids() → list

Read IDs of all jobs listed in jobID.dat or similar file

Call:
>>> job_ids = read_job_ids()
Outputs:
job_ids: list[str]

List of job identifiers

Versions:

  • 2024-06-13 @ddalle: v1.0

cape.cfdx.queue.sbatch(fname: str) → str | None

Submit a Slurm script and return the job number

Call:
>>> jobid = sbatch(fname)
Inputs:
fname: str

Name of Slurm script to submit

Outputs:
jobid: str | None

Slurm job ID if submission was successful

Versions:

  • 2022-10-05 @ddalle: v1.0

  • 2024-06-18 @ddalle: v2.0; string output

cape.cfdx.queue.scancel(jobID: str | int)

Delete a Slurm job by ID

Call:
>>> scancel(jobID)
Inputs:
pbs: int or list[int]

PBS job ID number if submission was successful

Versions:

  • 2022-10-05 @ddalle: v1.0

cape.cfdx.queue.squeue(u: str | None = None, j: str | int | None = None, server: str | None = None) → dict

Call squeue and process information

Call:
>>> jobs = squeue(u=None, j=None)
Inputs:
u: str

User name, defaults to os.environ[USER]

j: int

Specific job ID for which to check

server: {None} | str

Name of Slurm cluster

Outputs:
jobs: dict

Info on each job, jobs[jobID] for each submitted job

Versions:

  • 2023-06-16 @ddalle: v1.0

  • 2024-06-18 @ddalle: v1.1; add J

  • 2025-05-03 @ddalle: v1.2; switch J -> j, add server