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

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 

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

66 

67 

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

69 """ 

70 Convert a multi-statistics object to a stream. 

71 

72 :param multi: the multi-statistics object 

73 :return: the stream of strings 

74 """ 

75 if not isinstance(multi, MultiStatistics): 

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

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

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

79 yield from stat_to_stream(ss) 

80 

81 

82class MultiStatisticsSpace(Space): 

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

84 

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

86 """ 

87 Create a multi-statistics space. 

88 

89 :param instances: the instances 

90 """ 

91 if not isinstance(instances, tuple): 

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

93 for inst in instances: 

94 if not isinstance(inst, Instance): 

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

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

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

98 #: the instance names 

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

100 inst.name for inst in instances) 

101 

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

103 """ 

104 Copy one multi-statistics to another one. 

105 

106 :param dest: the destination multi-statistics 

107 :param source: the source multi-statistics 

108 """ 

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

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

111 

112 def create(self) -> MultiStatistics: 

113 """ 

114 Create an empty multi-statistics record. 

115 

116 :return: the empty multi-statistics record 

117 """ 

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

119 

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

121 """ 

122 Convert a multi-statistics to a string. 

123 

124 :param x: the packing 

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

126 """ 

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

128 

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

130 """ 

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

132 

133 :param x1: the first multi-statistics 

134 :param x2: the second multi-statistics 

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

136 """ 

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

138 x1.inst_names == x2.inst_names)) 

139 

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

141 """ 

142 Convert a string to a multi-statistics. 

143 

144 :param text: the string 

145 :return: the multi-statistics 

146 """ 

147 if not isinstance(text, str): 

148 raise type_error(text, "text", str) 

149 raise NotImplementedError 

150 

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

152 """ 

153 Check if a multi-statistics is valid. 

154 

155 :param x: the multi-statistics 

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

157 wrong type 

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

159 """ 

160 if not isinstance(x, MultiStatistics): 

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

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

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

164 for s in x.per_instance: 

165 if not isinstance(s, Statistics): 

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

167 if x.inst_names != self.__inst_names: 

168 raise ValueError("Wrong instance names.") 

169 

170 def n_points(self) -> int: 

171 """ 

172 Get the number of possible multi-statistics. 

173 

174 :return: just some arbitrary very large number 

175 """ 

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

177 

178 def __str__(self) -> str: 

179 """ 

180 Get the name of the multi-statistics space. 

181 

182 :return: the name 

183 """ 

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

185 

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

187 """ 

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

189 

190 :param logger: the logger for the parameters 

191 """ 

192 super().log_parameters_to(logger) 

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

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