llnl.util package

Subpackages

llnl.util.tty package

Submodules

llnl.util.argparsewriter module

class llnl.util.argparsewriter.ArgparseRstWriter(prog: str, out: ~typing.IO = <_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, aliases: bool = False, rst_levels: ~typing.Sequence[str] = ['=', '-', '^', '~', ':', '`'])[source]

Bases: ArgparseWriter

Write argparse output as rst sections.

begin_command(prog: str) → str[source]

Text to print before a command.

Parameters:: prog – Program name.
Returns:: Text before a command.

begin_optionals() → str[source]

Text to print before optional arguments.

Returns:: Optional arguments header.

begin_positionals() → str[source]

Text to print before positional arguments.

Returns:: Positional arguments header.

begin_subcommands(subcommands: List[Tuple[ArgumentParser, str, str]]) → str[source]

Table with links to other subcommands.

Parameters:: subcommands – List of subcommands.
Returns:: Subcommand linking text.

description(description: str) → str[source]

Description of a command.

Parameters:: description – Command description.
Returns:: Description of a command.

end_optionals() → str[source]

Text to print after optional arguments.

Returns:: Optional arguments footer.

end_positionals() → str[source]

Text to print after positional arguments.

Returns:: Positional arguments footer.

format(cmd: Command) → str[source]

Return the string representation of a single node in the parser tree.

Parameters:: cmd – Parsed information about a command or subcommand.
Returns:: String representation of a node.

optional(opts: str, help: str) → str[source]

Description of an optional argument.

Parameters:

opts – Optional argument.
help – Help text.

Returns:

Optional argument description.

positional(name: str, help: str) → str[source]

Description of a positional argument.

Parameters:

name – Argument name.
help – Help text.

Returns:

Positional argument description.

usage(usage: str) → str[source]

Example usage of a command.

Parameters:: usage – Command usage.
Returns:: Usage of a command.

class llnl.util.argparsewriter.ArgparseWriter(prog: str, out: ~typing.IO = <_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, aliases: bool = False)[source]

Bases: HelpFormatter, ABC

Analyze an argparse ArgumentParser for easy generation of help.

abstractmethod format(cmd: Command) → str[source]

Return the string representation of a single node in the parser tree.

Override this in subclasses to define how each subcommand should be displayed.

Parameters:: cmd – Parsed information about a command or subcommand.
Returns:: String representation of this subcommand.

parse(parser: ArgumentParser, prog: str) → Command[source]

Parse the parser object and return the relavent components.

Parameters:

parser – Command parser.
prog – Program name.

Returns:

Information about the command from the parser.

write(parser: ArgumentParser) → None[source]

Write out details about an ArgumentParser.

Parameters:: parser – Command parser.

class llnl.util.argparsewriter.Command(prog: str, description: str | None, usage: str, positionals: List[Tuple[str, Iterable[Any] | None, int | str | None, str]], optionals: List[Tuple[Sequence[str], List[str], str, int | str | None, str]], subcommands: List[Tuple[ArgumentParser, str, str]])[source]

Bases: object

Parsed representation of a command from argparse.

This is a single command from an argparse parser. ArgparseWriter creates these and returns them from parse(), and it passes one of these to each call to format() so that we can take an action for a single command.

llnl.util.filesystem module

class llnl.util.filesystem.BaseDirectoryVisitor[source]

Bases: object

Base class and interface for visit_directory_tree().

after_visit_dir(root: str, rel_path: str, depth: int) → None[source]

Called after recursion into rel_path finished. This function is not called when rel_path was not recursed into.

Parameters:

root – root directory
rel_path – relative path to current directory from root
depth – depth of current directory from the root directory

after_visit_symlinked_dir(root: str, rel_path: str, depth: int) → None[source]

Called after recursion into rel_path finished. This function is not called when rel_path was not recursed into.

Parameters:

root – root directory
rel_path – relative path to current symlink from root
depth – depth of current symlink from the root directory

before_visit_dir(root: str, rel_path: str, depth: int) → bool[source]

Return True from this function to recurse into the directory at os.path.join(root, rel_path). Return False in order not to recurse further.

Parameters:

root – root directory
rel_path – relative path to current directory from root
depth – depth of current directory from the root directory

Returns:

True when the directory should be recursed into. False when not

Return type:

bool

before_visit_symlinked_dir(root: str, rel_path: str, depth: int) → bool[source]

Return True to recurse into the symlinked directory and False in order not to. Note: rel_path is the path to the symlink itself. Following symlinked directories blindly can cause infinite recursion due to cycles.

Parameters:

root – root directory
rel_path – relative path to current symlink from root
depth – depth of current symlink from the root directory

Returns:

True when the directory should be recursed into. False when not

Return type:

bool

visit_file(root: str, rel_path: str, depth: int) → None[source]

Handle the non-symlink file at os.path.join(root, rel_path)

Parameters:

root – root directory
rel_path – relative path to current file from root
depth (int) – depth of current file from the root directory

visit_symlinked_file(root: str, rel_path: str, depth) → None[source]

Handle the symlink to a file at os.path.join(root, rel_path). Note: rel_path is the location of the symlink, not to what it is pointing to. The symlink may be dangling.

Parameters:

root – root directory
rel_path – relative path to current symlink from root
depth – depth of current symlink from the root directory

class llnl.util.filesystem.FileFilter(*filenames)[source]

Bases: object

Convenience class for calling filter_file a lot.

filter(regex: str, repl: str | Callable[[Match], str], string: bool = False, backup: bool = False, ignore_absent: bool = False, start_at: str | None = None, stop_at: str | None = None) → None[source]

class llnl.util.filesystem.FileList(files)[source]

Bases: Sequence

