Miscellaneous and Utility Code

gambit.util

Miscellaneous utility code.

gambit.util.misc

Utility code that doesn’t fit anywhere else.

gambit.util.misc.chunk_slices(n, size)

Iterate over slice objects which split a sequence of length n into chunks of size size.

Parameters:

n (int) – Length of sequence.
size (int) – Size of chunks (apart from last).

Return type:

Iterator[slice]

gambit.util.misc.join_list_human(strings, conj='and')

Join items into a single human-readable string with commas and the given conjunction.

Parameters:

strings (Iterable[str])
conj (str)

Return type:

str

gambit.util.misc.type_singledispatchmethod(func)

Similar to singledispatchmethod, but the first (non-self) argument is expected to be a type and dispatch occurs on the argument’s value.

Parameters:: func (Callable) – Default implementation. Signature should start with (self, cls: type, ...).
Returns:: Function with register and dispatch attributes similar to singledispatchmethod.
Return type:: Callable

gambit.util.misc.zip_strict(it1: Iterable[T], it2: Iterable[T2], /) → Iterator[tuple[T, T2]]

gambit.util.misc.zip_strict(it1: Iterable[T], it2: Iterable[T2], it3: Iterable[T3], /) → Iterator[tuple[T, T2, T3]]

gambit.util.misc.zip_strict(it1: Iterable[T], it2: Iterable[T2], it3: Iterable[T3], it4: Iterable[T4], /) → Iterator[tuple[T, T2, T3, T4]]

Like the builtin zip function but raises an error if any argument is exhausted before the others.

Parameters:: iterables – Any number of iterable objects.
Raises:: ValueError –

gambit.util.io

Utility code for reading/writing data files.

class gambit.util.io.FilePath: Alias for types which can represent a file system path (str or os.PathLike).

class gambit.util.io.ClosingIterator

Bases: Iterable[T]

Wraps an iterator which reads from a stream, closes the stream when finished.

Used to wrap return values from functions which do some sort of lazy IO operation (specifically Bio.SeqIO.parse()) and return an iterator which reads from a stream every time next() is called on it. The object is an iterator itself, but will close the stream automatically when it finishes. May also be used as a context manager which closes the stream on exit.

fobj: The underlying file-like object or stream which the instance is responsible for closing

iterator: The iterator which the instance wraps.

closed: Read-only boolean property, mirrors the same attribute of fobj.

Parameters:

iterable – Iterable to iterate over. The iterator attribute will be obtained from calling iter() on this.
fobj – File-like object to close when the iterator finishes, context is exited or the close() method is called.

__init__(iterable, fobj)

Parameters:: iterable (Iterable[T])

close()

Close the stream.

Just calls the close method on fobj.

gambit.util.io.guess_compression(fobj)

Guess the compression mode of an readable file-like object in binary mode.

Assumes the current position is at the beginning of the file.

Parameters:: fobj (BinaryIO)
Return type:: str

gambit.util.io.maybe_open(file_or_path, mode='r', **open_kw)

Open a file given a file path as an argument, but pass existing file objects though.

Intended to be used by API functions that take either type as an argument. If a file path is passed the function will need to call open to get the file object to use, and will need to close that object after it is done. If an existing file object is passed, it should be left to the caller of the function to close it afterwards. This function returns a context manager which performs the correct action for both opening and closing.

Parameters:

file_or_path (FilePath | IO) – A path-like object or open file object.
mode (str) – Mode to open file in.
**open_kw – Keyword arguments to open().

Returns:

Context manager which gives an open file object on enter and closes it on exit only if it was opened by this function.

Return type:

ContextManager[IO]

gambit.util.io.open_compressed(path, mode='rt', compression='auto', **kwargs)

Open a file with compression method specified by a string.

Parameters:

path (FilePath) – Path of file to open. May be string or path-like object.
mode (str) – Mode to open file in - similar to open(). Must be exactly two characters, the first in rwax and the second in``tb``.
compression (str) – Compression method. Allowed values are 'none', 'gzip', or 'auto'.
**kwargs – Additional text-specific keyword arguments identical to the following open() arguments: encoding, errors, and newlines.

Returns:

Open file object.

Return type:

IO

gambit.util.io.read_lines(file_or_path, strip=True, skip_empty=False)

Iterate over lines in text file.

Parameters:

file_or_path (FilePath | TextIO) – A path-like object or open file object.
strip (bool) – Strip whitespace from lines.
skip_empty (bool) – Omit empty lines.

Returns:

Iterator over lines with trailing newlines removed.

Return type:

Iterable[str]

gambit.util.io.write_lines(lines, file_or_path)

Write strings to text file, one per line.

Parameters:

lines (Iterable) – Iterable of lines to write.
file_or_path (FilePath | TextIO) – A path-like object or open file object.

gambit.util.json

Convert data to/from JSON format.

The dump() and load() functions in this module work just like their built-in equivalents in the json module, but support additional types such as attrs-defined classes.

class gambit.util.json.Jsonable

Bases: object

Mixin class that provides custom JSON conversion methods.

Either of the special methods __to_json__ and __from_json__ may be set to None to indicate that the default conversion process should be used.

abstract __to_json__(): Convert the instance to JSON-writable data (anything that can be passed to json.dump()).

abstract classmethod __from_json__(data): Create an instance from parsed JSON data.

gambit.util.json.dump(obj, f, **kw)

Write the JSON representation of an object to a file.

Parameters:

obj – Object to write.
f (TextIO) – Writeable file object in text mode.
**kw – Keyword arguments to json.dump().

gambit.util.json.dumps(obj, **kw)

Get the JSON representation of an object as a string.

Parameters:

obj – Object to write.
**kw – Keyword arguments to json.dumps().

Return type:

str

gambit.util.json.from_json(data, cls=typing.Any)

Load object from parsed JSON data.

Parameters:

data – Data parsed from JSON format.
cls – Type to load.

Return type:

Instance of cls

gambit.util.json.load(f, cls=typing.Any)

Load an object from a JSON file.

Parameters:

f (TextIO) – Readable file object in text mode.
cls – Type to load.

Return type:

Instance of cls

gambit.util.json.loads(s, cls=typing.Any)

Load an object from a JSON string.

Parameters:

s (str) – String containing JSON-encoded data.
cls – Type to load.

Return type:

Instance of cls

gambit.util.json.register_hooks(cls, unstructure, structure, withtype=False): Register converter unstructure and structure hooks in the same call.

gambit.util.json.register_structure_hook_notype(cls, f): Register converter structure hook taking no 2nd type argument.

gambit.util.json.to_json(obj)

Convert object to JSON-writable data (anything that can be passed to json.dump()).

Parameters:: obj – Object to convert.

gambit.util.indexing

Tools to implement and test Numpy-style advanced indexing for Sequence types.

class gambit.util.indexing.AdvancedIndexingMixin

Bases: object

Mixin class for sequence types which enables Numpy-style advanced indexing.

Implements a __getitem__ which takes care of identifying the type of index, performing bounds checking, and converting negative indices.

The following methods must be implemented by subtypes:

_getitem_int()
_getitem_int_array()

The following methods may optionally be overridden, but default to calling _getitem_int_array():

_getitem_slice()
_getitem_bool_array()

_check_index(i)

Check integer index is in bounds, converting negative indices to positive.

Parameters:: i (int)
Return type:: int

_getitem_bool_array(index)

Get subsequence given a boolean array.

Parameters:: index (Sequence[bool]) – Boolean Numpy array of same length as sequence.

_getitem_int(i)

Get single value given valid integer index.

Parameters:: i (int) – Positive in-bounds integer index.

_getitem_int_array(index)

Get subsequence given an array of integer indices.

Parameters:: index (Sequence[int]) – Array of integer indices, all positive and in-bounds.

_getitem_slice(index)

Get subsequence by indexing with a slice.

Parameters:

slice – Slice object. Guaranteed to be valid (start, stop, and step are integers or None and step size is non-zero).
index (slice)

gambit.util.progress

Abstract interface for progress meters.

class gambit.util.progress.AbstractProgressMeter

Bases: ABC

Abstract base class for an object which displays progress to the user.

Instances can be used as context managers, on exit the close() method is called.

n

Number of completed iterations. Do not modify directly, use the increment() and moveto() methods instead.

Type:: int

total

Expected total number of iterations.

Type:: int

closed

Whether the meter has been closed/completed.

Type:: int

close(): Stop displaying progress and perform whatever cleanup is necessary.

classmethod config(**kw)

Create a factory function which creates instances with the given default settings.

Keyword arguments are passed on to create().

Return type:: ProgressConfig

abstract classmethod create(total, *, initial=0, desc=None, file=None, **kw)

Factory function with standardized signature to create instances.

