cape.cfdx.archivist: File archiving and clean-up for cases

This module provides the class CaseArchivist, which conducts the operations of commands such as

pyfun --archive
pycart --clean
pyover --skeleton
class cape.cfdx.archivist.CaseArchivist(opts: ArchiveOpts, where: str | None = None, casename: str | None = None)

Bases: object

Class to archive a single CFD case

Call:
>>> a = CaseArchivist(opts, where=None, casename=None)
Inputs:
opts: ArchiveOpts

Case archiving options

where: {None} | str

Root of CFD case folder (default: CWD)

casename: {None} | str

Name of CFD case folder (default: last two levels of CWD)

abspath_archive(fname: str) → str

Return absolute path to a file within archive folder

Call:
>>> fabs = a.abspath_archive(fname)
Inputs:
a: CaseArchivist

Archive controller for one case

fname: str

Relative path to a file

Outputs:
fabs: str

Absolute path

Versions:

  • 2024-09-12 @ddalle: v1.0

abspath_local(fname: str) → str

Return absolute path to a file within local case folder

Call:
>>> fabs = a.abspath_local(fname)
Inputs:
a: CaseArchivist

Archive controller for one case

fname: str

Relative path to a file

Outputs:
fabs: str

Absolute path

Versions:

  • 2024-09-12 @ddalle: v1.0

archive(test: bool = False)

Run the --archive action

Call:
>>> a.archive(test=False)
Inputs:
a: CaseArchivist

Archive controller for one case

test: True | {False}

Option to log all actions but not actually copy/delete

Versions:

  • 2024-09-18 @ddalle: v1.0

archive_file(fname: str, parent: int = 0)

Copy a file to the archive

Call:
>>> a.archive_file(fname, parent=1)
Inputs:
a: CaseArchivist

Archive controller for one case

fname: str

Name of file to copy

parent: {0} | int

Additional depth of function name in log

Versions:

  • 2024-09-04 @ddalle: v1.0

  • 2024-09-15 @ddalle: v1.1; add mtime checks

archive_files(matchdict: dict, n: int)

Copy collection(s) of files from a single search

Call:
>>> a.archive_files(matchdict, n)
Inputs:
a: CaseArchivist

Archive controller for one case

matchdict: dict

List of files to archive for each regex group value

n: int

Number of files to keep for each list

Versions:

  • 2024-09-13 @ddalle: v1.0

archivedir
assert_archive()

Raise an exception if archive root does not exist

Call:
>>> a.assert_archive()
Inputs:
a: CaseArchivist

Archive controller for one case

Raises:

FileNotFoundError if a.archivedir does not exist

Versions:

  • 2024-09-04 @ddalle: v1.0

  • 2024-10-08 @ddalle: v1.1; call make_case_archivedir()

begin(safety: str = 'report', test: bool = False)

Initialize counters and lsits for new archiving action

Call:
>>> a.begin(safety="archive", test=False)
Inputs:
a: CaseArchivist

Archive controller for one case

safety: {"archive"} | str

Level of safety, determines which files to protect

test: True | {False}

If True, log which files would be deleted, but don’t actually delete, copy, or tar anything

Versions:

  • 2024-09-12 @ddalle: v1.0

casename
check_safety(filename: str) → bool

Check if it’s safe to delete a file using current settings

This function will check a._safety for the current safety level. It will also log a warning with the reason if it’s unsafe to delete.

Call:
>>> q = a.check_safety(filename)
Inputs:
a: CaseArchivist

Archive controller for one case

filename: str

Name of prospective file/folder to delete

Outputs:
q: bool

Whether it’s safe to delete a

Versions:

  • 2024-09-13 @ddalle: v1.0

clean(test: bool = False)

Run the --clean action

Call:
>>> a.clean(test=False)
Inputs:
a: CaseArchivist

Archive controller for one case

test: True | {False}

Option to log all actions but not actually copy/delete

Versions:

  • 2024-09-15 @ddalle: v1.0

delete_dir(filename: str)

Delete a single folder if allowed; log results

