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

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. 

7""" 

8 

9from itertools import chain 

10from typing import Callable, Final, Generator 

11 

12from moptipy.utils.logger import KEY_VALUE_SEPARATOR 

13from pycommons.io.csv import CSV_SEPARATOR, SCOPE_SEPARATOR 

14from pycommons.math.stream_statistics import ( 

15 KEY_MAXIMUM, 

16 KEY_MEAN_ARITH, 

17 KEY_MINIMUM, 

18 KEY_STDDEV, 

19 StreamStatistics, 

20) 

21from pycommons.strings.string_conv import num_or_none_to_str 

22from pycommons.types import check_int_range, type_error 

23 

24#: the name of the statistics key 

25COL_STAT: Final[str] = "stat" 

26#: the total column name 

27COL_TOTAL: Final[str] = "total" 

28#: the statistics rate 

29KEY_RATE: Final[str] = "rate" 

30#: the product column prefix 

31COL_PRODUCT_PREFIX: Final[str] = "product_" 

32#: the mean TRP row 

33ROW_TRP: Final[str] = "trp" 

34#: the fill rate row 

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

36#: the CWT row 

37ROW_CWT: Final[str] = "cwt" 

38#: the mean stock level row 

39ROW_STOCK_LEVEL_MEAN: Final[str] = \ 

40 f"stocklevel{SCOPE_SEPARATOR}{KEY_MEAN_ARITH}" 

41#: the fulfilled rate 

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

43#: the simulation time getter 

44ROW_SIMULATION_TIME: Final[str] = \ 

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

46 

47#: the statistics that we will print 

48__STATS: tuple[tuple[str, Callable[[ 

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

50 (KEY_MINIMUM, StreamStatistics.getter_or_none(KEY_MINIMUM)), 

51 (KEY_MEAN_ARITH, StreamStatistics.getter_or_none(KEY_MEAN_ARITH)), 

52 (KEY_MAXIMUM, StreamStatistics.getter_or_none(KEY_MAXIMUM)), 

53 (KEY_STDDEV, StreamStatistics.getter_or_none(KEY_STDDEV))) 

54 

55 

56class Statistics: 

57 """A statistics record based on production scheduling.""" 

58 

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

60 """ 

61 Create the statistics record. 

62 

63 :param n_products: the number of products 

64 """ 

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

66 #: the production time statistics per-product 

67 self.production_times: Final[list[ 

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

69 #: the overall production time statistics 

70 self.production_time: StreamStatistics | None = None 

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

72 #: on a per-product basis 

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

74 [None] * n_products) 

75 #: the overall fraction of immediately satisfied demands 

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

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

78 #: satisfied -- only counting demands that were actually satisfied 

79 self.waiting_times: Final[list[ 

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

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

82 #: satisfied -- only counting demands that were actually satisfied 

83 self.waiting_time: StreamStatistics | None = None 

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

85 self.fulfilled_rates: Final[list[ 

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

87 #: the fraction of demands that were fulfilled overall 

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

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

90 self.stock_levels: Final[list[ 

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

92 #: the overall average stock level 

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

94 #: the nano seconds used by the simulation 

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

96 

97 def __str__(self) -> str: 

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

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

100 

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

102 """ 

103 Copy the contents of another statistics record. 

104 

105 :param stat: the other statistics record 

106 """ 

107 if not isinstance(stat, Statistics): 

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

109 self.production_times[:] = stat.production_times 

110 self.production_time = stat.production_time 

111 self.immediate_rates[:] = stat.immediate_rates 

112 self.immediate_rate = stat.immediate_rate 

113 self.waiting_times[:] = stat.waiting_times 

114 self.waiting_time = stat.waiting_time 

115 self.fulfilled_rates[:] = stat.fulfilled_rates 

116 self.fulfilled_rate = stat.fulfilled_rate 

117 self.stock_levels[:] = stat.stock_levels 

118 self.stock_level = stat.stock_level 

119 self.simulation_time_nanos = stat.simulation_time_nanos 

120 

121 

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

123 """ 

124 Write a statistics record to a stream. 

125 

126 :param stats: the statistics record 

127 :return: the stream of data 

128 """ 

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

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

131 

132 yield str.join(CSV_SEPARATOR, chain(( 

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

134 n_products)))) 

135 

136 for key, alle, single in ( 

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

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

139 for stat, call in __STATS: 

140 yield str.join(CSV_SEPARATOR, chain(( 

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

142 map(nts, map(call, alle))))) 

143 yield str.join(CSV_SEPARATOR, chain(( 

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

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

146 yield str.join(CSV_SEPARATOR, chain(( 

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

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

149 yield str.join(CSV_SEPARATOR, chain(( 

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

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

152 yield f"{ROW_SIMULATION_TIME}{nts( 

153 stats.simulation_time_nanos / 1_000_000_000)}"