Utilities¶

Some useful utilities abstracting or enriching functionality of the main source.

types ¶

Classes:

Name	Description
`AudioCodec`	Object specifying audio codec options for FFmpeg.
`AudioFormatEnum`	Audio format enumeration for supported FFmpeg video formats.
`LoggingSpec`	Object specifying data storage options.
`VideoCodec`	Object specifying video codec options for FFmpeg.
`VideoFormatEnum`	Video format enumeration for supported FFmpeg video formats.

AudioCodec `dataclass` ¶

Object specifying audio codec options for FFmpeg.

AudioFormatEnum ¶

Bases: Enum


              flowchart TD
              hermes.utils.types.AudioFormatEnum[AudioFormatEnum]

              

              click hermes.utils.types.AudioFormatEnum href "" "hermes.utils.types.AudioFormatEnum"

Audio format enumeration for supported FFmpeg video formats.

TODO: Must be a tuple of (, ...), where: write format is one of: ffmpeg -formats

LoggingSpec `dataclass` ¶

Object specifying data storage options.

Parameters:

Name	Type	Description	Default
`log_dir` ¶	`str`	Path to the directory on disk to flush data to.	required
`experiment` ¶	`dict[str, str]`	Nested setup definition of Nodes across distributed hosts.	required
`log_time_s` ¶	`float`	Start time of saving data.	required
`ref_time_s` ¶	`float`	Reference time of the Broker to align all Nodes to.	required
`stream_period_s` ¶	`float`	Duration of periods over which to flush streamed accumulated data from memory to disk. Defaults to `30.0`.	`30.0`
`is_quiet` ¶	`bool`	Whether to print FFmpeg stats to the terminal. Defaults to `False`.	`False`
`stream_hdf5` ¶	`bool`	Whether to stream data into HDF5 files. Defaults to `False`.	`False`
`stream_video` ¶	`bool`	Whether to stream video data into MP4/MKV files. Defaults to `False`.	`False`
`stream_csv` ¶	`bool`	Whether to stream data into CSV files. Defaults to `False`.	`False`
`stream_audio` ¶	`bool`	Whether to stream audio data into MP3/WAV files. Defaults to `False`.	`False`
`dump_csv` ¶	`bool`	Weather to dump in-memory recorded data in CSV files. Defaults to `False`.	`False`
`dump_hdf5` ¶	`bool`	Weather to dump in-memory recorded data in HDF5 files. Defaults to `False`.	`False`
`dump_video` ¶	`bool`	Weather to dump in-memory recorded video data in MP4/MKV files. Defaults to `False`.	`False`
`dump_audio` ¶	`bool`	Weather to dump in-memory recorded audio data in MP3/WAV files. Defaults to `False`.	`False`
`video_codec` ¶	`VideoCodec`	Definition of the video codec to use for FFmpeg. Defaults to `None`.	`None`
`audio_codec` ¶	`AudioCodec`	Definition of the audio codec to use for FFmpeg. Defaults to `None`.	`None`

VideoCodec `dataclass` ¶

Object specifying video codec options for FFmpeg.

VideoFormatEnum ¶

Bases: Enum


              flowchart TD
              hermes.utils.types.VideoFormatEnum[VideoFormatEnum]

              

              click hermes.utils.types.VideoFormatEnum href "" "hermes.utils.types.VideoFormatEnum"

Video format enumeration for supported FFmpeg video formats.

Must be a tuple of (, ), where: write format is one of: ffmpeg -formats pixel color is one of: ffmpeg -pix_fmts

time_utils ¶

Classes:

Name	Description
`SystemTime`	Highly accurate system time Singleton that uses performance counters of the host device.

Functions:

Name	Description
`get_ref_time`	Gets the current process's `SystemTime` Singleton reference time.
`get_time`	Gets the highly accurate current system time.
`get_time_s_from_local_str`	Gets seconds since epoch from provided local time string.
`get_time_s_from_utc_time_no_date_str`	Gets local time in seconds since epoch from provided UTC time strings.
`get_time_str`	Gets a date string from seconds since epoch.
`init_time`	Initialize the current process's `SystemTime` Singleton with a common reference time.

