Skip to content

egse.system

cgse-common

This code is part of the cgse-common package.

egse.system

The system module defines convenience functions that provide information on system specific functionality like, file system interactions, timing, operating system interactions, etc.

The module has external dependencies to:

  • distro: for determining the Linux distribution
  • psutil: for system statistics
  • rich: for console output

Classes:

Name Description
AttributeDict

This class is and acts like a dictionary but has the additional functionality

NotSpecified

Class for NOT_SPECIFIED constant.

Sentinel

This Sentinel can be used as an alternative to None or other meaningful values in e.g. a function argument.

SignalCatcher

This class registers handler to signals. When a signal is caught, the handler is

Timer

Context manager to benchmark some lines of code.

Functions:

Name Description
all_logging_disabled

Context manager to temporarily disable logging messages during its execution.

capture_rich_output

Capture the output of a Rich console print of the given object. If the object is a known Rich renderable or if

chdir

Context manager to temporarily change directory.

check_argument_type

Check that the given object is of a specific (sub)type of the given target_class.

check_is_a_string

Checks if the given variable is a string and raises a TypeError if the check fails.

check_str_for_slash

Check if there is a slash in the given string, and raise a ValueError if so.

clear_average_execution_times

Clear out all function timing for this process.

do_every

This method executes a function periodically, taking into account

duration

Returns a timedelta object with the duration, i.e. time difference between dt_start and dt_end.

env_var

Context manager to run some code that need alternate settings for environment variables.

execution_time

A decorator to save the execution time of the function. Use this decorator

filter_by_attr

A helper that returns the elements from the iterable that meet all the traits passed in attrs.

find_class

Find and returns a class based on the fully qualified name.

flatten_dict

Flatten the given dictionary concatenating the keys with a colon ':'.

format_datetime

Format a datetime as YYYY-mm-ddTHH:MM:SS.μs+0000.

get_active_loggers

Retrieves information about active loggers and their respective log levels.

get_average_execution_time

Returns the average execution time of the given function. The function 'func' shall be previously executed using

get_average_execution_times

Returns a dictionary with key = <function name> and value = <average execution time>, for all function that

get_caller_breadcrumbs

Returns a string representing the calling sequence of this function. The string contains the calling sequence from

get_caller_info

Returns the filename, function name and lineno of the caller.

get_full_classname

Returns the fully qualified class name for the given object.

get_host_ip

Returns the IP address.

get_module_location

Returns the location of the module as a Path object.

get_os_name

Returns the name of the OS in lower case.

get_os_version

Return the version of the OS.

get_package_description

Returns the description of the package as specified in the projects metadata Summary.

get_package_location

Retrieves the file system locations associated with a Python package.

get_referenced_var_name

Returns a list of variable names that reference the given object.

get_system_architecture

Returns the machine type. This is a string describing the processor architecture,

get_system_name

Returns the name of the system in lower case.

get_system_stats

Gather system information about the CPUs and memory usage and return a dictionary with the

has_internet

Returns True if we have internet connection.

humanize_seconds

