Machines Simulation
Parts processing system showing resource allocation, machine utilization, and reliability with random breakdowns.
Level:Advanced
Emergence
Discover how interactions between parts can create properties and behaviors that no individual component possesses alone.
Explore EmergenceRouting parts to machines using tokens
A small fleet of machines take turns working on parts. We pass tokens around to announce when a machine is idle, which keeps the queue orderly even as the production rate rises and falls in a sine wave.
import math
import random
import json
from tys import probe, progress
Simulate a token-based routing system for machines.
def simulate(cfg: dict):
import simpy
env = simpy.Environment()
Parameters
num_machines = cfg["machines"]
base_rate = cfg["producer_rate_base"] # parts / sec
amp_rate = cfg["producer_rate_amp"] # sine amplitude
freq_rate = cfg["producer_rate_freq"] # cycles / sec
mean_service_time = cfg["service_time_mean"] # mean service time (exp)
breakdown_prob = cfg.get("breakdown_prob", 0.0) # chance of failure per part
repair_time_mean = cfg.get("repair_time_mean", 0.0) # avg downtime when failed
sim_time = cfg["sim_time"] # run length [sec]
initial_buffer = cfg.get("initial_buffer", 0)
Stores
buffer = simpy.Store(env)
idle_tokens = simpy.Store(env, capacity=num_machines)
buffer.items.extend([f'INIT_{i}' for i in range(initial_buffer)])
idle_tokens.items.extend(range(num_machines)) # one token per machine
util_time = [0.0] * num_machines # cumulative busy time
failures = [0] * num_machines # failure counts
done = env.event() # marks simulation end
Helper: sine-wave production rate Compute current production rate from a sine wave.
def sine(t: float) -> float:
return max(0.0, base_rate + amp_rate * math.sin(2 * math.pi * freq_rate * t))
Processes Generate parts at the current production rate.
def producer():
pid = 0
while True:
rate = sine(env.now)
probe("in_rate", env.now, rate)
for _ in range(int(rate)): # integer parts/second
buffer.put(f'P{pid}')
pid += 1
yield env.timeout(1)
env.process(producer())
dispatch_count = 0
Route parts to idle machines using tokens.
def router():
nonlocal dispatch_count
while True:
part_ev = buffer.get()
token_ev = idle_tokens.get()
out = yield env.all_of([part_ev, token_ev]) # {event: value}
part = out[part_ev]
mid = out[token_ev]
dispatch_count += 1
env.process(machine(mid, part))
env.process(router())
Process one part on a machine then release its token.
def machine(mid: int, part):
svc = random.expovariate(1.0 / mean_service_time)
yield env.timeout(svc)
util_time[mid] += svc
if breakdown_prob > 0 and random.random() < breakdown_prob:
failures[mid] += 1
probe("breakdown", env.now, mid)
down = random.expovariate(1.0 / repair_time_mean) if repair_time_mean > 0 else 0
if down > 0:
yield env.timeout(down)
probe("repair", env.now, mid)
yield idle_tokens.put(mid) # announce idle
Telemetry recorder running every 0.5 s.
def recorder():
nonlocal dispatch_count
interval = 0.5
while True:
buf_len = len(buffer.items)
probe("buffer_level", env.now, buf_len)
rate = dispatch_count / interval
probe("dispatch_rate", env.now, rate)
dispatch_count = 0
yield env.timeout(interval)
env.process(recorder())
Stop the simulation and report results.
def stopper():
yield env.timeout(sim_time)
done.succeed({
"buffer_final": len(buffer.items),
"utilisation": json.dumps({f"M{i}": round(util_time[i] / sim_time, 3) for i in range(num_machines)}),
"failures": json.dumps({f"M{i}": failures[i] for i in range(num_machines)})
})
env.process(stopper())
Run
env.run(until=done)
progress(100)
return done.value
def requirements():
return {
"builtin": ["micropip", "pyyaml"],
"external": ["simpy==4.1.1"],
}
machines: 3
producer_rate_base: 12
producer_rate_amp: 6
producer_rate_freq: 0.02
service_time_mean: 0.25
breakdown_prob: 0.05
repair_time_mean: 5
sim_time: 100
initial_buffer: 0