Parameters:

total (int) – Total number of iterations to completion.
initial (int) – Initial value of n.
desc (str | None) – Description to display to the user.
file (TextIO | None) – File-like object to write to. Defaults to sys.stderr.
**kw – Additional options depending on the subclass.

Return type:

AbstractProgressMeter

abstract increment(delta=1)

Increment the position of the meter by the given value.

Parameters:: delta (int)

abstract moveto(n)

Set the meter’s position to the given value.

Parameters:: n (int)

class gambit.util.progress.ClickProgressMeter

Bases: AbstractProgressMeter

Wrapper around a progress bar from the click library.

__init__(pbar)

close(): Stop displaying progress and perform whatever cleanup is necessary.

classmethod create(total, *, initial=0, desc=None, file=None, **kw)

Factory function with standardized signature to create instances.

Parameters:

total (int) – Total number of iterations to completion.
initial (int) – Initial value of n.
desc (str | None) – Description to display to the user.
file (TextIO | None) – File-like object to write to. Defaults to sys.stderr.
**kw – Additional options depending on the subclass.

increment(delta=1)

Increment the position of the meter by the given value.

Parameters:: delta (int)

moveto(n)

Set the meter’s position to the given value.

Parameters:: n (int)

class gambit.util.progress.NullProgressMeter

Bases: AbstractProgressMeter

Progress meter which does nothing.

close(): Stop displaying progress and perform whatever cleanup is necessary.

classmethod create(total, initial=0, **kw)

Factory function with standardized signature to create instances.

Parameters:

total (int) – Total number of iterations to completion.
initial (int) – Initial value of n.
desc – Description to display to the user.
file – File-like object to write to. Defaults to sys.stderr.
**kw – Additional options depending on the subclass.

increment(delta=1)

Increment the position of the meter by the given value.

Parameters:: delta (int)

moveto(n)

Set the meter’s position to the given value.

Parameters:: n (int)

class gambit.util.progress.ProgressConfig

Bases: object

Configuration settings used to create new progress meter instances.

This allows callers to pass the desired progress meter type and other settings to API functions which can then create the instance themselves within the function body by specifying the total length and other final options.

callable

The AbstractProgressMeter.create() method of a concrete progress meter type, or another callable with the same signature which returns a progress meter instance.

Type:: Callable[[int], gambit.util.progress.AbstractProgressMeter]

kw

Keyword arguments to pass to callable.

Type:: dict[str, Any]

__init__(callable, kw)

Parameters:

callable (Callable[[int], AbstractProgressMeter])
kw (dict[str, Any])

create(total, **kw)

Call the factory function with the stored keyword arguments to create a progress meter instance.

The signature of this function is identical to AbstractProgressMeter.create().

Parameters:: total (int)
Return type:: AbstractProgressMeter

update(*args, **kw)

Update keyword arguments and return a new instance.

Parameters:: args (Mapping[str, Any])

class gambit.util.progress.ProgressIterator

Bases: Iterator[T]

__init__(iterable, meter)

Parameters:

iterable (Iterable[T])
meter (AbstractProgressMeter)

class gambit.util.progress.TestProgressMeter

Bases: AbstractProgressMeter

Progress meter which displays no user information but does track progress information.

To be used for testing.

__init__(total, initial=0, allow_decrement=True, **kw)

Parameters:

total (int)
initial (int)
allow_decrement (bool)

close(): Stop displaying progress and perform whatever cleanup is necessary.

classmethod create(total, initial=0, **kw)

Factory function with standardized signature to create instances.

Parameters:

total (int) – Total number of iterations to completion.
initial (int) – Initial value of n.
desc – Description to display to the user.
file – File-like object to write to. Defaults to sys.stderr.
**kw – Additional options depending on the subclass.

increment(delta=1)

Increment the position of the meter by the given value.

Parameters:: delta (int)

moveto(n)

Set the meter’s position to the given value.

Parameters:: n (int)

class gambit.util.progress.TqdmProgressMeter

Bases: AbstractProgressMeter

Wrapper around a progress meter from the tqdm library.

__init__(pbar)

Parameters:: pbar – tqdm.std.tqdm instance to wrap.

close(): Stop displaying progress and perform whatever cleanup is necessary.

classmethod create(total, *, initial=0, desc=None, file=None, **kw)

Factory function with standardized signature to create instances.

Parameters:

