Coverage for moptipyapps/prodsched/multistatistics.py: 69%

1""" 

2A statistics record for multiple simulations. 

3 

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. 

12 

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. 

19 

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""" 

28 

29from dataclasses import dataclass 

30from typing import Final, Generator, Iterable, Self 

31 

32from moptipy.api.space import Space 

33from moptipy.utils.logger import ( 

34 KeyValueLogSection, 

35) 

36from pycommons.types import type_error 

37 

38from moptipyapps.prodsched.instance import Instance 

39from moptipyapps.prodsched.statistics import Statistics 

40from moptipyapps.prodsched.statistics import to_stream as stat_to_stream 

41 

42 

43@dataclass(order=False, frozen=True) 

44class MultiStatistics: 

45 """A set of statistics gathered over multiple instances.""" 

46 

47 #: the per-instance statistics 

48 per_instance: tuple[Statistics, ...] 

49 #: the instance names 

50 inst_names: tuple[str, ...] 

51 

52 def __init__(self, instances: tuple[Instance, ...], 

53 names: tuple[str, ...] | None = None) -> None: 

54 """ 

55 Create the multi-statistics object. 

56 

57 :param instances: the instances for which we create the statistics 

58 """ 

59 object.__setattr__(self, "per_instance", tuple( 

60 Statistics(inst.n_products, inst.n_stations) 

61 for inst in instances)) 

62 if names is None: 

63 names = tuple(inst.name for inst in instances) 

64 elif tuple.__len__(names) != tuple.__len__(instances): 

65 raise ValueError(f"names {names} do not fit") 

66 object.__setattr__(self, "inst_names", names) 

67 

68 def from_stream(self, stream: Iterable[str]) -> Self: 

69 """ 

70 Convert a stream of text to a multi-statistics object. 

71 

72 Warning: This method cannot restore the numbers `n` in the single 

73 stream statistics. 

74 

75 :param stream: the stream 

76 :return: the object itself 

77 """ 

78 pi: Final[tuple[Statistics, ...]] = self.per_instance 

79 n: Final[int] = tuple.__len__(pi) 

80 for statistics in pi: 

81 statistics.clear() 

82 source = iter(stream) 

83 needs: Final[set[int]] = set(range(n)) 

84 for srow in source: 

85 row = str.strip(srow) 

86 if not row.startswith("-"): 

87 continue 

88 space: int = row.index(" ", row.index(" ") + 1) 

89 colon: int = row.index(":", space + 1) 

90 prime_1: int = row.index("'", colon + 1) 

91 prime_2: int = row.index("'", prime_1 + 1) 

92 inst_name: str = str.strip(row[prime_1 + 1: prime_2]) 

93 inst_idx: int = int(row[space + 1:colon]) 

94 if inst_idx not in needs: 

95 raise ValueError(f"Instance data {inst_idx} / " 

96 f"{inst_name!r} already loaded.") 

97 if self.inst_names[inst_idx] != inst_name: 

98 raise ValueError( 

99 f"Name {inst_name!r} of instance {inst_idx} should " 

100 f"be {self.inst_names[inst_idx]!r}.") 

101 pi[inst_idx].from_stream(source) 

102 needs.remove(inst_idx) 

103 

104 if set.__len__(needs) != 0: 

105 raise ValueError(f"Data for instances {needs} is missing.") 

106 return self 

107 

108 

109def to_stream(multi: MultiStatistics) -> Generator[str, None, None]: 

110 """ 

111 Convert a multi-statistics object to a stream. 

112 

113 :param multi: the multi-statistics object 

114 :return: the stream of strings 

115 """ 

116 if not isinstance(multi, MultiStatistics): 

117 raise type_error(multi, "multi", MultiStatistics) 

118 for i, ss in enumerate(multi.per_instance): 

119 yield f"-------- Instance {i}: {multi.inst_names[i]!r} -------" 

120 yield from stat_to_stream(ss) 

121 

122 

123class MultiStatisticsSpace(Space): 

124 """An implementation of the `Space` API of for multiple statistics.""" 

125 

126 def __init__(self, instances: tuple[Instance, ...]) -> None: 