The number of seconds is represented as [#D]d [#H]h[#M]m[#S]s.MS where:

ignore_m_warning

Ignore RuntimeWarning by runpy that occurs when executing a module with python -m package.module,

is_in

Returns result of a in b.

is_in_ipython

Returns True if the code is running in IPython.

is_module

Returns True if the argument is a module or represents a module, False otherwise.

is_namespace

Checks if a module represents a namespace package.

is_not_in

Returns result of a not in b.

now

Returns a datetime object for the current time in UTC or local time.

ping

Sends a ping request to the given host.

read_last_line

Returns the last line of a (text) file.

read_last_lines

Return the last lines of a text file.

recursive_dict_update

Recursively update a dictionary this with the content of another dictionary other.

replace_environment_variable

Returns the input_string with all occurrences of ENV['var'].

sanity_check

Checks a boolean flag and raises an AssertionError with the provided message if the check fails.

save_average_execution_time

Executes the function 'func' with the given arguments and saves the execution time. All positional

str_to_datetime

Convert the given string to a datetime object.

time_in_ms

Returns the current time in milliseconds since the Epoch.

time_since_epoch_1958

Calculate the time since epoch 1958 for the given string representation of a datetime.

touch

Unix-like 'touch', i.e. create a file if it doesn't exist and set the modification time to the current time.

type_name

Returns the name of the type of var.

wait_until

Sleep until the given condition is fulfilled. The arguments are passed into the condition

waiting_for

Sleep until the given condition is fulfilled. The arguments are passed into the condition

Attributes:

Name Type Description
NOT_SPECIFIED

The constant that defines a not-specified value. Intended use is as a sentinel object.

SIGNAL_NAME

The signals that can be catch with the SignalCatcher.

attrdict

Shortcut for the AttributeDict class.

NOT_SPECIFIED module-attribute

NOT_SPECIFIED = NotSpecified()

The constant that defines a not-specified value. Intended use is as a sentinel object.

SIGNAL_NAME module-attribute

SIGNAL_NAME = {
    1: "SIGHUP",
    2: "SIGINT",
    3: "SIGQUIT",
    6: "SIGABRT",
    15: "SIGTERM",
    30: "SIGUSR1",
    31: "SIGUSR2",
}

The signals that can be catch with the SignalCatcher.

attrdict module-attribute

attrdict = AttributeDict

Shortcut for the AttributeDict class.

AttributeDict

AttributeDict(*args, label=None, **kwargs)

Bases: dict

This class is and acts like a dictionary but has the additional functionality that all keys in the dictionary are also accessible as instance attributes.

>>> ad = AttributeDict({'a': 1, 'b': 2, 'c': 3})

>>> assert ad.a == ad['a']
>>> assert ad.b == ad['b']
>>> assert ad.c == ad['c']

Similarly, adding or defining attributes will make them also keys in the dict.

>>> ad.d = 4  # creates a new attribute
>>> print(ad['d'])
4

NotSpecified

Class for NOT_SPECIFIED constant. Is used so that a parameter can have a default value other than None.

Evaluate to False when converted to boolean.

Sentinel

This Sentinel can be used as an alternative to None or other meaningful values in e.g. a function argument.

Usually, a sensible default would be to use None, but if None is a valid input parameter, you can use a Sentinel object and check in the function if the argument value is a Sentinel object.

Example
def get_info(server_socket, timeout: int = Sentinel()):
    if isinstance(timeout, Sentinel):
       raise ValueError("You should enter a valid timeout or None")

SignalCatcher

SignalCatcher()

This class registers handler to signals. When a signal is caught, the handler is executed and a flag for termination or user action is set to True. Check for this flag in your application loop.

  • Termination signals: 1=HUP, 2=INT, 3=QUIT, 6=ABORT, 15=TERM
  • User signals: 30=USR1, 31=USR2

Methods:

Name Description
clear

Call this method to clear the user signal after handling.

handler

Handle the known signals by setting the appropriate flag.

Attributes:

Name Type Description
signal_name

The name of the signal that was caught.

signal_number

The value of the signal that was caught.

signal_name property

signal_name

The name of the signal that was caught.

signal_number property

signal_number

The value of the signal that was caught.

clear

clear(term=False)

Call this method to clear the user signal after handling. Termination signals are not cleared by default since the application is supposed to terminate. Pass in a term=True to also clear the TERM signals, e.g. when you want to ignore some TERM signals.

handler

handler(signal_number, frame)

Handle the known signals by setting the appropriate flag.

Timer

Timer(name='Timer', precision=3, log_level=INFO)

Context manager to benchmark some lines of code.

When the context exits, the elapsed time is sent to the default logger (level=INFO).

Elapsed time can be logged with the log_elapsed() method and requested in fractional seconds by calling the class instance. When the contexts goes out of scope, the elapsed time will not increase anymore.

Log messages are sent to the logger (including egse_logger for egse.system) and the logging level can be passed in as an optional argument. Default logging level is INFO.

Parameters:

Name Type Description Default
name str

a name for the Timer, will be printed in the logging message

'Timer'
precision int

the precision for the presentation of the elapsed time (number of digits behind the comma)

3
log_level int

the log level to report the timing [default=INFO]

INFO
Example
with Timer("Some calculation") as timer:
    # do some calculations
    timer.log_elapsed()
    # do some more calculations
    print(f"Elapsed seconds: {timer()}")
Elapsed seconds: ...

Methods:

Name Description
get_elapsed

Returns the elapsed time for this timer as a float in seconds.

log_elapsed

Sends the elapsed time info to the default logger.

get_elapsed

get_elapsed()

Returns the elapsed time for this timer as a float in seconds.

log_elapsed

log_elapsed()

Sends the elapsed time info to the default logger.

all_logging_disabled

all_logging_disabled(highest_level=CRITICAL, flag=True)

Context manager to temporarily disable logging messages during its execution.

Parameters:

Name Type Description Default
highest_level int

The maximum logging level to be disabled. Defaults to logging.CRITICAL. Note: Adjust this only if a custom level greater than CRITICAL is defined.

CRITICAL
flag bool

If True, disables all logging; if False, no changes are made. Defaults to True.

True
Example
with all_logging_disabled():
    ...  # Your code with logging messages disabled
Note

This context manager is designed to prevent any logging messages triggered during its body from being processed. It temporarily disables logging and restores the previous state afterward.

capture_rich_output

capture_rich_output(obj)

Capture the output of a Rich console print of the given object. If the object is a known Rich renderable or if the object implements the __rich__() method, the output string will contain escape sequences to format the output when printed to a terminal.

This method is usually used to represent Rich output in a log file, e.g. to print a table in the log file.

Parameters:

Name Type Description Default
obj Any

any object

required

Returns:

Type Description
str

The output of the capture, a string that possibly contains escape sequences as a result of rendering rich text.

chdir

chdir(dirname=None)

Context manager to temporarily change directory.

Parameters:

Name Type Description Default
dirname str or Path

temporary folder name to switch to within the context

None
Example
with chdir('/tmp'):
    ...  # do stuff in this writable /tmp folder

check_argument_type

check_argument_type(
    obj, name, target_class, allow_none=False
)

Check that the given object is of a specific (sub)type of the given target_class. The target_class can be a tuple of types.

Parameters:

Name Type Description Default
obj object

any object

required
name str

the name of the object

required
target_class Union[type, Tuple[type]]

the required type of the object (can be a tuple of types)

required
allow_none bool

True if the object can be None

False

Raises:

Type Description
TypeError

when not of the required type or None when not allowed.

check_is_a_string

check_is_a_string(var, allow_none=False)

Checks if the given variable is a string and raises a TypeError if the check fails.

Parameters:

Name Type Description Default
var Any

The variable to be checked.

required
allow_none bool

If True, allows the variable to be None without raising an error. Defaults to False.

False

Raises:

Type Description
TypeError

If the variable is not a string or is None (when allow_none is False).

Example
check_is_a_string("example")
Note

This function is designed to validate that the input variable is a string. If allow_none is set to True, it allows the variable to be None without raising an error.

check_str_for_slash

check_str_for_slash(arg)

Check if there is a slash in the given string, and raise a ValueError if so.

Raises:

Type Description
ValueError

if the string contains a slash '/'.

clear_average_execution_times

clear_average_execution_times()

Clear out all function timing for this process.

do_every

do_every(period, func, *args, count=None)

This method executes a function periodically, taking into account that the function that is executed will take time also and using a simple sleep() will cause a drift. This method will not drift.

You can use this function in combination with the threading module to execute the function in the background, but be careful as the function func might not be thread safe.

timer_thread = threading.Thread(target=do_every, args=(10, func))
timer_thread.daemon = True
timer_thread.start()

Parameters:

Name Type Description Default
period float

a time interval between successive executions [seconds]

required
func callable

the function to be executed

required
*args tuple[int, ...]

optional arguments to be passed to the function

()
count int

if you do not need an endless loop, provide the number of iterations, if count=0 the function will not be executed.

None

duration

duration(dt_start, dt_end)

Returns a timedelta object with the duration, i.e. time difference between dt_start and dt_end.

Notes

If you need the number of seconds of your measurement, use the total_seconds() method of the timedelta object.

Even if you —by accident— switch the start and end time arguments, the duration will be calculated as expected.

Parameters:

Name Type Description Default
dt_start str | datetime

start time of the measurement

required
dt_end str | datetime

end time of the measurement

required

Returns:

Type Description
timedelta

The time difference (duration) between dt_start and dt_end.

env_var

env_var(**kwargs)

Context manager to run some code that need alternate settings for environment variables.

Parameters:

Name Type Description Default
**kwargs dict[str, str]

dictionary with environment variables that are needed

{}
Example
with env_var(PLATO_DATA_STORAGE_LOCATION="/Users/rik/data"):
   # do stuff that needs these alternate setting
   ...

execution_time

execution_time(func)

A decorator to save the execution time of the function. Use this decorator if you want —by default and always— have an idea of the average execution time of the given function.

Use this in conjunction with the get_average_execution_time() function to retrieve the average execution time for the given function.

filter_by_attr

filter_by_attr(elements, **attrs)

A helper that returns the elements from the iterable that meet all the traits passed in attrs.

The attributes are compared to their value with the operator.eq function. However, when the given value for an attribute is a tuple, the first element in the tuple is considered a comparison function and the second value the actual value. The attribute is then compared to the value using this function.

result = filter_by_attr(setups, camera__model="EM", site_id=(is_in, ("CSL", "INTA")))
The function is_in is defined as follows:
def is_in(a, b):
    return a in b
but you can of course also use a lambda function: lambda a, b: a in b.

One function is treated special, it is the built-in function hasattr. Using this function, the value can be True or False. Use this to return all elements in the iterable that have the attribute, or not. The following example returns all Setups where the gse.ogse.fwc_factor is not defined:

result = filter_by_attr(setups, camera__model="EM", gse__ogse__fwc_factor=(hasattr, False)))

