Car Service Simulation - SimPy Tutorial
Priority car wash example demonstrating SimPy features like PriorityResource, FilterStore, and interrupts.
Level:Intermediate
Delays
Learn about delays in systems, how they create oscillations, and their impact on system behavior and decision-making.
Explore DelaysRunning a busy garage with SimPy
This longer example strings together a handful of SimPy resource types inside a small car‑service shop. We'll see Priority queues, Containers, FilterStores and more working in concert to keep the vehicles moving.
import random
from typing import Any, Callable, Dict, List
from tys import probe, progress
Run the car service simulation.
def simulate(cfg: Dict[str, Any]) -> Dict[str, int]:
import simpy
Environment & shared state
random.seed(cfg.get("seed", 42))
env = simpy.Environment()
pump = simpy.Resource(env, capacity=cfg.get("num_pumps", 2))
wash = simpy.PriorityResource(env, capacity=cfg.get("wash_bays", 1))
fuel_capacity = cfg.get("fuel_capacity", 1000)
fuel_tank = simpy.Container(env, init=cfg.get("fuel_init", 800),
capacity=fuel_capacity)
parts_store: simpy.FilterStore[str] = simpy.FilterStore(
env, capacity=cfg.get("parts_capacity", 30)
)
part_types: List[str] = cfg.get("part_types", ["brake", "filter", "wiper"])
for _ in range(cfg.get("initial_parts", 15)):
parts_store.put(random.choice(part_types))
mechanic = simpy.PreemptiveResource(env, capacity=1)
tanker_en_route = False
cars_served = 0
sim_done = env.event()
Helper – periodic probe sampler Record a metric at a fixed interval.
def record(name: str, getter: Callable[[], Any], interval: float = 0.5):
while True:
probe(name, env.now, getter())
yield env.timeout(interval)
Car service steps Fill the car's fuel tank.
def refuel_car():
with pump.request() as req:
yield req
fuel_needed = random.randint(*cfg.get("fuel_range", (20, 60)))
yield fuel_tank.get(fuel_needed)
yield env.timeout(fuel_needed * cfg.get("fuel_time", 0.1))
Run the car through the wash bay.
def wash_car(priority: int):
with wash.request(priority=priority) as req:
yield req
yield env.timeout(cfg.get("wash_time", 60))
Retrieve a part from storage if available.
def fetch_spare_part():
needed_part = random.choice(part_types)
try:
yield parts_store.get(lambda p: p == needed_part)
except simpy.exceptions.FilterError:
pass
Perform a repair, may be interrupted by audit.
def repair_car():
with mechanic.request(priority=1) as req:
yield req
try:
yield env.timeout(cfg.get("repair_time", 90))
except simpy.Interrupt:
yield env.timeout(cfg.get("audit_time", 15))
Drive until breakdown or completion.
def drive_round_trip():
breakdown = env.timeout(random.randint(*cfg.get("breakdown_after", (60, 120))))
drive_done = env.timeout(cfg.get("drive_time", 180))
outcome = yield env.any_of([breakdown, drive_done])
if breakdown in outcome:
yield from repair_car()
yield drive_done
Car life‑cycle orchestrator Complete the service workflow for one car.
def car(idx: int):
nonlocal cars_served
vip = (idx % cfg.get("vip_frequency", 5) == 0)
priority = 0 if vip else 1
yield from refuel_car()
yield from wash_car(priority)
yield from fetch_spare_part()
yield from drive_round_trip()
cars_served += 1
progress(int(100 * cars_served / cfg["num_cars"]))
if cars_served >= cfg["num_cars"] and not sim_done.triggered:
sim_done.succeed({"cars_serviced": cars_served})
Car‑arrival generator with Condition guard Spawn cars at random intervals.
def generator():
nonlocal tanker_en_route
for i in range(cfg["num_cars"]):
if fuel_tank.level < 0.10 * fuel_capacity and not tanker_en_route:
cond = env.condition(lambda: fuel_tank.level >= 0.10 * fuel_capacity
or tanker_en_route)
yield cond
env.process(car(i))
inter = random.expovariate(1.0 / cfg.get("arrival_interval", 45))
yield env.timeout(inter)
Re‑stocking: fuel & parts + tanker flag maintenance Refill fuel and restock parts.
def restock():
nonlocal tanker_en_route
while True:
if fuel_tank.level < 0.50 * fuel_capacity and not tanker_en_route:
tanker_en_route = True
yield env.timeout(cfg.get("tanker_time", 120))
yield fuel_tank.put(fuel_capacity - fuel_tank.level)
tanker_en_route = False
if len(parts_store.items) < cfg["parts_capacity"] / 2:
for _ in range(cfg["parts_capacity"] - len(parts_store.items)):
parts_store.put(random.choice(part_types))
yield env.timeout(cfg.get("restock_interval", 30))
Authority audits – explicit interrupt on mechanic’s current user Randomly audit the mechanic and interrupt if needed.
def authority():
while True:
yield env.timeout(random.expovariate(1.0 / cfg.get("audit_interval", 300)))
for req in list(mechanic.users):PreemptiveResource.users holds Request events, not the actual process objects, so interrupt the associated process
req.proc.interrupt("audit")
Kick‑off background tasks & metrics
env.process(generator())
env.process(restock())
env.process(authority())
env.process(record("fuel_level", lambda: fuel_tank.level))
env.process(record("wash_q", lambda: len(wash.queue)))
env.process(record("parts", lambda: len(parts_store.items)))
env.run(until=sim_done)
return sim_done.value
def requirements() -> Dict[str, List[str]]:
return {
"builtin": ["micropip", "pyyaml"],
"external": ["simpy==4.1.1"],
}
seed: 42
num_pumps: 2
wash_bays: 1
fuel_capacity: 1000
fuel_init: 800
fuel_range:
- 20
- 60
fuel_time: 0.1
wash_time: 60
drive_time: 180
breakdown_after:
- 60
- 120
repair_time: 90
arrival_interval: 45
restock_interval: 30
tanker_time: 120
audit_interval: 300
audit_time: 15
parts_capacity: 30
initial_parts: 15
part_types:
- brake
- filter
- wiper
vip_frequency: 5
num_cars: 25