Resources Module

Resource management components for modeling limited resources, servers, and queues.

Server

Multi-server processing stations with automatic queue management.

class simcraft.Server[source]

Bases: Generic[T]

Multi-server processing station.

A Server models a processing station with one or more parallel servers (capacity), a waiting queue, and configurable service time distributions.

Parameters:

sim (Simulation) – Parent simulation
capacity (int) – Number of parallel servers
service_time (Union[float, Callable[[], float]]) – Fixed service time or function returning random service time
queue_capacity (int) – Maximum queue length (0 = unlimited)
name (str) – Optional name for the server

Examples

>>> def service_time():
...     return sim.rng.exponential(5.0)
...
>>> server = Server(sim, capacity=2, service_time=service_time)
>>> server.enqueue(customer)
>>> server.on_departure(lambda c: print(f"{c.id} departed"))

__init__(sim, capacity=1, service_time=1.0, queue_capacity=0, name='')[source]

Initialize server.

Parameters:

sim (Simulation)
capacity (int)
service_time (Union[float, Callable[[], float]])
queue_capacity (int)
name (str)

Return type:

None

property name: str: Get server name.

property capacity: int: Get server capacity.

property available_capacity: int: Get number of available servers.

property is_idle: bool: Check if all servers are idle.

property is_busy: bool: Check if all servers are busy.

property queue_length: int: Get current queue length.

property in_service_count: int: Get number of entities in service.

property total_in_system: int: Get total entities (queue + in service).

property stats: ServerStats: Get server statistics.

property queue: Queue[T]: Get the waiting queue.

property state: ServerState: Get current server state.

enqueue(entity)[source]

Add an entity to the server.

The entity will be served immediately if capacity is available, otherwise it will wait in the queue.

Parameters:: entity (T) – Entity to process
Returns:: True if entity was accepted, False if queue is full
Return type:: bool

preempt(entity)[source]

Preempt service for highest-priority entity.

Parameters:: entity (T) – Entity requesting preemption
Returns:: The preempted entity, or None if no preemption occurred
Return type:: Optional[T]

shutdown()[source]

Shut down the server.

Return type:: None

restart()[source]

Restart the server.

Return type:: None

on_arrival(callback)[source]

Set callback for entity arrivals.

Parameters:: callback (Callable[[T], None])
Return type:: None

on_service_start(callback)[source]

Set callback for service start.

Parameters:: callback (Callable[[T], None])
Return type:: None

on_departure(callback)[source]

Set callback for entity departures.

Parameters:: callback (Callable[[T], None])
Return type:: None

on_balk(callback)[source]

Set callback for balking entities.

Parameters:: callback (Callable[[T], None])
Return type:: None

reset_stats()[source]

Reset server and queue statistics.

Return type:: None

get_entities_in_service()[source]

Get list of entities currently in service.

Return type:: List[T]

class simcraft.resources.server.ServerStats[source]

Bases: object

Statistics for server performance.

arrivals

Total entities that arrived at the server

Type:: int

departures

Total entities that completed service

Type:: int

balked

Entities that left without service (queue full)

Type:: int

busy_time

Total time server was busy

Type:: float

idle_time

Total time server was idle

Type:: float

blocked_time

Total time server was blocked

Type:: float

down_time

Total time server was down

Type:: float

arrivals: int = 0

departures: int = 0

balked: int = 0

busy_time: float = 0.0

idle_time: float = 0.0

blocked_time: float = 0.0

down_time: float = 0.0

total_service_time: float = 0.0

record_state_change(time, new_state)[source]

Record a state change.

Parameters:

time (float)
new_state (ServerState)

Return type:

None

property utilization: float: Get server utilization (busy time / total active time).

property average_service_time: float: Get average service time.

property throughput_rate: float: Get throughput rate (departures per time unit).

reset()[source]

Reset all statistics.

Return type:: None

__init__(arrivals=0, departures=0, balked=0, busy_time=0.0, idle_time=0.0, blocked_time=0.0, down_time=0.0, total_service_time=0.0, _last_state_change=0.0, _current_state=ServerState.IDLE, _busy_count=0)

Parameters:

arrivals (int)
departures (int)
balked (int)
busy_time (float)
idle_time (float)
blocked_time (float)
down_time (float)
total_service_time (float)
_last_state_change (float)
_current_state (ServerState)
_busy_count (int)

Return type:

None

class simcraft.resources.server.ServerState[source]

