jvconnected.utils
- class jvconnected.utils.IOType(value)[source]
Bases:
Enum
Enum to distinguish between input and output types
- NONE = 1
Not set
- INPUT = 2
Input
- OUTPUT = 3
Output
- jvconnected.utils.async_callback(fn)[source]
Wrap a coroutine function or method (
async def
) where a sync function is expected.The decorated function or method will be wrapped in an
asyncio.Task
and scheduled on the current event loop (within the context of the callback).Any exceptions will be caught and forwarded to the event loop through
asyncio.loop.call_exception_handler()
.callback_event = asyncio.Event() @async_callback async def my_async_callback(*args, **kwargs): print(f'callback got: {args}, {kwargs}') callback_event.set() # Calling `my_async_callback` as a normal function my_async_callback('foo', bar='baz') # Run the loop until callback_event is set from inside the callback loop = asyncio.get_event_loop() loop.run_until_complete(callback_event.wait())
callback got: ('foo',), {'bar': 'baz'}
- class jvconnected.utils.IndexedDict(*args, **kwargs)[source]
Bases:
Dispatcher
A
dict
like container that tracks indices for its items- Events
- remove(key: Any)[source]
Remove an item
- Parameters
key (Any) – The dictionary key
- Returns
The item that was removed
- change_item_index(key: Any, new_index: int)[source]
Change the index for an existing item. If necessary, change indices for any conflicting items
- keys() Iterator[Any] [source]
Return an iterator of the dictionary keys, sorted by the item indices
- values() Iterator[Any] [source]
Return an iterator of the dictionary values, sorted by the item indices
- items() Iterator[Tuple[Any, Any]] [source]
Return an iterator of the dictionary key, value pairs, sorted by the item indices
- iter_indices(start_index: int = 0) Iterator[int] [source]
Iterate through sorted indices starting from the one given
- Parameters
start_index (int, optional) – The starting index, defaults to
0
- iter_consecutive_indices(start_index: int = 0) Iterator[int] [source]
Iterate through sorted indices starting from the one given, but stop at the first gap
- Parameters
start_index (int, optional) – The starting index, defaults to
0
- class jvconnected.utils.NamedItem(key: Any, item: Any)[source]
Bases:
object
Helper class for
NamedQueue
- class jvconnected.utils.NamedQueue(maxsize=0, *, loop=None)[source]
Bases:
Queue
A
asyncio.Queue
subclass that stores items by user-defined keys.The items placed on the queue must be instances of
NamedItem
. For convenience, there is acreate_item()
contructor method.- classmethod create_item(key: Any, item: Any) NamedItem [source]
Create a
NamedItem
to be put on the queue
- async put(item: NamedItem)[source]
Put a
NamedItem
into the queue.If the queue is full, wait until a free slot is available before adding item.
If an item with the same
key
already exists in the queue, it will be replaced.
- put_nowait(item: NamedItem)[source]
Put an item into the queue without blocking.
If no free slot is immediately available, raise QueueFull.