Skip to content

Instantly share code, notes, and snippets.

@mbergal
Created August 8, 2022 13:25
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save mbergal/caf5d70f6bbcbcd1e11f5d541d1c6ecb to your computer and use it in GitHub Desktop.
Save mbergal/caf5d70f6bbcbcd1e11f5d541d1c6ecb to your computer and use it in GitHub Desktop.
import simpy
from scipy.stats import truncnorm
def get_truncated_normal(mean=0, sd=1, low=0, upp=1):
return truncnorm((low - mean) / sd, (upp - mean) / sd, loc=mean, scale=sd)
exam_queue = []
def r(env: simpy.Environment, name, delay: float):
yield env.timeout(delay)
print(f"{env.now} push")
e = env.event()
exam_queue.append(e)
yield e
print(f"wait: {env.now - delay}")
def process_queue(env, name):
while True:
if len(exam_queue) > 0:
a = exam_queue.pop()
yield env.timeout(0.033333333)
a.succeed()
print(f"{env.now} pop")
else:
yield env.timeout(0.001)
env = simpy.Environment()
[env.process(r(env, "fast", get_truncated_normal(mean=8, sd=2, low=0, upp=24).rvs())) for _ in range(20000)]
[env.process(process_queue(env, "process")) for _ in range(100)]
env.run(until=25)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment