pykanto.utils.compute#

A collection of functions and decorators used to manage common operations (e.g., parallelisation, timing).

Functions

calc_chunks(len_iterable[, factor, ...])

Calculate chunk size to optimise parallel computing.

dictlist_to_dict(dictlist)

Builds a dictionary from a list of dictionaries.

flatten_list(lst)

Flattens a list using chain.

get_chunks(lst, n)

Yields successive n-sized chunks from list.

print_dict(dictionary)

Pretty print a class __dict__ attribute.

print_parallel_info(n, iterable_name, ...)

Prints information about a parallel process for the user.

timing(f)

Custom timer decorator.

to_iterator(obj_ids[, breaks])

Use ray paralell with a progress bar.

with_pbar(iterable[, desc])

Returns a custom progress bar.
pykanto.utils.compute.to_iterator(obj_ids, breaks: bool = True)[source]#

Use ray paralell with a progress bar. Modified from https://git.io/JMv6r.

pykanto.utils.compute.with_pbar(iterable: Iterable[Any], desc: str | None = None, **kwargs) → tqdm[source]#

Returns a custom progress bar. This is just a wrapper around tqdm.

Parameters

  • iterable (Iterable[Any]) – Object to iterate on.

  • desc (str, optional) – Description of what’s going on. Defaults to None.

Returns

progress bar.

Return type

tqdm

pykanto.utils.compute.print_dict(dictionary: Dict) → str[source]#

Pretty print a class __dict__ attribute.

Parameters

dictionary (Dict) – __dict__ attribute containing an object’s writable attributes.

Returns

Dictionary contents in a legible way.

Return type

str

pykanto.utils.compute.timing(f)[source]#

Custom timer decorator. Prints time info unless used within a KantoData where parameters.verbose = False.

pykanto.utils.compute.calc_chunks(len_iterable: int, factor: int = 2, n_workers: float | None = None, verbose: bool = False) → Chunkinfo[source]#

Calculate chunk size to optimise parallel computing. Adapted from https://stackoverflow.com/a/54032744.

returns [‘n_workers’, ‘len_iterable’, ‘n_chunks’,

‘chunksize’, ‘last_chunk’]

pykanto.utils.compute.get_chunks(lst: List[Any], n: int) → Iterator[Any][source]#

Yields successive n-sized chunks from list. Last chunk will be shorter if len(lst) % n != 0.

Parameters

  • lst (List[Any]) – List to return chunks from.

  • n (int) – Number of chunks

Yields

Iterator[Any] – n-sized chunks.

pykanto.utils.compute.flatten_list(lst: List[Any]) → List[Any][source]#

Flattens a list using chain.

pykanto.utils.compute.dictlist_to_dict(dictlist: List[Dict[str, Any]]) → Dict[str, Any][source]#

Builds a dictionary from a list of dictionaries.

Parameters

dictlist (List[Dict[str, Any]]) – List of dictionaries.

Returns

Dictionary with items == list elements.

Return type

Dict[str, Any]

pykanto.utils.compute.print_parallel_info(n: int, iterable_name: str, n_chunks: int, chunk_length: int) → None[source]#

Prints information about a parallel process for the user.

Parameters

  • n (int) – Total length of iterable.

  • iterable_name (str) – Description of iterable.

  • n_chunks (int) – Number of chunks that will be used.

  • chunk_length (int) – Length of each chunk.