usim.typing module

class usim.typing.BorrowedResources(resources: usim._basics.resource.BaseResources, debits: usim._basics._resource_level.ResourceLevels)

Bases: usim._basics.resource.BaseResources[usim._basics.resource.T]

Fixed supply of named resources temporarily taken from another resource supply

borrow(**amounts: usim._basics.resource.T) → usim._basics.resource.BorrowedResources[usim._basics.resource.T]

Temporarily borrow resources for a given context

Parameters

amounts – resource levels to borrow

Returns

async context to borrow resources

property limits

Upper limit of resource levels

class usim.typing.ClaimedResources(resources: usim._basics.resource.BaseResources, debits: usim._basics._resource_level.ResourceLevels)

Bases: usim._basics.resource.BorrowedResources[usim._basics.resource.T]

Fixed supply of resources temporarily taken without delay

class usim.typing.Condition

Bases: usim._primitives.notification.Notification

An asynchronous logical condition

Every Condition can be used both in an asynchronous and boolean context. In an asynchronous context, such as await, a Condition triggers when the Condition becomes True. In a boolean context, such as if, a Condition provides its current boolean value.

if condition:    # resume with current value
    print(condition, 'is met')
else:
    print(condition, 'is not met')

await condition  # resume when condition is True

async with until(condition):  # interrupt when condition is True
    ...

Every Condition supports the bitwise operators ~a (not), a & b (and), as well as a | b (or) to derive a new Condition. While it is possible to use the boolean operators not, and, as well as or, they immediately evaluate any Condition in a boolean context.

await (a & b)   # resume when both a and b are True
await (a | b)   # resume when one of a or b are True
await (a & ~b)  # resume when a is True and b is False

c = a & b  # derive new Condition...
await c    # that can be awaited

d = a and b  # force boolean evaluation

class usim.typing.Notification

Bases: object

Synchronisation point to which activities can subscribe

await notification  # hibernate until notified

async with until(notification):
    ...

class usim.typing.ResourceLevels(**kwargs: usim._basics._resource_level.T)

Bases: Generic[usim._basics._resource_level.T]

Common class for named resource levels

Representation for the levels of multiple named resources. Every set of resources, such as usim.Resources or usim.Capacities, specializes a ResourceLevels subclass with one attribute for each named resource. For example, Resources(a=3, b=4) uses a ResourceLevels with attributes a and b.

from usim import Resources

resources = Resources(a=3, b=4)
print(resources.levels.a)  # 3
print(resources.levels.b)  # 4
print(resources.levels.c)  # raises AttributeError

ResourceLevels subtypes allow no additional attributes other than their initial resources, but their values may be changed. Instantiating a subtype requires resource levels to be specified by keyword; missing resource are set to zero.

Each resource always uses the same ResourceLevels subtype. Binary operators for comparisons and arithmetic can be applied for instances of the same subtype.

levels_a + levels_b

levels_a - levels_b

Elementwise addition/subtraction of values.

levels_a > levels_b

levels_a >= levels_b

levels_a <= levels_b

levels_a < levels_b

Strict elementwise comparison of values. True if the comparison is satisfied by each element pair, False otherwise.

levels_a == levels_b

Total elementwise equality of values. True if each element pair is equal, False otherwise. The inverse of levels_a != levels_b.

levels_a != levels_b

Partial elementwise unequality of values. False if each element pair is equal, True otherwise. The inverse of levels_a == levels_b.

In addition, iteration on a ResourceLevels subtype yields field, value pairs. This is similar to dict.items().

for field, value in levels_a

Iterate over the current field, value pairs.

dict(levels_a)

Create dict of field: value pairs.

class usim.typing.Task(payload: Coroutine[Any, Any, usim._primitives.task.RT], parent: Scope, delay: Optional[float], at: Optional[float], volatile: bool)

Bases: Awaitable[usim._primitives.task.RT]

Concurrently running activity

A Task wraps a payload activity that is concurrently run in a parent Scope. This allows to store or pass on the Task in order to control the underlying activity. Other activities can await a Task to receive results or exceptions on completion, similar to a regular activity.

async def my_activity(delay):
    await (time + delay)
    return delay

await my_activity()  # await an activity

async with Scope() as scope:
    task = scope.do(my_activity())
    await task   # await Task of an activity

In contrast to a bare activity, it is possible to

• cancel() a Task before completion,

• await the result of a Task multiple times, and

• await that a Task is done.

Note

This class should not be instantiated directly. Always use a Scope to create it.

cancel(*token) → None

Cancel this task during the current time step

If the Task is running, a CancelTask is raised once the activity suspends. The activity may catch and react to CancelActivity, but should not suppress it.

If the Task is done before CancelTask is raised, the cancellation is ignored. This also means that cancelling an activity multiple times is allowed, but only the first successful cancellation is stored as the cancellation cause.

If the Task has not started running, it is cancelled immediately. This prevents any code execution, even before the first suspension.

Warning

The timing of cancelling a Task before it started running may change in the future.

property done: usim._primitives.task.Done

Condition whether the Task has stopped running. This includes completion, cancellation and failure.

parent

payload

property status: usim._primitives.task.TaskState

The current status of this activity