Sequence of absolute paths to files.

Provides a few convenience methods to manipulate file paths.

property basenames

Stable de-duplication of the base-names in the list

>>> l = LibraryList(['/dir1/liba.a', '/dir2/libb.a', '/dir3/liba.a'])
>>> l.basenames
['liba.a', 'libb.a']
>>> h = HeaderList(['/dir1/a.h', '/dir2/b.h', '/dir3/a.h'])
>>> h.basenames
['a.h', 'b.h']

Returns:: A list of base-names
Return type:: list

property directories

Stable de-duplication of the directories where the files reside.

>>> l = LibraryList(['/dir1/liba.a', '/dir2/libb.a', '/dir1/libc.a'])
>>> l.directories
['/dir1', '/dir2']
>>> h = HeaderList(['/dir1/a.h', '/dir1/b.h', '/dir2/c.h'])
>>> h.directories
['/dir1', '/dir2']

Returns:: A list of directories
Return type:: list

joined(separator=' ')[source]

class llnl.util.filesystem.HeaderList(files)[source]

Bases: FileList

Sequence of absolute paths to headers.

Provides a few convenience methods to manipulate header paths and get commonly used compiler flags or names.

add_macro(macro)[source]

Add a macro definition

Parameters:: macro (str) – The macro to add

property cpp_flags

Include flags + macro definitions

>>> h = HeaderList(['/dir1/a.h', '/dir1/b.h', '/dir2/c.h'])
>>> h.cpp_flags
'-I/dir1 -I/dir2'
>>> h.add_macro('-DBOOST_DYN_LINK')
>>> h.cpp_flags
'-I/dir1 -I/dir2 -DBOOST_DYN_LINK'

Returns:: A joined list of include flags and macro definitions
Return type:: str

property directories: Directories to be searched for header files.

property headers

Stable de-duplication of the headers.

Returns:: A list of header files
Return type:: list

property include_flags

Include flags

>>> h = HeaderList(['/dir1/a.h', '/dir1/b.h', '/dir2/c.h'])
>>> h.include_flags
'-I/dir1 -I/dir2'

Returns:: A joined list of include flags
Return type:: str

include_regex = re.compile('(.*?)(\\binclude\\b)(.*)')

property macro_definitions

Macro definitions

>>> h = HeaderList(['/dir1/a.h', '/dir1/b.h', '/dir2/c.h'])
>>> h.add_macro('-DBOOST_LIB_NAME=boost_regex')
>>> h.add_macro('-DBOOST_DYN_LINK')
>>> h.macro_definitions
'-DBOOST_LIB_NAME=boost_regex -DBOOST_DYN_LINK'

Returns:: A joined list of macro definitions
Return type:: str

property names

Stable de-duplication of header names in the list without extensions

>>> h = HeaderList(['/dir1/a.h', '/dir2/b.h', '/dir3/a.h'])
>>> h.names
['a', 'b']

Returns:: A list of files without extensions
Return type:: list

class llnl.util.filesystem.LibraryList(files)[source]

Bases: FileList

Sequence of absolute paths to libraries

Provides a few convenience methods to manipulate library paths and get commonly used compiler flags or names

property ld_flags

Search flags + link flags

>>> l = LibraryList(['/dir1/liba.a', '/dir2/libb.a', '/dir1/liba.so'])
>>> l.ld_flags
'-L/dir1 -L/dir2 -la -lb'

Returns:: A joined list of search flags and link flags
Return type:: str

property libraries

Stable de-duplication of library files.

Returns:: A list of library files
Return type:: list

property link_flags

Link flags for the libraries

>>> l = LibraryList(['/dir1/liba.a', '/dir2/libb.a', '/dir1/liba.so'])
>>> l.link_flags
'-la -lb'

Returns:: A joined list of link flags
Return type:: str

property names

Stable de-duplication of library names in the list

>>> l = LibraryList(['/dir1/liba.a', '/dir2/libb.a', '/dir3/liba.so'])
>>> l.names
['a', 'b']

Returns:: A list of library names
Return type:: list

property search_flags

Search flags for the libraries

>>> l = LibraryList(['/dir1/liba.a', '/dir2/libb.a', '/dir1/liba.so'])
>>> l.search_flags
'-L/dir1 -L/dir2'

Returns:: A joined list of search flags
Return type:: str

llnl.util.filesystem.ancestor(dir, n=1)[source]: Get the nth ancestor of a directory.

llnl.util.filesystem.can_access(file_name)[source]: True if we have read/write access to the file.

llnl.util.filesystem.change_sed_delimiter(old_delim, new_delim, *filenames)[source]

Find all sed search/replace commands and change the delimiter.

e.g., if the file contains seds that look like 's///', you can call change_sed_delimiter('/', '@', file) to change the delimiter to '@'.

Note that this routine will fail if the delimiter is ' or ". Handling those is left for future work.

Parameters:

old_delim (str) – The delimiter to search for
new_delim (str) – The delimiter to replace with
*filenames – One or more files to search and replace

llnl.util.filesystem.chgrp(path, group, follow_symlinks=True)[source]: Implement the bash chgrp function on a single path

llnl.util.filesystem.chmod_x(entry, perms)[source]: Implements chmod, treating all executable bits as set using the chmod utility’s +X option.

llnl.util.filesystem.copy(src, dest, _permissions=False)[source]

Copy the file(s) src to the file or directory dest.

If dest specifies a directory, the file will be copied into dest using the base filename from src.

src may contain glob characters.

Parameters:

src (str) – the file(s) to copy
dest (str) – the destination file or directory
_permissions (bool) – for internal use only

Raises:

OSError – if src does not match any files or directories
ValueError – if src matches multiple files but dest is not a directory

