gufe.utils

Functions

magic_open(path_or_stream)

Open a file path or stream as text, transparently decompressing when possible.

requires_package(package_name)

Helper function to denote that a function requires some optional dependency.

Classes

ensure_filelike(fn[, mode, force_close])

Context manager to convert pathlike or filelike to filelike.
gufe.utils.magic_open(path_or_stream: str | PathLike | BinaryIO | TextIO) → Iterator[TextIO]

Open a file path or stream as text, transparently decompressing when possible.

File paths and binary streams are inspected for compression using magic bytes rather than file extensions. gzip, bzip2, and lzma/xz are supported. Text streams are yielded unchanged and are assumed to already be decoded and, if applicable, already decompressed.

Parameters:

path_or_stream (str or os.PathLike or BinaryIO or TextIO input source to open as text.) –

  • If a file path is provided, the file is opened in binary mode and inspected for compression.

  • If a binary stream is provided, its leading bytes are inspected for compression.

  • If a text stream is provided, it is yielded unchanged without any compression detection.

Yields:

TextIO – A text-mode stream suitable for iteration and line-based reading.

  • For compressed inputs, this is a text wrapper around the appropriate decompressor.

  • For uncompressed binary inputs, this is an io.TextIOWrapper.

  • For text-mode inputs, this is the original stream.

Raises:

TypeError – Raised if a non-text stream is provided whose read() method does not return a bytes-like object.

Notes

Compression detection is only possible for file paths and binary streams. For text streams, decoding has already occurred, so the original byte signature is no longer available.

Seekable binary streams are rewound after header inspection so the full content remains available to the decompressor or text wrapper.

Non-seekable binary streams are fully buffered into an io.BytesIO object so that the header bytes consumed during detection are preserved.

Streams opened by this function are closed when the context manager exits. Caller-owned streams are never closed by this function.

Examples

Open a plain text file by path:

>>> with magic_open("data.txt") as f:
...     first_line = f.readline()

Open a compressed file by path:

>>> with magic_open("data.txt.gz") as f:
...     text = f.read()

Pass an already-open binary stream:

>>> with open("data.txt.gz", "rb") as raw:
...     with magic_open(raw) as f:
...         text = f.read()

Pass an already-open text stream:

>>> with open("data.txt", "rt") as f:
...     with magic_open(f) as text_stream:
...         first_line = text_stream.readline()
class gufe.utils.ensure_filelike(fn, mode=None, force_close=False)

Context manager to convert pathlike or filelike to filelike.

This makes it so that methods can allow a range of user inputs.

Parameters:

  • fn (PathLike or FileLike) – The user input to normalize.

  • mode (str or None) – The mode, if fn is pathlike. If fn is filelike, a warning will be emitted and the mode will be ignored.

  • force_close (bool, default False) – Whether to forcibly close the stream on exit. For pathlike inputs, the stream will always be closed. Filelike inputs will close if this parameter is True.

gufe.utils.requires_package(package_name: str) → Callable

Helper function to denote that a function requires some optional dependency. A function decorated with this decorator will raise MissingDependencyError if the package is not found by importlib.import_module().

Parameters:

package_name (str) – The directory path to enter within the context

Raises:

MissingDependencyError –