GEMstack.utils package

Other utilities common to onboard / offboard use.

Submodules

GEMstack.utils.config module

GEMstack.utils.config.load_config_recursive(fn: str) → dict

Loads a configuration file with !include and !relative_path directives.

GEMstack.utils.config.save_config(fn: str, config: dict) → None

Saves a configuration file.

GEMstack.utils.config.update_recursive(d: dict, u: dict, caution_callback=None) → dict

Updates a dictionary d with another dictionary u recursively.

The update happens in-place and d is returned.

GEMstack.utils.logging module

GEMstack.utils.loops module

class GEMstack.utils.loops.TimedLooper(dt=None, rate=None, warning_frequency='auto', name=None)

Bases: object

A class to easily control how timed loops are run.

Usage:

looper = TimedLooper(dt=0.01)
while looper:
    ... do stuff ...
    if need to stop:
        looper.stop()
        #or just call break

Note that if dt is too small (or rate is too large), the timing will not be accurate due to the system scheduler resolution.

If the code within the loop takes more than dt seconds to run, then a warning may be printed. To turn this off, set warnings=0 in the constructor. By default, this will print a warning on the first overrun, and every warning_frequency overruns thereafter.

Parameters

  • dt (float, optional) – the desired time between loops (in seconds)

  • rate (float, optional) – the number of times per second to run this loop (in Hz). dt = 1/rate. One of dt or rate must be specified.

  • warning_frequency (int, optional) – if the elapsed time between calls exceeds dt, a warning message will be printed at this frequency. Set this to 0 to disable warnings.

  • name (str, optional) – a descriptive name to be used in the warning string.

Warning: DO NOT attempt to save some time and call the TimedLooper() constructor as the condition of your while loop! I.e., do not do this:

while TimedLooper(dt=0.01):
    ...
iters()

Returns the total number of iters run

print_warning(s)

Override this to change how warning strings are printed, e.g. to add your own logger

stop()
time_elapsed()

Returns the total time elapsed from the start, in seconds

class GEMstack.utils.loops.TimedLooperAsync(dt=None, rate=None, warning_frequency='auto', name=None)

Bases: object

A class to easily control how timed loops are run. This is more accurate than asyncio.sleep(dt) and also maintains information about the loop.

Usage:

looper = TimedLooperAsync(dt=0.01)
while no need to stop:
    await looper()
    ... do stuff ...

Note that if dt is too small (or rate is too large), the timing will not be accurate due to the system scheduler resolution.

If the code within the loop takes more than dt seconds to run, then a warning may be printed. To turn this off, set warnings=0 in the constructor. By default, this will print a warning on the first overrun, and every warning_frequency overruns thereafter.

Parameters

  • dt (float, optional) – the desired time between loops (in seconds)

  • rate (float, optional) – the number of times per second to run this loop (in Hz). dt = 1/rate. One of dt or rate must be specified.

  • warning_frequency (int, optional) – if the elapsed time between calls exceeds dt, a warning message will be printed at this frequency. Set this to 0 to disable warnings.

  • name (str, optional) – a descriptive name to be used in the warning string.

Warning: DO NOT attempt to save some time and call the TimedLooperAsync() constructor as the condition of your while loop! I.e., do not do this:

while True:
    await TimedLooperAsync(dt=0.01)()
    ... do stuff...
iters()

Returns the total number of iters run

print_warning(s)

Override this to change how warning strings are printed, e.g. to add your own logger

time_elapsed()

Returns the total time elapsed from the start, in seconds

GEMstack.utils.serialization module

GEMstack.utils.settings module

GEMstack.utils.settings.get(path: ~typing.Union[str, ~typing.List[str]], defaultValue=<class 'KeyError'>) → Any

Retrieves a setting by a list of keys or a ‘.’-separated string.

GEMstack.utils.settings.load_settings(settings_file: Optional[str] = None)

Loads the settings object for the first time.

Order of operations is: - If settings_file is given, load it. - Otherwise, get the settings from defaults.SETTINGS - Look through the command line arguments to determine whether the user has

overridden any settings using –KEY=VALUE.

GEMstack.utils.settings.set(path: Union[str, List[str]], value: Any, leaf_only=True) → None

Sets a setting by a list of keys or a ‘.’-separated string.

If leaf_only=True (default), we prevent inadvertently deleting parts of the settings dictionary.

GEMstack.utils.settings.settings()

Returns all global settings, loading them if necessary.