When multiple attributes are specified, they are checked using logical AND, not logical OR. Meaning they have to meet every attribute passed in and not one of them.

To have a nested attribute search (i.e. search by gse.hexapod.ID) then pass in gse__hexapod__ID as the keyword argument.

If nothing is found that matches the attributes passed, then an empty list is returned.

When an attribute is not part of the iterated object, that attribute is silently ignored.

Parameters:

Name Type Description Default
elements Iterable

An iterable to search through.

required
attrs dict[str, Any]

Keyword arguments that denote attributes to search with.

{}

find_class

find_class(class_name)

Find and returns a class based on the fully qualified name.

A class name can be preceded with the string class//. This is used in YAML files where the class is then instantiated on load by the Setup.

Parameters:

Name Type Description Default
class_name str

a fully qualified name for the class

required

Returns:

Type Description
Type

The class object corresponding to the fully qualified class name.

Raises:

Type Description
AttributeError

when the class is not found in the module.

ValueError

when the class_name can not be parsed.

ModuleNotFoundError

if the module could not be found.

flatten_dict

flatten_dict(source_dict)

Flatten the given dictionary concatenating the keys with a colon ':'.

Parameters:

Name Type Description Default
source_dict dict

the original dictionary that will be flattened

required

Returns:

Type Description
dict

