in source/synthetic/create_synthetic_data.py [0:0]
def _generate_data(self, upto: int):
"""
Generate samples from the model up to a specific time
:param upto: the period to sample up to
:return: the last sample
"""
# pre-allocate some zeros up to the length requested
if len(self._data) < upto:
self._data.extend([0 for _ in range(upto - len(self._data) + 1)])
period = 0
last_percent = 0.0
while self._arrival_time < upto + 1:
p = np.random.random()
# get the rate for the current time
rate = self.rate_at(self._arrival_time)
# using the inverse CDF of the poisson distribution
# this algorithm is slow for small rates. consider
# using accept/reject in a future release
if rate == 0:
self._arrival_time += 1.0 / self._bucket_size_m
continue
else:
_inter_arrival_time = -np.log(1.0 - p) / rate
self._arrival_time += _inter_arrival_time
next_period = int(np.floor(self._arrival_time))
if next_period != period:
new_percent = 100 * period / (upto + 1)
if int(new_percent) != int(last_percent):
logger.debug(
"%3d percent complete on %s" % (int(new_percent), self.name)
)
last_percent = new_percent
period = next_period
# accommodate overshoot
if period >= len(self._data):
self._data.extend([0 for _ in range(period - len(self._data) + 1)])
self._data[period] += 1
logger.debug("%3d percent complete on %s" % (100, self.name))
self._last_updated = period - 1
return self._data[upto]