Call:
>>> a.delete_file(filename)
Inputs:
a: CaseArchivist

Archive controller for one case

filename: str

Name of file to delete

Versions:

  • 2024-09-13 @ddalle: v1.0

delete_dirs(matchdict: dict, n: int)

Delete collection(s) of folders from a single search

Call:
>>> a.delete_dirs(matchdict, n)
Inputs:
a: CaseArchivist

Archive controller for one case

matchdict: dict

List of dirs to delete for each regex group value

n: int

Number of files to keep for each list

Versions:

  • 2024-09-13 @ddalle: v1.0

delete_file(filename: str)

Delete a single file if allowed; log results

Call:
>>> a.delete_file(filename)
Inputs:
a: CaseArchivist

Archive controller for one case

filename: str

Name of file to delete

Versions:

  • 2024-09-13 @ddalle: v1.0

  • 2024-10-08 @ddalle: v1.1; rm ‘filename’ to STDOUT

delete_files(matchdict: dict, n: int)

Delete collection(s) of files from a single search

Call:
>>> a.delete_files(matchdict, n)
Inputs:
a: CaseArchivist

Archive controller for one case

matchdict: dict

List of files to delete for each regex group value

n: int

Number of files to keep for each list

Versions:

  • 2024-09-13 @ddalle: v1.0

find_keepfiles(searchopt: dict) → list

Generate list of files to keep based on a search option

Call:
>>> a.find_keepfiles(searchopt)
Inputs:
a: CaseArchivist

Archive controller for one case

searchopt: dict

Key is the regular expression (or glob), value is the number of files to protect that match the pattern

Outputs:
mtches: list[str]

List of files [and folders] to protect

Versions:

  • 2024-09-13 @ddalle: v1.0

get_archivedir() → str

Get path to case’s archive folder

Call:
>>> dirname = a.get_archivedir()
Inputs:
a: CaseArchivist

Archive controller for one case

Outputs:
dirname: str

Absolute path to archive folder for this case

Versions:

  • 2024-09-16 @ddalle: v1.0

get_funcname(frame: int = 1) → str

Get name of calling function, mostly for log messages

Call:
>>> funcname = runner.get_funcname(frame=1)
Inputs:
runner: CaseRunner

Controller to run one case of solver

frame: {1} | int

Depth of function to seek title of

Outputs:
funcname: str

Name of calling function

Versions:

  • 2024-08-16 @ddalle

get_logger() → ArchivistLogger

Get or create logger instance

Call:
>>> logger = runner.get_logger()
Inputs:
runner: CaseRunner

Controller to run one case of solver

Outputs:
logger: ArchivistLogger

Logger instance

Versions:

  • 2024-08-29 @ddalle: v1.0

get_tail_lines(filename: str, vdef: int = 10) → int

Get number of lines to keep for a PostTailFile

Call:
>>> nl = a.get_tail_lines(filename, vdef=10)
Inputs:
a: CaseArchivist

Archive controller for one case

filename: str

Name of file to tail

vdef: {10} | int

Default value for output

Outputs:
nl: int

Number of lines to keep for filename

Versions:

  • 2024-09-18 @ddalle: v1.0

get_tarfile(filename: str) → str

Add .tar, .tar.gz, .zip, etc. as needed to file

Call:
>>> ftar = a.get_tarfile(filename)
Inputs:
a: CaseArchivist

Archive controller for one case

filename: str

Name of file or folder

Outputs:
ftar: str

Name of file or folder with archive extension added

Versions:

  • 2024-09-14 @ddalle: v1.0

keep_file(filename: str)

Keep a file and log the action

Call:
>>> a.keep_file(filename)
Inputs:
a: CaseArchivist

Archive controller for one case

filename: str

Name of file to protect

Versions:

  • 2024-09-13 @ddalle: v1.0

log(msg: str, title: str | None = None, parent: int = 0)

Write a message to primary log

Call:
>>> a.log(msg, title, parent=0)
Inputs:
a: CaseArchivist

Archive controller for one case

msg: str

Primary content of message

title: {None} | str

Manual title (default is name of calling function)

parent: {0} | int

