Source code for galaxy.jobs.runners.util.cli.job.slurm_torque

import re

from .torque import Torque

__all__ = ("SlurmTorque",)


[docs]class SlurmTorque(Torque): """A CLI job executor for Slurm's Torque compatibility mode. This differs from real torque CLI in that -x command line is not available so job status needs to be parsed from qstat table instead of XML. """
[docs] def get_status(self, job_ids=None): return "qstat"
[docs] def parse_status(self, status, job_ids): rval = {} for line in status.strip().splitlines(): if line.startswith("Job ID"): continue line_parts = re.compile(r"\s+").split(line) if len(line_parts) < 5: continue id = line_parts[0] state = line_parts[4] if id in job_ids: # map PBS job states to Galaxy job states. rval[id] = self._get_job_state(state) return rval