Coverage for moptipyapps/prodsched/statistics.py: 93%
150 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-12 08:11 +0000
1"""
2A statistics record for the simulation.
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"""
13from itertools import chain
14from typing import Callable, Final, Generator, Iterable, Self
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 (
26 num_or_none_to_str,
27 str_to_num_or_none,
28)
29from pycommons.types import check_int_range, type_error
31#: the name of the statistics key
32COL_STAT: Final[str] = "stat"
33#: the total column name
34COL_TOTAL: Final[str] = "total"
35#: the statistics rate
36KEY_RATE: Final[str] = "rate"
37#: the product column prefix
38COL_PRODUCT_PREFIX: Final[str] = "product_"
39#: the mean TRP row
40ROW_TRP: Final[str] = "trp"
41#: the fill rate row
42ROW_SERVICE_LEVEL: Final[str] = "servicelevel"
43#: the CWT row
44ROW_CWT: Final[str] = "cwt"
45#: the mean stock level row
46ROW_STOCK_LEVEL_MEAN: Final[str] = \
47 f"stocklevel{SCOPE_SEPARATOR}{KEY_MEAN_ARITH}"
48#: the machine utilization row
49ROW_UTILIZATION_MEAN: Final[str] = \
50 f"utilization{SCOPE_SEPARATOR}{KEY_MEAN_ARITH}"
52#: the fulfilled rate
53ROW_FULFILLED_RATE: Final[str] = f"fulfilled{SCOPE_SEPARATOR}{KEY_RATE}"
54#: the simulation time getter
55ROW_SIMULATION_TIME: Final[str] = \
56 f"time{SCOPE_SEPARATOR}s{KEY_VALUE_SEPARATOR}"
58#: the statistics that we will print
59_STATS: tuple[tuple[str, Callable[[
60 StreamStatistics], int | float | None]], ...] = (
61 (KEY_MINIMUM, StreamStatistics.getter_or_none(KEY_MINIMUM)),
62 (KEY_MEAN_ARITH, StreamStatistics.getter_or_none(KEY_MEAN_ARITH)),
63 (KEY_MAXIMUM, StreamStatistics.getter_or_none(KEY_MAXIMUM)),
64 (KEY_STDDEV, StreamStatistics.getter_or_none(KEY_STDDEV)))
67class Statistics:
68 """
69 A statistics record based on production scheduling.
71 It provides the following statistics:
73 - :attr:`~service_levels`: The per-product-fillrate, i.e., the fraction
74 of demands of a given product that were immediately fulfilled when
75 arriving in the system (i.e., that were fulfilled by using product that
76 was available in the warehouse/in stock).
77 Higher values are good.
78 - :attr:`~service_level`: The overall fillrate, i.e., the total fraction
79 of demands that were immediately fulfilled upon arrival in the system
80 over all demands. That is, this is the fraction of demands that were
81 fulfilled by using product that was available in the warehouse/in stock.
82 Higher values are good.
83 - :attr:`~waiting_times`: The per-product waiting times ("CWT") for the
84 demands that came in but could *not* immediately be fulfilled. These are
85 the demands for a given product that were, so to say, not covered by the
86 fillrate/:attr:`~service_level`. If all demands of a product could
87 immediately be satisfied, then this is `None`.
88 Otherwise, smaller values are good.
89 - :attr:`~waiting_time`: The overall waiting times ("CWT") for the demands
90 that came in but could *not* immediately be fulfilled. These are all the
91 demands for a given product that were, so to say, not covered by the
92 fillrate/:attr:`~service_level`. If all demands could immediately be
93 satisfied, then this is `None`.
94 Otherwise, smaller values are good.
95 - :attr:`~production_times`: The per-product times that producing one unit
96 of the product takes from the moment that a production job is created
97 until it is completed. Smaller values of this "TRP" are better.
98 - :attr:`~production_time`: The overall statistics on the times that
99 producing one unit of any product takes from the moment that a
100 production job is created until it is completed. Smaller values this
101 "TRP" are better.
102 - :attr:`~fulfilled_rates`: The per-product fraction of demands that were
103 satisfied. Demands for a product may remain unsatisfied if they have not
104 been satisfied by the end of the simulation period. Larger values are
105 better.
106 - :attr:`~fulfilled_rate`: The fraction of demands that were satisfied.
107 Demands may remain unsatisfied if they have not been satisfied by the
108 end of the simulation period. Larger values are better.
109 - :attr:`~stock_levels`: The average amount of a given product in the
110 warehouse averaged over the simulation time. Smaller values are better.
111 - :attr:`~stock_level`: The total average amount units of any product in
112 the warehouse averaged over the simulation time. Smaller values are
113 better.
114 - :attr:`~utilizations`: The average per-workstation utilization. In other
115 words, the fraction of the non-warmup simulated time that each
116 workstation was busy.
117 - :attr:`~utilization`: The total average utilization, averaged over all
118 workstations.
119 - :attr:`~simulation_time_nanos`: The total time that the simulation took,
120 measured in nanoseconds.
122 Instances of this class are filled by
123 :class:`~moptipyapps.prodsched.statistics_collector.StatisticsCollector`
124 objects plugged into the
125 :class:`~moptipyapps.prodsched.simulation.Simulation`.
126 """
128 def __init__(self, n_products: int, n_stations: int) -> None:
129 """
130 Create the statistics record for a given number of products.
132 :param n_products: the number of products
133 :param n_stations: the number of stations
134 """
135 check_int_range(n_products, "n_products", 1, 1_000_000_000)
136 check_int_range(n_stations, "n_stations", 1, 1_000_000_000)
137 #: the production time (TRP) statistics per-product
138 self.production_times: Final[list[
139 StreamStatistics | None]] = [None] * n_products
140 #: the overall production time (TRP) statistics
141 self.production_time: StreamStatistics | None = None
142 #: the fraction of demands that were immediately satisfied,
143 #: on a per-product basis, i.e., the fillrate
144 self.service_levels: Final[list[int | float | None]] = (
145 [None] * n_products)
146 #: the overall fraction of immediately satisfied demands, i.e.,
147 #: the fillrate
148 self.service_level: int | float | None = None
149 #: the average waiting time for all demands that were not immediately
150 #: satisfied -- only counting demands that were actually satisfied,
151 #: i.e., the CWT
152 self.waiting_times: Final[list[
153 StreamStatistics | None]] = [None] * n_products
154 #: the overall waiting time for all demands that were not immediately
155 #: satisfied -- only counting demands that were actually satisfied,
156 #: i.e., the CWT
157 self.waiting_time: StreamStatistics | None = None
158 #: the fraction of demands that were fulfilled, on a per-product basis
159 self.fulfilled_rates: Final[list[
160 int | float | None]] = [None] * n_products
161 #: the fraction of demands that were fulfilled overall
162 self.fulfilled_rate: int | float | None = None
163 #: the average stock level, on a per-product basis
164 self.stock_levels: Final[list[
165 int | float | None]] = [None] * n_products
166 #: the overall average stock level
167 self.stock_level: int | float | None = None
168 #: the average machine utilization, on a per-station basis
169 self.utilizations: Final[list[
170 int | float | None]] = [None] * n_stations
171 #: the overall average utilization
172 self.utilization: int | float | None = None
173 #: the nanoseconds used by the simulation
174 self.simulation_time_nanos: int | float | None = None
176 def __str__(self) -> str:
177 """Convert this object to a string."""
178 return "\n".join(to_stream(self))
180 def clear(self) -> None:
181 """Clear all the data."""
182 n: Final[int] = list.__len__(self.production_times)
183 if n <= 0:
184 raise ValueError("Huh?")
185 for i in range(n):
186 self.production_times[i] = None
187 self.service_levels[i] = None
188 self.waiting_times[i] = None
189 self.fulfilled_rates[i] = None
190 self.stock_levels[i] = None
191 for i in range(list.__len__(self.utilizations)):
192 self.utilizations[i] = None
194 self.production_time = None
195 self.service_level = None
196 self.utilization = None
197 self.waiting_time = None
198 self.fulfilled_rate = None
199 self.stock_level = None
200 self.simulation_time_nanos = None
202 def copy_from(self, stat: "Statistics") -> None:
203 """
204 Copy the contents of another statistics record.
206 :param stat: the other statistics record
207 """
208 if not isinstance(stat, Statistics):
209 raise type_error(stat, "stat", Statistics)
210 self.production_times[:] = stat.production_times
211 self.production_time = stat.production_time
212 self.service_levels[:] = stat.service_levels
213 self.service_level = stat.service_level
214 self.waiting_times[:] = stat.waiting_times
215 self.waiting_time = stat.waiting_time
216 self.fulfilled_rates[:] = stat.fulfilled_rates
217 self.fulfilled_rate = stat.fulfilled_rate
218 self.stock_levels[:] = stat.stock_levels
219 self.stock_level = stat.stock_level
220 self.simulation_time_nanos = stat.simulation_time_nanos
221 self.utilizations[:] = stat.utilizations
222 self.utilization = stat.utilization
224 def from_stream(self, stream: Iterable[str]) -> Self:
225 """
226 Load the data from a stream.
228 Notice: The `n` values of the statistics records cannot be loaded.
229 They will be lost and just set to some more or less random number.
231 :param stream: the stream of data
232 :return: this object
233 """
234 self.clear()
236 n: Final[int] = list.__len__(self.production_times)
237 if n <= 0:
238 raise ValueError("Huh?")
240 keys: Final[set[str]] = {
241 f"{key}{SCOPE_SEPARATOR}{the_stat[0]}"
242 for key in (ROW_TRP, ROW_CWT) for the_stat in _STATS}
243 keys.update((ROW_SERVICE_LEVEL, ROW_FULFILLED_RATE,
244 ROW_STOCK_LEVEL_MEAN, ROW_UTILIZATION_MEAN))
245 sim_time_key: Final[str] = ROW_SIMULATION_TIME
247 data: dict[str, list[int | float | None]] = {}
248 sim_time: int | None = None
249 for srow in stream:
250 row = str.strip(srow)
251 if row.startswith(sim_time_key):
252 sim_time = check_int_range(
253 round(float(row[str.__len__(
254 sim_time_key):]) * 1_000_000_000),
255 sim_time_key, 0, 1_000_000_000_000_000_000_000_000)
256 if set.__len__(keys) <= 0:
257 break
258 continue
260 cols: list[str] = str.split(srow, CSV_SEPARATOR)
261 key: str = cols[0]
262 if (list.__len__(cols) <= (n + 1)) or (key not in keys):
263 continue
264 if key in data:
265 raise ValueError(f"Duplicate key '{key}'.")
266 data[key] = [str_to_num_or_none(cols[i]) for i in range(1, n + 2)]
267 keys.remove(key)
268 if (set.__len__(keys) <= 0) and (sim_time is not None):
269 break
271 if set.__len__(keys) > 0:
272 raise ValueError(f"Missing keys: {keys}")
273 if sim_time is None:
274 raise ValueError(f"Did not find key '{sim_time_key}'.")
276 self.simulation_time_nanos = sim_time
277 self.production_time = _split_data_stat(
278 data, ROW_TRP, self.production_times)
279 self.waiting_time = _split_data_stat(
280 data, ROW_CWT, self.waiting_times)
282 vals: list[int | float | None] = data[ROW_SERVICE_LEVEL]
283 self.service_level = vals[0]
284 self.service_levels[:] = vals[1:]
286 vals = data[ROW_STOCK_LEVEL_MEAN]
287 self.stock_level = vals[0]
288 self.stock_levels[:] = vals[1:]
290 vals = data[ROW_FULFILLED_RATE]
291 self.fulfilled_rate = vals[0]
292 self.fulfilled_rates[:] = vals[1:]
294 vals = data[ROW_UTILIZATION_MEAN]
295 self.utilization = vals[0]
296 self.utilizations[:] = vals[1:]
298 return self
301def _split_data_stat(data: dict[str, list[int | float | None]],
302 key: str,
303 dest: list[StreamStatistics | None]) \
304 -> StreamStatistics | None:
305 """
306 Split a data set.
308 :param data: the data set
309 :param dest: the destination list
310 :return: the main statistics, if any
311 """
312 key_min: Final[str] = f"{key}{SCOPE_SEPARATOR}{KEY_MINIMUM}"
313 key_mean: Final[str] = f"{key}{SCOPE_SEPARATOR}{KEY_MEAN_ARITH}"
314 key_max: Final[str] = f"{key}{SCOPE_SEPARATOR}{KEY_MAXIMUM}"
315 key_sd: Final[str] = f"{key}{SCOPE_SEPARATOR}{KEY_STDDEV}"
316 for i in range(1, list.__len__(dest) + 1):
317 dest[i - 1] = __stream_stats(data, key_min, key_mean, key_max,
318 key_sd, i)
319 return __stream_stats(data, key_min, key_mean, key_max, key_sd, 0)
322def __stream_stats(data: dict[str, list[int | float | None]],
323 key_min: str, key_mean: str, key_max: str, key_sd: str,
324 i: int) -> StreamStatistics | None:
325 """
326 Get a stream statistics.
328 :param data: the data array
329 :param key_min: the minimum key
330 :param key_mean: the mean key
331 :param key_max: the maximum key
332 :param key_sd: the standard deviation key
333 :param i: the index
334 :return: the statistics or `None`
335 """
336 the_min = data[key_min][i]
337 the_mean = data[key_mean][i]
338 the_max = data[key_max][i]
339 the_sd = data[key_sd][i]
340 if (the_min is None) or (the_max is None):
341 return None
342 if the_mean is None:
343 raise ValueError(
344 f"Invalid mean {the_mean} for min={the_min}, max={the_max}!")
345 return StreamStatistics(n=1 if the_sd is None else 100, minimum=the_min,
346 mean_arith=the_mean,
347 maximum=the_max, stddev=the_sd)
350def to_stream(stats: Statistics) -> Generator[str, None, None]:
351 """
352 Write a statistics record to a stream.
354 :param stats: the statistics record
355 :return: the stream of data
356 """
357 n_products: Final[int] = list.__len__(stats.production_times)
358 nts: Final[Callable[[int | float | None], str]] = num_or_none_to_str
360 yield str.join(CSV_SEPARATOR, chain((
361 COL_STAT, COL_TOTAL), (f"{COL_PRODUCT_PREFIX}{i}" for i in range(
362 n_products))))
364 for key, alle, single in (
365 (ROW_TRP, stats.production_times, stats.production_time),
366 (ROW_CWT, stats.waiting_times, stats.waiting_time)):
367 for stat, call in _STATS:
368 yield str.join(CSV_SEPARATOR, chain((
369 f"{key}{SCOPE_SEPARATOR}{stat}", nts(call(single))), (
370 map(nts, map(call, alle)))))
371 yield str.join(CSV_SEPARATOR, chain((
372 ROW_SERVICE_LEVEL, nts(stats.service_level)), (
373 map(nts, stats.service_levels))))
374 yield str.join(CSV_SEPARATOR, chain((
375 ROW_STOCK_LEVEL_MEAN, nts(stats.stock_level)), (
376 map(nts, stats.stock_levels))))
377 yield str.join(CSV_SEPARATOR, chain((
378 ROW_FULFILLED_RATE, nts(stats.fulfilled_rate)), (
379 map(nts, stats.fulfilled_rates))))
380 yield str.join(CSV_SEPARATOR, chain((
381 ROW_UTILIZATION_MEAN, nts(stats.utilization)), (
382 map(nts, stats.utilizations))))
383 yield f"{ROW_SIMULATION_TIME}{nts(
384 stats.simulation_time_nanos / 1_000_000_000)}"