total (int) – Total number of iterations to completion.
initial (int) – Initial value of n.
desc (str | None) – Description to display to the user.
file (TextIO | None) – File-like object to write to. Defaults to sys.stderr.
**kw – Additional options depending on the subclass.

increment(delta=1)

Increment the position of the meter by the given value.

Parameters:: delta (int)

moveto(n)

Set the meter’s position to the given value.

Parameters:: n (int)

gambit.util.progress.capture_progress(config)

Creates a ProgressConfig which captures references to the progress meter instances created with it.

This is intended to be used for testing other API functions which create progress meter instances internally that normally would not be accessible by the caller. The captured instance can be checked to ensure it has the correct attributes and went through the full range of iterations, for example.

Returns:: The first item is a modified ProgessConfig instance which can be passed to the function to be tested. The second is a list which is initially empty, and is populated with progress meter instances as they are created by it.
Return type:: Tuple[ProgressConfig, List[AbstractProgressMeter]]
Parameters:: config (ProgressConfig)

gambit.util.progress.check_progress(*, total=None, allow_decrement=False, check_closed=True)

Context manager which checks a progress meter is advanced to completion.

Returned context manager yields a ProgressConfig instance on enter, tests are run when context is exited. Expects that the config will be used to instantiate exactly one progress meter. Tests are performed with assert statements.

Parameters:

total (int | None) – Check that the progress meter is created with this total length.
allow_decrement (bool) – If false, raise an error if the created progress meter is moved backwards.
check_closed (bool) – Check that the progress meter was closed.

Raises:

AssertionError – If any checks fail.

Return type:

Iterator[ProgressConfig]

gambit.util.progress.default_progress_cls()

Get the default AbstractProgressMeter subclass to use.

Currently returns TqdmProgressMeter if tqdm is importable, otherwise prints a warning and returns NullProgressMeter.

Return type:: type

gambit.util.progress.get_progress(arg, total, initial=0, **kw)

Get a progress meter instance.

Meant to be used within other API funcs in which the caller wants to specify the type and parameters of the progress meter but cannot pass an actual instance because the total number of iterations is determined within the body of the function. Instead the API function can take a single progress argument which specifies this information, then create the instance by by calling this function with than information along with the total length.

Accepts the following types/values for the argument:

ProgressConfig
None - uses NullProgressMeter.
True - uses class returned by default_progress_cls().
False - same as None.
str key - Looks up progress bar class/factory function in REGISTRY.
AbstractProgressMeter subclass
callable - factory function. Must have same signature as AbstractProgressMeter.create().

Parameters:

arg (ProgressConfig | str | bool | type | Callable[[int], AbstractProgressMeter] | None) – See above.
total (int) – Length of progress meter to create.
initial (int) – Initial position of progress meter.
**kw – Additional keyword arguments to pass to progress meter class or factory function defined by arg..

Return type:

AbstractProgressMeter

gambit.util.progress.iter_progress(iterable, progress=True, total=None, **kw)

Display a progress meter while iterating over an object.

The returned iterator object can also be used as a context manager to ensure that the progress meter is closed properly even if iteration does not finish.

Parameters:

itarable – Iterable object.
progress (ProgressConfig | str | bool | type | Callable[[int], AbstractProgressMeter] | None) – Passed to get_progress().
total (int | None) – Total number of expected iterations. Defaults to len(iterable).
**kw – Additional keyword arguments to pass to progress meter factory.
iterable (Iterable[T])

Returns:

Iterator over values in iterable which advances a progress meter.

Return type:

ProgressIterator

gambit.util.progress.progress_config(arg, **kw)

Get a ProgressConfig instance from flexible argument types.

See get_progress() for description of allowed argument types/values.

Parameters:: arg (ProgressConfig | str | bool | type | Callable[[int], AbstractProgressMeter] | None)
Return type:: ProgressConfig

gambit.util.progress.register(key)

Decorator to register progress meter class or factory function under the given key.

Parameters:: key (str)

gambit.util.progress.ProgressFactoryFunc

Type alias for a callable which takes total and keyword arguments and returns an AbstractProgressMeter

alias of Callable[[int], AbstractProgressMeter]

gambit.util.progress.REGISTRY = {'click': <bound method ClickProgressMeter.create of <class 'gambit.util.progress.ClickProgressMeter'>>, 'tqdm': <bound method TqdmProgressMeter.create of <class 'gambit.util.progress.TqdmProgressMeter'>>}: TODO