isofit.debug.resource_tracker

Classes

`ResourceTracker`	Tracks system resources
`FileResources`	Subclass of ResourceTracker that writes the polled resources to a json list file

Functions

stream(→ dict)

Generator that yields parsed JSONL objects from a growing json file produced by

Module Contents

class ResourceTracker(callback: Callable[[dict], None], interval: float = 2, units: tuple = ('GB', 1024**3), cores: int | Literal['all'] = None, round: bool | int = 2, summarize: bool = True, allow_unsafe: bool = False)[source]

Tracks system resources

Parameters:

callback (Callable[[dict], None]) –
Function to call on each resource refresh. Signature must accept:

callable(dict) -> None

the first call will contain the non-changing values:

used_coresint
Number of CPU cores in use. See the ‘cores’ parameter for more

total_coresint
Total number of cores available on the system via os.cpu_count()

sys_mem_totalfloat
Total memory of the system

mem_unitstr
Unit label that the memory values are in

mem_valuefloat
The value used to convert the bytes to the mem_unit. This may be used to reverse the conversion

poll_intervalfloat
Resource polling interval

timestampfloat
The start timestamp of the resource tracker via time.time()

all calls afterwards will consist of:

pidint
Main process ID

namestr
Main process name

memfloat
Main process memory used

mem_totalfloat
Approximate total memory of the process (actual + shared)

mem_actualfloat
Approximate private memory in use by the process

mem_sharedfloat
Approximate shared memory

cpufloat
Main process CPU percentage over the interval

sys_cpu_per_corelist[float]
Per-core usage percentage over the interval

timestampfloat
Timestamp of the resource record via time.time()

statusstr
Main process status, eg. ‘running’, ‘sleeping’

childrenlist[dict]

Information of child processes:

pidint
Child process ID

namestr
Child process name

mem_totalfloat
Approximate total memory of the child process (actual + shared)

mem_actualfloat
Approximate private memory in use by the child process

mem_sharedfloat
Approximate shared memory

cpufloat
Child process CPU percentage over the interval

statusstr
Child process status, eg. ‘running’, ‘sleeping’

if summarize is enabled, these will also be included:

mem_app_totalfloat
Approximate total memory of the main process + children

mem_app_actualfloat
Approximate total private memory over all processes

mem_app_shared_avgfloat
Approximate average shared memory over all processes

mem_usedfloat
Memory in use by the system

mem_availfloat
Remaining available memory, defined as free + reclaimable

cpu_avgfloat
Average CPU percentage calculated as: sum(main + children) / cores

sys_cpufloat
System-wide CPU percentage over the interval
interval (int | float, default=2) – Interval frequency in seconds to check resources Must be greater than 0. Values less than 0.1 risk high CPU usage and skewing polled results The CPU usage is calculated as the percentage of CPU used over this interval
units (tuple[str, float], default=("GB", 1024**3)) –
Units to convert the memory values to. Must be in the form of (str, float) where the float is used to divide the bytes values that psutil returns Some possible conversions:
- (‘b’, 1/8) # Convert to bits (multiply by 8)
- (‘B’, 1) # No conversion, leave as the default bytes
- (‘KB’, 1024) # Kilobytes
- (‘MB’, 1024**2) # Megabytes
- (‘GB’, 1024**3) # Gigabytes, default
cores (int | 'all', default=1) – Number of cores being used by the source program. This is used for calculating the average CPU percentage. Can be passed ‘all’ to retrieve the os.cpu_count()
round (int | bool, default=2) – Round the memory variables to this many decimals. Set to False or 0 to disable True will be set to 1
summarize (bool, default=True) – Includes summary statistics such as the sum of all children
allow_unsafe (bool, default=False) – Bypasses the exception and allows unsafe interval values (less than 0.1) Not recommended

Examples

Basic usage:

import time
from isofit.debug.resource_tracker import ResourceTracker

def myFunction(info: dict):
    print(info)

rt = ResourceTracker(myFunction)
rt.start()
time.sleep(10)
rt.stop()

info[source]

thread = None[source]

callback[source]

interval = 2[source]

cores = None[source]

round = 2[source]

summarize = True[source]

_track()[source]: System resource tracker intended to be set in a thread

start()[source]: Starts the _track function in a thread

is_running()[source]: Checks if there is a thread running

stop()[source]: Sets the stop event to kill any running threads

class FileResources(file: str, /, reset: bool = False, **kwargs)[source]

Bases: ResourceTracker

Subclass of ResourceTracker that writes the polled resources to a json list file

Parameters:

file (str) – Path to a JSONL file to log resource information to
reset (bool, default=False) – If the file exists, reset it

Examples

Basic usage:

import time
from isofit.debug.resource_tracker import FileResources

fr = ResourceTracker("resources.jsonl", reset=True)
fr.start()
time.sleep(10)
fr.stop()