upstage.resources package#

Submodules#

upstage.resources.container module#

This file contains a ContinuousContainer.

exception ContainerEmptyError#

Bases: ContainerError

The container is empty or has a negative level.

exception ContainerError#

Bases: Exception

The container is in an invalid state.

property cause: Any#

Get the exception’s cause.

Returns:

The cause.

Return type:

Any

exception ContainerFullError#

Bases: ContainerError

The container has reach or exceeded its capacity.

class ContinuousContainer(env: Environment, capacity: int | float, init: int | float = 0.0, error_empty: bool = True, error_full: bool = True)#

Bases: object

A container that accepts continuous gets and puts.

add_user(user: _ContinuousEvent) → None#

Add a user to the container.

Parameters:

user (_ContinuousEvent) – The user event

property capacity: float#

Get the capacity of the container.

Returns:

The capacity.

Return type:

float

property env: Environment#

Get the environment of the container.

Returns:

The SimPy environment.

Return type:

Environment

get#

alias of ContinuousGet

property level: float#

Get the level of the container.

Returns:

The current amount remaining.

Return type:

float

put#

alias of ContinuousPut

property rate: float#

Get the current net rate.

Returns:

The net rate.

Return type:

float

remove_user(user: _ContinuousEvent) → None#

Remove a user of the container.

Parameters:

user (_ContinuousEvent) – The user event.

time_until_done(rate: float = 0.0) → float#

Calculate the time until the container is full or empty.

Parameters:

rate (float, optional) – Additional rate. Defaults to 0.0.

Returns:

Time until the container reaches a limit.

Return type:

float

time_until_level(level: float, rate: float = 0.0) → float#

Calculate the time until the containers reaches a value.

Parameters:

  • level (float) – The value to reach.

  • rate (float, optional) – Additional rate. Defaults to 0.0.

Returns:

The time to reach the level.

Return type:

float

class ContinuousGet(container: ContinuousContainer, rate: float, time: float, custom_callbacks: list[Callable[[], None]] | None = None)#

Bases: _ContinuousEvent

An event that gets amount from the container.

class ContinuousPut(container: ContinuousContainer, rate: float, time: float, custom_callbacks: list[Callable[[], None]] | None = None)#

Bases: _ContinuousEvent

An event that puts rate per unit time into the container.

Raise a ValueError if rate <= 0.

upstage.resources.monitoring module#

Stores that monitor/record their items over time.

class SelfMonitoringContainer(env: Environment, capacity: float = inf, init: float = 0.0)#

Bases: Container

A self-monitoring version of the SimPy Container.

class SelfMonitoringContinuousContainer(env: Environment, capacity: int | float, init: int | float = 0.0, error_empty: bool = True, error_full: bool = True)#

Bases: ContinuousContainer

A self-monitoring version of the Continuous Container.

class SelfMonitoringFilterStore(env: Environment, capacity: float | int = inf, item_func: Callable[[list[Any]], int] | None = None)#

Bases: FilterStore

A self-monitoring version of the SimPy FilterStore.

class SelfMonitoringReserveStore(env: Environment, capacity: float = inf, init: float = 0.0)#

Bases: ReserveStore

A self-monitoring version of the ReserveStore.

put(amount: float, capacity_increase: bool = False) → None#

Put some amount into the store.

Parameters:

  • amount (float) – The amount to put

  • capacity_increase (bool, optional) – Allow capacity to increase. Defaults to False.

take(requester: Any) → float#

Take some amount from the store, by a requester.

Parameters:

requester (Any) – The entity requesting the amount

Returns:

The aount requested.

Return type:

float

class SelfMonitoringSortedFilterStore(env: Environment, capacity: float | int = inf, item_func: Callable[[list[Any]], int] | None = None)#

Bases: SortedFilterStore, SelfMonitoringStore

A self-monitoring version of the SortedFilterStore.

class SelfMonitoringStore(env: Environment, capacity: float | int = inf, item_func: Callable[[list[Any]], int] | None = None)#

Bases: Store

A self-monitoring version of the SimPy Store.

upstage.resources.reserve module#

This file contains a Store that allows reservations.

class ReserveStore(env: Environment, init: float = 0.0, capacity: float = inf)#

Bases: object

A store that allows requests to be scheduled in advance.

This is not a true store (you can’t yield on a reserve slot!).

property available: float#

Return the amount remaining in the store.

Returns:

Amount remaining.

Return type:

float

cancel_request(requester: Any) → bool#

Have a request cancelled.

put(amount: float, capacity_increase: bool = False) → None#

Put some quantity back in.

property queued: list[Any]#

Get the queued requesters.

Returns:

List of requesters.

Return type:

list[Any]

property remaining: float#

Return the amount remaining in the store.

Returns:

Amount remaining

Return type:

float

reserve(requester: Any, quantity: float, expiration: float | None = None) → bool#

Reserve a quantity of storage.

take(requester: Any) → float#

If in queue, allow requester take the actual quantity.

upstage.resources.sorted module#

Filter stores that allow sorting of items.

class SortedFilterGet(get_location: ~upstage.resources.sorted.SortedFilterStore, filter: ~collections.abc.Callable[[~typing.Any], bool] = <function SortedFilterGet.<lambda>>, sorter: ~collections.abc.Callable[[~typing.Any], tuple[~typing.Any, ...]] | None = None, rehearsal_time_to_complete: float = 0.0)#

Bases: Get

A Get for a SortedFilterStore.

class SortedFilterStore(env: Environment, capacity: float | int = inf)#

Bases: Store

A store that supports the filtered and sorted retrieval of items.

Resource with capacity slots for storing arbitrary objects supporting filtered and sorted get requests. Like the Store, the capacity is unlimited by default and objects are put and retrieved from the store in a first-in first-out order.

Get requests can be customized with a filter function to only trigger for items for which said filter function returns True. They can further be customized with a sorter function that prioritizes which of the filtered items are to be returned. The prioritization happens through a minimization.

get#

alias of _SortedFilterStoreGet

Module contents#

This sub-module contains advanced stores and containers for UPSTAGE.