llnl.util.filesystem.copy_mode(src, dest)[source]: Set the mode of dest to that of src unless it is a link.

llnl.util.filesystem.copy_tree(src: str, dest: str, symlinks: bool = True, ignore: Callable[[str], bool] | None = None, _permissions: bool = False)[source]

Recursively copy an entire directory tree rooted at src.

If the destination directory dest does not already exist, it will be created as well as missing parent directories.

src may contain glob characters.

If symlinks is true, symbolic links in the source tree are represented as symbolic links in the new tree and the metadata of the original links will be copied as far as the platform allows; if false, the contents and metadata of the linked files are copied to the new tree.

If ignore is set, then each path relative to src will be passed to this function; the function returns whether that path should be skipped.

Parameters:

src (str) – the directory to copy
dest (str) – the destination directory
symlinks (bool) – whether or not to preserve symlinks
ignore (Callable) – function indicating which files to ignore
_permissions (bool) – for internal use only

Raises:

OSError – if src does not match any files or directories
ValueError – if src is a parent directory of dest

llnl.util.filesystem.filter_file(regex: str, repl: str | Callable[[Match], str], *filenames: str, string: bool = False, backup: bool = False, ignore_absent: bool = False, start_at: str | None = None, stop_at: str | None = None, encoding: str | None = 'utf-8') → None[source]

Like sed, but uses python regular expressions.

Filters every line of each file through regex and replaces the file with a filtered version. Preserves mode of filtered files.

As with re.sub, repl can be either a string or a callable. If it is a callable, it is passed the match object and should return a suitable replacement string. If it is a string, it can contain \1, \2, etc. to represent back-substitution as sed would allow.

Parameters:

regex – The regular expression to search for
repl – The string to replace matches with
*filenames – One or more files to search and replace string: Treat regex as a plain string. Default it False backup: Make backup file(s) suffixed with ~. Default is False
ignore_absent – Ignore any files that don’t exist. Default is False
start_at – Marker used to start applying the replacements. If a text line matches this marker filtering is started at the next line. All contents before the marker and the marker itself are copied verbatim. Default is to start filtering from the first line of the file.
stop_at – Marker used to stop scanning the file further. If a text line matches this marker filtering is stopped and the rest of the file is copied verbatim. Default is to filter until the end of the file.
encoding – The encoding to use when reading and writing the files. Default is None, which uses the system’s default encoding.

Finds all files matching the patterns from files starting from root. This function returns a deterministic result for the same input and directory structure when run multiple times. Symlinked directories are followed, and unique directories are searched only once. Each matching file is returned only once at lowest depth in case multiple paths exist due to symlinked directories.

Accepts any glob characters accepted by fnmatch:

Pattern	Meaning
`*`	matches one or more characters
`?`	matches any single character
`[seq]`	matches any character in `seq`
`[!seq]`	matches any character not in `seq`

Examples:

>>> find("/usr", "*.txt", recursive=True, max_depth=2)

finds all files with the extension .txt in the directory /usr and subdirectories up to depth 2.

>>> find(["/usr", "/var"], ["*.txt", "*.log"], recursive=True)

finds all files with the extension .txt or .log in the directories /usr and /var at any depth.

>>> find("/usr", "GL/*.h", recursive=True)

finds all header files in a directory GL at any depth in the directory /usr.

Parameters:

root – One or more root directories to start searching from
files – One or more filename patterns to search for
recursive – if False search only root, if True descends from roots. Defaults to True.
max_depth – if set, don’t search below this depth. Cannot be set if recursive is False

Returns a list of absolute, matching file paths.

llnl.util.filesystem.find_all_headers(root)[source]

Convenience function that returns the list of all headers found in the directory passed as argument.

Parameters:: root (str) – directory where to look recursively for header files
Returns:: List of all headers found in root and subdirectories.

llnl.util.filesystem.find_first(root: str, files: Iterable[str] | str, bfs_depth: int = 2) → str | None[source]

Find the first file matching a pattern.

The following

$ find /usr -name 'abc*' -o -name 'def*' -quit

is equivalent to:

>>> find_first("/usr", ["abc*", "def*"])

Any glob pattern supported by fnmatch can be used.

The search order of this method is breadth-first over directories, until depth bfs_depth, after which depth-first search is used.

Parameters:

root (str) – The root directory to start searching from
files (str or Iterable) – File pattern(s) to search for
bfs_depth (int) – (advanced) parameter that specifies at which depth to switch to depth-first search.

Returns:

The matching file or None when no file is found.

Return type:

str or None

llnl.util.filesystem.find_headers(headers, root, recursive=False)[source]

Returns an iterable object containing a list of full paths to headers if found.

Accepts any glob characters accepted by fnmatch:

Pattern	Meaning
	matches everything
?	matches any single character
[seq]	matches any character in `seq`
[!seq]	matches any character not in `seq`

Parameters:

headers (str or list) – Header name(s) to search for
root (str) – The root directory to start searching from
recursive (bool) – if False search only root folder, if True descends top-down from the root. Defaults to False.

Returns:

The headers that have been found

Return type:

HeaderList

llnl.util.filesystem.find_libraries(libraries, root, shared=True, recursive=False, runtime=True, max_depth: int | None = None)[source]

Returns an iterable of full paths to libraries found in a root dir.

Accepts any glob characters accepted by fnmatch:

Pattern	Meaning
	matches everything
?	matches any single character
[seq]	matches any character in `seq`
[!seq]	matches any character not in `seq`

Parameters:

