Skip to content

common

Base class for unist specific measurements that track minimin, maximum and average.

CommonStatistics

Bases: ExternalEvidence

Source code in mlte/measurement/common.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
class CommonStatistics(ExternalEvidence):
    DEFAULT_UNIT: Optional[Unit] = None

    def __init__(
        self, avg: float, min: float, max: float, unit: Optional[Unit]
    ):
        """

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

        self.avg = Quantity(avg, unit)
        """The average value."""

        self.min = Quantity(min, unit)
        """The minimum values."""

        self.max = Quantity(max, unit)
        """The maximum value."""

        self.unit = unit
        """The unit being used for all values."""

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

        :return: The JSON object
        """
        return {
            "avg": self.avg.magnitude,
            "min": self.min.magnitude,
            "max": self.max.magnitude,
            "unit": unit_to_str(self.unit),
        }

    @classmethod
    def deserialize(cls, data: dict[str, Any]) -> Any:
        """
        Deserialize from a JSON object.

        :param data: The JSON object

        :return: The deserialized instance
        """
        unit = str_to_unit(data["unit"])
        return cls(
            avg=data["avg"],
            min=data["min"],
            max=data["max"],
            unit=unit if unit else cls.DEFAULT_UNIT,
        )

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

    def __repr__(self) -> str:
        """Return a string representation for debugging."""
        return (
            f"avg={self.avg}, min={self.min}, max={self.max}, unit={self.unit}"
        )

    @classmethod
    def max_utilization_less_than(
        cls,
        threshold: float,
        unit: Optional[Unit] = None,
        success: str = "",
        failure: str = "",
    ) -> Validator:
        """
        Construct and invoke a validator for maximum resource utilization.

        :param threshold: The threshold value for maximum utilization
        :param unit: the unit the threshold comes in, as a value from Units; defaults to DEFAULT_UNIT

        :return: The Validator that can be used to validate a Value.
        """
        # This allows us to get the unit from a subclass
        if unit is None:
            unit = cls.DEFAULT_UNIT

        threshold_w_unit = Quantity(threshold, unit)
        bool_exp: Callable[[Any], bool] = (
            lambda stats: stats.max < threshold_w_unit
        )
        validator: Validator = Validator.build_validator(
            bool_exp=bool_exp,
            thresholds=[threshold_w_unit],
            success=success,
            failure=failure,
            default_success=f"Maximum utilization below threshold {threshold_w_unit}",
            default_failure=f"Maximum utilization exceeds threshold {threshold_w_unit}",
            input_types=[cls],
        )
        return validator

    @classmethod
    def average_utilization_less_than(
        cls,
        threshold: float,
        unit: Optional[Unit] = None,
        success: str = "",
        failure: str = "",
    ) -> Validator:
        """
        Construct and invoke a validator for average resource utilization.

        :param threshold: The threshold value for average utilization, in KB
        :param unit: the unit the threshold comes in, as a value from Units; defaults to Units.kilobyte

        :return: The Validator that can be used to validate a Value.
        """

        # This allows us to get the unit from a subclass
        if unit is None:
            unit = cls.DEFAULT_UNIT

        threshold_w_unit = Quantity(threshold, unit)
        bool_exp: Callable[[Any], bool] = (
            lambda stats: stats.avg < threshold_w_unit
        )
        validator: Validator = Validator.build_validator(
            bool_exp=bool_exp,
            thresholds=[threshold_w_unit],
            success=success,
            failure=failure,
            default_success=f"Average utilization below threshold {threshold_w_unit}",
            default_failure=f"Average utilization exceeds threshold {threshold_w_unit}",
            input_types=[cls],
        )
        return validator

avg = Quantity(avg, unit) instance-attribute

The average value.

max = Quantity(max, unit) instance-attribute

The maximum value.

min = Quantity(min, unit) instance-attribute

The minimum values.

unit = unit instance-attribute

The unit being used for all values.

__init__(avg, min, max, unit)

Parameters:

Name Type Description Default
avg float

The average value

 required
min float

The minimum value

 required
max float

The maximum value

 required
unit Optional[Unit]

the unit the values comes in, as a value from Units

 required
