Skip to content

local_process_cpu_utilization

mlte/measurement/cpu/local_process_cpu_utilization.py

CPU utilization measurement for local training processes.

CPUStatistics

Bases: ValueBase

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

Source code in mlte/measurement/cpu/local_process_cpu_utilization.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
class CPUStatistics(ValueBase):
    """
    The CPUStatistics class encapsulates data
    and functionality for tracking and updating
    CPU consumption statistics for a running process.
    """

    def __init__(
        self,
        evidence_metadata: EvidenceMetadata,
        avg: float,
        min: float,
        max: float,
    ):
        """
        Initialize a CPUStatistics instance.

        :param evidence_metadata: The generating measurement's metadata
        :param avg: The average utilization
        :param min: The minimum utilization
        :param max: The maximum utilization
        """
        super().__init__(self, evidence_metadata)

        self.avg = avg
        """The average CPU utilization, as a proportion."""

        self.min = min
        """The minimum CPU utilization, as a proportion."""

        self.max = max
        """The maximum CPU utilization, as a proportion."""

    def serialize(self) -> Dict[str, Any]:
        """
        Serialize an CPUStatistics to a JSON object.

        :return: The JSON object
        """
        return {"avg": self.avg, "min": self.min, "max": self.max}

    @staticmethod
    def deserialize(
        evidence_metadata: EvidenceMetadata, data: Dict[str, Any]
    ) -> CPUStatistics:
        """
        Deserialize an CPUStatistics from a JSON object.

        :param evidence_metadata: The generating measurement's metadata
        :param data: The JSON object

        :return: The deserialized instance
        """
        return CPUStatistics(
            evidence_metadata,
            avg=data["avg"],
            min=data["min"],
            max=data["max"],
        )

    def __str__(self) -> str:
        """Return a string representation of CPUStatistics."""
        s = ""
        s += f"Average: {self.avg:.2f}%\n"
        s += f"Minimum: {self.min:.2f}%\n"
        s += f"Maximum: {self.max:.2f}%"
        return s

    @classmethod
    def max_utilization_less_than(cls, threshold: float) -> Condition:
        """
        Construct and invoke a condition for maximum CPU utilization.

        :param threshold: The threshold value for maximum utilization, as percentage

        :return: The Condition that can be used to validate a Value.
        """
        condition: Condition = Condition.build_condition(
            bool_exp=lambda stats: stats.max < threshold,
            success=f"Maximum utilization below threshold {threshold:.2f}",
            failure=f"Maximum utilization exceeds threshold {threshold:.2f}",
        )
        return condition

    @classmethod
    def average_utilization_less_than(cls, threshold: float) -> Condition:
        """
        Construct and invoke a condition for average CPU utilization.

        :param threshold: The threshold value for average utilization, as percentage

        :return: The Condition that can be used to validate a Value.
        """
        condition: Condition = Condition.build_condition(
            bool_exp=lambda stats: stats.avg < threshold,
            success=f"Average utilization below threshold {threshold:.2f}",
            failure=f"Average utilization exceeds threshold {threshold:.2f}",
        )
        return condition

avg = avg instance-attribute

The average CPU utilization, as a proportion.

max = max instance-attribute

The maximum CPU utilization, as a proportion.

min = min instance-attribute

The minimum CPU utilization, as a proportion.

__init__(evidence_metadata, avg, min, max)

Initialize a CPUStatistics instance.

Parameters:

Name Type Description Default
evidence_metadata EvidenceMetadata

The generating measurement's metadata

 required
avg float

The average utilization

 required
min float

The minimum utilization

 required
max float

The maximum utilization

 required
Source code in mlte/measurement/cpu/local_process_cpu_utilization.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def __init__(
    self,
    evidence_metadata: EvidenceMetadata,
    avg: float,
    min: float,
    max: float,
):
    """
    Initialize a CPUStatistics instance.

    :param evidence_metadata: The generating measurement's metadata
    :param avg: The average utilization
    :param min: The minimum utilization
    :param max: The maximum utilization
    """
    super().__init__(self, evidence_metadata)

    self.avg = avg
    """The average CPU utilization, as a proportion."""

    self.min = min
    """The minimum CPU utilization, as a proportion."""

    self.max = max
    """The maximum CPU utilization, as a proportion."""

__str__()

Return a string representation of CPUStatistics.

Source code in mlte/measurement/cpu/local_process_cpu_utilization.py 
85
86
87
88
89
90
91
def __str__(self) -> str:
    """Return a string representation of CPUStatistics."""
    s = ""
    s += f"Average: {self.avg:.2f}%\n"
    s += f"Minimum: {self.min:.2f}%\n"
    s += f"Maximum: {self.max:.2f}%"
    return s

average_utilization_less_than(threshold) classmethod

Construct and invoke a condition for average CPU utilization.

Parameters:

Name Type Description Default
threshold float

