app

avail(user=typer.Option(getuser(), '-u', '--user', help="User name to check. Use '*' to check all users."), queue=typer.Option(None, '-q', '--queue', help='Queue to check. By default, all queues are checked.'), slurm=typer.Option(False if which('squeue') is None else True, '--slurm/--no-slurm', help='Check slurm jobs.'), sge=typer.Option(False if which('qstat') is None else True, '--sge/--no-sge', help='Check sge queues.'), gpu=typer.Option(False, '--gpu', help='Hides non gpu queues.'), hide=typer.Option(True, '--hide-full/--no-hide-full', help='Hide full queues.'), debug=typer.Option(False, help='Help debugging', hidden=True))

Alias: [yellow bold]hqavail[/yellow bold]. Get available queues

Source code in hpcman/queue/app.py

@app.command(
    rich_help_panel="Information", no_args_is_help=False, epilog="Produces a table similar to the `qstat` output."
)  # type: ignore
def avail(
    user: str = typer.Option(
        getuser(),
        "-u",
        "--user",
        help="User name to check. Use '*' to check all users.",
    ),
    queue: Optional[str] = typer.Option(
        None,
        "-q",
        "--queue",
        help="Queue to check. By default, all queues are checked.",
    ),
    slurm: bool = typer.Option(
        False if which("squeue") is None else True,
        "--slurm/--no-slurm",
        help="Check slurm jobs.",
    ),
    sge: bool = typer.Option(
        False if which("qstat") is None else True,
        "--sge/--no-sge",
        help="Check sge queues.",
    ),
    gpu: bool = typer.Option(
        False,
        "--gpu",
        help="Hides non gpu queues.",
    ),
    hide: bool = typer.Option(
        True,
        "--hide-full/--no-hide-full",
        help="Hide full queues.",
    ),
    debug: bool = typer.Option(False, help="Help debugging", hidden=True),
) -> None:
    """
    Alias: [yellow bold]hqavail[/yellow bold]. Get available queues
    """
    if gpu:
        sge = False
        slurm = True
    run_queue_avail(user=user, queue=queue, debug=debug, slurm=slurm, sge=sge, gpu=gpu, hide=hide)

error()

Get error queues, and explain state

Source code in hpcman/queue/app.py

@app.command(rich_help_panel="Information")  # type: ignore
def error() -> None:
    """
    Get error queues, and explain state
    """
    raise CommandNotImplemented

launch(command=typer.Argument(..., help='Command to launch.', show_choices=True, show_default=True), force=typer.Option(False, '--force', help='Force the submission of the job even if run is already present.', rich_help_panel='Extras'), queue=typer.Option(..., '-q', '--queue', help="The queue to use. Provide [yellow]'*'[/yellow] to explicitly select any available queue.", prompt='Please enter a queue name' if sys.stdin.isatty() else False, rich_help_panel='Required'), procs=typer.Option(1, '-p', '-P', '--procs', help='Number of processors to request.', rich_help_panel='Job'), freemem=typer.Option('4G', '-f', '--free', help='Free memory to request on the machine to run this job. (100M, 1G, 4G, 32G etc.)', rich_help_panel='Job'), maxmem=typer.Option(None, '-m', '--max', help='Maximum memory this job may use (kill if exceeded). (100M, 1G, 4G, 32G etc.)', rich_help_panel='Job'), maxfilesize=typer.Option('500G', '-F', '--filesize', help='Kill the job if any created file exceeds this size. (100M, 1G, 4G, 32G etc.)', rich_help_panel='Job'), path=typer.Option(environ.get('PATH'), '--path', help='The [green]$PATH[/green] to use for the submitted commands.', rich_help_panel='Extras', show_default='$PATH environment variable of current shell'), runname=typer.Option(None, '-r', '--runname', help='The run name and log output directory name.', rich_help_panel='Job', show_default='sge.{command}'), debug=typer.Option(False, help='Help debugging', hidden=True))

Launch interactive browser job type on a node

Source code in hpcman/queue/app.py