SystemTime ¶

Highly accurate system time Singleton that uses performance counters of the host device.

Performance counters of the host devices are not Unix float numbers since epoch, but a float starting from 0 on the boot-up of the device. This Singleton uses less accurate system time with the time.time() call and combines with the performance counters, to get highly accurate time of the host device.

This Singleton's methods should not be called directly. Use the global methods instead.

Each subprocess of the application must be initialized via init_time(...) with a reference time obtained in the parent process via get_ref_time().

get_ref_time ¶

get_ref_time() -> float

Gets the current process's SystemTime Singleton reference time.

get_time ¶

get_time() -> float

Gets the highly accurate current system time.

get_time_s_from_local_str ¶

get_time_s_from_local_str(
    time_local_str: str,
    input_time_format: str = "%H:%M:%S.%f",
    date_local_str: str | None = None,
    input_date_format: str = "%Y-%m-%d",
) -> float

Gets seconds since epoch from provided local time string.

Parameters:

Name	Type	Description	Default
`time_local_str` ¶	`str`	local time string to convert to seconds.	required
`input_time_format` ¶	`str`	Time string format of the provided time. Defaults to `%H:%M:%S.%f`.	`'%H:%M:%S.%f'`
`date_local_str` ¶	`str`	local date string to convert to seconds. Defaults to `None`.	`None`
`input_date_format` ¶	`str`	Date string format of the provided date. Defaults to `%Y-%m-%d`.	`'%Y-%m-%d'`

Returns:

Name	Type	Description
`float`	`float`	Seconds count since epoch, corresponding to the provided local time strings.

get_time_s_from_utc_time_no_date_str ¶

get_time_s_from_utc_time_no_date_str(
    time_utc_str: str,
    input_time_format: str = "%H:%M:%S.%f",
    date_utc_str: str | None = None,
    input_date_format: str = "%Y-%m-%d",
) -> float

Gets local time in seconds since epoch from provided UTC time strings.

Parameters:

Name	Type	Description	Default
`time_utc_str` ¶	`str`	UTC time string to convert to seconds.	required
`input_time_format` ¶	`str`	Time string format of the provided time. Defaults to `%H:%M:%S.%f`.	`'%H:%M:%S.%f'`
`date_utc_str` ¶	`str`	UTC date to convert to seconds. Defaults to `None`.	`None`
`input_date_format` ¶	`str`	Date string format of the provided date. Defaults to `%Y-%m-%d`.	`'%Y-%m-%d'`

Returns:

Name	Type	Description
`float`	`float`	Seconds count since epoch, corresponding to the provided UTC time strings.

get_time_str ¶

get_time_str(time_s: float = get_time(), format: str = '%Y-%m-%d_%H-%M-%S') -> str

Gets a date string from seconds since epoch.

Parameters:

Name	Type	Description	Default
`time_s` ¶	`float`	Time since epoch to convert to the human-readable string. Defaults to the current system time.	`get_time()`
`format` ¶	`str`	Format to construct the date string. Default to `%Y-%m-%d_%H-%M-%S`.	`'%Y-%m-%d_%H-%M-%S'`

Returns:

Name	Type	Description
`str`	`str`	Formatted date string corresponding to the provided time since epoch.

init_time ¶

init_time(ref_time: float) -> None

Initialize the current process's SystemTime Singleton with a common reference time.

Parameters:

Name	Type	Description	Default
`ref_time` ¶	`float`	Time obtained via `get_ref_time()` in the parent process to use as a reference for performance counters in the current process.	required

node_utils ¶

Functions:

Name	Description
`launch_node`	Launches callable `Node` objects using the user-provided specification.
`search_node_class`	Queries the current Python environment to match the requested `hermes.<module>`.

launch_node ¶

launch_node(spec: dict, input_queue: Queue[tuple[float, str]])

Launches callable Node objects using the user-provided specification.

Parameters:

Name	Type	Description	Default
`spec` ¶	`dict`	Specification containing at least package and `Node` names, and constructor arguments specific to that `Node`.	required
`input_queue` ¶	`Queue[tuple[float, str]]`	Multiprocessing queue to fan-in user keyboard inputs if the `Node` is interested to receive any.	required

search_node_class ¶

search_node_class(module_name: str, class_name: str) -> type[NodeInterface]

Queries the current Python environment to match the requested hermes.<module>.

Parameters:

Name	Type	Description	Default
`module_name` ¶	`str`	Name of the Python module containing the requested HERMES type `Node`.	required
`class_name` ¶	`str`	Name of the `Node` in the provided module to retrieve for construction.	required

mp_utils ¶

Functions:

Name	Description
`launch_broker`	Builds the `Broker` using provided configurations and manage all the components of the experiment.
`launch_callable`	Launches a callable object with the user-provided inputs.

launch_broker ¶

launch_broker(
    args: Namespace,
    node_specs: list[dict],
    input_queue: Queue[tuple[float, str]],
    is_ready_event: Event,
    is_quit_event: Event,
    is_done_event: Event,
    ref_time_s: float,
) -> None

Builds the Broker using provided configurations and manage all the components of the experiment.

Meant to be used as a target for a spawned Process or Thread.

Parameters:

Name	Type	Description	Default
`args` ¶	`Namespace`	HERMES top-level user input arguments for the experiment definition.	required
`node_specs` ¶	`list[dict]`	Configuration dictionaries of the Broker's locally managed `Node`s.	required
`input_queue` ¶	`Queue[tuple[float, str]]`	Multiprocessing queue to fan-out user keyboard inputs to all `Broker`s locally managed `Node`s.	required
`is_ready_event` ¶	`Event`	Multiprocessing synchronization primitive indicating completion of `Broker`s setup and handshaking with `Node`s.	required
`is_quit_event` ¶	`Event`	Multiprocessing synchronization primitive triggering to the `Broker` to gracefully wrap up and end.	required
`is_done_event` ¶	`Event`	Multiprocessing synchronization primitive indicating that the `Broker` is finished and experiment ended.	required
`ref_time_s` ¶	`float`	Main process reference system time obtained with `get_ref_time()` to use in all child processes for syncing the data.	required

launch_callable ¶

launch_callable(obj: Callable, *args, **kwargs) -> None

Launches a callable object with the user-provided inputs.

Parameters:

Name	Type	Description	Default
`args` ¶	`list`	Ordered unnamed inputs to provide to the callable object.	`()`
`kwargs` ¶	`dict`	Named key-value inputs to provide to the callable object.	`{}`

msgpack_utils ¶

Functions:

Name	Description
`convert_bytes_keys_to_strings`	Recursively decodes keys of the key-value mappings into strings.
`decode_ndarray`	Decodes received bytes and reconstructs detected NumPy arrays into original objects.
`deserialize`	Deserializes the received message to construct the original Python object.
`encode_ndarray`	Encodes NumPy contents of the provided object into bytes.
`serialize`	Serializes a Python dict-like object for ZeroMQ transmission.

convert_bytes_keys_to_strings ¶

convert_bytes_keys_to_strings(obj: object) -> object

Recursively decodes keys of the key-value mappings into strings.

Parameters:

Name	Type	Description	Default
`obj` ¶	`object`	Object whose keys to decode.	required

Returns:

Name	Type	Description
`object`	`object`	Processed object with keys converted into proper dictionary string fields.

decode_ndarray ¶

decode_ndarray(obj: object) -> object

Decodes received bytes and reconstructs detected NumPy arrays into original objects.

Parameters:

Name	Type	Description	Default
`obj` ¶	`object`	Raw bytes to convert into NumPy arrays if any detected.	required

Returns:

Name	Type	Description
`object`	`object`	Decoded key-value mapping with NumPy arrays reconstructed.

deserialize ¶

deserialize(msg: bytes) -> dict

Deserializes the received message to construct the original Python object.

Parameters:

Name	Type	Description	Default
`msg` ¶	`bytes`	Raw binary data containing the original Python object.	required

Returns:

