Skip to content

qoa_utils

qoa4ml.utils.qoa_utils

Functions

convert_to_gbyte(value)

Convert a value from bytes to gigabytes.

Parameters:

Name Type Description Default
value
float

The value in bytes to be converted.

 required

Returns:

Type Description
float

The converted value in gigabytes.

convert_to_kbyte(value)

Convert a value from bytes to kilobytes.

Parameters:

Name Type Description Default
value
float

The value in bytes to be converted.

 required

Returns:

Type Description
float

The converted value in kilobytes.

convert_to_mbyte(value)

Convert a value from bytes to megabytes.

Parameters:

Name Type Description Default
value
float

The value in bytes to be converted.

 required

Returns:

Type Description
float

The converted value in megabytes.

get_cgroup_version()

Retrieve the current cgroup version.

Returns:

Type Description
str

The cgroup version ("v1" or "v2").
Notes

Uses subprocess to execute the mount command and grep for cgroup version.

get_cpu_stat(stats, key)

Retrieve CPU usage statistics from Docker stats.

Parameters:

Name Type Description Default
stats
dict

The Docker stats dictionary.

 required
key
str

The key indicating the type of CPU statistic (e.g., "percentage").

 required

Returns:

Type Description
float

The CPU usage percentage, or -1 if the key is not recognized.
Notes
  • Calculates the CPU usage percentage based on the difference between the current and previous CPU usage.

get_dict_at(dictionary, i=0)

Retrieve the key-value pair at a specific index in a dictionary.

Parameters:

Name Type Description Default
dictionary
dict

The dictionary from which to retrieve the key-value pair.

 required
i
int

The index of the key-value pair to retrieve, default is 0.

 0

Returns:

Type Description
tuple

A tuple containing the key and the value at the specified index.

Raises:

Type Description
IndexError

If the index is out of range.
Notes
  • Logs an error and prints the exception traceback if an error occurs.

get_file_dir(file, to_string=True)

Get the directory of a file.

Parameters:

Name Type Description Default
file
str

The file path.

 required
to_string
bool

Flag to return the directory as a string, default is True.

 True

Returns:

Type Description
str or Path

The directory of the file as a string or Path object.

get_mem_stat(stats, key)

Retrieve memory usage statistics from Docker stats.

Parameters:

Name Type Description Default
stats
dict

The Docker stats dictionary.

 required
key
str

The key indicating the type of memory statistic (e.g., "used").

 required

Returns:

Type Description
int

The memory usage in bytes, or -1 if the key is not recognized.

get_parent_dir(file, parent_level=1, to_string=True)

Get the parent directory of a file by a specified number of levels.

Parameters:

Name Type Description Default
file
str

The file path.

 required
parent_level
int

The number of levels up to retrieve the parent directory, default is 1.

 1
to_string
bool

Flag to return the directory as a string, default is True.

 True

Returns:

Type Description
str or Path

The parent directory of the file as a string or Path object.

get_proc_cpu(pid=None)

Retrieve CPU usage statistics for a given process and its children.

Parameters:

Name Type Description Default
pid
int

The process ID to retrieve CPU stats for. If None, uses the current process ID.

 None

Returns:

Type Description
dict

Dictionary containing CPU stats for the process and its children.
Notes
  • The main process's stats are keyed by its PID.
  • Each child process's stats are keyed by its PID with a "c" suffix.

get_proc_mem(pid=None)

Retrieve memory usage statistics for a given process and its children.

Parameters:

Name Type Description Default
pid
int

The process ID to retrieve memory stats for. If None, uses the current process ID.

 None

Returns:

Type Description
dict

Dictionary containing memory stats for the process and its children.
Notes
  • The main process's stats are keyed by its PID.
  • Each child process's stats are keyed by its PID with a "c" suffix.

get_process_allowed_cpus()

Retrieve the list of CPU cores available to the process.

Returns:

Type Description
list[int]

A list of CPU core indices.
Notes
  • Uses the call process's PID (0) to get the CPU affinity.

get_process_allowed_memory()

Retrieve the memory limit allowed to the process.

Returns:

Type Description
Optional[float]

The memory limit in bytes, or None if unable to retrieve.
Notes
  • Supports both cgroup v1 and v2 formats to get the memory limit.

get_sys_cpu()

Retrieve system CPU statistics and times.

Returns:

Type Description
dict

Dictionary containing CPU stats and times.
Notes

Uses psutil to retrieve both CPU stats and times.