Extra levels to use for calling function name

Versions:

  • 2024-08-29 @ddalle: v1.0

logger
make_case_archivedir()

Create the case archive folder if needed

Call:
>>> a.make_case_archivedir()
Inputs:
a: CaseArchivist

Archive controller for one case

Versions:

  • 2024-09-04 @ddalle: v1.0

make_case_subdir(fdir: str)

Create a subfolder in the archive folder if needed

Call:
>>> a.make_case_subdir(fdir)
Inputs:
a: CaseArchivist

Archive controller for one case

fdir: str

Name of subfolder (can be compound)

Versions:

  • 2024-09-04 @ddalle: v1.0

opts
remove_local(fname: str, parent: int = 1)

Delete a local file

Call:
>>> a.remove_local(fname, parent=1)
Inputs:
a: CaseArchivist

Archive controller for one case

fname: str

Name of file to delete

parent: {0} | int

Additional depth of function name in log

Versions:

  • 2024-09-12 @ddalle: v1.0

rmtree_local(fdir: str, parent: int = 1)

Delete a local folder (recursively)

Call:
>>> a.rmtree_local(fdir, parent=1)
Inputs:
a: CaseArchivist

Archive controller for one case

fdir: str

Name of folder to delete

parent: {0} | int

Additional depth of function name in log

Versions:

  • 2024-09-12 @ddalle: v1.0

root_dir
save_reportfiles(searchopt: list | dict)

Save list of files to protect for "report" option

The idea is to generate and save a list of files that presently appear to be required in order to restart the case without unarchiving.

Call:
>>> a.save_reportfiles(searchopt)
>>> a.save_reportfiles(filelist)
Inputs:
a: CaseArchivist

Archive controller for one case

searchopt: dict

Key is the regular expression (or glob), value is the number of files to protect that match the pattern

filelist: list

Raw list of files to save

Versions:

  • 2024-09-13 @ddalle: v1.0

  • 2024-12-12 @ddalle: v1.1; add filelist

save_restartfiles(searchopt: list | dict)

Save list of files to protect for "restart" option

The idea is to generate and save a list of files that presently appear to be required in order to restart the case without unarchiving.

Call:
>>> a.save_restartfiles(searchopt)
>>> a.save_restartfiles(filelist)
Inputs:
a: CaseArchivist

Archive controller for one case

searchopt: dict

Key is the regular expression (or glob), value is the number of files to protect that match the pattern

filelist: list

Raw list of files to save

Versions:

  • 2024-09-13 @ddalle: v1.0

  • 2024-12-12 @ddalle: v1.1; add filelist

search(pat: str, links: bool = False) → dict

Search case folder for files matching a given pattern

Call:
>>> matchdict = a.search(pat, links=False)
Inputs:
a: CaseArchivist

Archive controller for one case

pat: str

Regular expression pattern

links: True | {False}

Option to allow links in output

Outputs:
matchdict: dict[list]

Mapping of files matching pat keyed by identifier for the groups in pat

matchdict[lbl]: list[str]

List of files matching pat with group values in lbl, sorted by ascending modification time

Versions:

  • 2024-09-11 @ddalle: v1.0

  • 2025-04-04 @ddalle: v1.1; add links

search_glob(pat: str, links: bool = False) → dict

Search case folder for files, using glob SearchMethod

Call:
>>> matchdict = a.search_glob(pat, links=False)
Inputs:
a: CaseArchivist

Archive controller for one case

pat: str

Regular expression pattern

links: True | {False}

Option to allow links in output

Outputs:
matchdict: dict[list]

Mapping of files matching pat keyed by identifier for the groups in pat

matchdict[lbl]: list[str]

List of files matching pat with group values in lbl, sorted by ascending modification time

Versions:

  • 2024-09-25 @ddalle: v1.0

search_regex(pat: str, links: bool = False) → dict

Search case folder for files, using regex SearchMethod

Call:
>>> matchdict = a.search_regex(pat)
Inputs:
a: CaseArchivist

Archive controller for one case

pat: str

Regular expression pattern

links: True | {False}