Source code in mlte/measurement/common.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def __init__(
    self, avg: float, min: float, max: float, unit: Optional[Unit]
):
    """

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

    self.avg = Quantity(avg, unit)
    """The average value."""

    self.min = Quantity(min, unit)
    """The minimum values."""

    self.max = Quantity(max, unit)
    """The maximum value."""

    self.unit = unit
    """The unit being used for all values."""

__repr__()

Return a string representation for debugging.

Source code in mlte/measurement/common.py 
79
80
81
82
83
def __repr__(self) -> str:
    """Return a string representation for debugging."""
    return (
        f"avg={self.avg}, min={self.min}, max={self.max}, unit={self.unit}"
    )

__str__()

Return a string representation.

Source code in mlte/measurement/common.py 
71
72
73
74
75
76
77
def __str__(self) -> str:
    """Return a string representation."""
    return (
        f"Average: {self.avg}\n"
        + f"Minimum: {self.min}\n"
        + f"Maximum: {self.max}"
    )

average_utilization_less_than(threshold, unit=None, success='', failure='') classmethod

Construct and invoke a validator for average resource utilization.

Parameters:

Name Type Description Default
threshold float

The threshold value for average utilization, in KB

 required
unit Optional[Unit]

the unit the threshold comes in, as a value from Units; defaults to Units.kilobyte

 None

Returns:

Type Description
Validator

The Validator that can be used to validate a Value.
Source code in mlte/measurement/common.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
@classmethod
def average_utilization_less_than(
    cls,
    threshold: float,
    unit: Optional[Unit] = None,
    success: str = "",
    failure: str = "",
) -> Validator:
    """
    Construct and invoke a validator for average resource utilization.

    :param threshold: The threshold value for average utilization, in KB
    :param unit: the unit the threshold comes in, as a value from Units; defaults to Units.kilobyte

    :return: The Validator that can be used to validate a Value.
    """

    # This allows us to get the unit from a subclass
    if unit is None:
        unit = cls.DEFAULT_UNIT

    threshold_w_unit = Quantity(threshold, unit)
    bool_exp: Callable[[Any], bool] = (
        lambda stats: stats.avg < threshold_w_unit
    )
    validator: Validator = Validator.build_validator(
        bool_exp=bool_exp,
        thresholds=[threshold_w_unit],
        success=success,
        failure=failure,
        default_success=f"Average utilization below threshold {threshold_w_unit}",
        default_failure=f"Average utilization exceeds threshold {threshold_w_unit}",
        input_types=[cls],
    )
    return validator

deserialize(data) classmethod

Deserialize from a JSON object.

Parameters:

Name Type Description Default
data dict[str, Any]

The JSON object

 required

Returns:

Type Description
Any

The deserialized instance
Source code in mlte/measurement/common.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
@classmethod
def deserialize(cls, data: dict[str, Any]) -> Any:
    """
    Deserialize from a JSON object.

    :param data: The JSON object

    :return: The deserialized instance
    """
    unit = str_to_unit(data["unit"])
    return cls(
        avg=data["avg"],
        min=data["min"],
        max=data["max"],
        unit=unit if unit else cls.DEFAULT_UNIT,
    )

max_utilization_less_than(threshold, unit=None, success='', failure='') classmethod

Construct and invoke a validator for maximum resource utilization.

Parameters:

Name Type Description Default
threshold float

The threshold value for maximum utilization

 required
unit Optional[Unit]

the unit the threshold comes in, as a value from Units; defaults to DEFAULT_UNIT

 None

Returns:

Type Description
Validator

The Validator that can be used to validate a Value.
Source code in mlte/measurement/common.py 
 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
@classmethod
def max_utilization_less_than(
    cls,
    threshold: float,
    unit: Optional[Unit] = None,
    success: str = "",
    failure: str = "",
) -> Validator:
    """
    Construct and invoke a validator for maximum resource utilization.

    :param threshold: The threshold value for maximum utilization
    :param unit: the unit the threshold comes in, as a value from Units; defaults to DEFAULT_UNIT

    :return: The Validator that can be used to validate a Value.
    """
    # This allows us to get the unit from a subclass
    if unit is None:
        unit = cls.DEFAULT_UNIT

    threshold_w_unit = Quantity(threshold, unit)
    bool_exp: Callable[[Any], bool] = (
        lambda stats: stats.max < threshold_w_unit
    )
    validator: Validator = Validator.build_validator(
        bool_exp=bool_exp,
        thresholds=[threshold_w_unit],
        success=success,
        failure=failure,
        default_success=f"Maximum utilization below threshold {threshold_w_unit}",
        default_failure=f"Maximum utilization exceeds threshold {threshold_w_unit}",
        input_types=[cls],
    )
    return validator

serialize()

Serialize to a JSON object.

Returns:

Type Description
dict[str, Any]

The JSON object
Source code in mlte/measurement/common.py 
41
42
43
44
45
46
47
48
49
50
51
52
def serialize(self) -> dict[str, Any]:
    """
    Serialize to a JSON object.

    :return: The JSON object
    """
    return {
        "avg": self.avg.magnitude,
        "min": self.min.magnitude,
        "max": self.max.magnitude,
        "unit": unit_to_str(self.unit),
    }