Common Utilities#

Common utility functions used throughout the project.

Timeout utilities are adapted from pnpnpn/timeout-decorator (MIT License) and from chaidiscovery/chai-lab

exception atomworks.common.ChildProcessError[source]#

Bases: Exception

Exception raised when a child process dies unexpectedly.

class atomworks.common.KeyToIntMapper[source]#

Bases: object

Maps keys to unique integers based on the order of the first appearance of the key.

This is useful for mapping id’s such as chain_id, chain_entity, molecule_iid, etc. to integers.

Example

>>> chain_id_to_int = KeyToIntMapper()
>>> chain_id_to_int("A")  # 0
>>> chain_id_to_int("C")  # 1
>>> chain_id_to_int("A")  # 0
>>> chain_id_to_int("B")  # 2

atomworks.common.as_list(value: Any) → list[source]#

Convert a value to a list.

Handles various types using duck typing:

Iterable objects (lists, tuples, strings, etc.): converted to list
Single values: wrapped in a list

Parameters:: value – The value to convert to a list.
Returns:: A list containing the value(s).

atomworks.common.default(obj: Any, default: Any) → Any[source]#

Return obj if not None, otherwise return default.

Parameters:

obj – The primary object to return.
default – The fallback value if obj is None.

Returns:

obj if it is not None, otherwise default.

atomworks.common.do_nothing(*args, **kwargs) → Callable[source]#

A decorator that does nothing and simply returns the original function.

This decorator can be used as a placeholder or for testing purposes when you want to conditionally apply decorators without changing the code structure.

Returns:: A decorator function that returns the original function unchanged.

Example

@do_nothing_decorator()
def my_function():
    return "Hello, World!"


# or:
do_nothing(bla=123, blub=456)(my_function)

atomworks.common.exists(obj: Any) → bool[source]#

Check that obj is not None.

Parameters:: obj – The object to check.
Returns:: True if obj is not None, False otherwise.

atomworks.common.immutable_lru_cache(maxsize: int = 128, typed: bool = False, deepcopy: bool = True, copy_func: Callable | None = None) → Callable[source]#

An immutable version of lru_cache for caching functions that return mutable objects.

Parameters:

maxsize – Maximum number of items to cache.
typed – Whether to treat different types as separate cache entries.
deepcopy – Whether to use deep copy for immutable caching.
copy_func – Custom copy function to use. If provided, overrides deepcopy parameter. Should be a callable that takes the cached object and returns a copy.

Returns:

A decorator that provides immutable caching functionality.

Example

>>> # Use biotite's fast copy for AtomArrays
>>> @immutable_lru_cache(maxsize=200, copy_func=lambda x: x.copy())
>>> def get_template(code):
...     return atom_array_from_ccd_code(code)

atomworks.common.listmap(func: Callable, *iterables) → list[source]#

Like map, but returns a list instead of an iterator.

Parameters:

func – The function to apply.
*iterables – Variable number of iterables to map over.

Returns:

A list containing the results of applying func to the iterables.

atomworks.common.not_isin(element: ndarray, array: ndarray, **isin_kwargs) → ndarray[source]#

Like ~np.isin, but more efficient.

Parameters:

element – The array to test.
array – The array of values to test against.
**isin_kwargs – Additional keyword arguments for np.isin.

Returns:

Boolean array indicating which elements are not in the array.

atomworks.common.string_to_md5_hash(s: str, truncate: int = 32) → str[source]#

Generate an MD5 hash of a string and return the first truncate characters.

Parameters:

s – The string to hash.
truncate – Number of characters to return from the hash.

Returns:

The truncated MD5 hash as a string.

atomworks.common.sum_string_arrays(*objs: ndarray | str) → ndarray[source]#

Sum a list of string arrays or strings into a single string array.

Concatenates the arrays and determines the shortest string length to set as dtype.

Parameters:: *objs – Variable number of string arrays or strings to sum.
Returns:: A single concatenated string array.

atomworks.common.timeout(timeout: float | int | None = None, strategy: Literal['signal', 'subprocess'] = 'subprocess') → Callable[source]#

Decorator to apply a timeout to a function.

The signal strategy is more efficient and slightly faster, but does not work in all contexts (e.g. with some C dependencies like RDKit, on certain operating systems). The subprocess strategy is always available, but slightly slower and with a higher overhead.

atomworks.common.timeout_using_signal(timeout: float | int | None) → Callable[source]#

Build a decorator that applies a timeout to a function using the signal module.

This decorator sets up a signal handler to raise a TimeoutError if the decorated function exceeds the specified timeout duration. It uses the SIGALRM signal to implement the timeout.

Use for example as: `python result = timeout_using_signal(timeout=10.0)(my_function)(*args, **kwargs) `

Parameters:: timeout (float | int | None) – The timeout duration in seconds.
Returns:: A decorator function that can be applied to other functions to add timeout functionality.
Return type:: Callable

atomworks.common.timeout_using_subprocess(timeout: float | int | None) → Callable[source]#

Force function to timeout after specified time.

The returned decorator uses a subprocess to execute the function, allowing for timeout functionality even for CPU-bound operations that cannot be interrupted by signals.

Parameters:

timeout (float | int | None) – The maximum time in seconds allowed for the function to execute.

Returns:

A decorator that can be applied to a function.

Return type:

Callable

Raises:

TimeoutError – If the function does not return before the timeout.
ChildProcessException – If the child process dies unexpectedly.

atomworks.common.to_hashable(element: Any) → Any[source]#

Convert an element to a hashable type.

Parameters:: element – The element to convert.
Returns:: The element if already hashable, otherwise converted to a tuple.

Common Utilities#

This Page