Name	Type	Description
`dict`	`dict`	Python object with the key-value pairs preserved and any NumPy arrays reconstructed.

encode_ndarray ¶

encode_ndarray(obj: object) -> object

Encodes NumPy contents of the provided object into bytes.

Parameters:

Name	Type	Description	Default
`obj` ¶	`object`	NumPy array to encode as serializeable key-value pair.	required

Returns:

Name	Type	Description
`object`	`object`	Encoded key-value mapping with NumPy arrays serialized to bytes.

serialize ¶

serialize(**kwargs) -> bytes

Serializes a Python dict-like object for ZeroMQ transmission.

Preserves named arguments as key-value pairs.

Parameters:

Name	Type	Description	Default
`kwargs` ¶	`dict`	Inputs to serialize using a custom encoding hook for NumPy arrays.	`{}`

Returns:

Name	Type	Description
`bytes`	`bytes`	Serialized binary data safe to transmit.

argparse_utils ¶

Classes:

Name	Description
`ParseExperimentKwargs`	Parsing object for experiment details specification.
`ParseNodeKwargs`	Parsing object for `Node` specifications.
`ParseStorageKwargs`	Parsing object for `Storage` specification.

Functions:

Name	Description
`parse_type`	Parsing utility to convert data types to native Python objects.
`validate_ip4`	Validates whether parsed value is a valid IPv4 address.
`validate_path`	Validates whether parsed file path exists.

ParseExperimentKwargs ¶

Bases: Action


              flowchart TD
              hermes.utils.argparse_utils.ParseExperimentKwargs[ParseExperimentKwargs]

              

              click hermes.utils.argparse_utils.ParseExperimentKwargs href "" "hermes.utils.argparse_utils.ParseExperimentKwargs"

Parsing object for experiment details specification.

ParseNodeKwargs ¶

Bases: Action


              flowchart TD
              hermes.utils.argparse_utils.ParseNodeKwargs[ParseNodeKwargs]

              

              click hermes.utils.argparse_utils.ParseNodeKwargs href "" "hermes.utils.argparse_utils.ParseNodeKwargs"

Parsing object for Node specifications.

ParseStorageKwargs ¶

Bases: Action


              flowchart TD
              hermes.utils.argparse_utils.ParseStorageKwargs[ParseStorageKwargs]

              

              click hermes.utils.argparse_utils.ParseStorageKwargs href "" "hermes.utils.argparse_utils.ParseStorageKwargs"

Parsing object for Storage specification.

parse_type ¶

parse_type(s: str) -> int | float | bool | str

Parsing utility to convert data types to native Python objects.

Parameters:

Name	Type	Description	Default
`s` ¶	`str`	Text to convert to the corresponding data type.	required

Returns:

Type	Description
`int \| float \| bool \| str`	int \| float \| bool \| str: Parsed primitive data type.

validate_ip4 ¶

validate_ip4(s: str) -> str

Validates whether parsed value is a valid IPv4 address.

Parameters:

Name	Type	Description	Default
`s` ¶	`str`	Parsed text value to test for validity as IPv4 address.	required

Returns:

Name	Type	Description
`str`	`str`	Unchanged text value to use as IP address.

Raises:

Type	Description
`ArgumentTypeError`	If the parsed value is not a valid IPv4 address.

validate_path ¶

validate_path(s: str) -> str

Validates whether parsed file path exists.

Parameters:

Name	Type	Description	Default
`s` ¶	`str`	File path to parse.	required

Returns:

Name	Type	Description
`str`	`str`	Unchanged path if it exists.

Raises:

Type	Description
`ArgumentTypeError`	If the provided path does not exist.

sensor_utils ¶

Functions:

Name	Description
`estimate_transmission_delay`	Estimates the mean transmission delay of the provided "ping" function.

estimate_transmission_delay ¶

estimate_transmission_delay(ping_fn: Callable, num_samples: int = 100) -> float

Estimates the mean transmission delay of the provided "ping" function.

Used with the custom provided ping_fn to measure the one-way delay to use for offsetting sensor data streams from the toa_s to obtain true sample time.