Bases: Enum

Server operational states.

IDLE = 1

BUSY = 2

BLOCKED = 3

DOWN = 4

class simcraft.resources.server.MultiStageServer[source]

Bases: Generic[T]

Multi-stage processing server.

Entities flow through a sequence of processing stages, each with its own service time and optional queue.

Parameters:

sim (Simulation) – Parent simulation
stage_configs (List[Dict]) – Configuration for each stage with keys: - capacity: int - service_time: float or Callable - queue_capacity: int (optional)
name (str) – Optional name

Examples

>>> stages = [
...     {"capacity": 1, "service_time": 5.0},
...     {"capacity": 2, "service_time": 3.0},
... ]
>>> server = MultiStageServer(sim, stages)
>>> server.enqueue(job)

__init__(sim, stage_configs, name='')[source]

Initialize multi-stage server.

Parameters:

sim (Simulation)
stage_configs (List[Dict[str, Any]])
name (str)

Return type:

None

property num_stages: int: Get number of stages.

enqueue(entity)[source]

Add entity to first stage.

Parameters:: entity (T) – Entity to process
Returns:: True if accepted
Return type:: bool

on_completion(callback)[source]

Set callback for entity completion of all stages.

Parameters:: callback (Callable[[T], None])
Return type:: None

get_stage(index)[source]

Get a specific stage.

Parameters:: index (int) – Stage index (0-based)
Returns:: The stage server
Return type:: Server[T]

Queue

FIFO and priority queues with statistics collection.

class simcraft.Queue[source]

Bases: Generic[T]

FIFO queue with statistics collection.

A basic first-in-first-out queue that tracks entry times and collects performance statistics.

Parameters:

sim (Simulation) – Parent simulation for time tracking
capacity (int) – Maximum queue capacity (0 = unlimited)
name (str) – Optional name for the queue

Examples

>>> queue = Queue(sim, capacity=10, name="WaitingRoom")
>>> queue.enqueue(customer)
>>> if not queue.is_empty:
...     next_customer = queue.dequeue()
>>> print(queue.stats.average_wait)

__init__(sim, capacity=0, name='')[source]

Initialize queue.

Parameters:

sim (Simulation)
capacity (int)
name (str)

Return type:

None

property name: str: Get queue name.

property stats: QueueStats: Get queue statistics.

property capacity: int: Get queue capacity (0 = unlimited).

property is_empty: bool: Check if queue is empty.

property is_full: bool: Check if queue is at capacity.

enqueue(item)[source]

Add item to the queue.

Parameters:: item (T) – Item to add
Returns:: True if item was added, False if queue is full
Return type:: bool

dequeue()[source]

Remove and return the first item.

Returns:: First item or None if queue is empty
Return type:: Optional[T]

peek()[source]

Return the first item without removing it.

Returns:: First item or None if queue is empty
Return type:: Optional[T]

remove(item)[source]

Remove a specific item from the queue.

Parameters:: item (T) – Item to remove
Returns:: True if item was found and removed
Return type:: bool

clear()[source]

Remove all items from the queue.

Returns:: List of removed items
Return type:: List[T]

contains(item)[source]

Check if item is in the queue.

Parameters:: item (T)
Return type:: bool

on_enqueue(callback)[source]

Set callback for enqueue events.

Parameters:: callback (Callable[[T], None])
Return type:: None

on_dequeue(callback)[source]

Set callback for dequeue events.

Parameters:: callback (Callable[[T], None])
Return type:: None

reset_stats()[source]

Reset statistics (keep items).

Return type:: None

class simcraft.resources.queue.QueueStats[source]

Bases: object

Statistics for queue performance.

entries

Total number of entities that entered the queue

Type:: int

exits

Total number of entities that left the queue

Type:: int

max_length

Maximum queue length observed

Type:: int

total_wait_time

Sum of all waiting times

Type:: float

area

Time-weighted queue length integral

Type:: float

entries: int = 0

exits: int = 0

max_length: int = 0

total_wait_time: float = 0.0

area: float = 0.0

record_entry(time)[source]

Record an entry to the queue.

Parameters:: time (float)
Return type:: None

record_exit(time, wait_time)[source]

Record an exit from the queue.

Parameters:

time (float)
wait_time (float)

Return type:

None

property average_length: float: Get time-average queue length.

property average_wait: float: Get average waiting time.

property current_length: int: Get current queue length.

