0076644aa32cc3719e2b6fb4f4ba08d4d1f4673e
[csit.git] / resources / libraries / python / MLRsearch / trimmed_stat.py
1 # Copyright (c) 2023 Cisco and/or its affiliates.
2 # Licensed under the Apache License, Version 2.0 (the "License");
3 # you may not use this file except in compliance with the License.
4 # You may obtain a copy of the License at:
5 #
6 #     http://www.apache.org/licenses/LICENSE-2.0
7 #
8 # Unless required by applicable law or agreed to in writing, software
9 # distributed under the License is distributed on an "AS IS" BASIS,
10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11 # See the License for the specific language governing permissions and
12 # limitations under the License.
13
14 """Module defining TrimmedStat class."""
15
16 from __future__ import annotations
17
18 from dataclasses import dataclass
19 from typing import Optional
20
21 from .discrete_load import DiscreteLoad
22 from .load_stats import LoadStats
23 from .target_spec import TargetSpec
24
25
26 @dataclass
27 class TrimmedStat(LoadStats):
28     """Load stats trimmed to a single target.
29
30     Useful mainly for reporting the overall results.
31     """
32
33     def __post_init__(self) -> None:
34         """Initialize load value and check there is one target to track."""
35         super().__post_init__()
36         if len(self.target_to_stat) != 1:
37             raise ValueError(f"No single target: {self.target_to_stat!r}")
38
39     @staticmethod
40     def for_target(stats: LoadStats, target: TargetSpec) -> TrimmedStat:
41         """Return new instance with only one target in the mapping.
42
43         :param stats: The load stats instance to trim.
44         :param target: The one target which should remain in the mapping.
45         :type stats: LoadStats
46         :type target: TargetSpec
47         :return: Newly created instance.
48         :rtype: TrimmedStat
49         """
50         return TrimmedStat(
51             rounding=stats.rounding,
52             int_load=stats.int_load,
53             target_to_stat={target: stats.target_to_stat[target]},
54         )
55
56     @property
57     def conditional_throughput(self) -> Optional[DiscreteLoad]:
58         """Compute conditional throughput from the load.
59
60         Target stat has dur_rat_sum and good_long.
61         The code here adds intended load and handles the case min load is hit.
62         If min load is not a lower bound, None is returned.
63
64         :return: Conditional throughput assuming self is a relevant lower bound.
65         :rtype: Optional[DiscreteLoad]
66         :raises RuntimeError: If target is unclear or load is spurious.
67         """
68         target = list(self.target_to_stat.keys())[0]
69         _, pes = self.estimates(target)
70         if not pes:
71             if int(self):
72                 raise RuntimeError(f"Not a lower bound: {self}")
73             return None
74         # TODO: Verify self is really the clo?
75         stat = self.target_to_stat[target]
76         loss_ratio = stat.dur_rat_sum / stat.good_long
77         ret = self * (1.0 - loss_ratio)
78         return ret