Parameters:

Name	Type	Description	Default
`ping_fn` ¶	`Callable`	User's function that wraps "ping" like functionality to the specific device.	required
`num_samples` ¶	`int`	Number of round-trip transmission to perform to average over. Defaults to `100`.	`100`

Returns:

Name	Type	Description
`float`	`float`	Estimated mean delay for samples from the device.

numpy_utils ¶

dict_utils ¶

angle_utils ¶

print_utils ¶

Classes:

Name	Description
`SystemTime`	Highly accurate system time Singleton that uses performance counters of the host device.

Functions:

Name	Description
`get_ref_time`	Gets the current process's `SystemTime` Singleton reference time.
`get_time`	Gets the highly accurate current system time.
`get_time_s_from_local_str`	Gets seconds since epoch from provided local time string.
`get_time_s_from_utc_time_no_date_str`	Gets local time in seconds since epoch from provided UTC time strings.
`get_time_str`	Gets a date string from seconds since epoch.
`init_time`	Initialize the current process's `SystemTime` Singleton with a common reference time.

SystemTime ¶

Highly accurate system time Singleton that uses performance counters of the host device.

Performance counters of the host devices are not Unix float numbers since epoch, but a float starting from 0 on the boot-up of the device. This Singleton uses less accurate system time with the time.time() call and combines with the performance counters, to get highly accurate time of the host device.

This Singleton's methods should not be called directly. Use the global methods instead.

Each subprocess of the application must be initialized via init_time(...) with a reference time obtained in the parent process via get_ref_time().

get_ref_time ¶

get_ref_time() -> float

Gets the current process's SystemTime Singleton reference time.

get_time ¶

get_time() -> float

Gets the highly accurate current system time.

get_time_s_from_local_str ¶

get_time_s_from_local_str(
    time_local_str: str,
    input_time_format: str = "%H:%M:%S.%f",
    date_local_str: str | None = None,
    input_date_format: str = "%Y-%m-%d",
) -> float

Gets seconds since epoch from provided local time string.

Parameters:

Name	Type	Description	Default
`time_local_str` ¶	`str`	local time string to convert to seconds.	required
`input_time_format` ¶	`str`	Time string format of the provided time. Defaults to `%H:%M:%S.%f`.	`'%H:%M:%S.%f'`
`date_local_str` ¶	`str`	local date string to convert to seconds. Defaults to `None`.	`None`
`input_date_format` ¶	`str`	Date string format of the provided date. Defaults to `%Y-%m-%d`.	`'%Y-%m-%d'`

Returns:

Name	Type	Description
`float`	`float`	Seconds count since epoch, corresponding to the provided local time strings.

get_time_s_from_utc_time_no_date_str ¶

get_time_s_from_utc_time_no_date_str(
    time_utc_str: str,
    input_time_format: str = "%H:%M:%S.%f",
    date_utc_str: str | None = None,
    input_date_format: str = "%Y-%m-%d",
) -> float

Gets local time in seconds since epoch from provided UTC time strings.

Parameters:

Name	Type	Description	Default
`time_utc_str` ¶	`str`	UTC time string to convert to seconds.	required
`input_time_format` ¶	`str`	Time string format of the provided time. Defaults to `%H:%M:%S.%f`.	`'%H:%M:%S.%f'`
`date_utc_str` ¶	`str`	UTC date to convert to seconds. Defaults to `None`.	`None`
`input_date_format` ¶	`str`	Date string format of the provided date. Defaults to `%Y-%m-%d`.	`'%Y-%m-%d'`

Returns:

Name	Type	Description
`float`	`float`	Seconds count since epoch, corresponding to the provided UTC time strings.

get_time_str ¶

get_time_str(time_s: float = get_time(), format: str = '%Y-%m-%d_%H-%M-%S') -> str

Gets a date string from seconds since epoch.

Parameters:

Name	Type	Description	Default
`time_s` ¶	`float`	Time since epoch to convert to the human-readable string. Defaults to the current system time.	`get_time()`
`format` ¶	`str`	Format to construct the date string. Default to `%Y-%m-%d_%H-%M-%S`.	`'%Y-%m-%d_%H-%M-%S'`

