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.
`Periodic`	A timer that periodically invokes a function in the background.
`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.
`TyperAsyncCommand`	Runs an asyncio Typer command.

Functions:

Name	Description
`all_logging_disabled`	Context manager to temporarily disable logging messages during its execution.
`await_me_maybe`	Invoke a callback with an arbitrary number of parameters.
`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_current_location`	Returns the location where this function is called, i.e. the filename, line number, and function name.
`get_full_classname`	Returns the fully qualified class name for the given object.
`get_host_ip`	Returns the IP address. If no IP address can be found, None will be returned and the caller can try
`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'].
`round_up`	Round a number up to a specified number of decimal places.
`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.

Periodic ¶

Periodic(
    interval,
    *,
    name=None,
    callback=None,
    repeat=None,
    skip=True,
    pause=False,
)

A timer that periodically invokes a function in the background.

If no callback is provided, a warning message will be logged. In a future, we might send out an event to the application (will need an event handler).

When the function execution takes longer then the interval there re several options:

if skip is True (default) the interval will take precedence and the Args: interval: The time between timer events, in seconds. name: A name to assign the event (for debugging), defaults to Periodic#. callback: A optional callback to invoke when the event is handled. repeat: The number of times to repeat the timer, or None to repeat forever. skip: Enable skipping of scheduled function calls that couldn't be sent in time. pause: Start the timer paused. Use resume() to activate the timer.

Methods:

Name	Description
`pause`	Pause the timer.
`reset`	Reset the timer, so it starts from the beginning.
`resume`	Resume a paused timer.
`start`	Start the timer.
`stop`	Stop the timer.

pause ¶

pause()

Pause the timer.

A paused timer will not send events until it is resumed.

reset ¶

reset()

Reset the timer, so it starts from the beginning.

resume ¶

resume()

Resume a paused timer.

start ¶

start()

Start the timer.

stop ¶

stop()

Stop the timer.

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.

TyperAsyncCommand ¶

TyperAsyncCommand(*args, **kwargs)

Bases: TyperCommand

Runs an asyncio Typer command.

Example:

@add.command(cls=TyperAsyncCommand)
async def start():
    ...

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.

await_me_maybe `async` ¶

await_me_maybe(callback, *params)

Invoke a callback with an arbitrary number of parameters.

The callback can be a coroutine (async def) or a plain old function. The await_me_maybe awaits the result of the callback if it's an awaitable, or simply returns the result if not.

Parameters:

Name	Type	Description	Default
`callback`	`Callable`	The callable to be invoked.	required

Returns:

Type	Description
`Any`	The return value of the invoked callable.

capture_rich_output ¶

capture_rich_output(obj, width=120)

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
`width`	`int`	the console width to use, None for full width	`120`

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_current_location ¶

get_current_location()

Returns the location where this function is called, i.e. the filename, line number, and function name.

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. If no IP address can be found, None will be returned and the caller can try to use localhost.

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

round_up ¶

round_up(n, decimals=0)

Round a number up to a specified number of decimal places.

This function rounds the input number upward (toward positive infinity) regardless of the value of the digits being rounded. It uses math.ceil() after multiplying by a power of 10 to achieve the specified precision.

Parameters:

Name	Type	Description	Default
`n`	`float or int`	The number to round up.	required
`decimals`	`int`	The number of decimal places to round to. Must be non-negative. Defaults to 0.	`0`

Returns:

Name	Type	Description
`float`		The rounded number with the specified precision.

Examples:

>>> round_up(3.14159, 3)
3.142
>>> round_up(3.1409, 3)
3.141
>>> round_up(-3.14159, 3)
-3.141
>>> round_up(5, 2)
5.0

Note

For negative numbers, "rounding up" means rounding toward zero, so -3.14159 rounded up to 3 decimals is -3.141.

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.

egse.system

egse.system ¶

NOT_SPECIFIED module-attribute ¶

SIGNAL_NAME module-attribute ¶

attrdict module-attribute ¶

AttributeDict ¶

NotSpecified ¶

Periodic ¶

pause ¶

reset ¶

resume ¶

start ¶

stop ¶

Sentinel ¶

SignalCatcher ¶

signal_name property ¶

signal_number property ¶

clear ¶

handler ¶

Timer ¶

get_elapsed ¶

log_elapsed ¶

TyperAsyncCommand ¶

all_logging_disabled ¶

await_me_maybe async ¶

capture_rich_output ¶

chdir ¶

check_argument_type ¶

check_is_a_string ¶

check_str_for_slash ¶

clear_average_execution_times ¶

do_every ¶

duration ¶

env_var ¶

execution_time ¶

filter_by_attr ¶

find_class ¶

flatten_dict ¶

format_datetime ¶

get_active_loggers ¶

get_average_execution_time ¶

get_average_execution_times ¶

get_caller_breadcrumbs ¶

get_caller_info ¶

get_current_location ¶

get_full_classname ¶

get_host_ip ¶

get_module_location ¶

get_os_name ¶

get_os_version ¶

get_package_description ¶

get_package_location ¶

get_referenced_var_name ¶

get_system_architecture ¶

get_system_name ¶

get_system_stats ¶

has_internet ¶

humanize_seconds ¶

ignore_m_warning ¶

is_in ¶

is_in_ipython ¶

is_module ¶

is_namespace ¶

is_not_in ¶

now ¶

ping ¶

read_last_line ¶

read_last_lines ¶

recursive_dict_update ¶

replace_environment_variable ¶

round_up ¶

sanity_check ¶

save_average_execution_time ¶

str_to_datetime ¶

time_in_ms ¶

time_since_epoch_1958 ¶

touch ¶

type_name ¶

wait_until ¶

waiting_for ¶

NOT_SPECIFIED `module-attribute` ¶

SIGNAL_NAME `module-attribute` ¶

attrdict `module-attribute` ¶

signal_name `property` ¶

signal_number `property` ¶

await_me_maybe `async` ¶