4.1.2.2. lib.utilities#

Module with base-classes that provide some useful methods that are not associated with a specific module.

4.1.2.2.1. Module Contents#

4.1.2.2.1.1. Classes#

Configuration

Configuration class, used by many objects to initialize an internal (read only) configuration, which cannot be changed after initialization.

Status

Class to document the status of a component. Sanity checks can change the status of an object and trigger an action, e.g. abortion of a measurement.

FunctionLogger

Class to log the activity of functions with an observer pattern.

Log

NumpyArrayEncoder

Helper class for JSON export of NumPy arrays.

Labeler

Converter

Importer

Exporter

MultithreadedCopier

BaseFunc

4.1.2.2.1.2. API#

class lib.utilities.Configuration(default_params, param_dict)#

Configuration class, used by many objects to initialize an internal (read only) configuration, which cannot be changed after initialization.

Parameters:
  • default_params (dict) – Dictionary of default parameters.

  • param_dict (dict) – Dictionary of parameters to be set.

Returns:

None

Return type:

None

Initialization

Initializes an instance of the Configuration class.

Parameters:
  • default_params (dict) – Dictionary of default parameters.

  • param_dict (dict) – Dictionary of parameters to be set.

Returns:

None

Return type:

None

set_params(param_dict)#

Update an instance of Configuration with a dictionary, e.g. to modify values after a fit. Dependent quantities are recalculated.

Parameters:

param_dict (dict) – Dictionary of parameters to be set.

Returns:

None

Return type:

None

do_sanity_check()#

Perform a sanity check on the Configuration object.

Returns:

None

Return type:

None

class lib.utilities.Status#

Class to document the status of a component. Sanity checks can change the status of an object and trigger an action, e.g. abortion of a measurement.

Variables:

ok (bool) – The status of the component. True if the sanity check was successful, False otherwise.

Initialization

Initializes an instance of the Status class.

Returns:

None

Return type:

None

class lib.utilities.FunctionLogger(table_name='test', data_base='log.sqlite', comment='test')#

Class to log the activity of functions with an observer pattern.

Variables:
  • table_name (str) – The name of the table in the database to store the log records.

  • data_base (str) – The name of the SQLite database file.

  • comment (str) – Additional comment for the log records.

  • git_hash (str) – The current Git hash.

  • conn (sqlite3.Connection) – The connection to the SQLite database.

  • registered_functions (List) – List of registered functions.

  • success (bool) – The success status of the logged function.

Initialization

Initializes an instance of the FunctionLogger class.

Parameters:
  • table_name (str) – The name of the table in the database to store the log records.

  • data_base (str) – The name of the SQLite database file.

  • comment (str) – Additional comment for the log records.

Returns:

None

Return type:

None

__call__(function)#

Decorator to wrap a function and log its activity.

Parameters:

function (function) – The function to be wrapped and logged.

Returns:

The wrapped function.

Return type:

function

register_function(function)#

Registers a function to be logged.

Parameters:

function (function) – The function to be registered.

Returns:

None

Return type:

None

log_function_call(function_name, args, kwargs)#

Logs a function call.

Parameters:
  • function_name (str) – The name of the function.

  • args (tuple) – The arguments passed to the function.

  • kwargs (dict) – The keyword arguments passed to the function.

Returns:

None

Return type:

None

rerun_f(date)#

Reruns a function call based on the specified date.

Parameters:

date (str) – The date of the function call to rerun.

Returns:

The result of the rerun function call.

Return type:

Any

Raises:

ValueError – If no function call is found with the specified date.

terminate()#

Terminates the connection to the SQLite database.

Returns:

None

Return type:

None

class lib.utilities.Log(comment)#

Initialization

Initializes a Log object.

Parameters:

comment (str) – The comment to be logged.

Returns:

None

Return type:

None

log_to_database(database)#

Logs a function call to the database.

Parameters:

database (str) – The name of the SQLite database file.

Returns:

None

Return type:

None

class lib.utilities.NumpyArrayEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)#

Bases: json.JSONEncoder

Helper class for JSON export of NumPy arrays.

This class is used as a custom encoder for JSON serialization of NumPy arrays. It handles the conversion of NumPy data types to their Python counterparts, as well as converting NumPy arrays to lists.

Usage:

json.dumps(data, cls=NumpyArrayEncoder)

Note:

To use this encoder, you need to import NumPy and JSONEncoder from the json module.

Example:

import json import numpy as np

data = {‘array’: np.array([1, 2, 3])} json_str = json.dumps(data, cls=NumpyArrayEncoder) print(json_str)

Output:

{“array”: [1, 2, 3]}

Initialization

Constructor for JSONEncoder, with sensible defaults.

If skipkeys is false, then it is a TypeError to attempt encoding of keys that are not str, int, float or None. If skipkeys is True, such items are simply skipped.

