realtime¶

Tools for interacting with realtime execution contexts.

class AsyncServer(options: Options | None = None, name: str | None = None, **kwargs)¶

Bases: BaseServer

A realtime execution context with asyncio-based OSC and process protocols.

Parameters:

options: Options | None = None¶: The context’s options.
**kwargs¶: Keyword arguments for options.

async boot(*, options: Options | None = None, **kwargs) → AsyncServer¶

Boot the server.

Parameters:

options: Options | None = None¶: The context’s options.
**kwargs¶: Keyword arguments for options.

async connect(*, options: Options | None = None, **kwargs) → AsyncServer¶

Connect to a running server.

Parameters:

options: Options | None = None¶: The context’s options.
**kwargs¶: Keyword arguments for options.

async disconnect() → AsyncServer¶: Disconnect from a running server.

async dump_tree(group: Group | None = None, include_controls: bool = True, sync: bool = True) → QueryTreeGroup | None¶

Dump the server’s node tree.

Emit /g_dumpTree requests.

Parameters:

group: Group | None = None¶: The group whose tree to query. Defaults to the root node.
include_controls: bool = True¶: Flag for including synth control values.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

async get_buffer(buffer: Buffer, *indices: int, sync: bool = True) → dict[int, float] | None¶

Get a buffer sample.

Emit /b_get requests.

Parameters:

buffer: Buffer¶: The buffer whose sample to get.
*indices: int¶: The sample indices to read.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

async get_buffer_range(buffer: Buffer, index: int, count: int, sync: bool = True) → Sequence[float] | None¶

Get a buffer sample range.

Emit /b_getn requests.

Parameters:

buffer: Buffer¶: The buffer whose samples to get.
index: int¶: The sample index to start reading at.
count: int¶: The number of samples to read.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

async get_bus(bus: Bus, sync: bool = True, use_shared_memory: bool = False) → float | None¶

Get a control bus value.

Emit /c_get requests.

Parameters:

bus: Bus¶: The control bus whose value to get.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.
use_shared_memory: bool = False¶: If true and sync=True, use the shared memory interface.

async get_bus_range(bus: Bus, count: int, sync: bool = True, use_shared_memory: bool = False) → Sequence[float] | None¶

Get a range of control bus values.

Emit /c_getn requests.

Parameters:

bus: Bus¶: The control bus to start reading at.
count: int¶: The number of contiguous buses whose values to get.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.
use_shared_memory: bool = False¶: If true and sync=True, use the shared memory interface.

async get_synth_control_range(synth: Synth, control: int | str, count: int, sync: bool = True) → Sequence[float | str] | None¶

Get a range of synth controls.

Emit /s_getn requests.

Parameters:

synth: Synth¶: The synth whose control to get.
control: int | str¶: The control to start reading at.
count: int¶: The number of contiguous controls to get.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

async get_synth_controls(synth: Synth, *controls: int | str, sync: bool = True) → dict[int | str, float] | None¶

Get a synth control.

Emit /s_get requests.

Parameters:

synth: Synth¶: The synth whose control to get.
*controls: int | str¶: The controls to get.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

async query_buffer(buffer: Buffer, sync: bool = True) → BufferInfo | None¶

Query a buffer.

Emit /b_query requests.

Parameters:

buffer: Buffer¶: The buffer to query.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

async query_node(node: Node, sync: bool = True) → NodeInfo | None¶

Query a node.

Emit /n_query requests.

Parameters:

node: Node¶: The node to query.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

async query_status(sync: bool = True) → StatusInfo | None¶

Query the server’s status.

Emit /status requests.

Parameters:

sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

async query_tree(group: Group | None = None, include_controls: bool = True, sync: bool = True) → QueryTreeGroup | None¶

Query the server’s node tree.

Emit /g_queryTree requests.

Parameters:

group: Group | None = None¶: The group whose tree to query. Defaults to the root node.
include_controls: bool = True¶: Flag for including synth control values.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

async query_version(sync: bool = True) → VersionInfo | None¶

Query the server’s version.

Emit /version requests.

Parameters:

sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

async quit(force: bool = False) → AsyncServer¶

Quit the server.

Emit /quit requests.

Parameters:

force: bool = False¶: Force the server to quit, even if this client doesn’t own the process.

async reboot() → AsyncServer¶: Reboot the server.

register_lifecycle_callback(event: ServerLifecycleEvent | SupportsInt | str | None | Iterable[ServerLifecycleEvent | SupportsInt | str | None], procedure: Callable[[ServerLifecycleEvent], Awaitable[None] | None], *, once: bool = False, args: tuple | None = None, kwargs: dict | None = None) → ServerLifecycleCallback¶: Register a server lifecycle callback.