The threshold value for average utilization, as percentage

 required

Returns:

Type Description
Condition

The Condition that can be used to validate a Value.
Source code in mlte/measurement/cpu/local_process_cpu_utilization.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
@classmethod
def average_utilization_less_than(cls, threshold: float) -> Condition:
    """
    Construct and invoke a condition for average CPU utilization.

    :param threshold: The threshold value for average utilization, as percentage

    :return: The Condition that can be used to validate a Value.
    """
    condition: Condition = Condition.build_condition(
        bool_exp=lambda stats: stats.avg < threshold,
        success=f"Average utilization below threshold {threshold:.2f}",
        failure=f"Average utilization exceeds threshold {threshold:.2f}",
    )
    return condition

deserialize(evidence_metadata, data) staticmethod

Deserialize an CPUStatistics from a JSON object.

Parameters:

Name Type Description Default
evidence_metadata EvidenceMetadata

The generating measurement's metadata

 required
data Dict[str, Any]

The JSON object

 required

Returns:

Type Description
CPUStatistics

The deserialized instance
Source code in mlte/measurement/cpu/local_process_cpu_utilization.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@staticmethod
def deserialize(
    evidence_metadata: EvidenceMetadata, data: Dict[str, Any]
) -> CPUStatistics:
    """
    Deserialize an CPUStatistics from a JSON object.

    :param evidence_metadata: The generating measurement's metadata
    :param data: The JSON object

    :return: The deserialized instance
    """
    return CPUStatistics(
        evidence_metadata,
        avg=data["avg"],
        min=data["min"],
        max=data["max"],
    )

max_utilization_less_than(threshold) classmethod

Construct and invoke a condition for maximum CPU utilization.

Parameters:

Name Type Description Default
threshold float

The threshold value for maximum utilization, as percentage

 required

Returns:

Type Description
Condition

The Condition that can be used to validate a Value.
Source code in mlte/measurement/cpu/local_process_cpu_utilization.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@classmethod
def max_utilization_less_than(cls, threshold: float) -> Condition:
    """
    Construct and invoke a condition for maximum CPU utilization.

    :param threshold: The threshold value for maximum utilization, as percentage

    :return: The Condition that can be used to validate a Value.
    """
    condition: Condition = Condition.build_condition(
        bool_exp=lambda stats: stats.max < threshold,
        success=f"Maximum utilization below threshold {threshold:.2f}",
        failure=f"Maximum utilization exceeds threshold {threshold:.2f}",
    )
    return condition

serialize()

Serialize an CPUStatistics to a JSON object.

Returns:

Type Description
Dict[str, Any]

The JSON object
Source code in mlte/measurement/cpu/local_process_cpu_utilization.py 
58
59
60
61
62
63
64
def serialize(self) -> Dict[str, Any]:
    """
    Serialize an CPUStatistics to a JSON object.

    :return: The JSON object
    """
    return {"avg": self.avg, "min": self.min, "max": self.max}

LocalProcessCPUUtilization

Bases: ProcessMeasurement

Measures CPU utilization for a local process.

Source code in mlte/measurement/cpu/local_process_cpu_utilization.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
class LocalProcessCPUUtilization(ProcessMeasurement):
    """Measures CPU utilization for a local process."""

    def __init__(self, identifier: str):
        """
        Initialize a new LocalProcessCPUUtilization measurement.

        :param identifier: A unique identifier for the measurement
        """
        super().__init__(self, identifier)
        if is_windows():
            raise RuntimeError(
                f"Measurement {self.metadata.identifier} is not supported on Windows."
            )

    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(
            self.metadata,
            avg=sum(stats) / len(stats),
            min=min(stats),
            max=max(stats),
        )

    @classmethod
    def value(self) -> Type[CPUStatistics]:
        """Returns the class type object for the Value produced by the Measurement."""
        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 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
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(
        self.metadata,
        avg=sum(stats) / len(stats),
        min=min(stats),
        max=max(stats),
    )

__init__(identifier)

Initialize a new LocalProcessCPUUtilization measurement.

Parameters:

Name Type Description Default
identifier str

A unique identifier for the measurement

 required
Source code in mlte/measurement/cpu/local_process_cpu_utilization.py 
134
135
136
137
138
139
140
141
142
143
144
def __init__(self, identifier: str):
    """
    Initialize a new LocalProcessCPUUtilization measurement.

    :param identifier: A unique identifier for the measurement
    """
    super().__init__(self, identifier)
    if is_windows():
        raise RuntimeError(
            f"Measurement {self.metadata.identifier} is not supported on Windows."
        )

value() classmethod

Returns the class type object for the Value produced by the Measurement.

Source code in mlte/measurement/cpu/local_process_cpu_utilization.py 
170
171
172
173
@classmethod
def value(self) -> Type[CPUStatistics]:
    """Returns the class type object for the Value produced by the Measurement."""
    return CPUStatistics