If ensure_ascii is true, the output is guaranteed to be str objects with all incoming non-ASCII characters escaped. If ensure_ascii is false, the output can contain non-ASCII characters.

If check_circular is true, then lists, dicts, and custom encoded objects will be checked for circular references during encoding to prevent an infinite recursion (which would cause an RecursionError). Otherwise, no such check takes place.

If allow_nan is true, then NaN, Infinity, and -Infinity will be encoded as such. This behavior is not JSON specification compliant, but is consistent with most JavaScript based encoders and decoders. Otherwise, it will be a ValueError to encode such floats.

If sort_keys is true, then the output of dictionaries will be sorted by key; this is useful for regression tests to ensure that JSON serializations can be compared on a day-to-day basis.

If indent is a non-negative integer, then JSON array elements and object members will be pretty-printed with that indent level. An indent level of 0 will only insert newlines. None is the most compact representation.

If specified, separators should be an (item_separator, key_separator) tuple. The default is (’, ‘, ‘: ‘) if indent is None and (‘,’, ‘: ‘) otherwise. To get the most compact JSON representation, you should specify (‘,’, ‘:’) to eliminate whitespace.

If specified, default is a function that gets called for objects that can’t otherwise be serialized. It should return a JSON encodable version of the object or raise a TypeError.

default(obj)#

Converts the specified object to a JSON serializable form.

Parameters:

obj (Any) – The object to be converted.

Returns:

The JSON serializable form of the object.

Return type:

Any

class lib.utilities.Labeler#

Initialization

Initializes a Labeler object.

Returns:

None

Return type:

None

stamp(filename)#

Creates a directory for the present date and a file string to save the artifacts with time and git-hash.

Parameters:

filename (str) – The name of the file to be saved.

Returns:

The file string and the directory path.

Return type:

tuple(str, str)

id_generator(size=6, chars=string.ascii_uppercase + string.digits)#

Generates a random ID string.

Parameters:
  • size (int) – The length of the generated string.

  • chars (str) – The characters to choose from.

Returns:

The generated ID string.

Return type:

str

class lib.utilities.Converter#

Initialization

Initializes a Converter object.

Returns:

None

Return type:

None

to_yaml(obj)#

Converts an object to YAML.

Parameters:

obj (Any) – The object to be converted.

Returns:

The serialized YAML output for the object, or None when saving to a file.

Return type:

str or None

class lib.utilities.Importer#

Initialization

Initializes an Importer object.

Returns:

None

Return type:

None

load_yaml(obj)#

Converts from YAML back to an object.

Parameters:

obj (str or object) – The path to the YAML file or the YAML object.

Returns:

The deserialized object.

Return type:

object

load_json(filename)#

Loads a dictionary from a JSON file, such as an experimental configuration.

Parameters:

filename (str) – The path to the JSON file.

Returns:

The loaded dictionary.

Return type:

dict

class lib.utilities.Exporter#
__Init__()#

Initializes an Exporter object.

Returns:

None

Return type:

None

write_yaml(obj, filename=None, out_dir=None)#

Export obj to a YAML file.

Parameters:
  • obj (Any) – The object to be exported.

  • filename (str or None) – The name of the YAML file to be created. If None, a timestamp will be used.

  • out_dir (str or None) – The directory to save the YAML file. If None, the current working directory will be used.

Returns:

The full path to the created YAML file.

Return type:

str

write_json(new_data, filestr=None, filename='')#

Writes data to a JSON file.

Parameters:
  • new_data (dict) – The data to be written to the JSON file.

  • filestr (str or None) – The path to the JSON file to be opened and data added. If None, a new file will be created.

  • filename (str) – The filename of the JSON file to be created. If no path is provided, a timestamp will be used.

Returns:

The full path to the created JSON file.

Return type:

str

class lib.utilities.MultithreadedCopier(max_threads)#

Initialization

Initializes a MultithreadedCopier object.

Parameters:

max_threads (int) – The maximum number of threads to use for copying.

Returns:

None

Return type:

None

copy(source, dest)#

Copy a file from source to destination using multiple threads.

Parameters:
  • source (str) – The path to the source file.

  • dest (str) – The destination directory.

Returns:

None

Return type:

None

__enter__()#

Enter the context manager.

Returns:

Self

Return type:

MultithreadedCopier

__exit__(exc_type, exc_val, exc_tb)#

Exit the context manager.

Parameters:
  • exc_type (type or None) – The exception type, if an exception occurred.

  • exc_val (Exception or None) – The exception value, if an exception occurred.

  • exc_tb (traceback or None) – The exception traceback, if an exception occurred.

Returns:

None

Return type:

None

class lib.utilities.BaseFunc#

Initialization

Initialize the BaseFunc class.

Returns:

None

join_dict_values(dict1, dict2)#