A new flattened dictionary.

Example
>>> d = {"A": 1, "B": {"E": {"F": 2}}, "C": {"D": 3}}
>>> flatten_dict(d)
{'A': 1, 'B:E:F': 2, 'C:D': 3}

>>> d = {"A": 'a', "B": {"C": {"D": 'd', "E": 'e'}, "F": 'f'}}
>>> flatten_dict(d)
{'A': 'a', 'B:C:D': 'd', 'B:C:E': 'e', 'B:F': 'f'}

format_datetime

format_datetime(dt=None, fmt=None, width=6, precision=3)

Format a datetime as YYYY-mm-ddTHH:MM:SS.μs+0000.

If the given argument is not timezone aware, the last part, i.e. +0000 will not be there.

If no argument is given, the timestamp is generated as datetime.datetime.now(tz=datetime.timezone.utc).

The dt argument can also be a string with the following values: today, yesterday, tomorrow, and 'day before yesterday'. The format will then be '%Y%m%d' unless specified.

Optionally, a format string can be passed in to customize the formatting of the timestamp. This format string will be used with the strftime() method and should obey those conventions.

Example
>>> format_datetime(datetime.datetime(2020, 6, 13, 14, 45, 45, 696138))
'2020-06-13T14:45:45.696'
>>> format_datetime(datetime.datetime(2020, 6, 13, 14, 45, 45, 696138), precision=6)
'2020-06-13T14:45:45.696138'
>>> format_datetime(datetime.datetime(2020, 6, 13, 14, 45, 59, 999501), precision=3)
'2020-06-13T14:45:59.999'
>>> format_datetime(datetime.datetime(2020, 6, 13, 14, 45, 59, 999501), precision=6)
'2020-06-13T14:45:59.999501'
>>> _ = format_datetime()
...
>>> format_datetime("yesterday")
'20220214'
>>> format_datetime("yesterday", fmt="%d/%m/%Y")
'14/02/2022'