libraries (str or list) – Library name(s) to search for
root (str) – The root directory to start searching from
shared (bool) – if True searches for shared libraries, otherwise for static. Defaults to True.
recursive (bool) – if False search only root folder, if True descends top-down from the root. Defaults to False.
max_depth (int) – if set, don’t search below this depth. Cannot be set if recursive is False
runtime (bool) – Windows only option, no-op elsewhere. If true, search for runtime shared libs (.DLL), otherwise, search for .Lib files. If shared is false, this has no meaning. Defaults to True.

Returns:

The libraries that have been found

Return type:

LibraryList

llnl.util.filesystem.find_system_libraries(libraries, shared=True)[source]

Searches the usual system library locations for libraries.

Search order is as follows:

/lib64
/lib
/usr/lib64
/usr/lib
/usr/local/lib64
/usr/local/lib

Accepts any glob characters accepted by fnmatch:

Pattern	Meaning
	matches everything
?	matches any single character
[seq]	matches any character in `seq`
[!seq]	matches any character not in `seq`

Parameters:

libraries (str or list) – Library name(s) to search for
shared (bool) – if True searches for shared libraries, otherwise for static. Defaults to True.

Returns:

The libraries that have been found

Return type:

LibraryList

llnl.util.filesystem.force_remove(*paths)[source]: Remove files without printing errors. Like rm -f, does NOT remove directories.

llnl.util.filesystem.force_symlink(src, dest)[source]

llnl.util.filesystem.getuid() → str | int[source]: Returns os getuid on non Windows On Windows returns 0 for admin users, login string otherwise This is in line with behavior from get_owner_uid which always returns the login string on Windows

llnl.util.filesystem.install(src, dest)[source]

Install the file(s) src to the file or directory dest.

Same as copy() with the addition of setting proper permissions on the installed file.

Parameters:

src (str) – the file(s) to install
dest (str) – the destination file or directory

Raises:

OSError – if src does not match any files or directories
ValueError – if src matches multiple files but dest is not a directory

llnl.util.filesystem.install_tree(src, dest, symlinks=True, ignore=None)[source]

Recursively install an entire directory tree rooted at src.

Same as copy_tree() with the addition of setting proper permissions on the installed files and directories.

Parameters:

src (str) – the directory to install
dest (str) – the destination directory
symlinks (bool) – whether or not to preserve symlinks
ignore (Callable) – function indicating which files to ignore

Raises:

OSError – if src does not match any files or directories
ValueError – if src is a parent directory of dest

llnl.util.filesystem.is_exe(path)[source]: True if path is an executable file.

llnl.util.filesystem.join_path(prefix, *args)[source]

llnl.util.filesystem.keep_modification_time(*filenames)[source]

Context manager to keep the modification timestamps of the input files. Tolerates and has no effect on non-existent files and files that are deleted by the nested code.

Parameters:: *filenames – one or more files that must have their modification timestamps unchanged

llnl.util.filesystem.library_extensions = ['a', 'la', 'so', 'tbd', 'dylib']: This generates the library filenames that may appear on any OS.

llnl.util.filesystem.mkdirp(*paths: str, mode: int | None = None, group: str | int | None = None, default_perms: str | None = None)[source]

Creates a directory, as well as parent directories if needed.

Parameters:

paths – paths to create with mkdirp
mode – optional permissions to set on the created directory – use OS default if not provided
group – optional group for permissions of final created directory – use OS default if not provided. Only used if world write permissions are not set
default_perms – one of ‘parents’ or ‘args’. The default permissions that are set for directories that are not themselves an argument for mkdirp. ‘parents’ means intermediate directories get the permissions of their direct parent directory, ‘args’ means intermediate get the same permissions specified in the arguments to mkdirp – default value is ‘args’

llnl.util.filesystem.partition_path(path, entry=None)[source]

Split the prefixes of the path at the first occurrence of entry and return a 3-tuple containing a list of the prefixes before the entry, a string of the prefix ending with the entry, and a list of the prefixes after the entry.

If the entry is not a node in the path, the result will be the prefix list followed by an empty string and an empty list.

llnl.util.filesystem.prefixes(path)[source]

Returns a list containing the path and its ancestors, top-to-bottom.

The list for an absolute path will not include an os.sep entry. For example, assuming os.sep is /, given path /ab/cd/efg the resulting paths will be, in order: /ab, /ab/cd, and /ab/cd/efg

The list for a relative path starting ./ will not include .. For example, path ./hi/jkl/mn results in a list with the following paths, in order: ./hi, ./hi/jkl, and ./hi/jkl/mn.

On Windows, paths will be normalized to use / and / will always be used as the separator instead of os.sep.

Parameters:: path (str) – the string used to derive ancestor paths
Returns:: A list containing ancestor paths in order and ending with the path

llnl.util.filesystem.remove_dead_links(root)[source]

Recursively removes any dead link that is present in root.

Parameters:: root (str) – path where to search for dead links

llnl.util.filesystem.remove_directory_contents(dir)[source]: Remove all contents of a directory.

llnl.util.filesystem.remove_if_dead_link(path)[source]

Removes the argument if it is a dead link.

Parameters:: path (str) – The potential dead link

llnl.util.filesystem.remove_linked_tree(path)[source]

Removes a directory and its contents.

If the directory is a symlink, follows the link and removes the real directory before removing the link.

This method will force-delete files on Windows

Parameters:: path (str) – Directory to be removed

llnl.util.filesystem.rename(src, dst)[source]

llnl.util.filesystem.set_executable(path)[source]

llnl.util.filesystem.set_install_permissions(path)[source]: Set appropriate permissions on the installed file.

llnl.util.filesystem.touch(path)[source]: Creates an empty file at the specified path.

llnl.util.filesystem.touchp(path)[source]: Like touch, but creates any parent directories needed for the file.

