Coverage for moptipyapps / prodsched / multistatistics.py: 69%
91 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-13 08:40 +0000
1"""
2A statistics record for multiple simulations.
4:class:`~MultiStatistics` are records that hold multiple simulation
5:class:`~moptipyapps.prodsched.statistics.Statistics`, each of which computed
6over a separate :class:`~moptipyapps.prodsched.simulation.Simulation` based on
7a separate :mod:`~moptipyapps.prodsched.instance` of the material flow
8problem.
9These records are filled with data by via the
10:class:`moptipyapps.prodsched.rop_multisimulation.ROPMultiSimulation`
11mechanism, which performs the multiple simulations.
13We cn use this record as the solution space when optimizing for the MFC
14scenario.
15Such a record holds comprehensive statistics across several simulation runs.
16This makes it suitable as source of data for objective functions
17(:class:`~moptipy.api.objective.Objective`).
18The objective functions can then access these statistics.
20Since we use :class:`~MultiStatistics` as solution space, we also need an
21implementation of moptipy's :class:`~moptipy.api.space.Space`-API to plug it
22into the optimization process.
23Sucha space implementation is provided as class
24:class:`~MultiStatisticsSpace`.
25It can create, copy, and serialize these objects to text, so that they can
26appear in the log files.
27"""
29from dataclasses import dataclass
30from typing import Final, Generator, Iterable, Self
32from moptipy.api.space import Space
33from moptipy.utils.logger import (
34 KeyValueLogSection,
35)
36from pycommons.types import type_error
38from moptipyapps.prodsched.instance import Instance
39from moptipyapps.prodsched.statistics import Statistics
40from moptipyapps.prodsched.statistics import to_stream as stat_to_stream
43@dataclass(order=False, frozen=True)
44class MultiStatistics:
45 """A set of statistics gathered over multiple instances."""
47 #: the per-instance statistics
48 per_instance: tuple[Statistics, ...]
49 #: the instance names
50 inst_names: tuple[str, ...]
52 def __init__(self, instances: tuple[Instance, ...],
53 names: tuple[str, ...] | None = None) -> None:
54 """
55 Create the multi-statistics object.
57 :param instances: the instances for which we create the statistics
58 """
59 object.__setattr__(self, "per_instance", tuple(
60 Statistics(inst.n_products) for inst in instances))
61 if names is None:
62 names = tuple(inst.name for inst in instances)
63 elif tuple.__len__(names) != tuple.__len__(instances):
64 raise ValueError(f"names {names} do not fit")
65 object.__setattr__(self, "inst_names", names)
67 def from_stream(self, stream: Iterable[str]) -> Self:
68 """
69 Convert a stream of text to a multi-statistics object.
71 Warning: This method cannot restore the numbers `n` in the single
72 stream statistics.
74 :param stream: the stream
75 :return: the object itself
76 """
77 pi: Final[tuple[Statistics, ...]] = self.per_instance
78 n: Final[int] = tuple.__len__(pi)
79 for statistics in pi:
80 statistics.clear()
81 source = iter(stream)
82 needs: Final[set[int]] = set(range(n))
83 for srow in source:
84 row = str.strip(srow)
85 if not row.startswith("-"):
86 continue
87 space: int = row.index(" ", row.index(" ") + 1)
88 colon: int = row.index(":", space + 1)
89 prime_1: int = row.index("'", colon + 1)
90 prime_2: int = row.index("'", prime_1 + 1)
91 inst_name: str = str.strip(row[prime_1 + 1: prime_2])
92 inst_idx: int = int(row[space + 1:colon])
93 if inst_idx not in needs:
94 raise ValueError(f"Instance data {inst_idx} / "
95 f"{inst_name!r} already loaded.")
96 if self.inst_names[inst_idx] != inst_name:
97 raise ValueError(
98 f"Name {inst_name!r} of instance {inst_idx} should "
99 f"be {self.inst_names[inst_idx]!r}.")
100 pi[inst_idx].from_stream(source)
101 needs.remove(inst_idx)
103 if set.__len__(needs) != 0:
104 raise ValueError(f"Data for instances {needs} is missing.")
105 return self
108def to_stream(multi: MultiStatistics) -> Generator[str, None, None]:
109 """
110 Convert a multi-statistics object to a stream.
112 :param multi: the multi-statistics object
113 :return: the stream of strings
114 """
115 if not isinstance(multi, MultiStatistics):
116 raise type_error(multi, "multi", MultiStatistics)
117 for i, ss in enumerate(multi.per_instance):
118 yield f"-------- Instance {i}: {multi.inst_names[i]!r} -------"
119 yield from stat_to_stream(ss)
122class MultiStatisticsSpace(Space):
123 """An implementation of the `Space` API of for multiple statistics."""
125 def __init__(self, instances: tuple[Instance, ...]) -> None:
126 """
127 Create a multi-statistics space.
129 :param instances: the instances
130 """
131 if not isinstance(instances, tuple):
132 raise type_error(instances, "instances", tuple)
133 for inst in instances:
134 if not isinstance(inst, Instance):
135 raise type_error(inst, "instance", Instance)
136 #: The instance to which the statistics and simulations belong.
137 self.instances: Final[tuple[Instance, ...]] = instances
138 #: the instance names
139 self.__inst_names: Final[tuple[str, ...]] = tuple(
140 inst.name for inst in instances)
142 def copy(self, dest: MultiStatistics, source: MultiStatistics) -> None:
143 """
144 Copy one multi-statistics to another one.
146 :param dest: the destination multi-statistics
147 :param source: the source multi-statistics
148 """
149 for i, d in enumerate(dest.per_instance):
150 d.copy_from(source.per_instance[i])
152 def create(self) -> MultiStatistics:
153 """
154 Create an empty multi-statistics record.
156 :return: the empty multi-statistics record
157 """
158 return MultiStatistics(self.instances, self.__inst_names)
160 def to_str(self, x: MultiStatistics) -> str:
161 """
162 Convert a multi-statistics to a string.
164 :param x: the packing
165 :return: a string corresponding to the multi-statistics
166 """
167 return "\n".join(to_stream(x))
169 def is_equal(self, x1: MultiStatistics, x2: MultiStatistics) -> bool:
170 """
171 Check if two multi-statistics have the same contents.
173 :param x1: the first multi-statistics
174 :param x2: the second multi-statistics
175 :return: `True` if both multi-statistics have the same content
176 """
177 return (x1 is x2) or ((x1.per_instance == x2.per_instance) and (
178 x1.inst_names == x2.inst_names))
180 def from_str(self, text: str) -> MultiStatistics:
181 """
182 Convert a string to a multi-statistics.
184 Warning: This method cannot restore the numbers `n` in the single
185 stream statistics. Therefore, the returned objects cannot be identical
186 to the stored objects...
188 :param text: the string
189 :return: the multi-statistics
190 """
191 result: Final[MultiStatistics] = self.create().from_stream(
192 text.split())
193 self.validate(result)
194 return result
196 def validate(self, x: MultiStatistics) -> None:
197 """
198 Check if a multi-statistics is valid.
200 :param x: the multi-statistics
201 :raises TypeError: if any component of the multi-statistics is of the
202 wrong type
203 :raises ValueError: if the multi-statistics is not feasible
204 """
205 if not isinstance(x, MultiStatistics):
206 raise type_error(x, "x", MultiStatistics)
207 if not isinstance(x.per_instance, tuple):
208 raise type_error(x.per_instance, "x.per_instance", tuple)
209 for s in x.per_instance:
210 if not isinstance(s, Statistics):
211 raise type_error(s, "x.per_instance[i]", Statistics)
212 if x.inst_names != self.__inst_names:
213 raise ValueError("Wrong instance names.")
215 def n_points(self) -> int:
216 """
217 Get the number of possible multi-statistics.
219 :return: just some arbitrary very large number
220 """
221 return 100 ** tuple.__len__(self.instances)
223 def __str__(self) -> str:
224 """
225 Get the name of the multi-statistics space.
227 :return: the name
228 """
229 return f"multistats_{tuple.__len__(self.instances)}"
231 def log_parameters_to(self, logger: KeyValueLogSection) -> None:
232 """
233 Log the parameters of the space to the given logger.
235 :param logger: the logger for the parameters
236 """
237 super().log_parameters_to(logger)
238 for i, inst in enumerate(self.__inst_names):
239 logger.key_value(f"inst_{i}", inst)