127 """ 

128 Create a multi-statistics space. 

129 

130 :param instances: the instances 

131 """ 

132 if not isinstance(instances, tuple): 

133 raise type_error(instances, "instances", tuple) 

134 for inst in instances: 

135 if not isinstance(inst, Instance): 

136 raise type_error(inst, "instance", Instance) 

137 #: The instance to which the statistics and simulations belong. 

138 self.instances: Final[tuple[Instance, ...]] = instances 

139 #: the instance names 

140 self.__inst_names: Final[tuple[str, ...]] = tuple( 

141 inst.name for inst in instances) 

142 

143 def copy(self, dest: MultiStatistics, source: MultiStatistics) -> None: 

144 """ 

145 Copy one multi-statistics to another one. 

146 

147 :param dest: the destination multi-statistics 

148 :param source: the source multi-statistics 

149 """ 

150 for i, d in enumerate(dest.per_instance): 

151 d.copy_from(source.per_instance[i]) 

152 

153 def create(self) -> MultiStatistics: 

154 """ 

155 Create an empty multi-statistics record. 

156 

157 :return: the empty multi-statistics record 

158 """ 

159 return MultiStatistics(self.instances, self.__inst_names) 

160 

161 def to_str(self, x: MultiStatistics) -> str: 

162 """ 

163 Convert a multi-statistics to a string. 

164 

165 :param x: the packing 

166 :return: a string corresponding to the multi-statistics 

167 """ 

168 return "\n".join(to_stream(x)) 

169 

170 def is_equal(self, x1: MultiStatistics, x2: MultiStatistics) -> bool: 

171 """ 

172 Check if two multi-statistics have the same contents. 

173 

174 :param x1: the first multi-statistics 

175 :param x2: the second multi-statistics 

176 :return: `True` if both multi-statistics have the same content 

177 """ 

178 return (x1 is x2) or ((x1.per_instance == x2.per_instance) and ( 

179 x1.inst_names == x2.inst_names)) 

180 

181 def from_str(self, text: str) -> MultiStatistics: 

182 """ 

183 Convert a string to a multi-statistics. 

184 

185 Warning: This method cannot restore the numbers `n` in the single 

186 stream statistics. Therefore, the returned objects cannot be identical 

187 to the stored objects... 

188 

189 :param text: the string 

190 :return: the multi-statistics 

191 """ 

192 result: Final[MultiStatistics] = self.create().from_stream( 

193 text.split()) 

194 self.validate(result) 

195 return result 

196 

197 def validate(self, x: MultiStatistics) -> None: 

198 """ 

199 Check if a multi-statistics is valid. 

200 

201 :param x: the multi-statistics 

202 :raises TypeError: if any component of the multi-statistics is of the 

203 wrong type 

204 :raises ValueError: if the multi-statistics is not feasible 

205 """ 

206 if not isinstance(x, MultiStatistics): 

207 raise type_error(x, "x", MultiStatistics) 

208 if not isinstance(x.per_instance, tuple): 

209 raise type_error(x.per_instance, "x.per_instance", tuple) 

210 for s in x.per_instance: 

211 if not isinstance(s, Statistics): 

212 raise type_error(s, "x.per_instance[i]", Statistics) 

213 if x.inst_names != self.__inst_names: 

214 raise ValueError("Wrong instance names.") 

215 

216 def n_points(self) -> int: 

217 """ 

218 Get the number of possible multi-statistics. 

219 

220 :return: just some arbitrary very large number 

221 """ 

222 return 100 ** tuple.__len__(self.instances) 

223 

224 def __str__(self) -> str: 

225 """ 

226 Get the name of the multi-statistics space. 

227 

228 :return: the name 

229 """ 

230 return f"multistats_{tuple.__len__(self.instances)}" 

231 

232 def log_parameters_to(self, logger: KeyValueLogSection) -> None: 

233 """ 

234 Log the parameters of the space to the given logger. 

235 

236 :param logger: the logger for the parameters 

237 """ 

238 super().log_parameters_to(logger) 

239 for i, inst in enumerate(self.__inst_names): 

240 logger.key_value(f"inst_{i}", inst)