jvconnected.utils

class jvconnected.utils.IndexedDict(*args, **kwargs)

Bases: pydispatch.dispatch.Dispatcher

A dict like container that tracks indices for its items

Events

on_item_added(key=key, item=item, index=index_)

Fired when an item is added

on_item_removed(key=key, item=item, index=index_)

Fired when an item is removed

on_item_index_changed(key=key, item=item, old_index=cur_index, new_index=new_index)

Fired when an item’s index changes

add(key: Any, item: Any, index_: int = - 1) → int

Add an item

Parameters

• key – The dictionary key

• item – The dictionary value

• index – The index for the item. If -1, the item will be appended to the end, otherwise it will be inserted at the specified index

Returns

The inserted item’s index

Return type

int

remove(key: Any)

Remove an item

Parameters

key – The dictionary key

Returns

The item that was removed

change_item_index(key: Any, new_index: int)

Change the index for an existing item. If necessary, change indices for any conflicting items

Parameters

• key – the dictionary key

• new_index (int) – New index for the item

compact_indices(start_index: int = 0, max_change: int = 1)

Remove gaps in indices

Parameters

• start_index (int, optional) – The index to start from

• max_change (int, optional) – Limit index changes to this amount

keys() → Iterator[Any]

Return an iterator of the dictionary keys, sorted by the item indices

values() → Iterator[Any]

Return an iterator of the dictionary values, sorted by the item indices

items() → Iterator[Tuple[Any, Any]]

Return an iterator of the dictionary key, value pairs, sorted by the item indices

iter_indices(start_index: int = 0) → Iterator[int]

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]

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

get(key: Any, default: Optional[Any] = None)

Get an item by key

get_by_index(index_: int, default: Optional[Any] = None)

Get an item by index

Parameters

• index (int) – The item index to get

• default (optional) – The default to return if no item exists with the given index, defaults to None

get_item_index(key: Any) → int

Get the index for the given key

class jvconnected.utils.NamedItem(key: Any, item: Any)

Bases: object

Helper class for NamedQueue

key: Any

The item key

item: Any

The item itself

class jvconnected.utils.NamedQueue(maxsize=0, *, loop=None)

Bases: asyncio.queues.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 a create_item() contructor method.

classmethod create_item(key: Any, item: Any) → jvconnected.utils.NamedItem

Create a NamedItem to be put on the queue

async put(item: jvconnected.utils.NamedItem)

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: jvconnected.utils.NamedItem)

Put an item into the queue without blocking.

If no free slot is immediately available, raise QueueFull.

async get() → jvconnected.utils.NamedItem

Remove and return an item from the queue.

If queue is empty, wait until an item is available.

get_nowait() → jvconnected.utils.NamedItem

Remove and return an item from the queue.

Return an item if one is immediately available, else raise QueueEmpty.