Option to allow links in output

Outputs:
matchdict: dict[list]

Mapping of files matching pat keyed by identifier for the groups in pat

matchdict[lbl]: list[str]

List of files matching pat with group values in lbl, sorted by ascending modification time

Versions:

  • 2024-09-25 @ddalle: v1.0

skeleton(test: bool = False)

Run the --skeleton action

Call:
>>> a.skeleton(test=False)
Inputs:
a: CaseArchivist

Archive controller for one case

test: True | {False}

Option to log all actions but not actually copy/delete

Versions:

  • 2024-09-18 @ddalle: v1.0

tail_file(filename: str)

Replace a tail with the contents of it’s last n lines

Call:
>>> a.tail_file(filename)
Inputs:
a: CaseArchivist

Archive controller for one case

filename: str

Name of file to tail

Versions:

  • 2024-09-18 @ddalle: v1.0

tail_files(matchdict: dict, n: int)

Replace collection(s) of files with their tails

Call:
>>> a.tail_files(matchdict, n)
Inputs:
a: CaseArchivist

Archive controller for one case

matchdict: dict

List of files to delete for each regex group value

n: int

Number of files to keep for each list

Versions:

  • 2024-09-18 @ddalle: v1.0

tar_archive(tarname: str, searchopt: dict)

Archive one or more files in a tarball/zip

Call:
>>> a.tar_archive(tarname, searchopt)
Inputs:
a: CaseArchivist

Archive controller for one case

tarname: str

Name of tar/zip file to create, without file extension

searchopt: dict

Patterns of file names to include in archive, key is regex/glob and value is n, number of files to include

Versions:

  • 2024-09-15 @ddalle: v1.0

tar_dirs_archive(matchdict: dict, n: int)

Tar individual folder(s) to archive

Call:
>>> a.tar_dirs_archive(matchdict, n)
Inputs:
a: CaseArchivist

Archive controller for one case

matchdict: dict

List of dirs to delete for each regex group value

n: int

Number of files to keep for each list

Versions:

  • 2024-09-17 @ddalle: v1.0

tar_dirs_local(matchdict: dict, n: int)

Tar individual folder(s) in case folder

Call:
>>> a.tar_dirs_local(matchdict, n)
Inputs:
a: CaseArchivist

Archive controller for one case

matchdict: dict

List of dirs to delete for each regex group value

n: int

Number of files to keep for each list

Versions:

  • 2024-09-17 @ddalle: v1.0

tar_local(tarname: str, searchopt: dict)

Archive one or more files in local case folder

Call:
>>> a.tar_local(tarname, searchopt)
Inputs:
a: CaseArchivist

Archive controller for one case

tarname: str

Name of tar/zip file to create, without file extension

searchopt: dict

Patterns of file names to include in archive, key is regex/glob and value is n, number of files to include

Versions:

  • 2024-09-15 @ddalle: v1.0

unarchive(test: bool = False)

Run the --unarchive action

Call:
>>> a.unarchive(test=False)
Inputs:
a: CaseArchivist

Archive controller for one case

test: True | {False}

Option to log all actions but not actually copy/delete

Versions:

  • 2024-09-20 @ddalle: v1.0

unarchive_file(fname: str, parent: int = 0)

Retrieve a file from the archive

Call:
>>> a.unarchive_file(fname, parent=1)
Inputs:
a: CaseArchivist

Archive controller for one case

fname: str

Name of file to copy

parent: {0} | int

Additional depth of function name in log

Versions:

  • 2024-09-20 @ddalle: v1.0

unarchive_partial()

Copy or untar archive files, as appropriate

Call:
>>> a.unarchive_partial()
Inputs:
a: CaseArchivist

Archive controller for one case

Versions:

  • 2024-09-20 @ddalle: v1.0

unarchive_whole()

Untar single-tarball archive

Call:
>>> a.unarchive_whole()
Inputs:
a: CaseArchivist

Archive controller for one case

Versions:

  • 2024-09-20 @ddalle: v1.0

untar_archive(tarname: str)

Untar a file from archive

