jvconnected.interfaces.midi.aioport

class jvconnected.interfaces.midi.aioport.BasePort(*args, **kwargs)

Bases: Dispatcher

Async wrapper for mido.ports

Parameters

name (str) – The port name

stopped

Type

asyncio.Event

Property running: bool = False

Current run state

running is a pydispatch.Property object.

EXECUTOR: ClassVar[concurrent.futures.ThreadPoolExecutor] = None

A concurrent.futures.ThreadPoolExecutor to use in the run_in_executor() method for all instances of all BasePort subclasses

Property name: str = None

The port name

name is a pydispatch.Property object.

static get_executor() → concurrent.futures.ThreadPoolExecutor

Get or create the EXECUTOR instance to use in the run_in_executor() method

async run_in_executor(fn: Callable) → Any

Call the given function in the EXECUTOR instance using asyncio.loop.run_in_executor() and return the result

This method is used to create and manipulate all mido ports to avoid blocking, threaded operations

async open() → bool

Open the midi port

Returns

True if the port was successfully opened

Return type

bool

async close()

Close the midi port

class jvconnected.interfaces.midi.aioport.InputPort(*args, **kwargs)

Bases: BasePort

Async wrapper around mido.ports.BaseInput

queue

Message queue for the port

Type

asyncio.Queue

async receive(timeout: Optional[Number] = None) → Optional[Message]

Wait for an incoming message

Parameters

timeout (float, optional) – Time to wait for a message. if None, wait until an item is available

Returns

An instance of mido.Message. If timeout was provided and no message was retrieved, None will be returned.

Return type

Optional[Message]

async receive_many(block: bool = True, timeout: Optional[Number] = None) → Optional[Union[bool, Tuple[Message]]]

Gather any/all available messages

Parameters

• block (bool, optional) – If True, wait_for_msg() is used initially to wait for the first available message. If False, only check for queued messages and return immediately if none exist. Default is True

• timeout (float, optional) – If block is True the timeout argument to pass to the wait_for_msg() method

Returns

If no messages were available (either block was False or the timeout was reached), None is returned.

In all other cases, a tuple of Messages

Return type

Optional[Union[bool, Tuple[Message]]]

async wait_for_msg(timeout: Optional[Number] = None) → bool

Wait until a message is available

Parameters

timeout (float, optional) – Maximum time to wait. If None, wait indefinitely

Returns

False if timeout was given and nothing was available before the timeout. Otherwise True is returned to indicate a message is available

Return type

bool

async queue_get(timeout: Optional[Number] = None) → Any

Convenience method for get() on the queue

Parameters

timeout (float, optional) – Time to wait for an item on the queue. if None, wait until an item is available

async queue_iter_get() → AsyncGenerator[Message, None]

Iterate over any/all messages available on the queue

task_done()

Convenience method for queue task_done()

class jvconnected.interfaces.midi.aioport.OutputPort(*args, **kwargs)

Bases: BasePort

Async wrapper around mido.ports.BaseOutput

queue

Message queue for the port. Since the output port operates in a separate thread, this is a thread-based queue (not async)

Type

queue.Queue

async open() → bool

Open the midi port

Returns

True if the port was successfully opened

Return type

bool

async send(msg: Message)

Send a message

The message will be placed on the queue and sent from a separate thread

Parameters

msg (Message) – The mido.Message to send

async send_many(*msgs)

Send multiple messages

The messages will be placed on the queue and sent from a separate thread

Parameters

*msgs – The Messages to send

class jvconnected.interfaces.midi.aioport.IOPort(*args, **kwargs)

Bases: BasePort