local_process_cpu_utilization

CPU utilization measurement for local training processes.

CPUStatistics

Bases: CommonStatistics

The CPUStatistics class encapsulates data and functionality for tracking and updating CPU utilization statistics for a running process.

Source code in mlte/measurement/cpu/local_process_cpu_utilization.py

class CPUStatistics(CommonStatistics):
    """
    The CPUStatistics class encapsulates data
    and functionality for tracking and updating
    CPU utilization statistics for a running process.
    """

    DEFAULT_UNIT = Units.percent

    def __init__(
        self,
        avg: float,
        min: float,
        max: float,
        unit: Unit = DEFAULT_UNIT,
    ):
        """
        Initialize a CPUStatistics instance.

        :param avg: The average utilization
        :param min: The minimum utilization
        :param max: The maximum utilization
        :param unit: the unit the values comes in, as a value from Units; defaults to Units.percent
        """
        super().__init__(avg, min, max, unit)

__init__(avg, min, max, unit=DEFAULT_UNIT)

Initialize a CPUStatistics instance.

Parameters:

Name	Type	Description	Default
avg	float	The average utilization	required
min	float	The minimum utilization	required
max	float	The maximum utilization	required
unit	Unit	the unit the values comes in, as a value from Units; defaults to Units.percent	DEFAULT_UNIT

Source code in mlte/measurement/cpu/local_process_cpu_utilization.py

def __init__(
    self,
    avg: float,
    min: float,
    max: float,
    unit: Unit = DEFAULT_UNIT,
):
    """
    Initialize a CPUStatistics instance.

    :param avg: The average utilization
    :param min: The minimum utilization
    :param max: The maximum utilization
    :param unit: the unit the values comes in, as a value from Units; defaults to Units.percent
    """
    super().__init__(avg, min, max, unit)

LocalProcessCPUUtilization

Bases: ProcessMeasurement

Measures CPU utilization for a local process.

Source code in mlte/measurement/cpu/local_process_cpu_utilization.py

class LocalProcessCPUUtilization(ProcessMeasurement):
    """Measures CPU utilization for a local process."""

    def __init__(
        self, identifier: Optional[str] = None, group: Optional[str] = None
    ):
        """
        Initialize a new LocalProcessCPUUtilization measurement.

        :param identifier: A unique identifier for the measurement
        :param group: An optional group id, if we want to group this measurement with others.
        """
        super().__init__(identifier, group)
        if is_windows():
            raise RuntimeError(
                f"Measurement for {self.evidence_metadata.test_case_id if self.evidence_metadata else 'this'} is not supported on Windows."
            )

    # Overriden.
    def __call__(self, pid: int, poll_interval: int = 1) -> CPUStatistics:
        """
        Monitor the CPU utilization of process at `pid` until exit.

        :param pid: The process identifier
        :param poll_interval: The poll interval in seconds

        :return: The collection of CPU usage statistics
        """
        stats = []
        while True:
            util = _get_cpu_usage(pid)
            if util < 0.0:
                break
            stats.append(util / 100.0)
            time.sleep(poll_interval)

        return CPUStatistics(
            avg=sum(stats) / len(stats),
            min=min(stats),
            max=max(stats),
            unit=Units.percent,
        )

    # Overriden.
    @classmethod
    def get_output_type(cls) -> type[CPUStatistics]:
        return CPUStatistics

__call__(pid, poll_interval=1)

Monitor the CPU utilization of process at pid until exit.

Parameters:

Name	Type	Description	Default
pid	int	The process identifier	required
poll_interval	int	The poll interval in seconds	1

Returns:

Type	Description
CPUStatistics	The collection of CPU usage statistics

Source code in mlte/measurement/cpu/local_process_cpu_utilization.py

def __call__(self, pid: int, poll_interval: int = 1) -> CPUStatistics:
    """
    Monitor the CPU utilization of process at `pid` until exit.

    :param pid: The process identifier
    :param poll_interval: The poll interval in seconds

    :return: The collection of CPU usage statistics
    """
    stats = []
    while True:
        util = _get_cpu_usage(pid)
        if util < 0.0:
            break
        stats.append(util / 100.0)
        time.sleep(poll_interval)

    return CPUStatistics(
        avg=sum(stats) / len(stats),
        min=min(stats),
        max=max(stats),
        unit=Units.percent,
    )

__init__(identifier=None, group=None)

Initialize a new LocalProcessCPUUtilization measurement.

Parameters:

Name	Type	Description	Default
identifier	Optional[str]	A unique identifier for the measurement	None
group	Optional[str]	An optional group id, if we want to group this measurement with others.	None

Source code in mlte/measurement/cpu/local_process_cpu_utilization.py

def __init__(
    self, identifier: Optional[str] = None, group: Optional[str] = None
):
    """
    Initialize a new LocalProcessCPUUtilization measurement.

    :param identifier: A unique identifier for the measurement
    :param group: An optional group id, if we want to group this measurement with others.
    """
    super().__init__(identifier, group)
    if is_windows():
        raise RuntimeError(
            f"Measurement for {self.evidence_metadata.test_case_id if self.evidence_metadata else 'this'} is not supported on Windows."
        )