coopstorage

A Python library for embedded storage tracking — models physical storage locations, containers, and the channel-processor rules that govern how items flow in and out of each slot.

Overview

coopstorage provides:

• Locations — physical storage slots with configurable capacity, dimensions, and channel-processor type
• Containers — items stored at locations, each with a UoM, optional contents, and resource qualifiers
• Channel processors — pluggable rules that control which slots are addable/removable (FIFO, LIFO, OMNI, NoFlow, push variants, etc.)
• Transfer requests — declarative add/move/remove operations resolved against qualifier criteria
• Location qualifiers — filter locations by capacity, dimensions, UoM, resource type, occupied state, channel accessibility, and more
• Live visualizer — browser-based isometric/top-down view with real-time SSE updates, slot overlays, UOM icons, and a legend

Channel Processor Types

Install

coopstorage ships a lean core with optional extras for the API server and persistence backends:

pip install coopstorage                       # core simulation engine only
pip install coopstorage[api]                  # + FastAPI visualizer/API server
pip install coopstorage[persistence]          # + SQL (Postgres/SQLite) and MongoDB stores
pip install coopstorage[api,persistence]      # both
pip install coopstorage[all]                  # everything

Quick Start

from coopstorage.storage.loc_load.storage import Storage
from coopstorage.storage.loc_load.location import Location
import coopstorage.storage.loc_load.dcs as dcs
import coopstorage.storage.loc_load.channel_processors as cps
from coopstorage.storage.loc_load.qualifiers import LocationQualifier
from coopstorage.storage.loc_load.transferRequest import TransferRequestCriteria

# Build storage with two locations
loc_a = Location(
    id='A',
    location_meta=dcs.LocationMeta(
        dims=(10, 10, 5),
        channel_processor=cps.FIFOFlowChannelProcessor(),
        capacity=5,
    ),
    coords=(0, 0, 0),
)
loc_b = Location(
    id='B',
    location_meta=dcs.LocationMeta(
        dims=(10, 10, 5),
        channel_processor=cps.LIFOFlowBackwardChannelProcessor(),
        capacity=5,
    ),
    coords=(15, 0, 0),
)

storage = Storage(locs=[loc_a, loc_b])

# Add a container to the first available location
storage.handle_transfer_requests([
    TransferRequestCriteria(
        new_container=dcs.Container(id='C001'),
        dest_loc_query_args=LocationQualifier(at_least_capacity=1, has_addable_position=True),
    )
])

Location Qualifiers

LocationQualifier filters candidate locations during transfer resolution:

from coopstorage.storage.loc_load.qualifiers import LocationQualifier, ContainerQualifier
from cooptools.qualifiers import PatternMatchQualifier, WhiteBlackListQualifier

# Destination: must have capacity, an accessible drop slot, and accept the container's UoM
LocationQualifier(
    at_least_capacity=1,
    has_addable_position=True,
)

# Source: must contain a container matching a pattern
LocationQualifier(
    has_all_containers=[ContainerQualifier(pattern=PatternMatchQualifier(regex='^C00'))]
)

Key parameters:

Visualizer

Start a live browser visualizer alongside a benchmark or simulation:

# Showcase: one of each channel processor type, lock-step add/remove
python run_viz_benchmark.py --mode showcase

# Continuous randomized simulation
python run_viz_benchmark.py --mode sim --config small

# Fixed benchmark workload
python run_viz_benchmark.py --mode benchmark --config medium

# Options
python run_viz_benchmark.py --delay 0.05 --log-level DEBUG --port 1219

The visualizer is served at http://localhost:1219/static/index.html.

Slot overlays:

Each container label shows a unique icon per UoM type with a live legend.

Running Tests

python -m pytest tests/ -v

License

MIT