PBS job details in human-readable form

job_details.py

#!/usr/bin/env python3

import sys
import subprocess
import re


def get_value(pattern: str, text: str, default: str = "?") -> str:
    """Extracts the first match of a regex group"""
    match = re.search(pattern, text)
    return match.group(1).strip() if match else default


def parse_memory(mem_str: str) -> str:
    """PBS memory -> Mb / Gb"""
    if not mem_str:
        return "N/A"

    match = re.search(r"(\d+)([a-zA-Z]+)", mem_str.lower())
    if not match:
        return mem_str

    value = float(match.group(1))
    unit = match.group(2)

    if unit == "b":
        value /= 1024
    elif unit == "mb":
        value *= 1024
    elif unit == "gb":
        value *= 1024**2

    # ---

    if value >= (1024**2):
        return f"{value / (1024**2):.1f} Gb"
    elif value >= 1024:
        return f"{value / 1024:.1f} Mb"
    else:
        return f"{value:.1f} Kb"


def main():
    if len(sys.argv) < 2:
        print("job-details <job_id>")
        sys.exit(1)

    job_id = sys.argv[1]

    try:
        output = subprocess.check_output(
            ["qstat", "-x", "-f", job_id],
            universal_newlines=True,
            stderr=subprocess.DEVNULL,
        )
    except subprocess.CalledProcessError:
        print(f"ERROR: Unable to retrieve details for job '{job_id}'")
        sys.exit(1)
    except FileNotFoundError:
        print("ERROR: `qstat` not found.")
        sys.exit(1)

    ncpus = get_value(r"resources_used\.ncpus\s*=\s*(\d+)", output, "N/A")
    cpupercent = get_value(r"resources_used\.cpupercent\s*=\s*(\d+)", output, "0")
    cput = get_value(r"resources_used\.cput\s*=\s*([\d:]+)", output)

    mem_raw = get_value(r"resources_used\.mem\s*=\s*(\d+[a-zA-Z]+)", output, "")
    vmem_raw = get_value(r"resources_used\.vmem\s*=\s*(\d+[a-zA-Z]+)", output, "")

    runtime = get_value(r"resources_used\.walltime\s*=\s*([\d:]+)", output)
    allowed = get_value(r"Resource_List\.walltime\s*=\s*([\d:]+)", output)

    server = get_value(r"server\s*=\s*(\S+)", output)
    exec_host = get_value(r"exec_host\s*=\s*(\S+)", output)

    print(f"CPU: {ncpus} x {cpupercent}%, {cput}")
    print(f"Memory: {parse_memory(mem_raw)}, virtual {parse_memory(vmem_raw)}")
    print(f"Runtime: {runtime} out of {allowed}")
    print(f"Location: {server}, {exec_host}")


if __name__ == "__main__":
    main()