feat(MLRseach): Update to v8 conditional throughput
[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         The conditional throughput has the same semantics as load,
61         so if load is unicirectional and user wants bidirectional
62         throughput, the manager has to compensate.
63
64         This only works correctly if the self load is a lower bound
65         for the self target, but this method does not check that.
66         Its should not matter, as MLRsearch controller only returns
67         the relevant lower bounds to the manager.
68
69         :return: Conditional throughput assuming self is a relevant lower bound.
70         :rtype: Optional[DiscreteLoad]
71         :raises RuntimeError: If target is unclear or load is spurious.
72         """
73         target = list(self.target_to_stat.keys())[0]
74         stat = self.target_to_stat[target]
75         loss_ratio = stat.pessimistic_loss_ratio()
76         ret = self * (1.0 - loss_ratio)
77         return ret