Call:
>>> a.untar_archive(tarname)
Inputs:
a: CaseArchivist

Archive controller for one case

tarname: str

Name of tar/zip file to create, without file extension

Versions:

  • 2024-09-21 @ddalle: v1.0

warn(msg: str, title: str | None = None, parent: int = 0)

Write a message to verbose log

Call:
>>> runner.warn(title, msg)
Inputs:
runner: CaseRunner

Controller to run one case of solver

msg: str

Primary content of message

title: {None} | str

Manual title (default is name of calling function)

parent: {0} | int

Extra levels to use for calling function name

Versions:

  • 2024-08-29 @ddalle: v1.0

cape.cfdx.archivist.getmtime(*filenames) → float

Get latest modification time of a file or folder

Call:
>>> t = _getmtime(*filenames)
Inputs:
filenames: tuple[str]

Name of one or more files or folders

Outputs:
total_size: int

Size of file or folder in bytes (0 if no such file)

Versions:

  • 2024-09-14 @ddalle: v1.0

cape.cfdx.archivist.getsize(file_or_folder: str) → int

Get size of file or folder, like du -sh

Call:
>>> total_size = getsize(file_or_folder)
Inputs:
file_or_folder: str

Name of file or folder

Outputs:
total_size: int

Size of file or folder in bytes (0 if no such file)

Versions:

  • 2024-09-12 @ddalle: v1.0

cape.cfdx.archivist.match_pats(name: str, pats: list) → bool

Match a single file name against a list of regular expressions

Call:
>>> q = match_pats(name, pats)
Inputs:
name: str

Name of file or other string to test

pats: list[str | re.Pattern]

List of patterns or compiled regexs

Outputs:
q: bool

Whether name matches any pattern in pats

Versions:

  • 2024-09-13 @ddalle: v1.0

cape.cfdx.archivist.reglob(pat: str) → list

Search for file and folder names matching regular expression

This function is constructed as a regular-expression version of glob.glob(), but it does not work with absolute paths.

Call:
>>> matchlist = reglob(pat)
Inputs:
pat: str

Regular expression pattern

Outputs:
matchlist: list[str]

Files and folders matching regular expression pat relative to current working directory

Versions:

  • 2024-09-02 @ddalle: v1.0

cape.cfdx.archivist.rematch(pat: str) → dict

Search for file and folder names matching regular expression

If the regex contains groups (parentheses), the results are grouped by the values of those groups. If the regex does not contain groups, the results are all in a single group called ''.

Call:
>>> matchdict = rematch(pat)
Inputs:
pat: str

Regular expression pattern

Outputs:
matchdict: dict[list]

Mapping of files matching pat keyed by identifier for the groups in pat

matchdict[lbl]: list[str]

List of files matching pat with group values identified in lbl

Versions:

  • 2024-09-02 @ddalle: v1.0

cape.cfdx.archivist.sort_by_mtime(file_list: list) → list

Sort a list of files by modification time

Call:
>>> sorted_list = sort_by_mtime(file_list)
Inputs:
file_list: list[str]

Original list of files

Outputs:
sorted_list: list[str]

List of files sorted by ascending modification time

Versions:

  • 2025-01-23 @ddalle: v1.0

cape.cfdx.archivist.use_glob(func)

Decorator to run a function "glob" for SearchMethod

Call:
>>> func = use_glob(func)
Wrapper Signature:
>>> v = a.func(*args, **kw)
Inputs:
func: func

Name of function

a: CaseArchivist

Archiving instance

args: tuple

Positional args to a.func()

kw: dict

Keyword args to a.func()

Versions:

  • 2024-09-25 @ddalle: v1.0

cape.cfdx.archivist.use_regex(func)

Decorator to run a function "regex" for SearchMethod

Call:
>>> func = use_regex(func)
Wrapper Signature:
>>> v = a.func(*args, **kw)
Inputs:
func: func

Name of function

a: CaseArchivist

Archiving instance

args: tuple

Positional args to a.func()

kw: dict

Keyword args to a.func()

Versions:

  • 2024-09-25 @ddalle: v1.0