usim.typing module
- class usim.typing.BorrowedResources(resources: usim._basics.resource.BaseResources, debits: usim._basics._resource_level.ResourceLevels)[source]
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] [source]
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)[source]
Bases:
usim._basics.resource.BorrowedResources
[usim._basics.resource.T
]Fixed supply of resources temporarily taken without delay
- class usim.typing.Condition[source]
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 asawait
, aCondition
triggers when theCondition
becomesTrue
. In a boolean context, such asif
, aCondition
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 asa | b
(or) to derive a newCondition
. While it is possible to use the boolean operatorsnot
,and
, as well asor
, they immediately evaluate anyCondition
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[source]
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)[source]
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
orusim.Capacities
, specializes aResourceLevels
subclass with one attribute for each named resource. For example,Resources(a=3, b=4)
uses aResourceLevels
with attributesa
andb
.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 oflevels_a != levels_b
.
- levels_a != levels_b
Partial elementwise unequality of values.
False
if each element pair is equal,True
otherwise. The inverse oflevels_a == levels_b
.
In addition, iteration on a
ResourceLevels
subtype yieldsfield, value
pairs. This is similar todict.items()
.- for field, value in levels_a
Iterate over the current
field, value
pairs.
- dict(levels_a)
Create
dict
offield: value
pairs.
- class usim.typing.Task(payload: Coroutine[Any, Any, usim._primitives.task.RT], parent: Scope, delay: Optional[float], at: Optional[float], volatile: bool)[source]
Bases:
Awaitable
[usim._primitives.task.RT
]Concurrently running activity
A
Task
wraps apayload
activity that is concurrently run in aparent
Scope
. This allows to store or pass on theTask
in order to control the underlying activity. Other activities canawait
aTask
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
await
the result of aTask
multiple times, and
- Note
This class should not be instantiated directly. Always use a
Scope
to create it.
- cancel(*token) None [source]
Cancel this task during the current time step
If the
Task
is running, aCancelTask
is raised once the activity suspends. The activity may catch and react toCancelActivity
, but should not suppress it.If the
Task
isdone
beforeCancelTask
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 theTask
has stopped running. This includes completion, cancellation and failure.
- parent
- payload
- property status: usim._primitives.task.TaskState
The current status of this activity