@app.command(rich_help_panel="Job Submission", no_args_is_help=True)
def launch(
    command: LaunchType = typer.Argument(
        ...,
        help="Command to launch.",
        show_choices=True,
        show_default=True,
    ),
    force: bool = typer.Option(
        False,
        "--force",
        help="Force the submission of the job even if run is already present.",
        rich_help_panel="Extras",
    ),
    queue: str = typer.Option(
        ...,
        "-q",
        "--queue",
        help="The queue to use. Provide [yellow]'*'[/yellow] to explicitly select any available queue.",
        prompt="Please enter a queue name" if sys.stdin.isatty() else False,
        rich_help_panel="Required",
    ),
    procs: int = typer.Option(
        1,
        "-p",
        "-P",
        "--procs",
        help="Number of processors to request.",
        rich_help_panel="Job",
    ),
    freemem: Optional[str] = typer.Option(
        "4G",
        "-f",
        "--free",
        help="Free memory to request on the machine to run this job. (100M, 1G, 4G, 32G etc.)",
        rich_help_panel="Job",
    ),
    maxmem: Optional[str] = typer.Option(
        None,
        "-m",
        "--max",
        help="Maximum memory this job may use (kill if exceeded). (100M, 1G, 4G, 32G etc.)",
        rich_help_panel="Job",
    ),
    maxfilesize: Optional[str] = typer.Option(
        "500G",
        "-F",
        "--filesize",
        help="Kill the job if any created file exceeds this size. (100M, 1G, 4G, 32G etc.)",
        rich_help_panel="Job",
    ),
    path: str = typer.Option(
        environ.get("PATH"),
        "--path",
        help="The [green]$PATH[/green] to use for the submitted commands.",
        rich_help_panel="Extras",
        show_default="$PATH environment variable of current shell",  # pyright: ignore
    ),
    runname: Optional[str] = typer.Option(
        None,
        "-r",
        "--runname",
        help="The run name and log output directory name.",
        rich_help_panel="Job",
        show_default="sge.{command}",  # pyright: ignore
    ),
    debug: bool = typer.Option(False, help="Help debugging", hidden=True),
) -> None:
    """
    Launch interactive browser job type on a node
    """
    launch_job(
        launchtype=command,
        force=force,
        queue=queue,
        procs=procs,
        freemem=freemem,
        maxmem=maxmem,
        maxfilesize=maxfilesize,
        path=path,
        runname=runname,
        debug=debug,
    )

queue()

Manage queuing system jobs, and get queue status

Source code in hpcman/queue/app.py

@app.callback()  # type: ignore
def queue() -> None:
    """
    Manage queuing system jobs, and get queue status
    """

stat(user=typer.Option(getuser(), '-u', '--user', help="User name to check. Use '*' to check all users."), queue=typer.Option(None, '-q', '--queue', help='Queue to check. By default, all queues are checked.'), slurm=typer.Option(False if which('squeue') is None else True, '--slurm/--no-slurm', help='Check slurm jobs.'), sge=typer.Option(False if which('qstat') is None else True, '--sge/--no-sge', help='Check sge queues.'), watch=typer.Option(False, '-w', '--watch', help='Watch for changes to the queue status'), debug=typer.Option(False, help='Help debugging', hidden=True))

Alias: [yellow bold]hqstat[/yellow bold]. Get job status

Source code in hpcman/queue/app.py

@app.command(
    rich_help_panel="Information", no_args_is_help=False, epilog="Produces a table similar to the `qstat` output."
)  # type: ignore
def stat(
    user: str = typer.Option(
        getuser(),
        "-u",
        "--user",
        help="User name to check. Use '*' to check all users.",
    ),
    queue: Optional[str] = typer.Option(
        None,
        "-q",
        "--queue",
        help="Queue to check. By default, all queues are checked.",
    ),
    slurm: bool = typer.Option(
        False if which("squeue") is None else True,
        "--slurm/--no-slurm",
        help="Check slurm jobs.",
    ),
    sge: bool = typer.Option(
        False if which("qstat") is None else True,
        "--sge/--no-sge",
        help="Check sge queues.",
    ),
    watch: bool = typer.Option(
        False,
        "-w",
        "--watch",
        help="Watch for changes to the queue status",
    ),
    debug: bool = typer.Option(False, help="Help debugging", hidden=True),
) -> None:
    """Alias: [yellow bold]hqstat[/yellow bold]. Get job status"""
    run_queue_status(user=user, queue=queue, watch=watch, debug=debug, slurm=slurm, sge=sge)