Parameters:

Name Type Description Default
dt datetime

a datetime object or an agreed string like yesterday, tomorrow, ...

None
fmt str

a format string that is accepted by strftime()

None
width int

the width to use for formatting the microseconds

6
precision int

the precision for the microseconds

3

Returns:

Type Description
str

a string representation of the current time in UTC, e.g. 2020-04-29T12:30:04.862+0000.

Raises:

Type Description
ValueError

will be raised when the given dt argument string is not understood.

get_active_loggers

get_active_loggers()

Retrieves information about active loggers and their respective log levels.

Returns a dictionary where keys are the names of active loggers, and values are the corresponding log levels in string format.

Returns:

Name Type Description
dict dict

A dictionary mapping logger names to their log levels.

Note

This function provides a snapshot of the currently active loggers and their log levels at the time of the function call.

get_average_execution_time

get_average_execution_time(func)

Returns the average execution time of the given function. The function 'func' shall be previously executed using the save_average_execution_time() function which remembers the last 100 execution times of the function. You can also decorate your function with @execution_time to permanently monitor it. The average time is a moving average over the last 100 times. If the function was never called before, 0.0 is returned.

This function can be used when setting a frequency to execute a certain function. When the average execution time of the function is longer than the execution interval, the frequency shall be decreased or the process will get stalled.

get_average_execution_times

get_average_execution_times()

Returns a dictionary with key = <function name> and value = <average execution time>, for all function that have been monitored in this process.

get_caller_breadcrumbs

get_caller_breadcrumbs(
    prefix="call stack: ", limit=5, with_filename=False
)

Returns a string representing the calling sequence of this function. The string contains the calling sequence from left to right. Each entry has the function name and the line number of the line being executed. When the with_filename is True, also the filename is printed before the function name. If the file is __init__.py, also the parent folder name is printed.

<filename>:<function name>[<lineno>] <— <filename>:<caller function name>[<lineno>]

Use this function for example if you need to find out when and where a function is called in your process.

Example
state.py:load_setup[126] <- state.py:setup[103] <- spw.py:__str__[167] <- nfeesim.py:run[575]

Parameters:

Name Type Description Default
prefix str

a prefix for the calling sequence [default='call stack: '].

'call stack: '
limit int

the maximum number of caller to go back up the calling stack [default=5].

5
with_filename bool

filename is included in the returned string when True [default=False].

False

Returns:

Type Description
str

A string containing the calling sequence.

get_caller_info

get_caller_info(level=1)

Returns the filename, function name and lineno of the caller.

The level indicates how many levels to go back in the stack. When level is 0 information about this function will be returned. That is usually not what you want so the default level is 1 which returns information about the function where the call to get_caller_info was made.

Parameters:

Name Type Description Default
level int

the number of levels to go back in the stack

1

Returns:

Type Description
CallerInfo

a namedtuple: CallerInfo['filename', 'function', 'lineno'].

get_full_classname

get_full_classname(obj)

Returns the fully qualified class name for the given object.

Parameters:

Name Type Description Default
obj object

The object for which to retrieve the fully qualified class name.

required

Returns:

Name Type Description
str str

The fully qualified class name, including the module.

Example
>>> get_full_classname("example")
'builtins.str'

>>> get_full_classname(42)
'builtins.int'
Note

The function considers various scenarios, such as objects being classes, built-ins, or literals like int, float, or complex numbers.

get_host_ip

get_host_ip()

Returns the IP address.

get_module_location

get_module_location(arg)

Returns the location of the module as a Path object.

The function can be given a string, which should then be a module name, or a function or module. For the latter two, the module name will be determined.

Parameters:

Name Type Description Default
arg Any

can be one of the following: function, module, string

required

Returns:

Type Description
Path | None

The location of the module as a Path object or None when the location can not be determined or an invalid argument was provided.

Example
>>> get_module_location('egse')
Path('/path/to/egse')