llnl.util.filesystem.traverse_tree(source_root: str, dest_root: str, rel_path: str = '', *, order: str = 'pre', ignore: Callable[[str], bool] | None = None, follow_nonexisting: bool = True, follow_links: bool = False)[source]

Traverse two filesystem trees simultaneously.

Walks the LinkTree directory in pre or post order. Yields each file in the source directory with a matching path from the dest directory, along with whether the file is a directory. e.g., for this tree:

root/
  a/
    file1
    file2
  b/
    file3

When called on dest, this yields:

('root',         'dest')
('root/a',       'dest/a')
('root/a/file1', 'dest/a/file1')
('root/a/file2', 'dest/a/file2')
('root/b',       'dest/b')
('root/b/file3', 'dest/b/file3')

Keyword Arguments:

order (str) – Whether to do pre- or post-order traversal. Accepted values are ‘pre’ and ‘post’
ignore (Callable) – function indicating which files to ignore. This will also ignore symlinks if they point to an ignored file (regardless of whether the symlink is explicitly ignored); note this only supports one layer of indirection (i.e. if you have x -> y -> z, and z is ignored but x/y are not, then y would be ignored but not x). To avoid this, make sure the ignore function also ignores the symlink paths too.
follow_nonexisting (bool) – Whether to descend into directories in src that do not exit in dest. Default is True
follow_links (bool) – Whether to descend into symlinks in src

llnl.util.filesystem.unset_executable_mode(path)[source]

llnl.util.filesystem.visit_directory_tree(root: str, visitor: BaseDirectoryVisitor, rel_path: str = '', depth: int = 0)[source]

Recurses the directory root depth-first through a visitor pattern using the interface from BaseDirectoryVisitor

Parameters:

root – path of directory to recurse into
visitor – what visitor to use
rel_path – current relative path from the root
depth – current depth from the root

llnl.util.filesystem.working_dir(dirname: str, *, create: bool = False)[source]

llnl.util.lang module

llnl.util.lang.ClassProperty

A type alias that represents either a classproperty descriptor or a constant value of the same type. This allows derived classes to override a computed class-level property with a constant value while retaining type compatibility.

alias of ClassPropertyType | classproperty[ClassPropertyType]

class llnl.util.lang.Const(value)[source]

Bases: object

Class level constant, raises when trying to set the attribute

value

class llnl.util.lang.DeprecatedProperty(name: str)[source]

Bases: object

Data descriptor to error or warn when a deprecated property is accessed.

Derived classes must define a factory method to return an adaptor for the deprecated property, if the descriptor is not set to error.

error_lvl = 0: 0 - Nothing 1 - Warning 2 - Error

factory(instance, owner)[source]

name

class llnl.util.lang.GroupedExceptionForwarder(context: str, handler: GroupedExceptionHandler, base: type)[source]

Bases: object

A contextmanager to capture exceptions and forward them to a GroupedExceptionHandler.

class llnl.util.lang.GroupedExceptionHandler[source]

Bases: object

A generic mechanism to coalesce multiple exceptions and preserve tracebacks.

forward(context: str, base: type = <class 'BaseException'>) → GroupedExceptionForwarder[source]: Return a contextmanager which extracts tracebacks and prefixes a message.

grouped_message(with_tracebacks: bool = True) → str[source]: Print out an error message coalescing all the forwarded errors.

class llnl.util.lang.HashableMap[source]

Bases: MutableMapping

This is a hashable, comparable dictionary. Hash is performed on a tuple of the values in the dictionary.

copy()[source]: Type-agnostic clone method. Preserves subclass type.

dict

class llnl.util.lang.ObjectWrapper(wrapped_object)[source]

Bases: object

Base class that wraps an object. Derived classes can add new behavior while staying undercover.

This class is modeled after the stackoverflow answer: * http://stackoverflow.com/a/1445289/771663

class llnl.util.lang.PriorityOrderedMapping[source]

Bases: Mapping[KT, VT]

Mapping that iterates over key according to an integer priority. If the priority is the same for two keys, insertion order is what matters.

The priority is set when the key/value pair is added. If not set, the highest current priority is used.

add(key: KT, *, value: VT, priority: int | None = None) → None[source]

Adds a key/value pair to the mapping, with a specific priority.

If the priority is None, then it is assumed to be the highest priority value currently in the container.

Raises:: ValueError – when the same priority is already in the mapping

remove(key: KT) → VT[source]

Removes a key from the mapping.

Returns:: The value associated with the key being removed
Raises:: KeyError – if the key is not in the mapping

reversed_keys()[source]: Iterates over keys from the highest priority, to the lowest.

reversed_values()[source]: Iterates over values from the highest priority, to the lowest.

class llnl.util.lang.Singleton(factory: Callable[[], object])[source]

Bases: object

Wrapper for lazily initialized singleton objects.

property instance

class llnl.util.lang.TypedMutableSequence(iterable)[source]

Bases: MutableSequence

Base class that behaves like a list, just with a different type.

Client code can inherit from this base class:

class Foo(TypedMutableSequence):
pass

and later perform checks based on types:

if isinstance(l, Foo):
# do something

insert(index, item)[source]: S.insert(index, value) – insert value before index

exception llnl.util.lang.UnhashableArguments[source]

Bases: TypeError

Raise when an @memoized function receives unhashable arg or kwarg values.

llnl.util.lang.attr_setdefault(obj, name, value)[source]: Like dict.setdefault, but for objects.

class llnl.util.lang.classproperty(callback: Callable[[Any], ClassPropertyType])[source]

Bases: Generic[ClassPropertyType]

Non-data descriptor to evaluate a class-level property. The function that performs the evaluation is injected at creation time and takes an owner (i.e., the class that originated the instance).