register_osc_callback(pattern: Sequence[float | str], procedure: Callable[[OscMessage], Awaitable[None] | None], *, failure_pattern: Sequence[float | str] | None = None, once: bool = False, args: tuple | None = None, kwargs: dict | None = None) → OscCallback¶: Register an OSC callback.

async reset() → AsyncServer¶: Reset the server’s state without quitting.

async sync(sync_id: int | None = None, timeout: float = 1.0) → AsyncServer¶

Sync the server.

Emit /sync requests.

Parameters:

sync_id: int | None = None¶: The sync ID to wait on.

async trace_node(node: Node) → None¶

Trace a node

Emit /n_trace requests.

Parameters:

node: Node¶: The node to trace.

unregister_osc_callback(callback: OscCallback) → None¶

Unregister an OSC callback.

Parameters:

callback: OscCallback¶: The callback to unregister.

property boot_future : Future[bool]¶

Get the server’s boot future.

Only reference this _after_ booting or connecting, as the future is created when booting or connecting.

property exit_future : Future[bool]¶

Get the server’s exit future.

Only reference this _after_ booting or connecting, as the future is created when booting or connecting.

property osc_protocol : AsyncOscProtocol¶: Get the server’s OSC protocol.

property process_protocol : AsyncProcessProtocol¶: Get the server’s process protocol.

class BaseServer(options: Options | None, name: str | None = None, **kwargs)¶

Bases: Context

Base class for realtime execution contexts.

Parameters:

options: Options | None¶: The context’s options.
**kwargs¶: Keyword arguments for options.

__contains__(object_: ContextObject) → bool¶

add_amplitude_scope(bus: Bus | BusGroup, add_action: AddAction | SupportsInt | str | None = AddAction.ADD_TO_TAIL, target_node: Node | None = None) → AmplitudeScope¶: Add an amplitude scope.

add_frequency_scope(bus: Bus, add_action: AddAction | SupportsInt | str | None = AddAction.ADD_TO_TAIL, fft_size: int = 4096, frequency_mode: 'linear' | 'logarithmic' = 'linear', rate: int = 4, target_node: Node | None = None) → FrequencyScope¶: Add a frequency scope.

send(message: Sequence | SupportsOsc | str) → None¶

Send a message to the execution context.

Parameters:

message: Sequence | SupportsOsc | str¶: The message to send.

set_latency(latency: float) → None¶

set the context’s latency.

Parameters:

latency: float¶: The latency in seconds.

unregister_lifecycle_callback(callback: ServerLifecycleCallback) → None¶

Unregister a lifecycle callback.

Parameters:

callback: ServerLifecycleCallback¶: The callback to unregister.

property default_group : Group¶: Get the server’s default group.

property is_owner : bool¶: Get the server’s ownership flag.

property shared_memory : ServerSHM | None¶: Get the server’s shared memory interface, if available.

property status : StatusInfo | None¶: Get the server’s last received status.

class Server(options: Options | None = None, name: str | None = None, **kwargs)¶

Bases: BaseServer

A realtime execution context with threading-based OSC and process protocols.

Parameters:

options: Options | None = None¶: The context’s options.
**kwargs¶: Keyword arguments for options.

boot(*, options: Options | None = None, **kwargs) → Server¶

Boot the server.

Parameters:

options: Options | None = None¶: The context’s options.
**kwargs¶: Keyword arguments for options.

connect(*, options: Options | None = None, **kwargs) → Server¶

Connect to a running server.

Parameters:

options: Options | None = None¶: The context’s options.
**kwargs¶: Keyword arguments for options.

disconnect() → Server¶: Disconnect from a running server.

dump_tree(group: Group | None = None, include_controls: bool = True, sync: bool = True) → QueryTreeGroup | None¶

Dump the server’s node tree.

Emit /g_dumpTree requests.

Parameters:

group: Group | None = None¶: The group whose tree to query. Defaults to the root node.
include_controls: bool = True¶: Flag for including synth control values.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

get_buffer(buffer: Buffer, *indices: int, sync: bool = True) → dict[int, float] | None¶

Get a buffer sample.

Emit /b_get requests.

Parameters:

buffer: Buffer¶: The buffer whose sample to get.
*indices: int¶: The sample indices to read.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

get_buffer_range(buffer: Buffer, index: int, count: int, sync: bool = True) → Sequence[float] | None¶

Get a buffer sample range.

Emit /b_getn requests.