reset()[source]

Reset all statistics.

Return type:: None

__init__(entries=0, exits=0, max_length=0, total_wait_time=0.0, area=0.0, _last_change_time=0.0, _current_length=0)

Parameters:

entries (int)
exits (int)
max_length (int)
total_wait_time (float)
area (float)
_last_change_time (float)
_current_length (int)

Return type:

None

class simcraft.resources.queue.PriorityQueue[source]

Bases: Generic[T]

Priority queue with statistics collection.

Items are dequeued in order of priority (lower value = higher priority).

Parameters:

sim (Simulation) – Parent simulation for time tracking
priority_fn (Optional[Callable[[T], float]]) – Function to extract priority from items (default uses 0)
capacity (int) – Maximum queue capacity (0 = unlimited)
name (str) – Optional name for the queue

Examples

>>> queue = PriorityQueue(sim, priority_fn=lambda x: x.priority)
>>> queue.enqueue(high_priority_job)
>>> queue.enqueue(low_priority_job)
>>> next_job = queue.dequeue()  # Returns high_priority_job

__init__(sim, priority_fn=None, capacity=0, name='')[source]

Initialize priority queue.

Parameters:

sim (Simulation)
priority_fn (Optional[Callable[[T], float]])
capacity (int)
name (str)

Return type:

None

property name: str: Get queue name.

property stats: QueueStats: Get queue statistics.

property is_empty: bool: Check if queue is empty.

property is_full: bool: Check if queue is at capacity.

enqueue(item, priority=None)[source]

Add item to the queue.

Parameters:

item (T) – Item to add
priority (Optional[float]) – Override priority (uses priority_fn if not specified)

Returns:

True if item was added, False if queue is full

Return type:

bool

dequeue()[source]

Remove and return the highest priority item.

Returns:: Highest priority item or None if empty
Return type:: Optional[T]

peek()[source]

Return the highest priority item without removing it.

Returns:: Highest priority item or None if empty
Return type:: Optional[T]

remove(item)[source]

Remove a specific item from the queue.

Parameters:: item (T) – Item to remove
Returns:: True if item was found and removed
Return type:: bool

update_priority(item, new_priority)[source]

Update the priority of an item.

Parameters:

item (T) – Item to update
new_priority (float) – New priority value

Returns:

True if item was found and updated

Return type:

bool

Resource

Seizable resources with acquire/release semantics.

class simcraft.Resource[source]

Bases: object

Limited resource with acquire/release semantics.

A Resource represents a limited item (or pool of items) that entities can acquire and later release. Supports priority-based waiting and timeouts.

Parameters:

sim (Simulation) – Parent simulation
capacity (int) – Number of resource units
name (str) – Optional resource name

Examples

>>> operator = Resource(sim, capacity=1, name="Operator")
>>>
>>> # Immediate acquisition
>>> if operator.acquire(job):
...     # Use resource
...     operator.release(job)
>>>
>>> # Request with callback
>>> def on_acquired(resource, job):
...     sim.schedule(release_op, delay=5.0, args=(job,))
>>> operator.request(job, callback=on_acquired)

__init__(sim, capacity=1, name='')[source]

Initialize resource.

Parameters:

sim (Simulation)
capacity (int)
name (str)

Return type:

None

property name: str: Get resource name.

property capacity: int: Get total capacity.

property available: int: Get available units.

property allocated: int: Get allocated units.

property waiting_count: int: Get number of waiting requests.

property stats: ResourceStats: Get resource statistics.

property is_available: bool: Check if any units are available.

acquire(requester, quantity=1)[source]

Immediately acquire resource units.

Parameters:

requester (T) – Entity requesting the resource
quantity (int) – Number of units to acquire

Returns:

True if acquisition succeeded, False otherwise

Return type:

bool

release(holder)[source]

Release resource units.

Parameters:: holder (T) – Entity currently holding the resource
Returns:: True if release succeeded, False if entity wasn’t holding
Return type:: bool

request(requester, quantity=1, priority=0, timeout=None, callback=None)[source]

Request resource units (may wait if not available).

Parameters:

requester (T) – Entity requesting the resource
quantity (int) – Number of units to request
priority (int) – Request priority (higher = processed first)
timeout (Optional[float]) – Maximum wait time (None = wait indefinitely)
callback (Optional[Callable]) – Function to call when acquired

Returns:

True if immediately acquired, False if waiting

