diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 0ea52f8..0ae94ab 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -12,9 +12,14 @@ jobs: runs-on: [self-hosted, Linux, X64] steps: + - name: Clean workspace + run: | + rm -rf "$GITHUB_WORKSPACE/.git" + - uses: actions/checkout@v5 with: fetch-depth: 0 + ref: ${{ github.event.release.tag_name }} - name: Fix git ownership run: git config --global --add safe.directory "$GITHUB_WORKSPACE" diff --git a/.gitignore b/.gitignore index a698b67..e3701aa 100644 --- a/.gitignore +++ b/.gitignore @@ -18,4 +18,7 @@ htmlcov/ docs/ # Version -src/qq_lib/_version.py \ No newline at end of file +src/qq_lib/_version.py + +# Playground +src/_playground*.py diff --git a/.readthedocs.yaml b/.readthedocs.yaml index 144c51c..d8b71fb 100644 --- a/.readthedocs.yaml +++ b/.readthedocs.yaml @@ -1,11 +1,11 @@ version: 2 build: - os: ubuntu-22.04 - tools: - python: "3.12" - commands: - - pip install uv - - uv pip install --system pdoc - - uv pip install --system . - - pdoc src/qq_lib -d google -o $READTHEDOCS_OUTPUT/html \ No newline at end of file + os: ubuntu-22.04 + tools: + python: "3.13" + commands: + - pip install uv + - uv pip install --system pdoc + - uv pip install --system . + - pdoc src/qq_lib -d google -o $READTHEDOCS_OUTPUT/html diff --git a/CHANGELOG.md b/CHANGELOG.md index ce958e8..ad53ea6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,18 @@ +## Version 0.12.0 +### Job collections +- **BREAKING CHANGE**: The `-s` option of `qq info` now corresponds to the `--server` option, not to the `--short` flag, for consistency with other commands. Use `--brief`/`-b` or the legacy long variant `--short` instead to get a brief summary of the job. +- The `qq killall` command has been deprecated in favor of `qq kill --all`, which has very similar behavior. It can still be used but will be completely removed in a future version of qq. +- `qq info`, `qq kill`, `qq sync`, `qq respawn`, `qq wipe`, and `qq go` can now accept one or more directories and operate on jobs within them. Job resolution is parallelized and therefore much faster than calling these commands individually. Some of these commands also accept the `--all` flag to operate on all unfinished jobs on the given batch server. +- `qq submit` now supports submitting multiple scripts at once. Submission is parallelized and therefore much faster than submitting scripts one by one. +- `qq clear` can now clear runtime files from multiple directories at once. +- *For more information about job collections, read [the manual](https://vachalab.github.io/qq-manual/job_collections.html).* + +### Other changes +- If a loop job does not create any archive files for the next cycle, `qq` creates an empty `.init` file fulfilling the loop job's requirements. +- If a loop job-specific or continuous job-specific option is used either on the command line or inside the submitted script, a warning is printed to inform the user that the option will be ignored. + +*** + ## Version 0.11 ### qq respawn - Failed or killed jobs can now be easily "respawned" using `qq respawn`. When respawning a job, qq will remove the working directory of the failed job, clear all runtime files, and resubmit the job with the same parameters as before. diff --git a/scripts/qq_scripts/gmx-eta b/scripts/qq_scripts/gmx-eta index 9711859..9f9dbe4 100755 --- a/scripts/qq_scripts/gmx-eta +++ b/scripts/qq_scripts/gmx-eta @@ -5,7 +5,7 @@ """ Get the estimated time of a Gromacs simulation finishing. -Version qq 0.11.0. +Version qq 0.12.0. Requires `uv`: https://docs.astral.sh/uv """ @@ -16,7 +16,7 @@ Requires `uv`: https://docs.astral.sh/uv # ] # # [tool.uv.sources] -# qq = { git = "https://github.com/VachaLab/qq.git", tag = "v0.11.0" } +# qq = { git = "https://github.com/VachaLab/qq.git", tag = "v0.12.0" } # /// import argparse diff --git a/scripts/qq_scripts/multi-check b/scripts/qq_scripts/multi-check index 9b7d9a0..9f30fc3 100755 --- a/scripts/qq_scripts/multi-check +++ b/scripts/qq_scripts/multi-check @@ -5,7 +5,8 @@ """ Check the state of qq jobs in multiple directories. -Version qq 0.11.0. +THIS SCRIPT IS DEPRECATED. YOU CAN USE `qq info -d ` INSTEAD. +Version qq 0.12.0. Requires `uv`: https://docs.astral.sh/uv """ @@ -16,7 +17,7 @@ Requires `uv`: https://docs.astral.sh/uv # ] # # [tool.uv.sources] -# qq = { git = "https://github.com/VachaLab/qq.git", tag = "v0.11.0" } +# qq = { git = "https://github.com/VachaLab/qq.git", tag = "v0.12.0" } # /// import argparse diff --git a/scripts/qq_scripts/multi-kill b/scripts/qq_scripts/multi-kill index e336780..da25ab6 100755 --- a/scripts/qq_scripts/multi-kill +++ b/scripts/qq_scripts/multi-kill @@ -5,7 +5,8 @@ """ Kill qq jobs in multiple directories. -Version qq 0.11.0. +THIS SCRIPT IS DEPRECATED. YOU CAN USE `qq kill -d ` INSTEAD. +Version qq 0.12.0. Requires `uv`: https://docs.astral.sh/uv """ @@ -16,7 +17,7 @@ Requires `uv`: https://docs.astral.sh/uv # ] # # [tool.uv.sources] -# qq = { git = "https://github.com/VachaLab/qq.git", tag = "v0.11.0" } +# qq = { git = "https://github.com/VachaLab/qq.git", tag = "v0.12.0" } # /// import argparse diff --git a/scripts/qq_scripts/multi-submit b/scripts/qq_scripts/multi-submit index fc90b68..a7386f2 100755 --- a/scripts/qq_scripts/multi-submit +++ b/scripts/qq_scripts/multi-submit @@ -5,7 +5,8 @@ """ Submit qq jobs from multiple directories. -Version qq 0.11.0. +THIS SCRIPT IS DEPRECATED. YOU CAN USE `qq submit /script.sh` INSTEAD. +Version qq 0.12.0. Requires `uv`: https://docs.astral.sh/uv """ @@ -16,7 +17,7 @@ Requires `uv`: https://docs.astral.sh/uv # ] # # [tool.uv.sources] -# qq = { git = "https://github.com/VachaLab/qq.git", tag = "v0.11.0" } +# qq = { git = "https://github.com/VachaLab/qq.git", tag = "v0.12.0" } # /// import argparse @@ -65,7 +66,7 @@ def submit_job(base_submitter: Submitter, directory: str) -> bool: with suppress(QQError): # remove runtime files in the directory, if possible - clearer = Clearer(Path(directory)) + clearer = Clearer([Path(directory)]) clearer.clear() # make sure that the directory is suitable for submission diff --git a/src/qq_lib/archive/archiver.py b/src/qq_lib/archive/archiver.py index d8b8448..105d0d7 100644 --- a/src/qq_lib/archive/archiver.py +++ b/src/qq_lib/archive/archiver.py @@ -74,7 +74,7 @@ def from_archive(self, dir: Path, cycle: int | None = None) -> None: QQError: If file transfer fails. """ if not ( - files := self._get_files( + files := self.get_files_matching_pattern( self._archive, self._input_machine, self._archive_format, cycle, False ) ): @@ -108,7 +108,11 @@ def to_archive(self, dir: Path) -> None: Raises: QQError: If file transfer or removal fails. """ - if not (files := self._get_files(dir, None, self._archive_format, None, False)): + if not ( + files := self.get_files_matching_pattern( + dir, None, self._archive_format, None, False + ) + ): logger.debug("Nothing to archive.") return @@ -153,7 +157,7 @@ def archive_runtime_files(self, job_name: str, cycle: int) -> None: QQError: If moving the runtime files fails. """ if not ( - files := self._get_files( + files := self.get_files_matching_pattern( self._input_dir, self._input_machine, # only use the stem of the job name, the extension will not be matched @@ -183,7 +187,7 @@ def archive_runtime_files(self, job_name: str, cycle: int) -> None: wait_seconds=CFG.archiver.retry_wait, ).run() - def _get_files( + def get_files_matching_pattern( self, directory: Path, host: str | None, @@ -205,7 +209,7 @@ def _get_files( include_qq_files (bool): Whether to include qq runtime files. Defaults to False. Returns: - list[Path]: A list of absolute paths to matching files. + list[Path]: A list of absolute (logical) paths to matching files. """ if cycle and is_printf_pattern(pattern): try: @@ -248,6 +252,16 @@ def _get_files( if regex.search(f.stem) and f.suffix not in CFG.suffixes.all_suffixes ] + def create_init_file(self, cycle: int) -> None: + """ + Create an empty init file for the given cycle. + Used as a fallback when no valid archive file is produced, ensuring + the next iteration of the loop job can proceed normally. + Args: + cycle (int): The index of the next cycle of the loop job. + """ + Path(f"{self._archive_format % cycle}.init").touch() + @staticmethod def _prepare_regex_pattern(pattern: str) -> re.Pattern[str]: """ diff --git a/src/qq_lib/batch/interface/interface.py b/src/qq_lib/batch/interface/interface.py index a0b81d5..be4f9b8 100644 --- a/src/qq_lib/batch/interface/interface.py +++ b/src/qq_lib/batch/interface/interface.py @@ -296,6 +296,29 @@ def get_batch_job(cls, job_id: str) -> TBatchJob: f"get_batch_job method is not implemented for {cls.__name__}" ) + @classmethod + def get_batch_jobs_from_ids(cls, job_ids: list[str]) -> list[TBatchJob]: + """ + Retrieve information about multiple jobs from the batch system. + + Batch jobs should be returned in the same order as they appear in `job_ids`. + A TBatchJob object should be returned for each job id, even if the job + no longer exists or its information is unavailable. + + Array jobs should NOT be expanded into their individual tasks. + + The default implementation is to call `get_batch_job` for each job id. + This implementation may be inefficient for large numbers of job ids and + should be overriden by subclasses. + + Args: + job_ids (list[str]): List of job identifiers. + + Returns: + list[TBatchJob]: List of TBatchJob objects, one for each job id. + """ + return [cls.get_batch_job(id) for id in job_ids] + @classmethod def get_unfinished_batch_jobs( cls, user: str, server: str | None = None diff --git a/src/qq_lib/batch/pbs/common.py b/src/qq_lib/batch/pbs/common.py index 2fe410c..bea6036 100644 --- a/src/qq_lib/batch/pbs/common.py +++ b/src/qq_lib/batch/pbs/common.py @@ -55,7 +55,9 @@ def parse_multi_pbs_dump_to_dictionaries( data = [] block, identifier = [], None - pattern = re.compile(rf"^\s*{keyword}:\s*(.*)$") if keyword else None + pattern = pattern = ( + re.compile(rf"^\s*(?:.*?{keyword}[:\s]+)(.*)$") if keyword else None + ) for line in text.splitlines(): # if the line is empty, start a new block diff --git a/src/qq_lib/batch/pbs/pbs.py b/src/qq_lib/batch/pbs/pbs.py index 36179f6..8a81e59 100644 --- a/src/qq_lib/batch/pbs/pbs.py +++ b/src/qq_lib/batch/pbs/pbs.py @@ -183,6 +183,20 @@ def job_kill_force(cls, job_id: str) -> None: def get_batch_job(cls, job_id: str) -> PBSJob: return PBSJob(job_id) + @classmethod + def get_batch_jobs_from_ids(cls, job_ids: list[str]) -> list[PBSJob]: + if not job_ids: + return [] + + command = f"qstat -fxw {' '.join(job_ids)}" + logger.debug(command) + return cls._get_batch_jobs_using_command( + command, + include_completed=True, + include_top_level_array=True, + ignore_exit_code=True, + ) + @classmethod def get_unfinished_batch_jobs( cls, user: str, server: str | None = None @@ -191,7 +205,12 @@ def get_unfinished_batch_jobs( if server: command += f" @{server}" logger.debug(command) - return cls._get_batch_jobs_using_command(command, include_completed=False) + return cls._get_batch_jobs_using_command( + command, + include_completed=False, + include_top_level_array=True, + ignore_exit_code=False, + ) @classmethod def get_batch_jobs(cls, user: str, server: str | None = None) -> list[PBSJob]: @@ -199,7 +218,12 @@ def get_batch_jobs(cls, user: str, server: str | None = None) -> list[PBSJob]: if server: command += f" @{server}" logger.debug(command) - return cls._get_batch_jobs_using_command(command, include_completed=True) + return cls._get_batch_jobs_using_command( + command, + include_completed=True, + include_top_level_array=True, + ignore_exit_code=False, + ) @classmethod def get_all_unfinished_batch_jobs(cls, server: str | None = None) -> list[PBSJob]: @@ -207,7 +231,12 @@ def get_all_unfinished_batch_jobs(cls, server: str | None = None) -> list[PBSJob if server: command += f" @{server}" logger.debug(command) - return cls._get_batch_jobs_using_command(command, include_completed=False) + return cls._get_batch_jobs_using_command( + command, + include_completed=False, + include_top_level_array=True, + ignore_exit_code=False, + ) @classmethod def get_all_batch_jobs(cls, server: str | None = None) -> list[PBSJob]: @@ -215,7 +244,12 @@ def get_all_batch_jobs(cls, server: str | None = None) -> list[PBSJob]: if server: command += f" @{server}" logger.debug(command) - return cls._get_batch_jobs_using_command(command, include_completed=True) + return cls._get_batch_jobs_using_command( + command, + include_completed=True, + include_top_level_array=True, + ignore_exit_code=False, + ) @classmethod def get_queues(cls, server: str | None = None) -> list[PBSQueue]: @@ -1000,7 +1034,11 @@ def _sync_directories( @classmethod def _get_batch_jobs_using_command( - cls, command: str, include_completed: bool + cls, + command: str, + include_completed: bool, + include_top_level_array: bool, + ignore_exit_code: bool, ) -> list[PBSJob]: """ Execute a shell command to retrieve information about PBS jobs and parse it. @@ -1009,6 +1047,10 @@ def _get_batch_jobs_using_command( command (str): The shell command to execute, typically a PBS query command. include_completed (bool): Include both completed and uncompleted jobs. If `False`, completed jobs are filtered out from the output. + include_top_level_array (bool): Include top-level array jobs in the output. + If `False`, top-level array jobs are filtered out from the output. + ignore_exit_code (bool): Ignore the exit code of the command. + If `False`, the command must return a zero exit code. Returns: list[PBSJob]: A list of `PBSJob` instances corresponding to the jobs @@ -1020,17 +1062,21 @@ def _get_batch_jobs_using_command( """ ... result = subprocess.run( - ["bash"], + # -oL (line-buffer stdout), -eL (line-buffer stderr) + # necessary for stdout and stderr merging + ["stdbuf", "-oL", "-eL", "bash"], input=command, text=True, check=False, - capture_output=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, errors="replace", ) - if result.returncode != 0: + if not ignore_exit_code and result.returncode != 0: raise QQError( - f"Could not retrieve information about jobs: {result.stderr.strip()}." + # standard error is written to stdout + f"Could not retrieve information about jobs: {result.stdout.strip()}." ) jobs = [] @@ -1039,8 +1085,7 @@ def _get_batch_jobs_using_command( ): job = PBSJob.from_dict(job_id, data) - # ignore top-level array jobs - if job.is_array_job(): + if not include_top_level_array and job.is_array_job(): continue # unless all jobs are to be shown, filter out completed jobs diff --git a/src/qq_lib/batch/slurm/slurm.py b/src/qq_lib/batch/slurm/slurm.py index 4018842..164ea96 100644 --- a/src/qq_lib/batch/slurm/slurm.py +++ b/src/qq_lib/batch/slurm/slurm.py @@ -14,6 +14,7 @@ from qq_lib.core.logger import get_logger from qq_lib.properties.depend import Depend from qq_lib.properties.resources import Resources +from qq_lib.properties.states import BatchState from .common import ( SACCT_FIELDS, @@ -152,6 +153,34 @@ def job_kill_force(cls, job_id: str) -> None: def get_batch_job(cls, job_id: str) -> SlurmJob: return SlurmJob(job_id) + @classmethod + def get_batch_jobs_from_ids(cls, job_ids: list[str]) -> list[SlurmJob]: + command = f"sacct --allocations --noheader --parsable2 -j {','.join(job_ids)} --format={SACCT_FIELDS}" + logger.debug(command) + + # sacct ignores IDs of nonexistent jobs - it will just not print any output for them + sacct_jobs = cls._get_batch_jobs_using_sacct_command(command) + sacct_jobs_dict = { + job.get_id(): job + for job in sacct_jobs + # filter out pending (queued) jobs - we want to use scontrol for them + if job.get_state() not in {BatchState.QUEUED, BatchState.HELD} + } + + # get information about the remaining jobs using scontrol + scontrol_jobs = cls._get_jobs_in_parallel( + [id for id in job_ids if id not in sacct_jobs_dict] + ) + scontrol_jobs_dict = {job.get_id(): job for job in scontrol_jobs} + + # merge all jobs but maintain their original order + all_jobs = [] + for id in job_ids: + if (job := sacct_jobs_dict.get(id)) or (job := scontrol_jobs_dict.get(id)): + all_jobs.append(job) + + return all_jobs + @classmethod def get_unfinished_batch_jobs( cls, user: str, server: str | None = None @@ -180,7 +209,7 @@ def get_batch_jobs(cls, user: str, server: str | None = None) -> list[SlurmJob]: # server unused _ = server - # get all jobs, except pending which are not available from sacct + # get all jobs, except pending for which full information is not available using sacct command = f"sacct -u {user} --allocations --noheader --parsable2 --array --format={SACCT_FIELDS}" logger.debug(command) @@ -625,7 +654,6 @@ def _get_batch_jobs_using_sacct_command(cls, command: str) -> list[SlurmJob]: QQError: If the command fails (non-zero return code) or if the output cannot be parsed into valid job information. """ - ... result = subprocess.run( ["bash"], input=command, @@ -670,7 +698,6 @@ def _get_batch_jobs_using_squeue_command(cls, command: str) -> list[SlurmJob]: QQError: If the command fails (non-zero return code) or if the output cannot be parsed into valid job information. """ - ... result = subprocess.run( ["bash"], input=command, @@ -687,22 +714,35 @@ def _get_batch_jobs_using_squeue_command(cls, command: str) -> list[SlurmJob]: ids = [line.strip() for line in result.stdout.split("\n") if line.strip()] + return Slurm._get_jobs_in_parallel(ids) + + @classmethod + def _get_jobs_in_parallel(cls, job_ids: list[str]) -> list[SlurmJob]: + """ + Constructs Slurm jobs in parallel. + + Args: + job_ids (list[str]): A list of job IDs to collect. + + Returns: + list[SlurmJob]: A list of SlurmJob objects. + """ + def get_job(job_id: str) -> SlurmJob: return SlurmJob(job_id) jobs: list[SlurmJob] = [] - # use ThreadPoolExecutor to get information about jobs in parallel with ThreadPoolExecutor( max_workers=CFG.slurm_options.jobs_scontrol_nthreads ) as executor: - future_to_id = {executor.submit(get_job, job_id): job_id for job_id in ids} + future_to_id = {executor.submit(get_job, id): id for id in job_ids} for future in as_completed(future_to_id): try: jobs.append(future.result()) except Exception as e: - job_id = future_to_id[future] - raise QQError(f"Failed to load job {job_id}: {e}.") from e + id = future_to_id[future] + raise QQError(f"Failed to load job {id}: {e}.") from e return jobs diff --git a/src/qq_lib/clear/clearer.py b/src/qq_lib/clear/clearer.py index f16b804..c83c00f 100644 --- a/src/qq_lib/clear/clearer.py +++ b/src/qq_lib/clear/clearer.py @@ -2,7 +2,10 @@ # Copyright (c) 2025-2026 Ladislav Bartos and Robert Vacha Lab +import threading from collections.abc import Iterable +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass from pathlib import Path from qq_lib.core.common import get_info_files, get_runtime_files @@ -15,70 +18,111 @@ logger = get_logger(__name__) +@dataclass +class _ClearResult: + """Accumulates results from clearing a single directory.""" + + deleted: int = 0 + excluded: int = 0 + + class Clearer: """ Handles detection and removal of qq runtime files from a directory. """ - def __init__(self, directory: Path): + def __init__(self, directories: list[Path]): """ - Initialize a Clearer for a specific directory. + Initialize a Clearer for one or more directories. Args: - directory (Path): The directory to clear qq runtime files from. + directories (list[Path]): The directories to clear qq runtime files from. """ - self._directory = directory + self._directories = directories def clear(self, force: bool = False) -> None: """ - Remove all qq runtime files from the directory that are safe to be removed. + Remove all qq runtime files from all directories that are safe to be removed. - Only qq files that do **not** correspond to an active or successfully - finished job will be removed, unless `force` is set to True. + Directories are cleared in parallel. Only qq files that do **not** + correspond to an active or successfully finished job will be removed, + unless `force` is set to True. A combined summary is logged at the end. Args: force (bool): If True, remove all qq runtime files, even if unsafe. """ - # get all qq runtime files - files = self._collect_runtime_files() - logger.debug(f"All qq runtime files: {files}.") - if not files: + results: list[_ClearResult] = [] + lock = threading.Lock() + + def clear_single(directory: Path) -> None: + result = Clearer._clear_directory(directory, force) + with lock: + results.append(result) + + with ThreadPoolExecutor( + max_workers=CFG.parallelization_options.clear_max_threads + ) as executor: + for directory in self._directories: + executor.submit(clear_single, directory) + + total_deleted = sum(r.deleted for r in results) + total_excluded = sum(r.excluded for r in results) + + if total_deleted == 0 and total_excluded == 0: logger.info("Nothing to clear.") return - # get files that should not be deleted - excluded: set[Path] = self._collect_excluded_files() if not force else set() - logger.debug(f"Files excluded from clearing: {excluded}.") - - # get files that are safe to be deleted - to_delete = files - excluded - logger.debug(f"Files to delete: {to_delete}.") - if not to_delete: + if total_deleted > 0: logger.info( - f"No qq files could be safely cleared. Rerun as '{CFG.binary_name} clear --force' to clear them forcibly." + f"Removed {total_deleted} qq file{'s' if total_deleted > 1 else ''}." ) - return - # remove the files that are safe to be deleted - Clearer._delete_files(to_delete) - logger.info( - f"Removed {len(to_delete)} qq file{'s' if len(to_delete) > 1 else ''}." - ) - if excluded: + if total_excluded > 0: logger.info( - f"{len(excluded)} qq files could not be safely cleared. Rerun as '{CFG.binary_name} clear --force' to clear them forcibly." + f"{total_excluded} qq file{'s' if total_excluded > 1 else ''} could not be safely cleared. " + f"Rerun as '{CFG.binary_name} clear --force' to clear them forcibly." ) - def _collect_runtime_files(self) -> set[Path]: + @staticmethod + def _clear_directory(directory: Path, force: bool) -> _ClearResult: + """ + Clear qq runtime files from a single directory and return the result. + + Args: + directory (Path): The directory to clear. + force (bool): If True, remove all qq runtime files, even if unsafe. + + Returns: + _ClearResult: The number of deleted and excluded files. + """ + files = Clearer._collect_runtime_files(directory) + logger.debug(f"All qq runtime files in '{directory}': {files}.") + if not files: + return _ClearResult() + + excluded = Clearer._collect_excluded_files(directory) if not force else set() + logger.debug(f"Files excluded from clearing in '{directory}': {excluded}.") + + to_delete = files - excluded + logger.debug(f"Files to delete in '{directory}': {to_delete}.") + + if to_delete: + Clearer._delete_files(to_delete) + + return _ClearResult(deleted=len(to_delete), excluded=len(excluded)) + + @staticmethod + def _collect_runtime_files(directory: Path) -> set[Path]: """ Collect all qq runtime files in the directory. Returns: set[Path]: Paths to all files matching qq-specific suffixes. """ - return set(get_runtime_files(self._directory)) + return set(get_runtime_files(directory)) - def _collect_excluded_files(self) -> set[Path]: + @staticmethod + def _collect_excluded_files(directory: Path) -> set[Path]: """ Collect qq runtime files that should **not** be deleted. @@ -90,7 +134,7 @@ def _collect_excluded_files(self) -> set[Path]: excluded = [] # iterate through info files - for file in get_info_files(self._directory): + for file in get_info_files(directory): try: informer = Informer.from_file(file) state = informer.get_real_state() @@ -105,14 +149,10 @@ def _collect_excluded_files(self) -> set[Path]: RealState.IN_AN_INCONSISTENT_STATE, ]: excluded.append(file) # qq info file + excluded.append(directory / informer.info.stdout_file) # script stdout + excluded.append(directory / informer.info.stderr_file) # script stderr excluded.append( - self._directory / informer.info.stdout_file - ) # script stdout - excluded.append( - self._directory / informer.info.stderr_file - ) # script stderr - excluded.append( - (self._directory / informer.info.job_name).with_suffix( + (directory / informer.info.job_name).with_suffix( CFG.suffixes.qq_out ) ) # qq out file diff --git a/src/qq_lib/clear/cli.py b/src/qq_lib/clear/cli.py index 6f4aa6f..b7e09de 100644 --- a/src/qq_lib/clear/cli.py +++ b/src/qq_lib/clear/cli.py @@ -7,7 +7,7 @@ import click -from qq_lib.core.click_format import GNUHelpColorsCommand +from qq_lib.core.click_format import QQOperatorCommand from qq_lib.core.config import CFG from qq_lib.core.error import QQError from qq_lib.core.logger import get_logger @@ -19,21 +19,21 @@ @click.command( short_help="Delete qq runtime files from a directory.", - help=f"""Delete qq runtime files from the current or specified directory. + help=f"""Delete qq runtime files from the current directory or specified directories. -By default, `{CFG.binary_name} clear` clears runtime files from the current directory. Use `--dir` to specify a different directory. +By default, `{CFG.binary_name} clear` clears runtime files from the current directory. Use `--dir` to specify a different directory or directories. By default, `{CFG.binary_name} clear` removes only those files that do not correspond to an active or successfully completed job. To force deletion of all qq files regardless of job status, use the `--force` flag.""", - cls=GNUHelpColorsCommand, + cls=QQOperatorCommand, help_options_color="bright_blue", ) @click.option( "-d", "--dir", - type=str, - help="Directory from which to clear qq runtime files.", - default=None, + type=click.Path(exists=True, file_okay=False, path_type=Path), + multiple=True, + help="One or more directories from which to clear qq runtime files. Supports globs.", ) @click.option( "--force", @@ -41,16 +41,16 @@ help="Force deletion of all qq runtime files, even if jobs are active or successfully completed.", default=False, ) -def clear(dir: str | None, force: bool) -> NoReturn: +def clear(dir: tuple[Path, ...], force: bool) -> NoReturn: """ - Delete qq runtime files in a specified directory or the current directory. + Delete qq runtime files in specified directories or the current directory. Only runtime files that do **not** correspond to an active or successfully completed job are deleted, unless the `force` option is used. """ try: - clearer = Clearer(Path(dir) if dir else Path()) + clearer = Clearer(list(dir) if dir else [Path()]) clearer.clear(force) sys.exit(0) except QQError as e: diff --git a/src/qq_lib/core/click_format.py b/src/qq_lib/core/click_format.py index f0a0a69..5cdebb8 100644 --- a/src/qq_lib/core/click_format.py +++ b/src/qq_lib/core/click_format.py @@ -10,6 +10,7 @@ """ from collections.abc import Sequence +from pathlib import Path import click from click import Context, HelpFormatter @@ -150,3 +151,64 @@ def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]: i += 1 args = processed return super().parse_args(ctx, args) + + +class GlobDirectoryMixin: + """ + Mixin that adds glob-expanding multi-value support for -d/--directory. + + Click does not support nargs=-1 on options. This mixin pre-processes the + argument list before Click's parser sees it, greedily consuming all + non-flag tokens following -d/--directory, glob-expanding each one, and + rewriting them as repeated -d pairs that Click's multiple=True + parser handles correctly. + + Must appear before the Click command class in the MRO so that its + parse_args runs first. + """ + + def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]: + """ + Rewrite -d/--dir tokens to support multiple values and globs. + + Greedily consumes all non-flag tokens following a -d/--dir flag, + glob-expands each one relative to its own parent directory, and emits + one -d pair per expanded result. + + Args: + ctx (click.Context): The current Click context. + args (list[str]): The raw argument list. + + Returns: + list[str]: The remaining unparsed arguments after delegation to the + parent parser. + """ + rewritten: list[str] = [] + i = 0 + while i < len(args): + if args[i] in ("-d", "--dir"): + i += 1 + while i < len(args) and not args[i].startswith("-"): + pattern = args[i] + p = Path(pattern) + if not p.name: + # pattern is a bare directory (e.g. "." or "/some/dir/") - no glob needed + rewritten.extend(["-d", pattern]) + else: + expanded = sorted(p.parent.glob(p.name)) + if expanded: + for path in expanded: + rewritten.extend(["-d", str(path)]) + else: + rewritten.extend(["-d", pattern]) + i += 1 + else: + rewritten.append(args[i]) + i += 1 + return super().parse_args(ctx, rewritten) # ty:ignore[unresolved-attribute] + + +class QQOperatorCommand(GlobDirectoryMixin, GNUHelpColorsCommand): + """GNUHelpColorsCommand with glob-expanding -d/--dir support.""" + + pass diff --git a/src/qq_lib/core/command_runner.py b/src/qq_lib/core/command_runner.py index 5a288bd..4549a1e 100644 --- a/src/qq_lib/core/command_runner.py +++ b/src/qq_lib/core/command_runner.py @@ -2,6 +2,7 @@ # Copyright (c) 2025-2026 Ladislav Bartos and Robert Vacha Lab +import getpass import logging import sys import threading @@ -10,7 +11,8 @@ from pathlib import Path from typing import Any, NoReturn, Self -from qq_lib.core.common import get_info_files +from qq_lib.batch.interface import BatchInterface, BatchJobInterface +from qq_lib.core.common import get_info_files, translate_server from qq_lib.core.config import CFG from qq_lib.core.error import QQError from qq_lib.info import Informer @@ -35,37 +37,50 @@ class CommandRunner: def __init__( self, - jobs: tuple[str, ...], + job_ids: tuple[str, ...], + directories: tuple[Path, ...], + all: bool, + server: str | None, callback: Callable, logger: logging.Logger, *args: Any, n_threads: int = 1, - directory: Path | None = None, **kwargs: Any, ): """ Initialize a CommandRunner. Args: - jobs (tuple[str, ...]): Job IDs provided on the command line. If empty, the `directory` - is searched for qq info files. + job_ids (tuple[str, ...]): List of batch job IDs to operate on. + directories (tuple[Path, ...]): List of directories to search for qq jobs in. + If `job_ids` and `directories` are both empty and `all` is `False`, the current directory is searched for jobs. + all (bool): Collect all qq jobs for the given user on the given server. + server (str | None): The batch server to collect qq jobs from. If `None`, the local batch server is used. + The server can be specified as a shortcut. callback (Callable): The operation to perform on each resolved Informer. The informer is passed as the first argument, followed by `*args` and `**kwargs`. logger (logging.Logger): Logger instance used for error and critical messages. Should be the module-level logger of the calling CLI module. *args (Any): Additional positional arguments forwarded to `callback`. n_threads (int): Number of threads for parallel informer resolution. Defaults to 1 (serial). - directory (Path | None): Directory to search for qq info files. If `None`, the current directory is used. **kwargs (Any): Additional keyword arguments forwarded to `callback`. """ - self._n_threads = n_threads - self._directory = directory or Path.cwd() - self._logger = logger - self._jobs = jobs + self._job_ids = job_ids + self._directories = directories + self._all = all + self._user = getpass.getuser() + self._server = translate_server(server) if server else None + if not self._job_ids and not self._directories and not all: + self._directories = [Path.cwd()] + self._callback = callback self._args = args self._kwargs = kwargs + + self._batch_system = BatchInterface.from_env_var_or_guess() + self._logger = logger self._exception_handlers: dict[type[Exception], Callable] = {} + self._n_threads = n_threads self.n_jobs = 0 self.current_iteration = 0 @@ -132,30 +147,80 @@ def _build_targets(self) -> list[Callable[[], Informer]]: Raises: QQError: If no job IDs were given and no info files were found in the current directory. """ + if self._server and not self._all: + self._logger.warning( + "Server is only used with --all. Ignoring server option." + ) + self._server = None + targets: list[Callable[[], Informer]] = [] - def _resolve_and_prepare(resolve: Callable[[], Informer]) -> Informer: - """Resolve an informer and reload its batch info.""" - informer = resolve() - informer.load_batch_info() - return informer + batch_jobs = [] + if self._job_ids: + batch_jobs.extend( + self._batch_system.get_batch_jobs_from_ids(list(self._job_ids)) + ) + if self._all: + batch_jobs.extend( + self._batch_system.get_unfinished_batch_jobs(self._user, self._server) + ) - if self._jobs: - for job in self._jobs: - targets.append( - lambda j=job: _resolve_and_prepare(lambda: Informer.from_job_id(j)) - ) - else: - info_files = get_info_files(self._directory) - if not info_files: + if batch_jobs: + targets.extend(self._build_targets_from_batch_jobs(batch_jobs)) + + if self._directories: + targets.extend(self._build_targets_from_files(list(self._directories))) + + if not targets: + if not self._job_ids and not self._all: raise QQError("No qq job info file found.") - for info in info_files: - targets.append( - lambda f=info: _resolve_and_prepare(lambda: Informer.from_file(f)) - ) + raise QQError("No jobs found.") return targets + def _build_targets_from_batch_jobs( + self, + jobs: list[BatchJobInterface], + ) -> list[Callable[[], Informer]]: + """ + Build preparation targets from a list of batch jobs. + + Args: + jobs (list[BatchJobInterface]): The batch jobs to resolve. + + Returns: + list[Callable[[], Informer]]: Per-job preparation callables. + """ + return [lambda bj=batch_job: Informer.from_batch_job(bj) for batch_job in jobs] + + def _build_targets_from_files( + self, + directories: list[Path], + ) -> list[Callable[[], Informer]]: + """ + Build preparation targets from qq info files across multiple directories. + + Directories are searched in order. Each target loads an `Informer` + from a file and queries the batch system for its info individually. + + Args: + directories (list[Path]): Directories to search for info files. + + Returns: + list[Callable[[], Informer]]: Per-job preparation callables. + """ + + def _resolve_and_prepare(path: Path) -> Informer: + informer = Informer.from_file(path) + informer.load_batch_info() + return informer + + info_files: list[Path] = [] + for directory in directories: + info_files.extend(get_info_files(directory)) + + return [lambda f=info_file: _resolve_and_prepare(f) for info_file in info_files] + def _run_pipeline(self, targets: list[Callable[[], Informer]]) -> None: """ Run the prepare-then-execute pipeline. @@ -220,7 +285,7 @@ def prepare(index: int, target: Callable[[], Informer]) -> None: self._execute(result) else: raise ValueError( - f"Unexpected result type: {type(result)}. This is a bug, please repport it." + f"Unexpected result type: {type(result)}. This is a bug, please report it." ) def _execute(self, informer: Informer) -> None: diff --git a/src/qq_lib/core/common.py b/src/qq_lib/core/common.py index 4593fc5..454e0a9 100644 --- a/src/qq_lib/core/common.py +++ b/src/qq_lib/core/common.py @@ -99,7 +99,7 @@ def get_info_file(directory: Path) -> Path: """ Locate the qq job info file in a directory. - This function searches for files matching the `QQ_INFO_SUFFIX` in the + This function searches for files with suffix `CFG.suffixes.qq_info` in the provided directory. It raises an error if none or multiple info files are found. Args: @@ -124,7 +124,7 @@ def get_info_files(directory: Path) -> list[Path]: """ Retrieve all qq job info files in a directory. - This function searches for files matching the `QQ_INFO_SUFFIX` in the + This function searches for files with suffix `CFG.suffixes.qq_info` in the provided directory. The files are sorted by their last modification time (with the newest modified file being last in the list). diff --git a/src/qq_lib/core/config.py b/src/qq_lib/core/config.py index 98bac9c..a2b28b1 100644 --- a/src/qq_lib/core/config.py +++ b/src/qq_lib/core/config.py @@ -182,6 +182,18 @@ class FullInfoPanelSettings: rule_style: str = "white" +@dataclass(frozen=True) +class BriefInfoSettings: + """Settings for brief info display.""" + + # Color of the job ID in brief info. + job_id_color: str = "default" + # Color of the directory path in brief info. + dir_path_color: str = "cyan" + # Color of the loop info in brief info. + loop_info_color: str = "grey70" + + @dataclass(frozen=True) class PresenterSettings: """Settings for Presenter.""" @@ -196,6 +208,9 @@ class PresenterSettings: default_factory=FullInfoPanelSettings ) + # Settings for the brief info display. + brief_info: BriefInfoSettings = field(default_factory=BriefInfoSettings) + # Style used for the keys in job status/info panel. key_style: str = "default bold" # Style used for values in job status/info panel. @@ -468,7 +483,13 @@ class ParallelizationOptions: """Options associated with multithreaded execution.""" # Maximal number of threads used to collect job information. - job_info_max_threads: int = 8 + job_info_max_threads: int = 16 + + # Maximal number of threads used to submit jobs in parallel. + submission_max_threads: int = 8 + + # Maximal number of threads used to clear runtime files in parallel. + clear_max_threads: int = 8 @dataclass(frozen=True) diff --git a/src/qq_lib/go/cli.py b/src/qq_lib/go/cli.py index b332377..cb55052 100644 --- a/src/qq_lib/go/cli.py +++ b/src/qq_lib/go/cli.py @@ -1,12 +1,13 @@ # Released under MIT License. # Copyright (c) 2025-2026 Ladislav Bartos and Robert Vacha Lab +from pathlib import Path from typing import NoReturn import click from rich.console import Console -from qq_lib.core.click_format import GNUHelpColorsCommand +from qq_lib.core.click_format import QQOperatorCommand from qq_lib.core.command_runner import CommandRunner from qq_lib.core.config import CFG from qq_lib.core.error import ( @@ -28,18 +29,20 @@ @click.command( short_help="Open a shell in a job's working directory.", help=f"""Open a new shell in the working directory of the specified qq job, or in the -working directory of the job submitted from the current directory. +working directories of qq jobs found in the specified directories. -{click.style("JOB_ID", fg="green")} One or more IDs of jobs whose working directories should be entered. Optional. +{click.style("JOB_ID...", fg="green")} One or more IDs of jobs whose working directories should be entered. Optional. -If no JOB_ID is specified, `{CFG.binary_name} go` searches for qq jobs in the current directory. +If no JOB_ID and no directory are specified, `{CFG.binary_name} go` searches for qq jobs in the current directory. If multiple suitable jobs are provided or found, `{CFG.binary_name} go` opens a shell for each job in turn. +You can combine JOB_IDs with directories. All JOB_IDs must be specified before the `--dir` option. + Uses `cd` for local directories or `ssh` if the working directory is on a remote host. Note that this command does not change the working directory of the current shell; it always opens a new shell at the destination. """, - cls=GNUHelpColorsCommand, + cls=QQOperatorCommand, help_options_color="bright_blue", ) @click.argument( @@ -50,12 +53,37 @@ default=None, nargs=-1, ) -def go(jobs: tuple[str, ...]) -> NoReturn: +@click.option( + "-d", + "--dir", + type=click.Path(exists=True, file_okay=False, path_type=Path), + multiple=True, + help="One or more directories to search for qq jobs in. Supports globs.", +) +@click.option( + "-a", + "--all", + is_flag=True, + help="Open a shell in the working directories of all your unfinished jobs.", +) +@click.option( + "-s", + "--server", + default=None, + help="Operate on jobs on the specified batch server. If not specified, the current server is used. Only used with --all.", +) +def go( + jobs: tuple[str, ...], dir: tuple[Path, ...], all: bool, server: str | None +) -> NoReturn: """ - Go to the working directory (directories) of the specified qq job(s) or qq job(s) submitted from this directory. + Go to the working directory (directories) of the specified qq job(s) + or qq job(s) submitted from the specified directories. """ CommandRunner( jobs, + dir, + all, + server, _go_to_job, logger, n_threads=CFG.parallelization_options.job_info_max_threads, diff --git a/src/qq_lib/info/cli.py b/src/qq_lib/info/cli.py index a968db5..e8b8306 100644 --- a/src/qq_lib/info/cli.py +++ b/src/qq_lib/info/cli.py @@ -1,12 +1,13 @@ # Released under MIT License. # Copyright (c) 2025-2026 Ladislav Bartos and Robert Vacha Lab +from pathlib import Path from typing import NoReturn import click from rich.console import Console -from qq_lib.core.click_format import GNUHelpColorsCommand +from qq_lib.core.click_format import QQOperatorCommand from qq_lib.core.command_runner import CommandRunner from qq_lib.core.config import CFG from qq_lib.core.error import QQError @@ -18,15 +19,32 @@ logger = get_logger(__name__) +# TODO: remove before version 1.0 +# This is to help users to adjust to the breaking change in version 0.12. +class InfoCommand(QQOperatorCommand): + def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]: + if "-s" in args: + next_arg_index = args.index("-s") + 1 + next_arg = args[next_arg_index] if next_arg_index < len(args) else None + if next_arg is None or next_arg.startswith("-"): + raise click.UsageError( + "'-s' now means '--server', not '--short'. " + "Use '-s ' to specify a batch server. " + "Or '--short'/'--brief'/'-b' to display only the job ID and current state." + ) + return super().parse_args(ctx, args) + + @click.command( short_help="Display information about a job.", - help=f"""Display information about the state and properties of the specified qq jobs, -or of qq jobs found in the current directory. + help=f"""Display information about the state and properties of the specified qq jobs or of qq jobs found in the specified directories. + +{click.style("JOB_ID...", fg="green")} One or more IDs of jobs to display information for. Optional. -{click.style("JOB_ID", fg="green")} One or more IDs of jobs to display information for. Optional. +If no JOB_ID and no directory are specified, `{CFG.binary_name} info` searches for qq jobs in the current directory. -If no JOB_ID is specified, `{CFG.binary_name} info` searches for qq jobs in the current directory.""", - cls=GNUHelpColorsCommand, +You can combine JOB_IDs with directories. All JOB_IDs must be specified before the `--dir` option.""", + cls=InfoCommand, help_options_color="bright_blue", ) @click.argument( @@ -36,34 +54,70 @@ nargs=-1, ) @click.option( - "-s", "--short", is_flag=True, help="Display only the job ID and current state." + "-d", + "--dir", + type=click.Path(exists=True, file_okay=False, path_type=Path), + multiple=True, + help="One or more directories to search for qq jobs in. Supports globs.", +) +@click.option( + "-a", + "--all", + is_flag=True, + help="Print info for all your unfinished jobs.", +) +@click.option( + "-s", + "--server", + default=None, + help="Collect jobs from the specified batch server. If not specified, the current server is used. Only used with --all.", +) +@click.option( + "-b", + "--brief", + "--short", + is_flag=True, + help="Display a brief summary of the job.", ) -def info(jobs: tuple[str, ...], short: bool) -> NoReturn: +def info( + jobs: tuple[str, ...], + dir: tuple[Path, ...], + all: bool, + server: str | None, + brief: bool, +) -> NoReturn: """ Get information about the specified qq jobs or qq jobs submitted from this directory. """ CommandRunner( jobs, + dir, + all, + server, _info_for_job, logger, - short, + brief, + len(dir) > 0, n_threads=CFG.parallelization_options.job_info_max_threads, ).on_exception(QQError, handle_general_qq_error).run() -def _info_for_job(informer: Informer, short: bool) -> None: +def _info_for_job(informer: Informer, brief: bool, dirs_provided: bool) -> None: """ Display information about a qq job associated with the specified Informer. Args: informer (Informer): Informer associated with the job. - short (bool): If True, print only the job ID and the current job state. + brief (bool): If True, print only the job ID, its directory + relative to the current directory (if `dirs_provided`), + and the current job state. If False, print the full formatted information panel. + dirs_provided (bool): If true, at least one directory was provided. """ presenter = Presenter(informer) console = Console() - if short: - console.print(presenter.get_short_info()) + if brief: + console.print(presenter.get_brief_info(dirs_provided)) else: panel = presenter.create_full_info_panel(console) console.print(panel) diff --git a/src/qq_lib/info/informer.py b/src/qq_lib/info/informer.py index 0c90f7f..94d07fd 100644 --- a/src/qq_lib/info/informer.py +++ b/src/qq_lib/info/informer.py @@ -1,9 +1,9 @@ # Released under MIT License. # Copyright (c) 2025-2026 Ladislav Bartos and Robert Vacha Lab -from datetime import datetime -from pathlib import Path -from typing import Self +from __future__ import annotations + +from typing import TYPE_CHECKING, Self from qq_lib.batch.interface import BatchInterface, BatchJobInterface from qq_lib.core.common import construct_info_file_path @@ -12,6 +12,10 @@ from qq_lib.properties.info import Info from qq_lib.properties.states import BatchState, NaiveState, RealState +if TYPE_CHECKING: + from datetime import datetime + from pathlib import Path + logger = get_logger(__name__) @@ -119,6 +123,26 @@ def from_batch_job(cls, batch_job: BatchJobInterface) -> Self: informer._batch_info = batch_job return informer + @staticmethod + def set_batch_info_in_bulk(informers: list[Informer]) -> None: + """ + Set the batch info for a list of informers in bulk, + by querying the batch system as few times as possible. + + If the corresponding batch job no longer exists, batch info is still set, + but the BatchJobInterface is empty. + + Args: + informers (list[Informer]): The list of informers to set the batch info for. + """ + if not informers: + return + + job_ids = [informer.info.job_id for informer in informers] + batch_jobs = informers[0].batch_system.get_batch_jobs_from_ids(job_ids) + for informer, batch_job in zip(informers, batch_jobs): + informer._batch_info = batch_job + def to_file(self, file: Path, host: str | None = None) -> None: """ Export the job information to a file. @@ -226,6 +250,12 @@ def load_batch_info(self) -> None: if self._batch_info is None: self._batch_info = self.batch_system.get_batch_job(self.info.job_id) + def set_batch_info(self, batch_info: BatchJobInterface) -> None: + """ + Set the batch job information. + """ + self._batch_info = batch_info + def get_batch_state(self) -> BatchState: """ Return the job's state as reported by the batch system. diff --git a/src/qq_lib/info/presenter.py b/src/qq_lib/info/presenter.py index c729d69..2fb91eb 100644 --- a/src/qq_lib/info/presenter.py +++ b/src/qq_lib/info/presenter.py @@ -3,6 +3,8 @@ from collections.abc import Sequence from datetime import datetime +from pathlib import Path +from warnings import deprecated from rich.align import Align from rich.console import Console, Group @@ -141,20 +143,56 @@ def create_full_info_panel(self, console: Console | None = None) -> Group: return Group(Text(""), full_panel, Text("")) - def get_short_info(self) -> Text: + def get_brief_info(self, print_dir: bool) -> Text: """ Return a concise, colorized summary of the job's current state. Returns: - Text: A Rich `Text` object containing the job ID followed by the - current state, colorized according to the `RealState`. + Text: A Rich `Text` object containing brief information about the job. """ state = self._informer.get_real_state() - return ( - Text(self._informer.info.job_id) - + " " - + Text(str(state), style=state.color) + job_id = Text( + self._informer.info.job_id, + style=CFG.presenter.brief_info.job_id_color, ) + state_text = Text(str(state), style=state.color) + + if print_dir: + dir_path = Text( + str( + self._informer.info.input_dir.resolve().relative_to( + Path.cwd(), walk_up=True + ) + ), + style=CFG.presenter.brief_info.dir_path_color, + ) + full_text = job_id + " [" + dir_path + "] " + state_text + else: + full_text = job_id + " " + state_text + + if loop_info := self._informer.info.loop_info: + full_text += ( + Text(" [") + + Text( + f"{loop_info.current}/{loop_info.end}", + style=CFG.presenter.brief_info.loop_info_color, + ) + + "]" + ) + + return full_text + + @deprecated("This function has been deprecated, use get_brief_info instead.") + def get_short_info(self, print_dir: bool) -> Text: + """ + Return a concise, colorized summary of the job's current state. + + This function is deprecated, use `get_brief_info` instead. + + Returns: + Text: A Rich `Text` object containing brief information about the job. + """ + return self.get_brief_info(print_dir) def _create_basic_info_table(self) -> Table: """ diff --git a/src/qq_lib/kill/cli.py b/src/qq_lib/kill/cli.py index fcf0680..f3b39d4 100644 --- a/src/qq_lib/kill/cli.py +++ b/src/qq_lib/kill/cli.py @@ -1,12 +1,13 @@ # Released under MIT License. # Copyright (c) 2025-2026 Ladislav Bartos and Robert Vacha Lab +from pathlib import Path from typing import NoReturn import click from rich.console import Console -from qq_lib.core.click_format import GNUHelpColorsCommand +from qq_lib.core.click_format import QQOperatorCommand from qq_lib.core.command_runner import CommandRunner from qq_lib.core.common import ( yes_or_no_prompt, @@ -30,11 +31,13 @@ @click.command( short_help="Terminate a job.", - help=f"""Terminate the specified qq jobs, or all qq jobs in the current directory. + help=f"""Terminate the specified qq jobs or all qq jobs in the specified directories. -{click.style("JOB_ID", fg="green")} One or more IDs of jobs to terminate. Optional. +{click.style("JOB_ID...", fg="green")} One or more IDs of jobs to terminate. Optional. -If no JOB_ID is specified, `{CFG.binary_name} kill` searches for qq jobs in the current directory. +If no JOB_ID and no directory are specified, `{CFG.binary_name} kill` searches for qq jobs in the current directory. + +You can combine JOB_IDs with directories. All JOB_IDs must be specified before the `--dir` option. By default, `{CFG.binary_name} kill` prompts for confirmation before terminating a job. @@ -43,7 +46,7 @@ When the `--force` flag is used, `{CFG.binary_name} kill` attempts to terminate any job regardless of its state, including jobs that are, according to qq, already finished or killed. This can be useful for removing lingering or stuck jobs.""", - cls=GNUHelpColorsCommand, + cls=QQOperatorCommand, help_options_color="bright_blue", ) @click.argument( @@ -55,16 +58,42 @@ nargs=-1, ) @click.option( - "-y", "--yes", is_flag=True, help="Terminate the job without confirmation." + "-d", + "--dir", + type=click.Path(exists=True, file_okay=False, path_type=Path), + multiple=True, + help="One or more directories to search for qq jobs in. Supports globs.", +) +@click.option( + "-a", + "--all", + is_flag=True, + help="Terminate all your unfinished jobs.", +) +@click.option( + "-s", + "--server", + default=None, + help="Terminate jobs on the specified batch server. If not specified, the current server is used. Only used with --all.", +) +@click.option( + "-y", "--yes", is_flag=True, help="Terminate the job(s) without confirmation." ) @click.option( "--force", is_flag=True, - help="Terminate the job forcibly, ignoring its current state and without confirmation.", + help="Terminate the job(s) forcibly, ignoring its current state and without confirmation.", ) -def kill(jobs: tuple[str, ...], yes: bool = False, force: bool = False) -> NoReturn: +def kill( + jobs: tuple[str, ...], + dir: tuple[Path, ...], + all: bool, + server: str | None, + yes: bool = False, + force: bool = False, +) -> NoReturn: """ - Terminate the specified qq job(s) or qq job(s) submitted from the current directory. + Terminate the specified qq job(s) or qq job(s) submitted from the specified directories. Details Killing a job sets its state to "killed". This is handled either by `qq kill` or @@ -88,6 +117,9 @@ def kill(jobs: tuple[str, ...], yes: bool = False, force: bool = False) -> NoRet """ CommandRunner( jobs, + dir, + all, + server, kill_job, logger, force, diff --git a/src/qq_lib/killall/cli.py b/src/qq_lib/killall/cli.py index 3e55113..9e2e8e2 100644 --- a/src/qq_lib/killall/cli.py +++ b/src/qq_lib/killall/cli.py @@ -25,11 +25,12 @@ @click.command( short_help="Terminate all your jobs.", - help="""Terminate all your submitted qq jobs. + help=f"""Terminate all your submitted qq jobs. {click.style("THIS COMMAND IS DEPRECATED! Use `qq kill --all` instead.", fg="red")} This command is only able to terminate qq jobs, all other jobs are not affected by it.""", cls=GNUHelpColorsCommand, help_options_color="bright_blue", + hidden=True, ) @click.option( "-y", "--yes", is_flag=True, help="Terminate the jobs without confirmation." @@ -49,6 +50,9 @@ def killall( yes: bool = False, force: bool = False, server: str | None = None ) -> NoReturn: try: + logger.warning( + "This command is deprecated! It will be removed in a future version of qq. Use `qq kill --all` instead." + ) BatchSystem = BatchInterface.from_env_var_or_guess() if server: @@ -86,6 +90,9 @@ def killall( else: logger.info("Operation aborted.") + logger.warning( + "This command is deprecated! It will be removed in a future version of qq. Use `qq kill --all` instead." + ) sys.exit(0) except QQError as e: logger.error(e) diff --git a/src/qq_lib/properties/resubmit_host.py b/src/qq_lib/properties/resubmit_host.py index 4a0ca3e..841e116 100644 --- a/src/qq_lib/properties/resubmit_host.py +++ b/src/qq_lib/properties/resubmit_host.py @@ -19,7 +19,7 @@ class ResubmitHost(ABC): Subclasses: InputHost: Resolves to the original input machine. WorkHost: Resolves to the current working node. - CustomHost: Resolves to an explicitly specified hostname. + ExplicitHost: Resolves to an explicitly specified hostname. """ @classmethod diff --git a/src/qq_lib/respawn/cli.py b/src/qq_lib/respawn/cli.py index c77fa2c..f0d69ce 100644 --- a/src/qq_lib/respawn/cli.py +++ b/src/qq_lib/respawn/cli.py @@ -1,12 +1,13 @@ # Released under MIT License. # Copyright (c) 2025-2026 Ladislav Bartos and Robert Vacha Lab +from pathlib import Path from typing import NoReturn import click from rich.console import Console -from qq_lib.core.click_format import GNUHelpColorsCommand +from qq_lib.core.click_format import QQOperatorCommand from qq_lib.core.command_runner import CommandRunner from qq_lib.core.config import CFG from qq_lib.core.error import QQError, QQNotSuitableError @@ -24,16 +25,18 @@ @click.command( short_help="Respawn a failed/killed job.", - help=f"""Respawn the specified qq jobs, or all qq jobs in the current directory. + help=f"""Respawn the specified qq jobs, or all qq jobs in the specified directories. -{click.style("JOB_ID", fg="green")} One or more IDs of jobs to respawn. Optional. +{click.style("JOB_ID...", fg="green")} One or more IDs of jobs to respawn. Optional. -If no JOB_ID is specified, `{CFG.binary_name} respawn` searches for qq jobs in the current directory. +If no JOB_ID and no directory are specified, `{CFG.binary_name} respawn` searches for qq jobs in the current directory. + +You can combine JOB_IDs with directories. All JOB_IDs must be specified before the `--dir` option. Respawning resubmits a failed or killed job to the batch system with its original parameters. This is useful when a job fails due to a node failure, an unexpected walltime limit, a random crash, or various other types of premature termination.""", - cls=GNUHelpColorsCommand, + cls=QQOperatorCommand, help_options_color="bright_blue", ) @click.argument( @@ -44,9 +47,19 @@ default=None, nargs=-1, ) -def respawn(jobs: tuple[str, ...]) -> NoReturn: +@click.option( + "-d", + "--dir", + type=click.Path(exists=True, file_okay=False, path_type=Path), + multiple=True, + help="One or more directories to search for qq jobs in. Supports globs.", +) +def respawn(jobs: tuple[str, ...], dir: tuple[Path, ...]) -> NoReturn: CommandRunner( jobs, + dir, + False, + None, _respawn_job, logger, n_threads=CFG.parallelization_options.job_info_max_threads, diff --git a/src/qq_lib/respawn/respawner.py b/src/qq_lib/respawn/respawner.py index 5ce9cae..39fa3be 100644 --- a/src/qq_lib/respawn/respawner.py +++ b/src/qq_lib/respawn/respawner.py @@ -60,7 +60,7 @@ def respawn(self) -> str: logger.warning(f"Failed to remove working directory: {e}") # clear files from the input directory - clearer = Clearer(input_dir) + clearer = Clearer([input_dir]) clearer.clear() # submit a new job diff --git a/src/qq_lib/run/runner.py b/src/qq_lib/run/runner.py index 4d71ac4..73b15c0 100644 --- a/src/qq_lib/run/runner.py +++ b/src/qq_lib/run/runner.py @@ -271,7 +271,7 @@ def finalize(self) -> None: logger.debug( f"Script exit code is '{self._process.returncode}'. Archiving files." ) - self._archiver.to_archive(self._work_dir) + self._archive_files_from_work_dir() # transfer files back to the input (submission) directory if self._use_scratch: @@ -690,7 +690,7 @@ def _resubmit(self) -> None: if self._informer.info.job_type == JobType.LOOP: if not (loop_info := self._informer.info.loop_info): - logger.warning( + raise QQError( "Loop info is undefined while resubmiting a loop job. This is a bug!" ) return @@ -707,6 +707,38 @@ def _resubmit(self) -> None: logger.info(f"Job resubmitted successfully as '{job_id}'.") + def _archive_files_from_work_dir(self) -> None: + """ + Archive files from the working directory. + + If no file exists for the next loop cycle, creates an empty init file to ensure the loop job continues normally. + """ + if not self._archiver: + raise QQError("Archiver is undefined while archiving files. This is a bug!") + + if not (loop_info := self._informer.info.loop_info): + raise QQError( + "Loop info is undefined while archiving files. This is a bug!" + ) + + # get the files to archive corresponding to the next loop job cycle + if not self._archiver.get_files_matching_pattern( + self._work_dir, + None, + loop_info.archive_format, + loop_info.current + 1, + False, + ): + # if there are no files matching the next loop job cycle, create an empty .init file + # so that the loop job continues normally + logger.debug( + f"Creating .init file for loop job cycle {loop_info.current + 1}." + ) + self._archiver.create_init_file(loop_info.current + 1) + + # archive all files matching the archive format + self._archiver.to_archive(self._work_dir) + def _get_explicitly_included_files_in_work_dir(self) -> list[Path]: """ Return absolute paths to files and directories in the working directory diff --git a/src/qq_lib/submit/cli.py b/src/qq_lib/submit/cli.py index dc07d27..955edf0 100644 --- a/src/qq_lib/submit/cli.py +++ b/src/qq_lib/submit/cli.py @@ -1,10 +1,10 @@ # Released under MIT License. # Copyright (c) 2025-2026 Ladislav Bartos and Robert Vacha Lab - import sys +from concurrent.futures import ThreadPoolExecutor from pathlib import Path -from typing import NoReturn +from typing import Any, NoReturn import click from click.shell_completion import CompletionItem @@ -33,8 +33,8 @@ def complete_script( Return completion items for script files and directories matching the incomplete string. If incomplete is a directory or resolves to one, lists files and subdirectories - inside it. Otherwise, lists files and directories in the current directory - matching the prefix. + inside it. Otherwise, lists files and directories in the parent directory + matching the prefix, expanding any glob patterns in the parent path. """ incomplete_path = Path(incomplete) @@ -45,9 +45,20 @@ def complete_script( search_dir = incomplete_path.parent prefix = incomplete_path.name + try: + items = search_dir.iterdir() + except FileNotFoundError: + # parent contains a glob pattern - expand it and collect from all matches + items = ( + child + for expanded in Path(search_dir.parent).glob(search_dir.name) + if expanded.is_dir() + for child in expanded.iterdir() + ) + return [ CompletionItem(str(path), type="file") - for path in search_dir.iterdir() + for path in items if path.name.startswith(prefix) ] @@ -56,22 +67,26 @@ def complete_script( @click.command( short_help="Submit a job to the batch system.", help=f""" -Submit a qq job to a batch system from the command line. +Submit one or more qq jobs to the batch system from the command line. -{click.style("SCRIPT", fg="green")} Path to the script to submit. +{click.style("SCRIPT...", fg="green")} One or more paths to the scripts to submit. All the options can also be specified inside the submitted script itself using qq directives of this format: `# qq