get_sys_cpu_metadata()

Retrieve metadata information about the system CPU.

Returns:

Type Description
dict

Dictionary containing CPU frequency and thread count.

get_sys_cpu_util()

Retrieve system CPU utilization for each core.

Returns:

Type Description
dict

Dictionary containing CPU utilization for each core.

get_sys_mem()

Retrieve system memory statistics.

Returns:

Type Description
dict

Dictionary containing memory stats.

get_sys_net()

Retrieve system network I/O statistics.

Returns:

Type Description
dict

Dictionary containing network I/O stats.

is_numpyarray(obj)

Check if an object is a NumPy array.

Parameters:

Name Type Description Default
obj
Any

The object to check.

 required

Returns:

Type Description
bool

True if the object is a NumPy array, False otherwise.

load_config(file_path)

Load a configuration file.

Parameters:

Name Type Description Default
file_path
str

The path to the configuration file.

 required

Returns:

Type Description
dict

The loaded configuration dictionary.
Notes

Supports JSON and YAML file formats. Logs a warning if the format is unsupported.

make_folder(temp_path)

Create a folder if it doesn't already exist.

Parameters:

Name Type Description Default
temp_path
str

The path of the folder to be created.

 required

Returns:

Type Description
bool

True if the folder exists or is created successfully, False otherwise.
Notes

If the folder already exists, nothing is done.

merge_report(f_report, i_report, prio=True)

Merge two report dictionaries.

Parameters:

Name Type Description Default
f_report
dict

The first report dictionary.

 required
i_report
dict

The second report dictionary.

 required
prio
bool

Flag to determine which report takes priority in case of conflict, default is True.

 True

Returns:

Type Description
dict

The merged report dictionary.
Notes
  • If both reports are dictionaries, merges them recursively.
  • If there is a conflict and prio is True, the value from f_report is used; otherwise, the value from i_report is used.

report_proc_child_cpu(process)

Retrieve CPU usage statistics for a given process and its children.

Parameters:

Name Type Description Default
process
Process

The process to retrieve CPU stats for.

 required

Returns:

Type Description
dict

Dictionary containing CPU stats for the process and its children.
Notes

This function can be time-consuming as it recursively evaluates all child processes.

report_proc_cpu(process)

Retrieve CPU usage statistics for a given process.

Parameters:

Name Type Description Default
process
Process

The process to retrieve CPU stats for.

 required

Returns:

Type Description
dict

Dictionary containing CPU stats for the process.

report_proc_mem(process)

Retrieve memory usage statistics for a given process.

Parameters:

Name Type Description Default
process
Process

The process to retrieve memory stats for.

 required

Returns:

Type Description
dict

Dictionary containing memory stats for the process.

set_logger_level(logging_level)

Set the logging level for the application logger.

Parameters:

Name Type Description Default
logging_level
int

The desired logging level: 0 - NOTSET 1 - DEBUG 2 - INFO 3 - WARNING 4 - ERROR 5 - CRITICAL

 required

Raises:

Type Description
ValueError

If the logging level is not between 0 and 5.

sys_monitor(client, interval)

Start monitoring system reports.

Parameters:

Name Type Description Default
client
Any

The client to send the report to.

 required
interval
int

The time interval in seconds between reports.

 required
Notes
  • Starts a new thread to generate system reports at regular intervals.

system_report(client, interval, to_mb=True, to_gb=False, to_kb=False)

Generate a system report and send it to the client at regular intervals.

Parameters:

Name Type Description Default
client
Any

The client to send the report to.

 required
interval
int

The time interval in seconds between reports.

 required
to_mb
bool

Convert network I/O values to megabytes, default is True.

 True
to_gb
bool

Convert network I/O values to gigabytes, default is False.

 False
to_kb
bool

Convert network I/O values to kilobytes, default is False.

 False
Notes
  • Collects CPU, memory, and network I/O statistics.
  • Logs errors if any occur during data collection or report sending.
  • Sleeps for the specified interval between reports.

to_json(file_path, conf)

Save a configuration to a JSON file.

Parameters:

Name Type Description Default
file_path
str

The path to the file where the configuration should be saved.

 required
conf
dict

The configuration dictionary to save.

 required

to_yaml(file_path, conf)

Save a configuration to a YAML file.

Parameters:

Name Type Description Default
file_path
str

The path to the file where the configuration should be saved.

 required
conf
dict

The configuration dictionary to save.

 required