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

1""" 

2A statistics record for multiple simulations. 

3 

4We use this record as the solution space when optimizing for the MFC scenario. 

5It stores the statistics of several simulation runs. 

6The objective functions can then access these statistics. 

7A space instance is provided that can create, copy, and serialize these 

8objects to text, so that they can appear in the log files. 

9""" 

10 

11from dataclasses import dataclass 

12from typing import Final, Generator, Iterable 

13 

14from moptipy.api.space import Space 

15from moptipy.utils.logger import ( 

16 KeyValueLogSection, 

17) 

18from pycommons.types import type_error 

19 

20from moptipyapps.prodsched.instance import Instance 

21from moptipyapps.prodsched.statistics import Statistics 

22from moptipyapps.prodsched.statistics import to_stream as stat_to_stream 

23 

24 

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

26class MultiStatistics: 

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

28 

29 #: the per-instance statistics 

30 per_instance: tuple[Statistics, ...] 

31 #: the instance names 

32 inst_names: tuple[str, ...] 

33 

34 def __init__(self, instances: Iterable[Instance]) -> None: 

35 """ 

36 Create the multi-statistics object. 

37 

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

39 """ 

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

41 Statistics(inst.n_products) for inst in instances)) 

42 object.__setattr__(self, "inst_names", tuple( 

43 inst.name for inst in instances)) 

44 

45 

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

47 """ 

48 Convert a multi-statistics object to a stream. 

49 

50 :param multi: the multi-statistics object 

51 :return: the stream of strings 

52 """ 

53 if not isinstance(multi, MultiStatistics): 

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

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

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

57 yield from stat_to_stream(ss) 

58 

59 

60class MultiStatisticsSpace(Space): 

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

62 

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

64 """ 

65 Create a multi-statistics space. 

66 

67 :param instances: the instances 

68 """ 

69 if not isinstance(instances, tuple): 

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

71 for inst in instances: 

72 if not isinstance(inst, Instance): 

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

74 #: The instance to which the packings apply. 

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

76 

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

78 """ 

79 Copy one multi-statistics to another one. 

80 

81 :param dest: the destination multi-statistics 

82 :param source: the source multi-statistics 

83 """ 

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

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

86 

87 def create(self) -> MultiStatistics: 

88 """ 

89 Create an empty multi-statistics record. 

90 

91 :return: the empty multi-statistics record 

92 """ 

93 return MultiStatistics(self.instances) 

94 

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

96 """ 

97 Convert a multi-statistics to a string. 

98 

99 :param x: the packing 

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

101 """ 

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

103 

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

105 """ 

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

107 

108 :param x1: the first multi-statistics 

109 :param x2: the second multi-statistics 

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

111 """ 

112 return (x1 is x2) or (x1.per_instance == x2.per_instance) 

113 

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

115 """ 

116 Convert a string to a multi-statistics. 

117 

118 :param text: the string 

119 :return: the multi-statistics 

120 """ 

121 if not isinstance(text, str): 

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

123 raise NotImplementedError 

124 

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

126 """ 

127 Check if a multi-statistics is valid. 

128 

129 :param x: the multi-statistics 

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

131 wrong type 

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

133 """ 

134 if not isinstance(x, MultiStatistics): 

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

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

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

138 for s in x.per_instance: 

139 if not isinstance(s, Statistics): 

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

141 

142 def n_points(self) -> int: 

143 """ 

144 Get the number of possible multi-statistics. 

145 

146 :return: just some arbitrary number 

147 """ 

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

149 

150 def __str__(self) -> str: 

151 """ 

152 Get the name of the multi-statistics space. 

153 

154 :return: the name 

155 """ 

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

157 

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

159 """ 

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

161 

162 :param logger: the logger for the parameters 

163 """ 

164 super().log_parameters_to(logger) 

165 for i, inst in enumerate(self.instances): 

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