llnl.util.lang.decorator_with_or_without_args(decorator)[source]

Allows a decorator to be used with or without arguments, e.g.:

# Calls the decorator function some args
@decorator(with, arguments, and=kwargs)

or:

# Calls the decorator function with zero arguments
@decorator

llnl.util.lang.dedupe(sequence, key=None)[source]

Yields a stable de-duplication of an hashable sequence by key

Parameters:

sequence – hashable sequence to be de-duplicated
key – callable applied on values before uniqueness test; identity by default.

Returns:

stable de-duplication of the sequence

Examples

Dedupe a list of integers:

[x for x in dedupe([1, 2, 1, 3, 2])] == [1, 2, 3]

[x for x in llnl.util.lang.dedupe([1,-2,1,3,2], key=abs)] == [1, -2, 3]

llnl.util.lang.done = <object object>: sentinel for testing that iterators are done in lazy_lexicographic_ordering

llnl.util.lang.elide_list(line_list: List[str], max_num: int = 10) → List[str][source]

Takes a long list and limits it to a smaller number of elements, replacing intervening elements with ‘…’. For example:

elide_list(["1", "2", "3", "4", "5", "6"], 4)

gives:

["1", "2", "3", "...", "6"]

llnl.util.lang.ensure_last(lst, *elements)[source]: Performs a stable partition of lst, ensuring that elements occur at the end of lst in specified order. Mutates lst. Raises ValueError if any elements are not already in lst.

llnl.util.lang.fnmatch_translate_multiple(named_patterns: Dict[str, str]) → str[source]: Similar to fnmatch.translate, but takes an ordered dictionary where keys are pattern names, and values are filename patterns. The output is a regex that matches any of the patterns in order, and named capture groups are used to identify which pattern matched.

llnl.util.lang.get_entry_points(*, group: str)[source]

Wrapper for importlib.metadata.entry_points

Parameters:: group – entry points to select
Returns:: EntryPoints for group or empty list if unsupported

llnl.util.lang.index_by(objects, *funcs)[source]

Create a hierarchy of dictionaries by splitting the supplied set of objects on unique values of the supplied functions.

Values are used as keys. For example, suppose you have four objects with attributes that look like this:

a = Spec("boost %gcc target=skylake")
b = Spec("mrnet %intel target=zen2")
c = Spec("libelf %xlc target=skylake")
d = Spec("libdwarf %intel target=zen2")

list_of_specs = [a,b,c,d]
index1 = index_by(list_of_specs, lambda s: str(s.target),
                  lambda s: s.compiler)
index2 = index_by(list_of_specs, lambda s: s.compiler)

index1 now has two levels of dicts, with lists at the leaves, like this:

{ 'zen2'    : { 'gcc' : [a], 'xlc' : [c] },
  'skylake' : { 'intel' : [b, d] }
}

And index2 is a single level dictionary of lists that looks like this:

{ 'gcc'    : [a],
  'intel'  : [b,d],
  'xlc'    : [c]
}

If any elements in funcs is a string, it is treated as the name of an attribute, and acts like getattr(object, name). So shorthand for the above two indexes would be:

index1 = index_by(list_of_specs, 'arch', 'compiler')
index2 = index_by(list_of_specs, 'compiler')

You can also index by tuples by passing tuples:

index1 = index_by(list_of_specs, ('target', 'compiler'))

Keys in the resulting dict will look like (‘gcc’, ‘skylake’).

llnl.util.lang.key_ordering(cls)[source]

Decorates a class with extra methods that implement rich comparison operations and __hash__. The decorator assumes that the class implements a function called _cmp_key(). The rich comparison operations will compare objects using this key, and the __hash__ function will return the hash of this key.

If a class already has __eq__, __ne__, __lt__, __le__, __gt__, or __ge__ defined, this decorator will overwrite them.

Raises:: TypeError – If the class does not have a _cmp_key method

llnl.util.lang.lazy_eq(lseq, rseq)[source]

Equality comparison for two lazily generated sequences.

See lazy_lexicographic_ordering.

llnl.util.lang.lazy_lexicographic_ordering(cls, set_hash=True)[source]

Decorates a class with extra methods that implement rich comparison.

This is a lazy version of the tuple comparison used frequently to implement comparison in Python. Given some objects with fields, you might use tuple keys to implement comparison, e.g.:

class Widget:
    def _cmp_key(self):
        return (
            self.a,
            self.b,
            (self.c, self.d),
            self.e
        )

    def __eq__(self, other):
        return self._cmp_key() == other._cmp_key()

    def __lt__(self):
        return self._cmp_key() < other._cmp_key()

    # etc.

Python would compare Widgets lexicographically based on their tuples. The issue there for simple comparators is that we have to bulid the tuples and we have to generate all the values in them up front. When implementing comparisons for large data structures, this can be costly.

Lazy lexicographic comparison maps the tuple comparison shown above to generator functions. Instead of comparing based on pre-constructed tuple keys, users of this decorator can compare using elements from a generator. So, you’d write:

@lazy_lexicographic_ordering
class Widget:
    def _cmp_iter(self):
        yield a
        yield b
        def cd_fun():
            yield c
            yield d
        yield cd_fun
        yield e

    # operators are added by decorator

There are no tuples preconstructed, and the generator does not have to complete. Instead of tuples, we simply make functions that lazily yield what would’ve been in the tuple. The @lazy_lexicographic_ordering decorator handles the details of implementing comparison operators, and the Widget implementor only has to worry about writing _cmp_iter, and making sure the elements in it are also comparable.

Some things to note:

If a class already has __eq__, __ne__, __lt__, __le__, __gt__, __ge__, or __hash__ defined, this decorator will overwrite them.

