feat: add cleanup routines for files in working directory

CHANGELOG.md

@@ -19,6 +19,7 @@
 - Add the keyword `dumpactints` to `BlockOutput` (#245).
 - Added fallback keyword argument for `get_final_energy`, `get_gradient`, and `get_structure` that allows to parse these properties from the `.out` file if no JSON output is available (#237).
 - Added `get_frequencies`, `get_imaginary_frequencies`, `is_pes_minimum`, and `is_pes_transition_state` to the `Output` class (#247).
+- Add `functions to clean up files created by ORCA jobs.(#262)
 - `Output.parse()`, `Output.parse_property()`, `Output.parse_gbw()`, and `Output.__init__()` now accept a `strict` parameter (default True). When set to False, output fields that fail Pydantic validation are silently set to None and a UserWarning is emitted instead of raising a ValidationError (#248).
 - Added `rmsd` and `rmsd_kabsch` to calculate RMSD without and with rotational alignment to `Structure` class (#230). 
 - Added `calc_rotational_constants` to calculate molecular rotational constants to `Structure` class (#230).

src/opi/output/core.py

@@ -4,7 +4,7 @@
 """
 
 import json
-from collections.abc import Callable, Sequence
+from collections.abc import Callable, Iterable, Sequence
 from pathlib import Path
 from typing import Any, cast
 from warnings import warn
@@ -536,6 +536,67 @@ def get_file(self, suffix: str, /, *, basename: str = "") -> Path:
             basename = self.basename
         return self.working_dir / (basename + suffix)
 
+    def _delete_files(
+        self, basename: str | None = None, *, suffixes: Sequence[str] = tuple()
+    ) -> None:
+        """
+        Delete files in `working_dir` belonging to the job with the given basename.
+
+        Parameters
+        ----------
+        basename : str | None, default: None
+            Basename of the job whose files should be deleted.
+            If not given, `self.basename` is used.
+        suffixes : Sequence[str] | None, default: None
+            If given, treated as glob patterns matched against `{basename}*{suffix}`;
+            only matching files are deleted.
+            If None, all files matching the basename are deleted.
+        """
+        basename = basename if basename else self.basename
+        if not basename:
+            raise ValueError("No basename specified")
+        if "/" in basename or "\\" in basename:
+            raise ValueError(f"Basename must not contain path separators: {basename!r}")
+        files: Iterable[Path]
+        if not suffixes:
+            files = self.working_dir.glob(f"{basename}*")
+        else:
+            files = (
+                file
+                for suffix in suffixes
+                for file in self.working_dir.glob(f"{basename}*{suffix}")
+            )
+        for file in files:
+            if file.is_file():
+                file.unlink(missing_ok=True)
+
+    def cleanup_files(self, basename: str | None = None) -> None:
+        """
+        Delete all files in `working_dir` belonging to the job with the given basename. This will also include files whose
+        names contain the basename, for example, if the basename is `job` and there exists files with basename 'job_1',
+        the files will be deleted.
+
+        Parameters
+        ----------
+        basename : str | None, default: None
+            Basename of the job whose files should be deleted.
+            If not given, `self.basename` is used.
+        """
+        self._delete_files(basename)
+
+    def cleanup_temp_files(self, basename: str | None = None) -> None:
+        """
+        Delete temporary files in `working_dir` belonging to the job with the given basename.
+
+        Parameters
+        ----------
+        basename : str | None, default: None
+            Basename of the job whose temporary files should be deleted.
+            If not given, `self.basename` is used.
+        """
+        temp_file_suffixes: tuple[str, ...] = (".tmp", ".proc", ".tmp.*", ".proc.*")
+        self._delete_files(basename, suffixes=temp_file_suffixes)
+
     def _get_version(self) -> "OrcaVersion":
         """Gets the ORCA version from the property-JSON file."""
 

tests/unit/test_output_cleanup.py

@@ -0,0 +1,108 @@
+from pathlib import Path
+
+import pytest
+
+from opi.output.core import Output
+
+"""
+Unit tests for Output cleanup functions.
+
+This module contains tests for the file-deletion helpers:
+- `Output.cleanup_files()`
+- `Output.cleanup_temp_files()`
+- `Output._delete_files()`
+"""
+
+
+def _make_output(basename: str, working_dir: Path) -> Output:
+    output = Output(basename, working_dir=working_dir, version_check=False)
+    return output
+
+
+def _create_file(directory: Path, *names: str) -> None:
+    for name in names:
+        (directory / name).touch()
+
+
+@pytest.mark.unit
+@pytest.mark.output
+def test_cleanup_files_deletes_matching_basename_files(tmp_path: Path):
+    """`cleanup_files()` deletes all files matching the basename, including files whose
+    name only contains the basename as a prefix, and leaves unrelated files untouched."""
+    _create_file(tmp_path, "job.out", "job.gbw", "job_1.gbw", "unrelated.txt")
+    output = _make_output("job", tmp_path)
+
+    output.cleanup_files()
+
+    remaining = {p.name for p in tmp_path.iterdir()}
+    assert remaining == {"unrelated.txt"}
+
+
+@pytest.mark.unit
+@pytest.mark.output
+def test_cleanup_files_explicit_basename_overrides_self_basename(tmp_path: Path):
+    """`cleanup_files(basename=...)` deletes files for the given basename, not `self.basename`."""
+    _create_file(tmp_path, "job_1.out", "job_2.out")
+    output = _make_output("job_1", tmp_path)
+
+    output.cleanup_files(basename="job_2")
+
+    remaining = {p.name for p in tmp_path.iterdir()}
+    assert remaining == {"job_1.out"}
+
+
+@pytest.mark.unit
+@pytest.mark.output
+def test_cleanup_files_raises_without_basename(tmp_path: Path):
+    """`cleanup_files()` raises `ValueError` rather than deleting everything in `working_dir`."""
+    _create_file(tmp_path, "unrelated.txt")
+    output = _make_output("", tmp_path)
+
+    with pytest.raises(ValueError, match="No basename specified"):
+        output.cleanup_files()
+
+    assert (tmp_path / "unrelated.txt").exists()
+
+
+@pytest.mark.unit
+@pytest.mark.output
+@pytest.mark.parametrize("basename", ["../secret", "sub/job", "sub\\job"])
+def test_cleanup_files_rejects_path_separators_in_basename(tmp_path: Path, basename: str):
+    """A basename containing a path separator must be rejected rather than letting the
+    glob pattern escape `working_dir`."""
+    working_dir = tmp_path / "workdir"
+    working_dir.mkdir()
+    _create_file(tmp_path, "secret_file.txt")
+    output = _make_output(basename, working_dir)
+
+    with pytest.raises(ValueError, match="path separators"):
+        output.cleanup_files()
+
+    assert (tmp_path / "secret_file.txt").exists()
+
+
+@pytest.mark.unit
+@pytest.mark.output
+def test_cleanup_temp_files_deletes_only_temp_suffixes(tmp_path: Path):
+    """`cleanup_temp_files()` only deletes `.tmp.*`/`.proc.*` files, not other job files."""
+    _create_file(tmp_path, "job.out", "job.gbw", "job.tmp", "job.proc", "job_scan.tmp")
+    output = _make_output("job", tmp_path)
+
+    output.cleanup_temp_files()
+
+    remaining = {p.name for p in tmp_path.iterdir()}
+    assert remaining == {"job.out", "job.gbw"}
+
+
+@pytest.mark.unit
+@pytest.mark.output
+def test_delete_files_empty_suffixes_deletes_all(tmp_path: Path):
+    """An empty `suffixes` sequence behaves like no filter at all: all files matching
+    the basename are deleted, same as omitting `suffixes`."""
+    _create_file(tmp_path, "job.out", "job.gbw", "unrelated.txt")
+    output = _make_output("job", tmp_path)
+
+    output._delete_files(None, suffixes=())
+
+    remaining = {p.name for p in tmp_path.iterdir()}
+    assert remaining == {"unrelated.txt"}