Store gives you the next item. FilterStore lets you be picky.
The Problem Store Can’t Solve
Regular Store is FIFO—first in, first out. But what if you need a specific item?
# With regular Store
store = simpy.Store(env)
store.put({'type': 'red', 'id': 1})
store.put({'type': 'blue', 'id': 2})
store.put({'type': 'red', 'id': 3})
# You can only get item 1, then 2, then 3
# No way to say "give me a blue one"
FilterStore fixes this.
Basic Usage
import simpy
env = simpy.Environment()
store = simpy.FilterStore(env, capacity=10)
# Put some items
yield store.put({'type': 'red', 'value': 1})
yield store.put({'type': 'blue', 'value': 2})
yield store.put({'type': 'red', 'value': 3})
# Get specifically a blue item
blue = yield store.get(lambda x: x['type'] == 'blue')
print(blue) # {'type': 'blue', 'value': 2}
The filter function receives each item and returns True if it matches.
How Filtering Works
When you get() with a filter:
- FilterStore checks each item against your filter
- First matching item is returned
- If no match, the get blocks until a matching item arrives
# Will block until a green item is put
green = yield store.get(lambda x: x['type'] == 'green')
Real-World Example: Parts Bin
class Part:
def __init__(self, part_type, size, quality):
self.part_type = part_type
self.size = size
self.quality = quality
def __repr__(self):
return f"Part({self.part_type}, {self.size}, {self.quality})"
def supplier(env, bin):
"""Delivers random parts."""
types = ['bolt', 'nut', 'washer']
sizes = ['small', 'medium', 'large']
while True:
part = Part(
random.choice(types),
random.choice(sizes),
random.uniform(0.8, 1.0)
)
yield bin.put(part)
yield env.timeout(1)
def assembler(env, bin, needs_type, needs_size):
"""Assembles using specific parts."""
while True:
# Get exactly what we need
part = yield bin.get(
lambda p: p.part_type == needs_type and p.size == needs_size
)
print(f"Got {part} at {env.now}")
yield env.timeout(5) # Assembly time
env = simpy.Environment()
parts_bin = simpy.FilterStore(env)
env.process(supplier(env, parts_bin))
env.process(assembler(env, parts_bin, 'bolt', 'medium'))
env.process(assembler(env, parts_bin, 'nut', 'small'))
env.run(until=50)
Multiple Criteria
Combine conditions in the filter:
# Get a large red item
item = yield store.get(
lambda x: x['type'] == 'red' and x['size'] == 'large'
)
# Get any item with value > 10
item = yield store.get(lambda x: x['value'] > 10)
# Get by multiple possible types
item = yield store.get(lambda x: x['type'] in ['red', 'blue'])
Default Get (No Filter)
Without a filter, FilterStore behaves like regular Store:
# Get any item (FIFO)
item = yield store.get()
Checking Available Items
Peek at what’s in the store:
# All items
print(store.items)
# Check if any match
matching = [x for x in store.items if x['type'] == 'blue']
print(f"Blue items available: {len(matching)}")
But remember: checking doesn’t reserve. Another process might grab it first.
Non-Blocking Filter Get
What if you don’t want to wait?
def try_get_blue(env, store, timeout):
get_event = store.get(lambda x: x['type'] == 'blue')
timeout_event = env.timeout(timeout)
result = yield get_event | timeout_event
if get_event in result:
return get_event.value
else:
get_event.cancel()
return None
Priority with Filtering
Need priority AND filtering? Combine patterns:
class PrioritizedPart:
def __init__(self, priority, part_type, part_id):
self.priority = priority
self.part_type = part_type
self.part_id = part_id
def get_highest_priority_of_type(store, part_type):
"""Get highest priority item of specific type."""
matching = [p for p in store.items if p.part_type == part_type]
if not matching:
return None
matching.sort(key=lambda x: x.priority)
return matching[0]
Or use a custom store class.
FilterStore vs PriorityStore
| Feature | FilterStore | PriorityStore |
|---|---|---|
| Selection | By custom filter | By priority |
| Multiple criteria | Yes | No |
| Get any item | Yes | Yes (lowest priority) |
| Complexity | Higher | Lower |
Use FilterStore when selection criteria are complex. Use PriorityStore when you just need ordering.
Example: Taxi Dispatch
class Ride:
def __init__(self, pickup, destination, passengers):
self.pickup = pickup
self.destination = destination
self.passengers = passengers
def dispatch_taxi(env, ride_queue, taxi_type, capacity):
"""Taxi only takes rides it can handle."""
while True:
# Only get rides within capacity
ride = yield ride_queue.get(lambda r: r.passengers <= capacity)
print(f"{taxi_type} takes ride for {ride.passengers} from {ride.pickup}")
yield env.timeout(10)
env = simpy.Environment()
rides = simpy.FilterStore(env)
# Small taxi (up to 4 passengers)
env.process(dispatch_taxi(env, rides, "SmallTaxi", 4))
# Large taxi (up to 7 passengers)
env.process(dispatch_taxi(env, rides, "LargeTaxi", 7))
# Generate rides
def generate_rides(env, queue):
for i in range(20):
ride = Ride(f"Location{i}", f"Dest{i}", random.randint(1, 6))
yield queue.put(ride)
yield env.timeout(2)
env.process(generate_rides(env, rides))
env.run(until=100)
Performance Consideration
FilterStore checks items linearly. With thousands of items, this matters.
For high-performance scenarios:
- Keep stores small
- Use multiple stores by category
- Consider custom data structures
Common Gotchas
Filter Never Matches
# Blocks forever if no blue items ever arrive
blue = yield store.get(lambda x: x['type'] == 'blue')
Add timeouts for safety.
Modifying Items
item = yield store.get(lambda x: x['id'] == 5)
item['status'] = 'processed' # Changes the object
# If you put it back, filters will see the change
yield store.put(item)
Be careful about mutating items.
Summary
FilterStore:
- Lets you get specific items, not just FIFO
- Filter function returns True for matching items
- Blocks until a match is found
- Perfect for typed queues, dispatch systems, inventory
When FIFO isn’t enough, filter.