Parameters:

buffer: Buffer¶: The buffer whose samples to get.
index: int¶: The sample index to start reading at.
count: int¶: The number of samples to read.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

get_bus(bus: Bus, sync: bool = True, use_shared_memory: bool = False) → float | None¶

Get a control bus value.

Emit /c_get requests.

Parameters:

bus: Bus¶: The control bus whose value to get.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.
use_shared_memory: bool = False¶: If true and sync=True, use the shared memory interface.

get_bus_range(bus: Bus, count: int, sync: bool = True, use_shared_memory: bool = False) → Sequence[float] | None¶

Get a range of control bus values.

Emit /c_getn requests.

Parameters:

bus: Bus¶: The control bus to start reading at.
count: int¶: The number of contiguous buses whose values to get.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.
use_shared_memory: bool = False¶: If true and sync=True, use the shared memory interface.

get_synth_control_range(synth: Synth, control: int | str, count: int, sync: bool = True) → Sequence[float | str] | None¶

Get a range of synth controls.

Emit /s_getn requests.

Parameters:

synth: Synth¶: The synth whose control to get.
control: int | str¶: The control to start reading at.
count: int¶: The number of contiguous controls to get.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

get_synth_controls(synth: Synth, *controls: int | str, sync: bool = True) → dict[int | str, float] | None¶

Get a synth control.

Emit /s_get requests.

Parameters:

synth: Synth¶: The synth whose control to get.
*controls: int | str¶: The controls to get.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

query_buffer(buffer: Buffer, sync: bool = True) → BufferInfo | None¶

Query a buffer.

Emit /b_query requests.

Parameters:

buffer: Buffer¶: The buffer to query.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

query_node(node: Node, sync: bool = True) → NodeInfo | None¶

Query a node.

Emit /n_query requests.

Parameters:

node: Node¶: The node to query.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

query_status(sync: bool = True) → StatusInfo | None¶

Query the server’s status.

Emit /status requests.

Parameters:

sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

query_tree(group: Group | None = None, include_controls: bool = True, sync: bool = True) → QueryTreeGroup | None¶

Query the server’s node tree.

Emit /g_queryTree requests.

Parameters:

group: Group | None = None¶: The group whose tree to query. Defaults to the root node.
include_controls: bool = True¶: Flag for including synth control values.
sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

query_version(sync: bool = True) → VersionInfo | None¶

Query the server’s version.

Emit /version requests.

Parameters:

sync: bool = True¶: If true, communicate the request immediately. Otherwise bundle it with the current request context.

quit(force: bool = False) → Server¶

Quit the server.

Emit /quit requests.

Parameters:

force: bool = False¶: Force the server to quit, even if this client doesn’t own the process.

reboot() → Server¶: Reboot the server.

register_lifecycle_callback(event: ServerLifecycleEvent | SupportsInt | str | None | Iterable[ServerLifecycleEvent | SupportsInt | str | None], procedure: Callable[[ServerLifecycleEvent], None], *, once: bool = False, args: tuple | None = None, kwargs: dict | None = None) → ServerLifecycleCallback¶: Register a server lifecycle callback.

register_osc_callback(pattern: Sequence[float | str], procedure: Callable[[OscMessage], None], *, failure_pattern: Sequence[float | str] | None = None, once: bool = False, args: tuple | None = None, kwargs: dict | None = None) → OscCallback¶: Register an OSC callback.

reset() → Server¶: Reset the server’s state without quitting.

sync(sync_id: int | None = None, timeout: float = 1.0) → Server¶

Sync the server.

Emit /sync requests.

Parameters:

sync_id: int | None = None¶: The sync ID to wait on.

trace_node(node: Node) → None¶

Trace a node

Emit /n_trace requests.

Parameters:

node: Node¶: The node to trace.

unregister_osc_callback(callback: OscCallback) → None¶

Unregister an OSC callback.

Parameters:

callback: OscCallback¶: The callback to unregister.

property boot_future : Future[bool]¶

Get the server’s boot future.

Only reference this _after_ booting or connecting, as the future is created when booting or connecting.

property exit_future : Future[bool]¶

Get the server’s exit future.

Only reference this _after_ booting or connecting, as the future is created when booting or connecting.

property osc_protocol : ThreadedOscProtocol¶: Get the server’s OSC protocol.

property process_protocol : ThreadedProcessProtocol¶: Get the server’s process protocol.

class ServerLifecycleCallback(context, events, procedure, once, args, kwargs)¶

Bases: NamedTuple

unregister() → None¶

exception FailWarning¶