Return type:

bool

cancel_request(requester)[source]

Cancel a pending request.

Parameters:: requester (T) – Entity that made the request
Returns:: True if request was found and cancelled
Return type:: bool

shutdown()[source]

Shut down the resource.

Return type:: None

restart()[source]

Restart the resource.

Return type:: None

reset_stats()[source]

Reset statistics.

Return type:: None

class simcraft.resources.resource.ResourceStats[source]

Bases: object

Statistics for resource usage.

acquisitions

Total number of successful acquisitions

Type:: int

releases

Total number of releases

Type:: int

timeouts

Number of acquisition timeouts

Type:: int

total_busy_time

Total time resources were in use

Type:: float

total_idle_time

Total time resources were available

Type:: float

acquisitions: int = 0

releases: int = 0

timeouts: int = 0

total_busy_time: float = 0.0

total_idle_time: float = 0.0

total_wait_time: float = 0.0

record_acquisition(time, wait_time=0.0)[source]

Record a resource acquisition.

Parameters:

time (float)
wait_time (float)

Return type:

None

record_release(time, hold_time)[source]

Record a resource release.

Parameters:

time (float)
hold_time (float)

Return type:

None

property utilization: float: Get average resource utilization.

property average_hold_time: float: Get average hold time.

property average_wait_time: float: Get average wait time for acquisition.

reset()[source]

Reset all statistics.

Return type:: None

__init__(acquisitions=0, releases=0, timeouts=0, total_busy_time=0.0, total_idle_time=0.0, total_wait_time=0.0, _area_allocated=0.0, _last_change_time=0.0, _current_allocated=0, _capacity=1)

Parameters:

acquisitions (int)
releases (int)
timeouts (int)
total_busy_time (float)
total_idle_time (float)
total_wait_time (float)
_area_allocated (float)
_last_change_time (float)
_current_allocated (int)
_capacity (int)

Return type:

None

class simcraft.resources.resource.ResourceState[source]

Bases: Enum

Resource states.

AVAILABLE = 1

ALLOCATED = 2

DOWN = 3

class simcraft.resources.resource.PreemptiveResource[source]

Bases: Resource

Resource with preemption support.

Lower priority holders can be preempted by higher priority requests.

Parameters:

sim (Simulation) – Parent simulation
capacity (int) – Number of resource units
name (str) – Optional resource name

Examples

>>> machine = PreemptiveResource(sim, capacity=1)
>>> machine.acquire(low_priority_job, priority=1)
>>> machine.acquire(high_priority_job, priority=10)  # Preempts

__init__(sim, capacity=1, name='')[source]

Initialize preemptive resource.

Parameters:

sim (Simulation)
capacity (int)
name (str)

Return type:

None

acquire(requester, quantity=1, priority=0)[source]

Acquire resource units, possibly preempting lower priority holders.

Parameters:

requester (T) – Entity requesting the resource
quantity (int) – Number of units to acquire
priority (int) – Priority level

Returns:

True if acquisition succeeded

Return type:

bool

release(holder)[source]

Release resource and clean up priority tracking.

Parameters:: holder (T)
Return type:: bool

on_preempt(callback)[source]

Set callback for preemption events.

Parameters:: callback (Callable[[T], None])
Return type:: None

Resource Pool

Pools of distinguishable resources.

class simcraft.ResourcePool[source]

Bases: Generic[R]

Pool of individual, distinguishable resources.

Unlike a simple capacity-based Resource, ResourcePool manages individual resource objects that can be selected based on various policies (first available, round-robin, least utilized, or custom selection function).

Parameters:

sim (Simulation) – Parent simulation
name (str) – Pool name
selection_policy (PoolSelectionPolicy) – Policy for selecting resources

Examples

>>> class AGV:
...     def __init__(self, id: str, location: tuple):
...         self.id = id
...         self.location = location
...
>>> pool = ResourcePool(sim, name="AGVPool")
>>> pool.add_resource(AGV("AGV1", (0, 0)), id="AGV1")
>>> pool.add_resource(AGV("AGV2", (10, 0)), id="AGV2")
>>>
>>> # Custom selection: nearest AGV
>>> def nearest_to(target):
...     def selector(available):
...         return min(available, key=lambda r: distance(r.location, target))
...     return selector
>>>
>>> agv = pool.acquire(job, selector=nearest_to((5, 5)))

