# Copyright (c) 2026, PyRETIS Development Team.
# Distributed under the LGPLv2.1+ License. See LICENSE for more info.
"""Removal of PyRETIS run artifacts (the ``pyretis clean`` command).
This module implements the cleanup that used to live in the per-example
``Makefile`` ``clean`` targets. It deletes the artifacts a PyRETIS run
leaves behind so an example (or any run directory) can be reset to its
committed inputs.
What is removed is the union of:
* a built-in set of always-generated PyRETIS artifacts (see
:data:`DEFAULT_FIND_FILES` and :data:`DEFAULT_FIND_DIRS`), and
* whatever an optional per-directory :data:`CONFIG_NAME` file adds.
The per-directory config (TOML) mirrors the variables the old shared
``common-clean.mk`` used::
[clean]
use_defaults = true # apply the built-in defaults (default: true)
find_files = ["*.tpr"] # extra file-name globs (recursive find -delete)
find_dirs = ["dump"] # extra directory names (recursive rm -rf)
rm_paths = ["lammps/system.data"] # root-relative rm -rf globs
keep = ["load/0"] # never delete these (or their ancestors)
``keep`` protects committed inputs that would otherwise match (for example
the committed ``load/0`` .. ``load/7`` initial paths of the validation
suite). Anything that is a kept path, lies inside a kept path, or is an
ancestor directory of a kept path is left untouched.
"""
import fnmatch
import glob
import os
import shutil
CONFIG_NAME = 'clean.toml'
# Built-in defaults, mirroring the always-applied set of the old
# ``common-clean.mk``: the artifacts every routed PyRETIS run can leave
# behind (byte-code caches, the translated infinite-swapping config, the
# coordinator restart files, the kick-init load directory and the
# per-worker run directories).
DEFAULT_FIND_FILES = ('*.pyc', '*.pyo', 'output.toml*', 'infswap.toml',
'restart.toml*')
DEFAULT_FIND_DIRS = ('__pycache__', 'load', 'worker[0-9]*')
# NB: the PyRETIS ensemble output directories (``000``, ``001``, ...) and
# committed ``out.toml`` references are NOT deleted by default. Several
# examples commit reference ``NNN`` / ``load`` / ``out.toml`` files as
# inputs, so removing those is left to each directory's ``clean.toml``
# (``rm_paths = ["0*"]``, ``find_files = ["out.toml*"]``, ... -- exactly as
# the old per-example Makefiles did), where it cannot touch a committed
# sibling by accident.
# ``find`` in common-clean.sh skipped anything under a ``results`` folder;
# keep that exclusion so curated reference results are never deleted.
EXCLUDED_DIR_NAMES = ('results',)
[docs]
def _read_toml(path):
"""Parse a TOML file, returning the top-level mapping.
Parameters
----------
path : string
Path to the TOML file.
Returns
-------
dict
The parsed configuration.
"""
try:
import tomllib as toml_reader
except ImportError:
import tomli as toml_reader
with open(path, 'rb') as handle:
return toml_reader.load(handle)
[docs]
def load_clean_config(directory):
"""Load the cleanup configuration for a directory.
Reads the built-in defaults and merges in the optional
:data:`CONFIG_NAME` file found in ``directory``.
Parameters
----------
directory : string
The directory to be cleaned.
Returns
-------
dict
A configuration with the keys ``find_files``, ``find_dirs``,
``rm_paths`` and ``keep`` (all lists of strings).
"""
find_files = []
find_dirs = []
rm_paths = []
keep = []
use_defaults = True
config_path = os.path.join(directory, CONFIG_NAME)
if os.path.isfile(config_path):
section = _read_toml(config_path).get('clean', {})
use_defaults = bool(section.get('use_defaults', True))
find_files.extend(section.get('find_files', []))
find_dirs.extend(section.get('find_dirs', []))
rm_paths.extend(section.get('rm_paths', []))
keep.extend(section.get('keep', []))
if use_defaults:
find_files = list(DEFAULT_FIND_FILES) + find_files
find_dirs = list(DEFAULT_FIND_DIRS) + find_dirs
return {
'find_files': find_files,
'find_dirs': find_dirs,
'rm_paths': rm_paths,
'keep': keep,
}
[docs]
def _normalise_keep(directory, keep):
"""Return the set of absolute, normalised kept paths.
Parameters
----------
directory : string
The root directory of the clean.
keep : list of str
The configured ``keep`` entries (root-relative, may be globs).
Returns
-------
set of str
Absolute normalised paths that must be protected.
"""
protected = set()
for pattern in keep:
matches = glob.glob(os.path.join(directory, pattern))
if matches:
for match in matches:
protected.add(os.path.abspath(match))
else:
# Keep the literal path too, so a keep entry protects a path
# even before it exists / even if it is not a glob.
protected.add(os.path.abspath(os.path.join(directory, pattern)))
return protected
[docs]
def _is_protected(path, protected):
"""Return True if ``path`` must not be deleted given ``protected``.
A path is protected when it equals a kept path, lies inside one, or
is an ancestor directory of one (so we never delete a parent that
still holds something we must keep).
Parameters
----------
path : string
The candidate path to delete.
protected : set of str
Absolute normalised kept paths.
Returns
-------
boolean
True when the candidate must be left untouched.
"""
candidate = os.path.abspath(path)
for kept in protected:
if candidate == kept:
return True
# candidate is inside a kept path.
if candidate.startswith(kept + os.sep):
return True
# candidate is an ancestor of a kept path.
if kept.startswith(candidate + os.sep):
return True
return False
[docs]
def _under_excluded_dir(path, root):
"""Return True if ``path`` lies under an excluded directory name.
Parameters
----------
path : string
The path to test.
root : string
The clean root (its own name is not considered).
Returns
-------
boolean
True when a path component (below ``root``) is excluded.
"""
relative = os.path.relpath(path, root)
parts = relative.split(os.sep)
# Drop the final component (the file/dir itself); only its parent
# directories decide the ``results`` exclusion, matching find's
# ``! -path '*/results/*'``.
return any(part in EXCLUDED_DIR_NAMES for part in parts[:-1])
[docs]
def _remove_file(path, dry_run, removed):
"""Delete a file (unless dry-run) and record it."""
removed.append(path)
if not dry_run:
os.remove(path)
[docs]
def _remove_tree(path, dry_run, removed):
"""Delete a file or directory tree (unless dry-run) and record it."""
removed.append(path)
if dry_run:
return
if os.path.isdir(path) and not os.path.islink(path):
shutil.rmtree(path)
else:
os.remove(path)
[docs]
def clean_directory(directory='.', dry_run=False):
"""Remove PyRETIS run artifacts from a directory.
Parameters
----------
directory : string, optional
The directory to clean. Defaults to the current directory.
dry_run : boolean, optional
When True, nothing is deleted; the would-be-removed paths are
only collected and returned.
Returns
-------
list of str
The paths that were removed (or, for a dry run, that would be
removed), sorted for a stable report.
"""
directory = os.path.abspath(directory)
if not os.path.isdir(directory):
raise ValueError(f'Clean directory "{directory}" does not exist.')
config = load_clean_config(directory)
protected = _normalise_keep(directory, config['keep'])
removed = []
_clean_find_files(directory, config['find_files'], protected, dry_run,
removed)
_clean_find_dirs(directory, config['find_dirs'], protected, dry_run,
removed)
_clean_rm_paths(directory, config['rm_paths'], protected, dry_run,
removed)
return sorted(set(removed))
[docs]
def _clean_find_files(directory, patterns, protected, dry_run, removed):
"""Delete files whose name matches one of ``patterns`` (recursive)."""
if not patterns:
return
for current, _, files in os.walk(directory):
for name in files:
path = os.path.join(current, name)
if _under_excluded_dir(path, directory):
continue
if _is_protected(path, protected):
continue
if any(fnmatch.fnmatch(name, pat) for pat in patterns):
_remove_file(path, dry_run, removed)
[docs]
def _clean_find_dirs(directory, patterns, protected, dry_run, removed):
"""Delete directories whose name matches one of ``patterns``.
The walk is top-down so a matched directory is removed before its
children are visited; ``dirs`` is pruned in place to avoid descending
into something already scheduled for removal.
"""
if not patterns:
return
for current, dirs, _ in os.walk(directory, topdown=True):
keep_dirs = []
for name in dirs:
path = os.path.join(current, name)
if _under_excluded_dir(path, directory):
keep_dirs.append(name)
continue
if not any(fnmatch.fnmatch(name, pat) for pat in patterns):
keep_dirs.append(name)
continue
if _is_protected(path, protected):
# The directory holds (or is) something to keep: descend
# into it so non-protected children can still be removed.
keep_dirs.append(name)
continue
_remove_tree(path, dry_run, removed)
dirs[:] = keep_dirs
[docs]
def _clean_rm_paths(directory, patterns, protected, dry_run, removed):
"""Delete root-relative paths/globs with ``rm -rf`` semantics."""
for pattern in patterns:
for match in glob.glob(os.path.join(directory, pattern)):
if _is_protected(match, protected):
continue
if not os.path.exists(match) and not os.path.islink(match):
continue
_remove_tree(match, dry_run, removed)