>>> get_module_location(egse.system)
Path('/path/to/egse/system')
Note

If the module is not found or is not a valid module, None is returned.

Warning

If the module is a namespace, None will be returned. Use the function is_namespace() to determine if the 'module' is a namespace.

get_os_name

get_os_name()

Returns the name of the OS in lower case.

If no name could be determined, 'unknown' is returned.

Returns:

Name Type Description
os str

'macos', 'centos'

get_os_version

get_os_version()

Return the version of the OS.

If no version could be determined, 'unknown' is returned.

Returns:

Name Type Description
version str

as '10.15' or '8.0' or 'unknown'

get_package_description

get_package_description(package_name)

Returns the description of the package as specified in the projects metadata Summary.

Example
>>> get_package_description('cgse-common')
'Software framework to support hardware testing'

get_package_location

get_package_location(module)

Retrieves the file system locations associated with a Python package.

This function takes a module, module name, or fully qualified module path, and returns a list of Path objects representing the file system locations associated with the package. If the module is a namespace package, it returns the paths of all namespaces; otherwise, it returns the location of the module.

Parameters:

Name Type Description Default
module Union[FunctionType, ModuleType, str]

The module or module name to retrieve locations for.

required

Returns:

Type Description
List[Path]

List[Path]: A list of Path objects representing the file system locations.

Note

If the module is not found or is not a valid module, an empty list is returned.

get_referenced_var_name

get_referenced_var_name(obj)

Returns a list of variable names that reference the given object. The names can be both in the local and global namespace of the object.

Parameters:

Name Type Description Default
obj Any

object for which the variable names are returned

required

Returns:

Type Description
List[str]

a list of variable names.

get_system_architecture

get_system_architecture()

Returns the machine type. This is a string describing the processor architecture, like 'i386' or 'arm64', but the exact string is not defined. An empty string can be returned when the type cannot be determined.

get_system_name

get_system_name()

Returns the name of the system in lower case.

Returns:

Name Type Description
name str

'linux', 'darwin', 'windows', ...

get_system_stats

get_system_stats()

Gather system information about the CPUs and memory usage and return a dictionary with the following information:

  • cpu_load: load average over a period of 1, 5,and 15 minutes given in in percentage (i.e. related to the number of CPU cores that are installed on your system) [percentage]
  • cpu_count: physical and logical CPU count, i.e. the number of CPU cores (incl. hyper-threads)
  • total_ram: total physical ram available [bytes]
  • avail_ram: the memory that can be given instantly to processes without the system going into swap. This is calculated by summing different memory values depending on the platform [bytes]
  • boot_time: the system boot time expressed in seconds since the epoch [s]
  • since: boot time of the system, aka Up time [str]

Returns:

Type Description
dict

a dictionary with CPU and memory statistics.

has_internet

has_internet(host='8.8.8.8', port=53, timeout=3.0)

Returns True if we have internet connection.

Parameters:

Name Type Description Default
host str

hostname or IP address [default: 8.8.8.8 (google-public-dns-a.google.com)]

'8.8.8.8'
port int

53 [service: tcp]

53
timeout float

the time to block before failing on a connection

3.0
Note

This might give the following error codes:

  • [Errno 51] Network is unreachable
  • [Errno 61] Connection refused (because the port is blocked?)
  • timed out

humanize_seconds

humanize_seconds(seconds, include_micro_seconds=True)