submit(command=typer.Argument(... if sys.stdin.isatty() else '-', help="Command to submit. Reading from stdin assumed. Provide [yellow]'-'[/yellow] to explicitly use stdin.", show_default=False, metavar='"COMMAND"'), runname=typer.Option(..., '-r', '--runname', help="The run name and log output directory name. Provide [yellow]'-'[/yellow] for automatic name generation.", prompt='Please enter a job run name' if sys.stdin.isatty() else False, rich_help_panel='Required'), queue=typer.Option(..., '-q', '--queue', help="The queue to use. Provide [yellow]'*'[/yellow] to explicitly select any available queue.", prompt='Please enter a queue name' if sys.stdin.isatty() else False, rich_help_panel='Required'), jobtype=typer.Option(JobType.BATCH if sys.stdin.isatty() else JobType.ARRAY, '-t', '--type', help='Type of job to submit. [bold]Note: Default depends on inputs.[/bold] [yellow]batch[/yellow] is default except when reading from stdin, then [yellow]array[/yellow] is default.', rich_help_panel='Job', case_sensitive=False, show_default=f"{JobType.BATCH.value}'; '{JobType.ARRAY.value}' when reading stdin'"), procs=typer.Option(1, '-p', '-P', '--procs', help='Number of processors to request. Match with threads/cores used in command. Exposed as [green]$NPROCS[/green] in the submit script.', rich_help_panel='Job'), freemem=typer.Option('4G', '-f', '--free', help='Free memory to request on the machine to run this job. (100M, 1G, 4G, 32G etc.)', rich_help_panel='Job'), maxmem=typer.Option(None, '-m', '--max', help='Maximum memory this job may use (kill if exceeded). (100M, 1G, 4G, 32G etc.)', rich_help_panel='Job'), maxfilesize=typer.Option('500G', '-F', '--filesize', help='Kill the job if any created file exceeds this size. (100M, 1G, 4G, 32G etc.)', rich_help_panel='Job'), force=typer.Option(False, '--force', help='Force the submission of the job even if --runname is already present. (as [green]SGE_Array[/green] default)', rich_help_panel='Extras'), path=typer.Option(environ.get('PATH'), '--path', help='The [green]$PATH[/green] to use for the submitted commands.', rich_help_panel='Extras', show_default='$PATH environment variable of current shell'), hold=typer.Option(None, '--hold', help='Hold the job until specified jobs (either job name or job ID) are completed. If multiple, must be comma-delimited.', rich_help_panel='Extras'), holdauto=typer.Option(False, '--hold-auto', help='Hold the execution of the job until all other jobs in dir have finished. (Uses [green].hpcman_jobnums[/green] file)', rich_help_panel='Extras'), concurrency=typer.Option(50, '-b', '--concurrency', help="Array job maximum task concurrency. ('batch size'; unused in batch mode)", rich_help_panel='Job'), queuetype=typer.Option(QueueType.SGE if 'vaughan' in str(environ.get('HOSTNAME', '')) else QueueType.SLURM, help='Type of queuing system.', rich_help_panel='Queue', hidden=False, show_default=QueueType.SGE.value if 'vaughan' in str(environ.get('HOSTNAME', '')) else QueueType.SLURM.value), validatepath=typer.Option(True, '--validate-path/--no-validate-path', help='Validate the provided [green]$PATH[/green] value. Turn off at own risk. You may need to disable if you want to use the default [green]$PATH[/green] of your bash shell, or if you want to do some testing.', rich_help_panel='Extras'), cache=typer.Option(False, '--cache', help='Cache information for the submitted job to [green]~/hpcman/cache[/green].', rich_help_panel='Extras', hidden=True), localdrive=typer.Option(None, '--local-drive', help=f"'Use the local drive for the submitted job. If set to [yellow]pertask[/yellow], uses [green]$TMPDIR[/green] ([yellow]'{environ.get('TMPDIR')}'[/yellow]) as the prefix for the tempdir (`mktemp -d -p`). If set to [yellow]shared[/yellow], uses [green]$TMPDIR/$USER/$DIRNAME[/green] ([yellow]$TMPDIR/'{environ.get('USER')}'/'{Path.cwd().name}'[/yellow]) as the tempdir, where [green]$DIRNAME[/green] is the name of the current directory.'", rich_help_panel='Local drive'), localprefix=typer.Option(None, '--local-prefix', help='Override the default prefix of the tempdir. See the help for --local-drive.', rich_help_panel='Local drive'), mirrortype=typer.Option(MirrorType.LINK, '--mirror-type', help='Type of mirroring to local drive. Only used if --local-drive is set.', rich_help_panel='Local drive', show_default=MirrorType.LINK.value), copyresults=typer.Option(None, '--copy-results/--no-copy-results', help='Copy final results back to submission directory. Defaults to [magenta]True[/magenta] if --local-drive [yellow]pertask[/yellow] and [magenta]False[/magenta] if --local-drive [yellow]shared[/yellow].', rich_help_panel='Local drive', show_default=False), dryrun=typer.Option(False, '--dry-run', help='Do not actually submit the job, but set up the submit directory,', rich_help_panel='Extras'), watch=typer.Option(False, '--watch', help='Run `hpcman queue stat --watch` to monitor the status of the jobs.', rich_help_panel='Extras'), conda=typer.Option(False, '--conda', help='Enable use of conda (e.g. [green]conda activate ...[/green]).', rich_help_panel='Conda'), condaexe=typer.Option('/local/cluster/miniconda3_base/bin/conda', '--conda-exe', help='Path to [yellow]conda[/yellow] executable to enable.', rich_help_panel='Conda', exists=True, file_okay=True, dir_okay=False, readable=True, resolve_path=True), debug=typer.Option(False, help='Help debugging', hidden=True))