Returns:

Name	Type	Description
`str`	`str`	Formatted date string corresponding to the provided time since epoch.

init_time ¶

init_time(ref_time: float) -> None

Initialize the current process's SystemTime Singleton with a common reference time.

Parameters:

Name	Type	Description	Default
`ref_time` ¶	`float`	Time obtained via `get_ref_time()` in the parent process to use as a reference for performance counters in the current process.	required

Utilities¶

types ¶

AudioCodec dataclass ¶

AudioFormatEnum ¶

LoggingSpec dataclass ¶

log_dir ¶

experiment ¶

log_time_s ¶

ref_time_s ¶

stream_period_s ¶

is_quiet ¶

stream_hdf5 ¶

stream_video ¶

stream_csv ¶

stream_audio ¶

dump_csv ¶

dump_hdf5 ¶

dump_video ¶

dump_audio ¶

video_codec ¶

audio_codec ¶

VideoCodec dataclass ¶

VideoFormatEnum ¶

time_utils ¶

SystemTime ¶

get_ref_time ¶

get_time ¶

get_time_s_from_local_str ¶

time_local_str ¶

input_time_format ¶

date_local_str ¶

input_date_format ¶

get_time_s_from_utc_time_no_date_str ¶

time_utc_str ¶

input_time_format ¶

date_utc_str ¶

input_date_format ¶

get_time_str ¶

time_s ¶

format ¶

init_time ¶

ref_time ¶

node_utils ¶

launch_node ¶

spec ¶

input_queue ¶

search_node_class ¶

module_name ¶

class_name ¶

mp_utils ¶

launch_broker ¶

args ¶

node_specs ¶

input_queue ¶

is_ready_event ¶

is_quit_event ¶

is_done_event ¶

ref_time_s ¶

launch_callable ¶

args ¶

kwargs ¶

msgpack_utils ¶

convert_bytes_keys_to_strings ¶

obj ¶

decode_ndarray ¶

obj ¶

deserialize ¶

msg ¶

encode_ndarray ¶

obj ¶

serialize ¶

kwargs ¶

argparse_utils ¶

ParseExperimentKwargs ¶

ParseNodeKwargs ¶

ParseStorageKwargs ¶

parse_type ¶

s ¶

validate_ip4 ¶

s ¶

AudioCodec `dataclass` ¶

LoggingSpec `dataclass` ¶

`log_dir` ¶

`experiment` ¶

`log_time_s` ¶

`ref_time_s` ¶

`stream_period_s` ¶

`is_quiet` ¶

`stream_hdf5` ¶

`stream_video` ¶

`stream_csv` ¶

`stream_audio` ¶

`dump_csv` ¶

`dump_hdf5` ¶

`dump_video` ¶

`dump_audio` ¶

`video_codec` ¶

`audio_codec` ¶

VideoCodec `dataclass` ¶

`time_local_str` ¶

`input_time_format` ¶

`date_local_str` ¶

`input_date_format` ¶

`time_utc_str` ¶

`input_time_format` ¶

`date_utc_str` ¶

`input_date_format` ¶

`time_s` ¶

`format` ¶

`ref_time` ¶

`spec` ¶

`input_queue` ¶

`module_name` ¶

`class_name` ¶

`args` ¶

`node_specs` ¶

`input_queue` ¶

`is_ready_event` ¶

`is_quit_event` ¶

`is_done_event` ¶

`ref_time_s` ¶

`args` ¶

`kwargs` ¶

`obj` ¶

`obj` ¶

`msg` ¶

`obj` ¶

`kwargs` ¶

`s` ¶

`s` ¶

`s` ¶

`ping_fn` ¶

`num_samples` ¶

`time_local_str` ¶

`input_time_format` ¶

`date_local_str` ¶

`input_date_format` ¶

`time_utc_str` ¶

`input_time_format` ¶

`date_utc_str` ¶

`input_date_format` ¶

`time_s` ¶

`format` ¶

`ref_time` ¶