The number of seconds is represented as [#D]d [#H]h[#M]m[#S]s.MS where:

  • #D is the number of days if days > 0
  • #H is the number of hours if hours > 0
  • #M is the number of minutes if minutes > 0 or hours > 0
  • #S is the number of seconds
  • MS is the number of microseconds

Parameters:

Name Type Description Default
seconds float

the number of seconds

required
include_micro_seconds bool

True if microseconds shall be included

True
Example
>>> humanize_seconds(20)
'20s.000'
>>> humanize_seconds(10*24*60*60)
'10d 00s.000'
>>> humanize_seconds(10*86400 + 3*3600 + 42.023)
'10d 03h00m42s.023'
>>> humanize_seconds(10*86400 + 3*3600 + 42.023, include_micro_seconds=False)
'10d 03h00m42s'

Returns:

Type Description
str

a string representation for the number of seconds.

ignore_m_warning

ignore_m_warning(modules=None)

Ignore RuntimeWarning by runpy that occurs when executing a module with python -m package.module, while that module is also imported.

The original warning message is:

'<package.module>' found in sys.modules after import of package '<package'>,
but prior to execution of '<package.module>'

is_in

is_in(a, b)

Returns result of a in b.

is_in_ipython

is_in_ipython()

Returns True if the code is running in IPython.

is_module

is_module(module)

Returns True if the argument is a module or represents a module, False otherwise.

Parameters:

Name Type Description Default
module str | ModuleType

a module or module name.

required

Returns:

Type Description
bool

True if the argument is a module, False otherwise.

is_namespace

is_namespace(module)

Checks if a module represents a namespace package.

Parameters:

Name Type Description Default
module str | ModuleType

The module to be checked.

required

Returns:

Type Description
bool

True if the argument is a namespace package, False otherwise.

Note

A namespace package is a special kind of package that spans multiple directories or locations, but doesn't contain an __init__.py file in any of its directories.

Technically, a namespace package is defined as a module that has a __path__ attribute and no __file__ attribute.

A namespace package allows for package portions to be distributed independently.

is_not_in

is_not_in(a, b)

Returns result of a not in b.

now

now(utc=True)

Returns a datetime object for the current time in UTC or local time.

ping

ping(host, timeout=3.0)

Sends a ping request to the given host.

Remember that a host may not respond to a ping (ICMP) request even if the host name is valid.

Parameters:

Name Type Description Default
host str

hostname or IP address (as a string)

required
timeout float

timeout in seconds

3.0

Returns:

Type Description
bool

True when host responds to a ping request.

Reference

SO – Pinging servers in Python

read_last_line

read_last_line(filename, max_line_length=5000)

Returns the last line of a (text) file.

The argument max_line_length should be at least the length of the last line in the file, because this value is used to backtrack from the end of the file as an optimization.

Parameters:

Name Type Description Default
filename Path | str

the filename as a string or Path

required
max_line_length int

the maximum length of the lines in the file

5000

read_last_lines

read_last_lines(filename, num_lines)

Return the last lines of a text file.

Parameters:

Name Type Description Default
filename str | Path

Filename.

required
num_lines int

Number of lines at the back of the file that should be read and returned.

required

Returns:

Type Description
List[str]

Last lines of a text file as a list of strings. An empty list is returned when the file doesn't exist.

Raises:

Type Description
AssertionError

when the requested num_lines is zero (0) or a negative number.

recursive_dict_update

recursive_dict_update(this, other)

Recursively update a dictionary this with the content of another dictionary other.

Any key in this dictionary will be recursively updated with the value of the same key in the other dictionary.

Please note that the update will be in-place, i.e. the this dictionaory will be changed/updated.

>>> global_settings = {"A": "GA", "B": "GB", "C": "GC"}
>>> local_settings = {"B": "LB", "D": "LD"}
>>> {**global_settings, **local_settings}
{'A': 'GA', 'B': 'LB', 'C': 'GC', 'D': 'LD'}

>>> global_settings = {"A": "GA", "B": "GB", "C": "GC", "R": {"X": "GX", "Y": "GY"}}
>>> local_settings = {"B": "LB", "D": "LD", "R": {"Y": "LY"}}
>>> recursive_dict_update(global_settings, local_settings)
{'A': 'GA', 'B': 'LB', 'C': 'GC', 'R': {'X': 'GX', 'Y': 'LY'}, 'D': 'LD'}

>>> global_settings = {"A": {"B": {"C": {"D": 42}}}}
>>> local_settings = {"A": {"B": {"C": 13, "D": 73}}}
>>> recursive_dict_update(global_settings, local_settings)
{'A': {'B': {'C': 13, 'D': 73}}}

Parameters:

Name Type Description Default
this dict

The origin dictionary

required
other dict

Changes that shall be applied to this

required

Returns:

Type Description
dict

The original this dictionary with the recursive updates.

replace_environment_variable

replace_environment_variable(input_string)

Returns the input_string with all occurrences of ENV['var'].

>>> replace_environment_variable("ENV['HOME']/data/CSL")
'/Users/rik/data/CSL'

Parameters:

Name Type Description Default
input_string str

the string to replace

required

sanity_check

sanity_check(flag, msg)

Checks a boolean flag and raises an AssertionError with the provided message if the check fails.

This function serves as a replacement for the 'assert' statement in production code. Using this ensures that your checks are not removed during optimizations.

Parameters:

Name Type Description Default
flag bool

The boolean flag to be checked.

required
msg str

The message to be included in the AssertionError if the check fails.

required

Raises:

Type Description
AssertionError

If the flag is False.

Example
>>> sanity_check(x > 0, "x must be greater than 0")
Note

This function is designed for production code to perform runtime checks that won't be removed during optimizations.

save_average_execution_time

save_average_execution_time(func, *args, **kwargs)

Executes the function 'func' with the given arguments and saves the execution time. All positional arguments (in args) and keyword arguments (in kwargs) are passed into the function. The execution time is saved in a deque of maximum 100 elements. When more times are added, the oldest times are discarded. This function is used in conjunction with the get_average_execution_time() function.

str_to_datetime

str_to_datetime(datetime_string)

Convert the given string to a datetime object.

Parameters:

Name Type Description Default
datetime_string str

String representing a datetime, in the format %Y-%m-%dT%H:%M:%S.%f%z.

required

Returns:

Type Description
datetime

a datetime object.

time_in_ms

time_in_ms()

Returns the current time in milliseconds since the Epoch.

Note

if you are looking for a high performance timer, you should really be using perf_counter() instead of this function.

time_since_epoch_1958

time_since_epoch_1958(datetime_string)

Calculate the time since epoch 1958 for the given string representation of a datetime.

Parameters:

Name Type Description Default
datetime_string str

String representing a datetime, in the format %Y-%m-%dT%H:%M:%S.%f%z.

required

Returns:

Type Description
float

Time since the 1958 epoch [s].

touch

touch(path)

Unix-like 'touch', i.e. create a file if it doesn't exist and set the modification time to the current time.

Parameters:

Name Type Description Default
path Path | str

full path to the file, can start with ~ which is automatically expanded.

required

type_name

type_name(var)

Returns the name of the type of var.

wait_until

wait_until(
    condition,
    *args,
    interval=0.1,
    timeout=1.0,
    verbose=False,
    **kwargs,
)

Sleep until the given condition is fulfilled. The arguments are passed into the condition callable which is called in a while loop until the condition is met or the timeout is reached.

Note that the condition can be a function, method or callable class object. An example of the latter is:

class SleepUntilCount:
    def __init__(self, end):
        self._end = end
        self._count = 0

    def __call__(self, *args, **kwargs):
        self._count += 1
        if self._count >= self._end:
            return True
        else:
            return False

Parameters:

Name Type Description Default
condition Callable

a callable that returns True when the condition is met, False otherwise

required
interval float

the sleep interval between condition checks [s, default=0.1]

0.1
timeout float

the period after which the function returns, even when the condition is not met [s, default=1]

1.0
verbose bool

log debugging messages if True

False
*args list

any arguments that will be passed into the condition function

()
**kwargs dict

any keyword arguments that will be passed into the condition function

{}

Returns:

Type Description
bool

True when function timed out, False otherwise.

waiting_for

waiting_for(
    condition,
    *args,
    interval=0.1,
    timeout=1.0,
    verbose=False,
    **kwargs,
)

Sleep until the given condition is fulfilled. The arguments are passed into the condition callable which is called in a while loop until the condition is met or the timeout is reached.

Note that the condition can be a function, method or callable class object. An example of the latter is:

class SleepUntilCount:
    def __init__(self, end):
        self._end = end
        self._count = 0

    def __call__(self, *args, **kwargs):
        self._count += 1
        if self._count >= self._end:
            return True
        else:
            return False

Parameters:

Name Type Description Default
condition Callable

a callable that returns True when the condition is met, False otherwise

required
interval float

the sleep interval between condition checks [s, default=0.1]

0.1
timeout float

the period after which the function returns, even when the condition is not met [s, default=1]

1.0
verbose bool

log debugging messages if True

False
*args list

any arguments that will be passed into the condition function

()
**kwargs dict

any keyword arguments that will be passed into the condition function

{}

Returns:

Type Description
float

The duration until the condition was met.

Raises:

Type Description
TimeoutError

when the condition was not fulfilled within the timeout period.