Coverage for moptipyapps / prodsched / statistics.py: 98%

1""" 

2A statistics record for the simulation. 

3 

4This module provides a record with statistics derived from one single 

5MFC simulation. It can store values such as the mean fill rate or the 

6mean stock level. 

7Such statistics records are filled in by instances of the 

8:class:`~moptipyapps.prodsched.statistics_collector.StatisticsCollector` 

9plugged into the 

10:class:`~moptipyapps.prodsched.simulation.Simulation`. 

11""" 

12 

13from itertools import chain 

14from typing import Callable, Final, Generator 

15 

16from moptipy.utils.logger import KEY_VALUE_SEPARATOR 

17from pycommons.io.csv import CSV_SEPARATOR, SCOPE_SEPARATOR 

18from pycommons.math.stream_statistics import ( 

19 KEY_MAXIMUM, 

20 KEY_MEAN_ARITH, 

21 KEY_MINIMUM, 

22 KEY_STDDEV, 

23 StreamStatistics, 

24) 

25from pycommons.strings.string_conv import num_or_none_to_str 

26from pycommons.types import check_int_range, type_error 

27 

28#: the name of the statistics key 

29COL_STAT: Final[str] = "stat" 

30#: the total column name 

31COL_TOTAL: Final[str] = "total" 

32#: the statistics rate 

33KEY_RATE: Final[str] = "rate" 

34#: the product column prefix 

35COL_PRODUCT_PREFIX: Final[str] = "product_" 

36#: the mean TRP row 

37ROW_TRP: Final[str] = "trp" 

38#: the fill rate row 

39ROW_FILL_RATE: Final[str] = f"fill{SCOPE_SEPARATOR}{KEY_RATE}" 

40#: the CWT row 

41ROW_CWT: Final[str] = "cwt" 

42#: the mean stock level row 

43ROW_STOCK_LEVEL_MEAN: Final[str] = \ 

44 f"stocklevel{SCOPE_SEPARATOR}{KEY_MEAN_ARITH}" 

45#: the fulfilled rate 

46ROW_FULFILLED_RATE: Final[str] = f"fulfilled{SCOPE_SEPARATOR}{KEY_RATE}" 

47#: the simulation time getter 

48ROW_SIMULATION_TIME: Final[str] = \ 

49 f"time{SCOPE_SEPARATOR}s{KEY_VALUE_SEPARATOR}" 

50 

51#: the statistics that we will print 

52__STATS: tuple[tuple[str, Callable[[ 

53 StreamStatistics], int | float | None]], ...] = ( 

54 (KEY_MINIMUM, StreamStatistics.getter_or_none(KEY_MINIMUM)), 

55 (KEY_MEAN_ARITH, StreamStatistics.getter_or_none(KEY_MEAN_ARITH)), 

56 (KEY_MAXIMUM, StreamStatistics.getter_or_none(KEY_MAXIMUM)), 

57 (KEY_STDDEV, StreamStatistics.getter_or_none(KEY_STDDEV))) 

58 

59 

60class Statistics: 

61 """ 

62 A statistics record based on production scheduling. 

63 

64 It provides the following statistics: 

65 

66 - :attr:`~immediate_rates`: The per-product-fillrate, i.e., the fraction 

67 of demands of a given product that were immediately fulfilled when 

68 arriving in the system (i.e., that were fulfilled by using product that 

69 was available in the warehouse/in stock). 

70 Higher values are good. 

71 - :attr:`~immediate_rate`: The overall fillrate, i.e., the total fraction 

72 of demands that were immediately fulfilled upon arrival in the system 

73 over all demands. That is, this is the fraction of demands that were 

74 fulfilled by using product that was available in the warehouse/in stock. 

75 Higher values are good. 

76 - :attr:`~waiting_times`: The per-product waiting times ("CWT") for the 

77 demands that came in but could *not* immediately be fulfilled. These are 

78 the demands for a given product that were, so to say, not covered by the 

79 fillrate/:attr:`~immediate_rate`. If all demands of a product could 

80 immediately be satisfied, then this is `None`. 

81 Otherwise, smaller values are good. 

82 - :attr:`~waiting_time`: The overall waiting times ("CWT") for the demands 

83 that came in but could *not* immediately be fulfilled. These are all the 

84 demands for a given product that were, so to say, not covered by the 

85 fillrate/:attr:`~immediate_rate`. If all demands could immediately be 

86 satisfied, then this is `None`. 

87 Otherwise, smaller values are good. 

88 - :attr:`~production_times`: The per-product times that producing one unit 

89 of the product takes from the moment that a production job is created 

90 until it is completed. Smaller values of this "TRP" are better. 

91 - :attr:`~production_time`: The overall statistics on the times that 

92 producing one unit of any product takes from the moment that a 

93 production job is created until it is completed. Smaller values this 

94 "TRP" are better. 

95 - :attr:`~fulfilled_rates`: The per-product fraction of demands that were 

96 satisfied. Demands for a product may remain unsatisfied if they have not 

97 been satisfied by the end of the simulation period. Larger values are 

98 better. 

99 - :attr:`~fulfilled_rate`: The fraction of demands that were satisfied. 

100 Demands may remain unsatisfied if they have not been satisfied by the 

101 end of the simulation period. Larger values are better. 

102 - :attr:`~stock_levels`: The average amount of a given product in the 

103 warehouse averaged over the simulation time. Smaller values are better. 

104 - :attr:`~stock_level`: The total average amount units of any product in 

105 the warehouse averaged over the simulation time. Smaller values are 

106 better. 

107 - :attr:`~simulation_time_nanos`: The total time that the simulation took, 

108 measured in nanoseconds. 

109 

110 Instances of this class are filled by 

111 :class:`~moptipyapps.prodsched.statistics_collector.StatisticsCollector` 

112 objects plugged into the 

113 :class:`~moptipyapps.prodsched.simulation.Simulation`. 

114 """ 