Join the values of dict1 with dict2.

Parameters:
  • dict1 (dict) – The first dictionary.

  • dict2 (dict) – The second dictionary.

Returns:

The joined dictionary.

Return type:

dict

remove_trailing_zeros(array)#

Remove trailing zeros from an array.

Parameters:

array (list or numpy.ndarray) – The input array.

Returns:

The array with trailing zeros removed.

Return type:

list or numpy.ndarray

calculate_norms(list_with_nans)#

Calculate the L2 norm for each non-NaN element in a given list.

Parameters:

list_with_nans (list) – A list of floats or NaNs.

Returns:

A list containing the L2 norm for each non-NaN element.

Return type:

list

Example:

>>> obj = BaseFunc()
>>> obj.calculate_norms([np.array([1,2,3]), np.nan])
[3.7416563645999, nan]
flatten_list(lst)#

Flatten a nested list.

Parameters:

lst (list) – The input list.

Returns:

The flattened list.

Return type:

list

Example:

>>> obj = BaseFunc()
>>> obj.flatten_list([1, [2, [3, 4]], 5])
[1, 2, 3, 4, 5]
nan_mask_list(list1, list2, list3=None)#

Return a copy of list1 with values set to NaN where they differ from list2 or list3 if provided.

Parameters:
  • list1 (list) – The list to be masked.

  • list2 (list) – The subset of list1 that will be used to mask it.

  • list3 (list or None) – The subset that will be used to mask list1 in place of list2 if not None and equal in length to list2.

Returns:

A new list with values set to NaN where they differ from list2 or list3.

Return type:

list

Raises:

ValueError – If list2 is not a subset of list1 or list2 and list3 are not equal in length.

Example:

>>> obj = BaseFunc()
>>> list1 = [1, 2, 3, 4, 5, 6]
>>> list2 = [3, 4, 5]
>>> obj.nan_mask_list(list1, list2, list3=[0, 1, 2])
[nan, nan, 0, 1, 2, nan]
to_tuple(lst)#

Convert a nested list to a tuple.

Parameters:

lst (list) – The input list.

Returns:

The converted tuple.

Return type:

tuple

Example:

>>> obj = BaseFunc()
>>> obj.to_tuple([1, [2, [3, 4]], 5])
(1, (2, (3, 4)), 5)
_moving_average(a, n=10)#

Calculate the moving average of an array.

Parameters:
  • a (list or numpy.ndarray) – The input array.

  • n (int) – The window size for the moving average. Default is 10.

Returns:

The moving average of the input array.

Return type:

numpy.ndarray

Example:

>>> obj = BaseFunc()
>>> arr = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
>>> obj._moving_average(arr, n=3)
array([2., 3., 4., 5., 6., 7., 8.])
weighted_median(values, weights)#

Calculate the weighted median of a list of values.

Parameters:
  • values (list) – The input values.

  • weights (list) – The weights corresponding to the input values.

Returns:

The weighted median.

Return type:

float

Example:

>>> obj = BaseFunc()
>>> values = [1, 2, 3, 4, 5]
>>> weights = [0.1, 0.2, 0.3, 0.2, 0.2]
>>> obj.weighted_median(values, weights)
3.0
match_pattern(string, pattern, match='partial')#

Check if a string matches a pattern.

Parameters:
  • string (str) – The input string.

  • pattern (str) – The pattern to match against.

  • match (str) – The type of match to perform. Default is ‘partial’. - ‘full’: Check for a full match. - ‘partial’: Check for a partial match.

Returns:

True if the string matches the pattern, False otherwise.

Return type:

bool

Example:

>>> obj = BaseFunc()
>>> obj.match_pattern('hello', 'hello', match='full')
True
>>> obj.match_pattern('hello world', 'hello', match='partial')
True
find_files(in_folder, condition, max_files=100, max_depth=10)#

Iterate through a nested directory and find files that meet a given condition.

Parameters:
  • in_folder (str) – The root directory to start the iteration from.

  • condition (callable) – The condition that files must meet to be included.

  • max_files (int) – The maximum number of files to be analyzed in one directory. Defaults to 100.

  • max_depth (int) – The maximum depth of subdirectories to be iterated through. Defaults to 10.

Returns:

A list of paths to the found files.

Return type:

list

Example:

>>> obj = BaseFunc()
>>> def file_condition(filename):
...     return filename.endswith('.txt')
>>> obj.find_files('/path/to/directory', file_condition, max_files=50, max_depth=5)
['/path/to/directory/file1.txt', '/path/to/directory/subdir/file2.txt', ...]
get_git_root(path)#

Get the root directory of a Git repository.

Given a file or directory path within a Git repository, this method returns the root directory of that repository.

Parameters:

path (str) – File or directory path within the Git repository.

Returns:

Root directory of the Git repository.

Return type:

str