If set_hash is False, this will not overwrite __hash__.

This class uses Python 2 None-comparison semantics. If you yield None and it is compared to a non-None type, None will always be less than the other object.

Raises:: TypeError – If the class does not have a _cmp_iter method

llnl.util.lang.lazy_lt(lseq, rseq)[source]

Less-than comparison for two lazily generated sequences.

See lazy_lexicographic_ordering.

llnl.util.lang.list_modules(directory, **kwargs)[source]: Lists all of the modules, excluding __init__.py, in a particular directory. Listed packages have no particular order.

llnl.util.lang.load_module_from_file(module_name, module_path)[source]

Loads a python module from the path of the corresponding file.

If the module is already in sys.modules it will be returned as is and not reloaded.

Parameters:

module_name (str) – namespace where the python module will be loaded, e.g. foo.bar
module_path (str) – path of the python file containing the module

Returns:

A valid module object

Raises:

ImportError – when the module can’t be loaded
FileNotFoundError – when module_path doesn’t exist

llnl.util.lang.match_predicate(*args)[source]

Utility function for making string matching predicates.

Each arg can be a: * regex * list or tuple of regexes * predicate that takes a string.

This returns a predicate that is true if: * any arg regex matches * any regex in a list or tuple of regexes matches. * any predicate in args matches.

llnl.util.lang.memoized(func)[source]: Decorator that caches the results of a function, storing them in an attribute of that function.

llnl.util.lang.nullcontext(*args, **kwargs)[source]: Empty context manager. TODO: replace with contextlib.nullcontext() if we ever require python 3.7.

llnl.util.lang.pretty_date(time, now=None)[source]

Convert a datetime or timestamp to a pretty, relative date.

Parameters:

time (datetime.datetime or int) – date to print prettily
now (datetime.datetime) – datetime for ‘now’, i.e. the date the pretty date is relative to (default is datetime.now())

Returns:

pretty string like ‘an hour ago’, ‘Yesterday’,: ’3 months ago’, ‘just now’, etc.

Return type:

(str)

Adapted from https://stackoverflow.com/questions/1551382.

llnl.util.lang.pretty_seconds(seconds)[source]

Seconds to string with appropriate units

Parameters:: seconds (float) – Number of seconds
Returns:: Time string with units
Return type:: str

llnl.util.lang.pretty_seconds_formatter(seconds)[source]

llnl.util.lang.pretty_string_to_date(date_str, now=None)[source]

Parses a string representing a date and returns a datetime object.

Parameters:: date_str (str) – string representing a date. This string might be in different format (like YYYY, YYYY-MM, YYYY-MM-DD, YYYY-MM-DD HH:MM, YYYY-MM-DD HH:MM:SS) or be a pretty date (like yesterday or two months ago)
Returns:: datetime object corresponding to date_str
Return type:: (datetime.datetime)

llnl.util.lang.stable_args(*args, **kwargs)[source]: A key factory that performs a stable sort of the parameters.

llnl.util.lang.stable_partition(input_iterable: Iterable[T], predicate_fn: Callable[[T], bool]) → Tuple[List[T], List[T]][source]

Partition the input iterable according to a custom predicate.

Parameters:

input_iterable – input iterable to be partitioned.
predicate_fn – predicate function accepting an iterable item as argument.

Returns:

Tuple of the list of elements evaluating to True, and list of elements evaluating to False.

llnl.util.lang.tuplify(seq)[source]: Helper for lazy_lexicographic_ordering().

llnl.util.lang.union_dicts(*dicts)[source]

Use update() to combine all dicts into one.

This builds a new dictionary, into which we update() each element of dicts in order. Items from later dictionaries will override items from earlier dictionaries.

Parameters:: dicts (list) – list of dictionaries

Return: (dict): a merged dictionary containing combined keys and: values from dicts.

llnl.util.lang.uniq(sequence)[source]

Remove strings of duplicate elements from a list.

This works like the command-line uniq tool. It filters strings of duplicate elements in a list. Adjacent matching elements are merged into the first occurrence.

For example:

uniq([1, 1, 1, 1, 2, 2, 2, 3, 3]) == [1, 2, 3]
uniq([1, 1, 1, 1, 2, 2, 2, 1, 1]) == [1, 2, 1]

llnl.util.link_tree module

LinkTree class for setting up trees of symbolic links.

class llnl.util.link_tree.LinkTree(source_root)[source]

Bases: object

Class to create trees of symbolic links from a source directory.

LinkTree objects are constructed with a source root. Their methods allow you to create and delete trees of symbolic links back to the source tree in specific destination directories. Trees comprise symlinks only to files; directries are never symlinked to, to prevent the source directory from ever being modified.

find_conflict(dest_root, ignore=None, ignore_file_conflicts=False)[source]: Returns the first file in dest that conflicts with src

find_dir_conflicts(dest_root, ignore)[source]

get_file_map(dest_root, ignore)[source]

merge(dest_root, ignore_conflicts=False, ignore=None, link=<built-in function symlink>, relative=False)[source]

Link all files in src into dest, creating directories: if necessary.

Keyword Args:

ignore_conflicts (bool): if True, do not break when the target exists;: return a list of files that could not be linked
ignore (callable): callable that returns True if a file is to be: ignored in the merge (by default ignore nothing)

link (callable): function to create links with (defaults to llnl.util.symlink)

relative (bool): create all symlinks relative to the target: (default False)

merge_directories(dest_root, ignore)[source]

unmerge(dest_root, ignore=None, remove_file=<function remove_link>)[source]

Unlink all files in dest that exist in src.

Unlinks directories in dest if they are empty.

unmerge_directories(dest_root, ignore)[source]

llnl.util.lock module