115 

116 def __init__(self, n_products: int) -> None: 

117 """ 

118 Create the statistics record for a given number of products. 

119 

120 :param n_products: the number of products 

121 """ 

122 check_int_range(n_products, "n_products", 1, 1_000_000_000) 

123 #: the production time (TRP) statistics per-product 

124 self.production_times: Final[list[ 

125 StreamStatistics | None]] = [None] * n_products 

126 #: the overall production time (TRP) statistics 

127 self.production_time: StreamStatistics | None = None 

128 #: the fraction of demands that were immediately satisfied, 

129 #: on a per-product basis, i.e., the fillrate 

130 self.immediate_rates: Final[list[int | float | None]] = ( 

131 [None] * n_products) 

132 #: the overall fraction of immediately satisfied demands, i.e., 

133 #: the fillrate 

134 self.immediate_rate: int | float | None = None 

135 #: the average waiting time for all demands that were not immediately 

136 #: satisfied -- only counting demands that were actually satisfied, 

137 #: i.e., the CWT 

138 self.waiting_times: Final[list[ 

139 StreamStatistics | None]] = [None] * n_products 

140 #: the overall waiting time for all demands that were not immediately 

141 #: satisfied -- only counting demands that were actually satisfied, 

142 #: i.e., the CWT 

143 self.waiting_time: StreamStatistics | None = None 

144 #: the fraction of demands that were fulfilled, on a per-product basis 

145 self.fulfilled_rates: Final[list[ 

146 int | float | None]] = [None] * n_products 

147 #: the fraction of demands that were fulfilled overall 

148 self.fulfilled_rate: int | float | None = None 

149 #: the average stock level, on a per-product basis 

150 self.stock_levels: Final[list[ 

151 int | float | None]] = [None] * n_products 

152 #: the overall average stock level 

153 self.stock_level: int | float | None = None 

154 #: the nanoseconds used by the simulation 

155 self.simulation_time_nanos: int | float | None = None 

156 

157 def __str__(self) -> str: 

158 """Convert this object to a string.""" 

159 return "\n".join(to_stream(self)) 

160 

161 def copy_from(self, stat: "Statistics") -> None: 

162 """ 

163 Copy the contents of another statistics record. 

164 

165 :param stat: the other statistics record 

166 """ 

167 if not isinstance(stat, Statistics): 

168 raise type_error(stat, "stat", Statistics) 

169 self.production_times[:] = stat.production_times 

170 self.production_time = stat.production_time 

171 self.immediate_rates[:] = stat.immediate_rates 

172 self.immediate_rate = stat.immediate_rate 

173 self.waiting_times[:] = stat.waiting_times 

174 self.waiting_time = stat.waiting_time 

175 self.fulfilled_rates[:] = stat.fulfilled_rates 

176 self.fulfilled_rate = stat.fulfilled_rate 

177 self.stock_levels[:] = stat.stock_levels 

178 self.stock_level = stat.stock_level 

179 self.simulation_time_nanos = stat.simulation_time_nanos 

180 

181 

182def to_stream(stats: Statistics) -> Generator[str, None, None]: 

183 """ 

184 Write a statistics record to a stream. 

185 

186 :param stats: the statistics record 

187 :return: the stream of data 

188 """ 

189 n_products: Final[int] = list.__len__(stats.production_times) 

190 nts: Final[Callable[[int | float | None], str]] = num_or_none_to_str 

191 

192 yield str.join(CSV_SEPARATOR, chain(( 

193 COL_STAT, COL_TOTAL), (f"{COL_PRODUCT_PREFIX}{i}" for i in range( 

194 n_products)))) 

195 

196 for key, alle, single in ( 

197 (ROW_TRP, stats.production_times, stats.production_time), 

198 (ROW_CWT, stats.waiting_times, stats.waiting_time)): 

199 for stat, call in __STATS: 

200 yield str.join(CSV_SEPARATOR, chain(( 

201 f"{key}{SCOPE_SEPARATOR}{stat}", nts(call(single))), ( 

202 map(nts, map(call, alle))))) 

203 yield str.join(CSV_SEPARATOR, chain(( 

204 ROW_FILL_RATE, nts(stats.immediate_rate)), ( 

205 map(nts, stats.immediate_rates)))) 

206 yield str.join(CSV_SEPARATOR, chain(( 

207 ROW_STOCK_LEVEL_MEAN, nts(stats.stock_level)), ( 

208 map(nts, stats.stock_levels)))) 

209 yield str.join(CSV_SEPARATOR, chain(( 

210 ROW_FULFILLED_RATE, nts(stats.fulfilled_rate)), ( 

211 map(nts, stats.fulfilled_rates)))) 

212 yield f"{ROW_SIMULATION_TIME}{nts( 

213 stats.simulation_time_nanos / 1_000_000_000)}"