Coverage for moptipyapps/prodsched/statistics_collector.py: 97%
137 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-12 08:11 +0000
1r"""
2A tool for collecting statistics from an MFC simulation.
4A statistics collector :class:`~StatisticsCollector` is a special
5:class:`~moptipyapps.prodsched.simulation.Listener`
6that can be plugged into a
7:class:`~moptipyapps.prodsched.simulation.Simulation`.
8During the execution of the simulation, it gathers statistics about what is
9going on. It finally stores these into a
10:class:`~moptipyapps.prodsched.statistics.Statistics` record.
11Such a record can then be used to understand the key characteristics of the
12behavior of the simulation in on a given
13:class:`~moptipyapps.prodsched.instance.Instance`.
15The simulation listeners (:class:`~moptipyapps.prodsched.simulation.Listener`)
16offer a method to pipe out data from arbitrary subclasses of
17:class:`~moptipyapps.prodsched.simulation.Simulation`.
18This means that they allow us to access data in a unified way, regardless of
19which manufacturing logic or production scheduling we actually implement.
21In the case of the :class:`~StatisticsCollector` implemented here, we
22implement the :class:`~moptipyapps.prodsched.simulation.Listener`-API to
23fill a :class:`~moptipyapps.prodsched.statistics.Statistics` record with data.
24Such records offer the standard statistics that Thürer et al. used in their
25works.
26In other words, we can make such statistics available and accessible,
27regardless of how our simulation schedules the production.
29Moreover, multiple such :class:`~moptipyapps.prodsched.statistics.Statistics`
30records, filled with data from simulations over multiple different instances
31(:mod:`~moptipyapps.prodsched.instance`) can be combined in a
32:class:`~moptipyapps.prodsched.multistatistics.MultiStatistics` record.
33This record, which will comprehensively represent performance over several
34independent instances, then can be used as basis for objective functions
35(:class:`~moptipy.api.objective.Objective`) such as those given in
36:mod:`~moptipyapps.prodsched.objectives`.
38>>> instance = Instance(
39... name="test2", n_products=2, n_customers=1, n_stations=2, n_demands=5,
40... time_end_warmup=21, time_end_measure=10000,
41... routes=[[0, 1], [1, 0]],
42... demands=[[0, 0, 1, 10, 20, 90], [1, 0, 0, 5, 22, 200],
43... [2, 0, 1, 7, 30, 200], [3, 0, 1, 6, 60, 200],
44... [4, 0, 0, 125, 5, 2000]],
45... warehous_at_t0=[2, 1],
46... station_product_unit_times=[[[10.0, 50.0, 15.0, 100.0],
47... [ 5.0, 20.0, 7.0, 35.0, 4.0, 50.0]],
48... [[ 5.0, 24.0, 7.0, 80.0],
49... [ 3.0, 21.0, 6.0, 50.0,]]])
52>>> from moptipyapps.prodsched.simulation import Simulation
53>>> statistics = Statistics(instance.n_products, instance.n_stations)
54>>> collector = StatisticsCollector(instance)
55>>> collector.set_dest(statistics)
56>>> simulation = Simulation(instance, collector)
57>>> simulation.ctrl_run()
58>>> from moptipyapps.prodsched.statistics import to_stream
59>>> text = list(to_stream(statistics))
60>>> print("\n".join(text[:-1]))
61stat;total;product_0;product_1
62trp.min;229.57142857142858;455.6;229.57142857142858
63trp.mean;304.8888888888889;455.6;246.92307692307693
64trp.max;455.6;455.6;267.1666666666667
65trp.sd;97.56326157594913;0;19.50721118346466
66cwt.min;1603;2278;1603
67cwt.mean;1829.3333333333333;2278;1605
68cwt.max;2278;2278;1607
69cwt.sd;388.56187838403986;;2.8284271247461903
70servicelevel;0;0;0
71stocklevel.mean;0.6078765407355446;0.4497444633730835;0.15813207736246118
72fulfilled.rate;1;1;1
73utilization.mean;0.12711694558573003;0.1645455456458563;0.08968834552560377
75>>> statistics_2 = Statistics(instance.n_products, instance.n_stations)
76>>> statistics_2 = statistics_2.from_stream(text)
77>>> print("\n".join(list(to_stream(statistics_2))[:-1]))
78stat;total;product_0;product_1
79trp.min;229.57142857142858;455.6;229.57142857142858
80trp.mean;304.8888888888889;455.6;246.92307692307693
81trp.max;455.6;455.6;267.1666666666667
82trp.sd;97.56326157594913;0;19.50721118346466
83cwt.min;1603;2278;1603
84cwt.mean;1829.3333333333333;2278;1605
85cwt.max;2278;2278;1607
86cwt.sd;388.56187838403986;;2.8284271247461903
87servicelevel;0;0;0
88stocklevel.mean;0.6078765407355446;0.4497444633730835;0.15813207736246118
89fulfilled.rate;1;1;1
90utilization.mean;0.12711694558573003;0.1645455456458563;0.08968834552560377
91"""
93from time import time_ns
94from typing import Final
96from pycommons.math.stream_statistics import (
97 StreamStatistics,
98 StreamStatisticsAggregate,
99)
100from pycommons.math.streams import StreamSum
101from pycommons.types import type_error
103from moptipyapps.prodsched.instance import (
104 Demand,
105 Instance,
106)
107from moptipyapps.prodsched.simulation import Job, Listener
108from moptipyapps.prodsched.statistics import Statistics
111class StatisticsCollector(Listener):
112 """A listener for simulation events that collects basic statistics."""
114 def __init__(self, instance: Instance) -> None:
115 """
116 Initialize the listener.
118 :param instance: the instance
119 """
120 if not isinstance(instance, Instance):
121 raise type_error(instance, "instance", Instance)
122 #: the destination to write to.
123 self.__dest: Statistics | None = None
124 #: the end of the warmup period
125 self.__warmup: Final[float] = instance.time_end_warmup
126 #: the total time window length
127 self.__total: Final[float] = instance.time_end_measure
128 #: the number of products
129 self.__n_products: Final[int] = instance.n_products
130 #: the measurable demands per product
131 self.__n_mdpb: Final[tuple[int, ...]] = (
132 instance.n_measurable_demands_per_product)
134 n_products: Final[int] = instance.n_products
135 #: the internal per-product production time records
136 self.__production_times: Final[tuple[
137 StreamStatisticsAggregate[StreamStatistics], ...]] = tuple(
138 StreamStatistics.aggregate() for _ in range(n_products))
139 #: the internal total production time
140 self.__production_time: Final[StreamStatisticsAggregate[
141 StreamStatistics]] = StreamStatistics.aggregate()
142 #: the immediately satisfied products: satisfied vs. total
143 self.__immediately_satisfied: Final[list[int]] = [0] * n_products
144 #: the statistics of the waiting time for not-immediately satisfied
145 #: products, per product
146 self.__waiting_times: Final[tuple[
147 StreamStatisticsAggregate[StreamStatistics], ...]] = tuple(
148 StreamStatistics.aggregate() for _ in range(n_products))
149 #: the total waiting time for non-immediately satisfied products
150 self.__waiting_time: Final[StreamStatisticsAggregate[
151 StreamStatistics]] = StreamStatistics.aggregate()
152 #: the number of fulfilled jobs, per-product
153 self.__fulfilled: Final[list[int]] = [0] * n_products
154 #: the stock levels on a per-product basis
155 self.__stock_levels: Final[tuple[StreamSum, ...]] = tuple(
156 StreamSum() for _ in range(n_products))
157 #: the total stock level sum
158 self.__stock_level: Final[StreamSum] = StreamSum()
159 #: the utilization on a per-machine basis
160 self.__utilizations: Final[tuple[StreamSum, ...]] = tuple(
161 StreamSum() for _ in range(instance.n_stations))
162 #: the total machine utilization sum
163 self.__utilization: Final[StreamSum] = StreamSum()
164 #: the last time machine became busy
165 self.__busy_start: Final[list[float | None]] = (
166 [None] * instance.n_stations)
167 #: the number of products currently in warehouse and since when
168 #: they were there
169 self.__in_warehouse: Final[tuple[list[float], ...]] = tuple([
170 0.0, 0.0] for _ in range(n_products))
171 #: the stat time of the simulation
172 self.__start: int = -1
174 def set_dest(self, dest: Statistics) -> None:
175 """
176 Set the statistics record to fill.
178 :param dest: the destination
179 """
180 if not isinstance(dest, Statistics):
181 raise type_error(dest, "dest", Statistics)
182 self.__dest = dest
184 def start(self) -> None:
185 """Clear all the data of the collector."""
186 if self.__dest is None:
187 raise ValueError("Need destination statistics!")
188 self.__start = time_ns()
189 n_products: Final[int] = self.__n_products
190 for i in range(n_products):
191 self.__production_times[i].reset()
192 self.__immediately_satisfied[i] = 0
193 self.__waiting_times[i].reset()
194 self.__fulfilled[i] = 0
195 self.__stock_levels[i].reset()
196 wh = self.__in_warehouse[i]
197 wh[0] = 0.0
198 wh[1] = 0.0
199 for i, m in enumerate(self.__utilizations):
200 m.reset()
201 self.__busy_start[i] = None
202 self.__production_time.reset()
203 self.__waiting_time.reset()
204 self.__stock_level.reset()
205 self.__utilization.reset()
207 def product_in_warehouse(
208 self, time: float, product_id: int, amount: int,
209 is_in_measure_period: bool) -> None:
210 """
211 Report a change of the amount of products in the warehouse.
213 :param time: the current time
214 :param product_id: the product ID
215 :param amount: the new absolute total amount of that product in the
216 warehouse
217 :param is_in_measure_period: is this event inside the measurement
218 period?
219 """
220 iwh: Final[list[float]] = self.__in_warehouse[product_id]
221 if is_in_measure_period:
222 value: float = (time - max(iwh[1], self.__warmup)) * iwh[0]
223 self.__stock_levels[product_id].add(value)
224 self.__stock_level.add(value)
225 iwh[0] = amount
226 iwh[1] = time
228 def produce_at_begin(
229 self, time: float, station_id: int, job: Job) -> None:
230 """
231 Report the start of the production of a certain product at a station.
233 :param time: the current time
234 :param station_id: the station ID
235 :param job: the production job
236 """
237 self.__busy_start[station_id] = time
239 def produce_at_end(
240 self, time: float, station_id: int, job: Job) -> None:
241 """
242 Report the completion of the production of a product at a station.
244 :param time: the current time
245 :param station_id: the station ID
246 :param job: the production job
247 """
248 if time > self.__warmup:
249 add = time - max(self.__warmup, self.__busy_start[
250 station_id])
251 self.__utilizations[station_id].add(add)
252 self.__utilization.add(add)
253 self.__busy_start[station_id] = None
254 if job.measure and job.completed:
255 am: Final[int] = job.amount
256 tt: float = time - job.arrival
257 if am <= 1:
258 self.__production_times[job.product_id].add(tt)
259 self.__production_time.add(tt)
260 else: # deal with amounts > 1
261 tt /= am
262 for _ in range(am):
263 self.__production_times[job.product_id].add(tt)
264 self.__production_time.add(tt)
266 def demand_satisfied(
267 self, time: float, demand: Demand) -> None:
268 """
269 Report that a given demand has been satisfied.
271 :param time: the time index when the demand was satisfied
272 :param demand: the demand that was satisfied
273 """
274 if demand.measure:
275 at: Final[float] = demand.arrival
276 pid: Final[int] = demand.product_id
277 if at >= time:
278 self.__immediately_satisfied[pid] += 1
279 else:
280 tt: Final[float] = time - at
281 self.__waiting_times[pid].add(tt)
282 self.__waiting_time.add(tt)
283 self.__fulfilled[pid] += 1
285 def finished(self, time: float) -> None:
286 """
287 Fill the collected statistics into the statistics record.
289 :param time: the time when we are finished
290 """
291 dest: Final[Statistics | None] = self.__dest
292 if dest is None:
293 raise ValueError("Lost destination statistics record?")
294 total: Final[tuple[int, ...]] = self.__n_mdpb
296 for i, stat in enumerate(self.__production_times):
297 dest.production_times[i] = stat.result_or_none()
298 dest.production_time = self.__production_time.result_or_none()
300 sn: int = 0
301 sf: int = 0
302 st: int = 0
303 for i, n in enumerate(self.__immediately_satisfied):
304 t = total[i]
305 dest.service_levels[i] = n / t
306 sn += n
307 f = self.__fulfilled[i]
308 dest.fulfilled_rates[i] = f / t
309 sf += f
310 st += t
311 dest.service_level = sn / st
312 dest.fulfilled_rate = sf / st
314 for i, stat in enumerate(self.__waiting_times):
315 dest.waiting_times[i] = stat.result_or_none()
316 dest.waiting_time = self.__waiting_time.result_or_none()
318 slm: StreamSum = self.__stock_level
319 tt: Final[float] = self.__total
320 twl: Final[float] = tt - self.__warmup
321 wu: Final[float] = self.__warmup
322 for i, sm in enumerate(self.__stock_levels):
323 wh = self.__in_warehouse[i]
324 v: float = (tt - max(wh[1], wu)) * wh[0]
325 sm.add(v)
326 slm.add(v)
327 dest.stock_levels[i] = sm.result() / twl
328 dest.stock_level = slm.result() / twl
330 slm = self.__utilization
331 for i, sm in enumerate(self.__utilizations):
332 wx = self.__busy_start[i]
333 if wx is not None:
334 v = tt - max(wu, wx)
335 sm.add(v)
336 slm.add(v)
337 dest.utilizations[i] = sm.result() / twl
338 dest.utilization = slm.result() / (twl * tuple.__len__(
339 self.__utilizations))
341 dest.simulation_time_nanos = time_ns() - self.__start