__init__(sim, name='', selection_policy=PoolSelectionPolicy.FIRST_AVAILABLE)[source]

Initialize resource pool.

Parameters:

sim (Simulation)
name (str)
selection_policy (PoolSelectionPolicy)

Return type:

None

property name: str: Get pool name.

property size: int: Get total number of resources in pool.

property available_count: int: Get number of available resources.

property allocated_count: int: Get number of allocated resources.

property stats: PoolStats: Get pool statistics.

property is_empty: bool: Check if pool has no resources.

property has_available: bool: Check if any resources are available.

add_resource(resource, id=None)[source]

Add a resource to the pool.

Parameters:

resource (R) – Resource to add
id (Optional[str]) – Unique identifier (generated if not provided)

Returns:

Resource identifier

Return type:

str

remove_resource(id)[source]

Remove a resource from the pool.

Parameters:: id (str) – Resource identifier
Returns:: The removed resource, or None if not found
Return type:: Optional[R]

get_resource(id)[source]

Get a resource by ID.

Parameters:: id (str) – Resource identifier
Returns:: The resource, or None if not found
Return type:: Optional[R]

acquire(requester, selector=None)[source]

Acquire a resource from the pool.

Parameters:

requester (Any) – Entity requesting the resource
selector (Optional[Callable]) – Custom function to select from available resources

Returns:

Selected resource, or None if none available

Return type:

Optional[R]

request(requester, callback, selector=None)[source]

Request a resource (wait if not available).

Parameters:

requester (Any) – Entity requesting the resource
callback (Callable) – Function to call when resource is acquired
selector (Optional[Callable]) – Custom selection function

Returns:

True if immediately acquired, False if waiting

Return type:

bool

release(resource)[source]

Release a resource back to the pool.

Parameters:: resource (R) – Resource to release
Returns:: True if successfully released
Return type:: bool

set_selection_policy(policy, custom_selector=None)[source]

Set resource selection policy.

Parameters:

policy (PoolSelectionPolicy) – Selection policy
custom_selector (Optional[Callable]) – Custom selector for CUSTOM policy

Return type:

None

get_available()[source]

Get list of available resources.

Returns:: Available resources
Return type:: List[R]

get_allocated()[source]

Get list of allocated resources.

Returns:: Allocated resources
Return type:: List[R]

get_utilization(resource_id)[source]

Get utilization for a specific resource.

Parameters:: resource_id (str) – Resource identifier
Returns:: Utilization ratio (0-1)
Return type:: float

get_average_utilization()[source]

Get average utilization across all resources.

Returns:: Average utilization ratio
Return type:: float

reset_stats()[source]

Reset all statistics.

Return type:: None

class simcraft.resources.pool.PooledResource[source]

Bases: Generic[R]

Wrapper for a resource in a pool.

resource

The actual resource object

Type:: R

id

Unique identifier

Type:: str

is_available

Whether resource is currently available

Type:: bool

allocations

Number of times this resource has been allocated

Type:: int

total_busy_time

Total time this resource has been allocated

Type:: float

resource: R

id: str

is_available: bool = True

allocations: int = 0

total_busy_time: float = 0.0

__init__(resource, id, is_available=True, allocations=0, total_busy_time=0.0, _allocated_at=None, _allocated_to=None)

Parameters:

resource (R)
id (str)
is_available (bool)
allocations (int)
total_busy_time (float)
_allocated_at (float | None)
_allocated_to (Any | None)

Return type:

None

class simcraft.resources.pool.PoolStats[source]

Bases: object

Statistics for resource pool.

total_acquisitions

Total successful acquisitions

Type:: int

total_releases

Total releases

Type:: int

failed_acquisitions

Failed acquisition attempts

Type:: int

total_acquisitions: int = 0

total_releases: int = 0

failed_acquisitions: int = 0

total_wait_time: float = 0.0

property success_rate: float: Get acquisition success rate.

__init__(total_acquisitions=0, total_releases=0, failed_acquisitions=0, total_wait_time=0.0, _waiting_count=0)

Parameters:

total_acquisitions (int)
total_releases (int)
failed_acquisitions (int)
total_wait_time (float)
_waiting_count (int)

Return type:

None

class simcraft.resources.pool.PoolSelectionPolicy[source]

Bases: Enum

Policies for selecting resources from pool.

FIRST_AVAILABLE = 1

ROUND_ROBIN = 2

LEAST_UTILIZED = 3

CUSTOM = 4