Alias: [yellow bold]hqsub[/yellow bold]. Submit queueing job "COMMAND". Multiple lines of commands supported using piped input from STDIN. If providing as argument and there are spaces in the command, use quotes to provide the command, either '' or "" will work! Do not use -c '', just provide the command as argument.

Source code in hpcman/queue/app.py

@app.command(  # type: ignore
    rich_help_panel="Job Submission",
    no_args_is_help=True,
    epilog="Emulates some functions of [green]SGE_Batch[/green] and [green]SGE_Array[/green]. Functions have been "
    "intentionally removed and/or modified in some cases to enable new features and improve reliability. Please submit "
    "a support ticket request at https://shell.cqls.oregonstate.edu/support to provide feedback on this software.\n\n"
    + "\n\n".join(submit_examples),
)
def submit(
    command: str = typer.Argument(
        ... if sys.stdin.isatty() else "-",
        help="Command to submit. Reading from stdin assumed. Provide [yellow]'-'[/yellow] to explicitly use stdin.",
        show_default=False,
        metavar='"COMMAND"',
    ),
    runname: str = typer.Option(
        ...,
        "-r",
        "--runname",
        help="The run name and log output directory name. Provide [yellow]'-'[/yellow] for automatic name generation.",
        prompt="Please enter a job run name" if sys.stdin.isatty() else False,
        rich_help_panel="Required",
    ),
    queue: str = typer.Option(
        ...,
        "-q",
        "--queue",
        help="The queue to use. Provide [yellow]'*'[/yellow] to explicitly select any available queue.",
        prompt="Please enter a queue name" if sys.stdin.isatty() else False,
        rich_help_panel="Required",
    ),
    jobtype: JobType = typer.Option(
        JobType.BATCH if sys.stdin.isatty() else JobType.ARRAY,
        "-t",
        "--type",
        help="Type of job to submit. [bold]Note: Default depends on inputs.[/bold] [yellow]batch[/yellow] is default "
        "except when reading from stdin, then [yellow]array[/yellow] is default.",
        rich_help_panel="Job",
        case_sensitive=False,
        show_default=f"{JobType.BATCH.value}; {JobType.ARRAY.value} when reading stdin",  # pyright: ignore
    ),
    procs: int = typer.Option(
        1,
        "-p",
        "-P",
        "--procs",
        help="Number of processors to request. Match with threads/cores used in command. Exposed as "
        "[green]$NPROCS[/green] in the submit script.",
        rich_help_panel="Job",
    ),
    freemem: Optional[str] = typer.Option(
        "4G",
        "-f",
        "--free",
        help="Free memory to request on the machine to run this job. (100M, 1G, 4G, 32G etc.)",
        rich_help_panel="Job",
    ),
    maxmem: Optional[str] = typer.Option(
        None,
        "-m",
        "--max",
        help="Maximum memory this job may use (kill if exceeded). (100M, 1G, 4G, 32G etc.)",
        rich_help_panel="Job",
    ),
    maxfilesize: Optional[str] = typer.Option(
        "500G",
        "-F",
        "--filesize",
        help="Kill the job if any created file exceeds this size. (100M, 1G, 4G, 32G etc.)",
        rich_help_panel="Job",
    ),
    force: bool = typer.Option(
        False,
        "--force",
        help="Force the submission of the job even if --runname is already present. (as [green]SGE_Array[/green] "
        "default)",
        rich_help_panel="Extras",
    ),
    path: str = typer.Option(
        environ.get("PATH"),
        "--path",
        help="The [green]$PATH[/green] to use for the submitted commands.",
        rich_help_panel="Extras",
        show_default="$PATH environment variable of current shell",  # pyright: ignore
    ),
    hold: Optional[str] = typer.Option(
        None,
        "--hold",
        help="Hold the job until specified jobs (either job name or job ID) are completed. If multiple, must be "
        "comma-delimited.",
        rich_help_panel="Extras",
    ),
    holdauto: bool = typer.Option(
        False,
        "--hold-auto",
        help="Hold the execution of the job until all other jobs in dir have finished. "
        "(Uses [green].hpcman_jobnums[/green] file)",
        rich_help_panel="Extras",
    ),
    # hold_jids: Optional[str] = typer.Option(
    #     None,
    #     "--hold-jids",
    #     help="Hold the execution of the job until jobs listed have finished. (e.g. --hold-jids "
    #     "[yellow]123456,123457)[/yellow] [red]Note: This setting name was changed compared to "
    #     "[green]SGE_Array[/green] --hold_jids[/red]",
    #     rich_help_panel="Extras",
    # ),
    # hold_names: Optional[str] = typer.Option(
    #     None,
    #     "--hold-names",
    #     help="Hold the execution of the job until jobs listed have finished. (e.g. --hold-names "
    #     "[yellow]fasta_format,blast...[/yellow]) [red]Note: This setting name was changed compared to "
    #     "[green]SGE_Array[/green] --hold_names[/red]",
    #     rich_help_panel="Extras",
    # ),
    concurrency: int = typer.Option(
        50,
        "-b",
        "--concurrency",
        help="Array job maximum task concurrency. ('batch size'; unused in batch mode)",
        rich_help_panel="Job",
    ),
    # tasklist: str = typer.Option(
    #     "", "-t", "--tasklist", help="Array Job Range to submit (e.g. 1-100).",
    #     rich_help_panel="Job",
    # ),
    # priority: int = typer.Option(
    #     0, "-p", "--priority", help="The priority of the job submitted. "
    #     "(e.g. beteween -1023-0, reduces priority)",
    #     rich_help_panel="Job",
    # ),
    queuetype: QueueType = typer.Option(
        QueueType.SGE if "vaughan" in str(environ.get("HOSTNAME", "")) else QueueType.SLURM,
        help="Type of queuing system.",
        rich_help_panel="Queue",
        hidden=False,  # Unhide when new queue types added
        show_default=QueueType.SGE.value if "vaughan" in str(environ.get("HOSTNAME", "")) else QueueType.SLURM.value,
    ),
    validatepath: bool = typer.Option(
        True,
        "--validate-path/--no-validate-path",
        help="Validate the provided [green]$PATH[/green] value. Turn off at own risk. You may need to disable if you "
        "want to use the default [green]$PATH[/green] of your bash shell, or if you want to do some testing.",
        rich_help_panel="Extras",
    ),
    cache: bool = typer.Option(
        False,
        "--cache",
        help="Cache information for the submitted job to [green]~/hpcman/cache[/green].",
        rich_help_panel="Extras",
        hidden=True,  # Unhide when caching is better supported
    ),
    localdrive: Optional[LocalType] = typer.Option(
        None,
        "--local-drive",
        help="Use the local drive for the submitted job. If set to [yellow]pertask[/yellow], uses "
        f"[green]$TMPDIR[/green] ([yellow]{environ.get('TMPDIR')}[/yellow]) as the prefix for the tempdir "
        "(`mktemp -d -p`). If set to [yellow]shared[/yellow], uses [green]$TMPDIR/$USER/$DIRNAME[/green] "
        f"([yellow]$TMPDIR/{environ.get('USER')}/{Path.cwd().name}[/yellow]) as the tempdir, where "
        "[green]$DIRNAME[/green] is the name of the current directory.",
        rich_help_panel="Local drive",
    ),
    localprefix: Optional[str] = typer.Option(
        None,
        "--local-prefix",
        help="Override the default prefix of the tempdir. See the help for --local-drive.",
        rich_help_panel="Local drive",
    ),
    mirrortype: MirrorType = typer.Option(
        MirrorType.LINK,
        "--mirror-type",
        help="Type of mirroring to local drive. Only used if --local-drive is set.",
        rich_help_panel="Local drive",
        show_default=MirrorType.LINK.value,  # pyright: ignore
    ),
    copyresults: Optional[bool] = typer.Option(
        None,
        "--copy-results/--no-copy-results",
        help="Copy final results back to submission directory. Defaults to [magenta]True[/magenta] if --local-drive "
        "[yellow]pertask[/yellow] and [magenta]False[/magenta] if --local-drive [yellow]shared[/yellow].",
        rich_help_panel="Local drive",
        show_default=False,
    ),
    dryrun: Optional[bool] = typer.Option(
        False,
        "--dry-run",
        help="Do not actually submit the job, but set up the submit directory,",
        rich_help_panel="Extras",
    ),
    watch: Optional[bool] = typer.Option(
        False,
        "--watch",
        help="Run `hpcman queue stat --watch` to monitor the status of the jobs.",
        rich_help_panel="Extras",
    ),
    conda: bool = typer.Option(
        False,
        "--conda",
        help="Enable use of conda (e.g. [green]conda activate ...[/green]).",
        rich_help_panel="Conda",
    ),
    condaexe: Path = typer.Option(
        "/local/cluster/miniconda3_base/bin/conda",
        "--conda-exe",
        help="Path to [yellow]conda[/yellow] executable to enable.",
        rich_help_panel="Conda",
        exists=True,
        file_okay=True,
        dir_okay=False,
        readable=True,
        resolve_path=True,
    ),
    debug: bool = typer.Option(False, help="Help debugging", hidden=True),
) -> None:
    """
    Alias: [yellow bold]hqsub[/yellow bold]. Submit queueing job "COMMAND".
    Multiple lines of commands supported using piped input from STDIN. If
    providing as argument and there are spaces in the command, use quotes to
    provide the command, either '' or "" will work!  Do not use -c '', just
    provide the command as argument.
    """
    if command == "-" and sys.stdin.isatty():
        rprint("[red]No command input[/red] detected from stdin.")
        rprint("Please provide a command or pipe commands to stdin.")
        rprint("[red]Exiting...[/red]")
        exit(1)
    if debug:
        load_dotenv(override=True)
        pathlist = [x for x in dict.fromkeys(environ["PATH"].split(":")) if x]
    else:
        if path:
            pathlist = [x for x in dict.fromkeys(path.split(":")) if x]
        else:
            pathlist = []
    command = sys.stdin.read().strip() if command == "-" else command
    timestamp = generate_runname(command, datetime.now().strftime("%Y-%m-%d_%H-%M-%S"))
    runname = timestamp if runname == "-" else runname
    ncommands = len(command.split("\n"))
    if localdrive:
        if localprefix is None:
            if localdrive is LocalType.PERTASK:
                localprefix = "$TMPDIR"
            elif localdrive is LocalType.SHARED:
                localprefix = f"$TMPDIR/{environ.get('USER')}/{Path.cwd().name}"
        if copyresults is None:
            copyresults = localdrive is LocalType.PERTASK
    if holdauto:
        holdadd = ",".join(get_holds())
        if hold:
            hold = f"{hold},{holdadd}"
        else:
            hold = holdadd
    try:
        job: Job = Job(
            command=command,
            force=force,
            jobtype=jobtype,
            queue=queue,
            queuetype=queuetype,
            runname=Path(runname),
            freemem=freemem,  # pyright: ignore
            maxmem=maxmem,  # pyright: ignore
            procs=procs,
            filesize=maxfilesize,  # pyright: ignore
            concurrency=concurrency if jobtype is JobType.ARRAY else None,
            ncommands=ncommands if jobtype is JobType.ARRAY else None,
            tasklist=None,
            priority=None,
            path=pathlist,
            validatepath=validatepath,
            timestamp=timestamp,
            cache=cache,
            localdrive=localdrive,
            mirrortype=mirrortype,
            localprefix=localprefix,
            copyresults=copyresults,
            dryrun=dryrun,
            hold=hold,
            conda=conda,
            condaexe=condaexe,
        )
    except ValidationError as e:
        rprint(str(e))
        locs: List[str] = [str(err["loc"][0]) for err in e.errors()]
        rprint(f"[red]Please check your provided settings for these options:[/red] {locs}")
        exit(1)
    if debug:
        rprint(job)
    submit_job(job, debug)
    if watch:
        slurm = queuetype.value == "SLURM"
        sge = queuetype.value == "SGE"
        run_queue_status(user=getuser(), queue=None, debug=debug, watch=True, slurm=slurm, sge=sge)

usage()

Get past queue usage information

Source code in hpcman/queue/app.py

@app.command(rich_help_panel="Information")  # type: ignore
def usage() -> None:
    """
    Get past queue usage information
    """
    raise CommandNotImplemented