exception llnl.util.lock.CantCreateLockError(path)[source]

Bases: LockPermissionError

Attempt to create a lock in an unwritable location.

class llnl.util.lock.Lock(path: str, *, start: int = 0, length: int = 0, default_timeout: float | None = None, debug: bool = False, desc: str = '')[source]

Bases: object

This is an implementation of a filesystem lock using Python’s lockf.

In Python, lockf actually calls fcntl, so this should work with any filesystem implementation that supports locking through the fcntl calls. This includes distributed filesystems like Lustre (when flock is enabled) and recent NFS versions.

Note that this is for managing contention over resources between processes and not for managing contention between threads in a process: the functions of this object are not thread-safe. A process also must not maintain multiple locks on the same file (or, more specifically, on overlapping byte ranges in the same file).

acquire_read(timeout: float | None = None) → bool[source]

Acquires a recursive, shared lock for reading.

Read and write locks can be acquired and released in arbitrary order, but the POSIX lock is held until all local read and write locks are released.

Returns True if it is the first acquire and actually acquires the POSIX lock, False if it is a nested transaction.

acquire_write(timeout: float | None = None) → bool[source]

Acquires a recursive, exclusive lock for writing.

Read and write locks can be acquired and released in arbitrary order, but the POSIX lock is held until all local read and write locks are released.

Returns True if it is the first acquire and actually acquires the POSIX lock, False if it is a nested transaction.

cleanup() → None[source]

downgrade_write_to_read(timeout: float | None = None) → None[source]

Downgrade from an exclusive write lock to a shared read.

Raises:: LockDowngradeError – if this is an attempt at a nested transaction

is_write_locked() → bool[source]

Check if the file is write locked

Returns:: True if the path is write locked, otherwise, False
Return type:: (bool)

release_read(release_fn: Callable[[], bool] | None = None) → bool[source]

Releases a read lock.

Parameters:: release_fn (Callable) – function to call before the last recursive lock (read or write) is released.

If the last recursive lock will be released, then this will call release_fn and return its result (if provided), or return True (if release_fn was not provided).

Otherwise, we are still nested inside some other lock, so do not call the release_fn and, return False.

Does limited correctness checking: if a read lock is released when none are held, this will raise an assertion error.

release_write(release_fn: Callable[[], bool] | None = None) → bool[source]

Releases a write lock.

Parameters:: release_fn (Callable) – function to call before the last recursive write is released.

If the last recursive write lock will be released, then this will call release_fn and return its result (if provided), or return True (if release_fn was not provided). Otherwise, we are still nested inside some other write lock, so do not call the release_fn, and return False.

Does limited correctness checking: if a read lock is released when none are held, this will raise an assertion error.

upgrade_read_to_write(timeout: float | None = None) → None[source]

Attempts to upgrade from a shared read lock to an exclusive write.

Raises:: LockUpgradeError – if this is an attempt at a nested transaction

exception llnl.util.lock.LockDowngradeError(path)[source]

Bases: LockError

Raised when unable to downgrade from a write to a read lock.

exception llnl.util.lock.LockError[source]

Bases: Exception

Raised for any errors related to locks.

exception llnl.util.lock.LockPermissionError[source]

Bases: LockError

Raised when there are permission issues with a lock.

exception llnl.util.lock.LockROFileError(path)[source]

Bases: LockPermissionError

Tried to take an exclusive lock on a read-only file.

exception llnl.util.lock.LockTimeoutError(lock_type, path, time, attempts)[source]

Bases: LockError

Raised when an attempt to acquire a lock times out.

Bases: object

Simple nested transaction context manager that uses a file lock.

Parameters:

lock (Lock) – underlying lock for this transaction to be accquired on enter and released on exit
acquire (Callable or contextlib.contextmanager) – function to be called after lock is acquired, or contextmanager to enter after acquire and leave before release.
release (Callable) – function to be called before release. If acquire is a contextmanager, this will be called after exiting the nexted context and before the lock is released.
timeout (float) – number of seconds to set for the timeout when accquiring the lock (default no timeout)

If the acquire_fn returns a value, it is used as the return value for __enter__, allowing it to be passed as the as argument of a with statement.

If acquire_fn returns a context manager, its __enter__ function will be called after the lock is acquired, and its __exit__ funciton will be called before release_fn in __exit__, allowing you to nest a context manager inside this one.

Timeout for lock is customizable.

exception llnl.util.lock.LockUpgradeError(path)[source]

Bases: LockError

Raised when unable to upgrade from a read to a write lock.

Bases: LockTransaction

LockTransaction context manager that does a read and releases it.

Bases: LockTransaction

LockTransaction context manager that does a write and releases it.

llnl.util.multiproc module

This implements a parallel map operation but it can accept more values than multiprocessing.Pool.apply() can. For example, apply() will fail to pickle functions if they’re passed indirectly as parameters.

class llnl.util.multiproc.Barrier(n, timeout=None)[source]

Bases: object

Simple reusable semaphore barrier.

Python 2 doesn’t have multiprocessing barriers so we implement this.

See https://greenteapress.com/semaphores/LittleBookOfSemaphores.pdf, p. 41.

wait()[source]

llnl.util.symlink module

exception llnl.util.symlink.AlreadyExistsError[source]

Bases: SymlinkError

Link path already exists.

exception llnl.util.symlink.SymlinkError[source]

Bases: RuntimeError

Exception class for errors raised while creating symlinks, junctions and hard links

llnl.util.symlink.resolve_link_target_relative_to_the_link(link)[source]: os.path.isdir uses os.path.exists, which for links will check the existence of the link target. If the link target is relative to the link, we need to construct a pathname that is valid from our cwd (which may not be the same as the link’s directory)