osc

class AsyncOscProtocol(*, name: str | None = None, on_connect_callback: Callable | None = None, on_disconnect_callback: Callable | None = None, on_panic_callback: Callable | None = None)

Bases: DatagramProtocol, OscProtocol

activate_healthcheck() → None
async connect(ip_address: str, port: int, *, healthcheck: HealthCheck | None = None)
connection_lost(exc) → None

Called when the connection is lost or closed.

The argument is an exception object or None (the latter meaning a regular EOF is received or the connection was aborted or closed).

connection_made(transport) → None

Called when a connection is made.

The argument is the transport representing the pipe connection. To receive data, wait for data_received() calls. When the connection is closed, connection_lost() is called.

datagram_received(data, addr) → None

Called when some datagram is received.

async disconnect() → None
error_received(exc) → None

Called when a send or receive operation raises an OSError.

(Other than BlockingIOError or InterruptedError.)

register(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 a callback.

send(message: Sequence | SupportsOsc | str) → None
unregister(callback: OscCallback) → None
class Capture(osc_protocol: OscProtocol)

Bases: object

__enter__() → Capture
__exit__(exc_type, exc_value, traceback) → None
__getitem__(i: int | slice) → CaptureEntry | list[CaptureEntry]
__iter__() → Iterator[CaptureEntry]
__len__() → int
add_entry(timestamp: float, label: 'R' | 'S', message: OscBundle | OscMessage, raw_message: Sequence | SupportsOsc | str | None = None) → None
filtered(sent: bool = True, received: bool = True, status: bool = False) → list[CaptureEntry]
class CaptureEntry(timestamp, label, message, raw_message)

Bases: NamedTuple

class HealthCheck(request_pattern: list[str], response_pattern: list[str], active: bool = True, timeout: float = 1.0, backoff_factor: float = 1.5, max_attempts: int = 5)

Bases: object

class OscBundle(timestamp: float | None = None, *, contents: Sequence[OscBundle | OscMessage])

Bases: object

An OSC bundle.

>>> import supriya.osc
>>> message_one = supriya.osc.OscMessage("/one", 1)
>>> message_two = supriya.osc.OscMessage("/two", 2)
>>> message_three = supriya.osc.OscMessage("/three", 3)

>>> inner_bundle = supriya.osc.OscBundle(
...     timestamp=1401557034.5,
...     contents=(message_one, message_two),
... )
>>> inner_bundle
OscBundle(timestamp=1401557034.5, contents=[OscMessage('/one', 1), OscMessage('/two', 2)])

>>> print(inner_bundle)
size 56
   0   23 62 75 6e  64 6c 65 00  d7 34 8e aa  80 00 00 00   |#bundle..4......|
  16   00 00 00 10  2f 6f 6e 65  00 00 00 00  2c 69 00 00   |..../one....,i..|
  32   00 00 00 01  00 00 00 10  2f 74 77 6f  00 00 00 00   |......../two....|
  48   2c 69 00 00  00 00 00 02                             |,i......|

>>> outer_bundle = supriya.osc.OscBundle(
...     contents=(inner_bundle, message_three),
... )
>>> outer_bundle
OscBundle(contents=[OscBundle(timestamp=1401557034.5, contents=[OscMessage('/one', 1), OscMessage('/two', 2)]), OscMessage('/three', 3)])

>>> print(outer_bundle)
size 96
   0   23 62 75 6e  64 6c 65 00  00 00 00 00  00 00 00 01   |#bundle.........|
  16   00 00 00 38  23 62 75 6e  64 6c 65 00  d7 34 8e aa   |...8#bundle..4..|
  32   80 00 00 00  00 00 00 10  2f 6f 6e 65  00 00 00 00   |......../one....|
  48   2c 69 00 00  00 00 00 01  00 00 00 10  2f 74 77 6f   |,i........../two|
  64   00 00 00 00  2c 69 00 00  00 00 00 02  00 00 00 10   |....,i..........|
  80   2f 74 68 72  65 65 00 00  2c 69 00 00  00 00 00 03   |/three..,i......|

>>> datagram = outer_bundle.to_datagram()

>>> decoded_bundle = supriya.osc.OscBundle.from_datagram(datagram)
>>> decoded_bundle
OscBundle(contents=[OscBundle(timestamp=1401557034.5, contents=[OscMessage('/one', 1), OscMessage('/two', 2)]), OscMessage('/three', 3)])

>>> decoded_bundle == outer_bundle
True
classmethod from_datagram(datagram: bytes) → OscBundle
classmethod partition(messages: list[OscMessage], timestamp: float | None = None) → list[OscBundle]
to_datagram(realtime: bool = True) → bytes
to_list()
to_osc() → OscBundle
class OscCallback(protocol, pattern, procedure, failure_pattern, once, args, kwargs)

Bases: NamedTuple

unregister() → None
class OscMessage(address: Enum | int | str, *contents: Enum | OscBundle | OscMessage | Sequence[Enum | OscBundle | OscMessage | Sequence[OscArgument] | bool | bytes | float | str | None] | bool | bytes | float | str | None)

Bases: object

An OSC message.

>>> from supriya.osc import OscMessage
>>> osc_message = OscMessage("/g_new", 0, 0)
>>> osc_message
OscMessage('/g_new', 0, 0)

>>> datagram = osc_message.to_datagram()
>>> OscMessage.from_datagram(datagram)
OscMessage('/g_new', 0, 0)

>>> print(osc_message)
size 20
   0   2f 67 5f 6e  65 77 00 00  2c 69 69 00  00 00 00 00   |/g_new..,ii.....|
  16   00 00 00 00                                          |....|

>>> osc_message = OscMessage("/foo", True, [None, [3.25]], OscMessage("/bar"))
>>> osc_message
OscMessage('/foo', True, [None, [3.25]], OscMessage('/bar'))

>>> datagram = osc_message.to_datagram()
>>> OscMessage.from_datagram(datagram)
OscMessage('/foo', True, [None, [3.25]], OscMessage('/bar'))

>>> print(osc_message)
size 40
   0   2f 66 6f 6f  00 00 00 00  2c 54 5b 4e  5b 66 5d 5d   |/foo....,T[N[f]]|
  16   62 00 00 00  40 50 00 00  00 00 00 0c  2f 62 61 72   |b...@P....../bar|
  32   00 00 00 00  2c 00 00 00                             |....,...|

>>> osc_message = supriya.osc.OscMessage(
...     "/foo",
...     1,
...     2.5,
...     supriya.osc.OscBundle(
...         contents=(
...             supriya.osc.OscMessage("/bar", "baz", 3.0),
...             supriya.osc.OscMessage("/ffff", False, True, None),
...         )
...     ),
...     ["a", "b", ["c", "d"]],
... )
>>> osc_message
OscMessage('/foo', 1, 2.5, OscBundle(contents=[OscMessage('/bar', 'baz', 3.0), OscMessage('/ffff', False, True, None)]), ['a', 'b', ['c', 'd']])

>>> datagram = osc_message.to_datagram()
>>> OscMessage.from_datagram(datagram)
OscMessage('/foo', 1, 2.5, OscBundle(contents=[OscMessage('/bar', 'baz', 3.0), OscMessage('/ffff', False, True, None)]), ['a', 'b', ['c', 'd']])

>>> print(osc_message)
size 112
   0   2f 66 6f 6f  00 00 00 00  2c 69 66 62  5b 73 73 5b   |/foo....,ifb[ss[|
  16   73 73 5d 5d  00 00 00 00  00 00 00 01  40 20 00 00   |ss]]........@ ..|
  32   00 00 00 3c  23 62 75 6e  64 6c 65 00  00 00 00 00   |...<#bundle.....|
  48   00 00 00 01  00 00 00 14  2f 62 61 72  00 00 00 00   |......../bar....|
  64   2c 73 66 00  62 61 7a 00  40 40 00 00  00 00 00 10   |,sf.baz.@@......|
  80   2f 66 66 66  66 00 00 00  2c 46 54 4e  00 00 00 00   |/ffff...,FTN....|
  96   61 00 00 00  62 00 00 00  63 00 00 00  64 00 00 00   |a...b...c...d...|
classmethod from_datagram(datagram: bytes) → OscMessage
to_datagram() → bytes
to_list()
to_osc() → OscMessage
class OscProtocol(*, name: str | None = None, on_connect_callback: Callable | None = None, on_disconnect_callback: Callable | None = None, on_panic_callback: Callable | None = None)

Bases: object

activate_healthcheck() → None
capture() → Capture
disconnect() → Awaitable[None] | None
register(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
send(message: Sequence | SupportsOsc | str) → None
unregister(callback: OscCallback) → None
class ThreadedOscProtocol(*, name: str | None = None, on_connect_callback: Callable | None = None, on_disconnect_callback: Callable | None = None, on_panic_callback: Callable | None = None)

Bases: OscProtocol

class Handler(request, client_address, server)
class Server(server_address, RequestHandlerClass, bind_and_activate=True)
activate_healthcheck() → None
connect(ip_address: str, port: int, *, healthcheck: HealthCheck | None = None) → None
disconnect() → None
register(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 a callback.

send(message: Sequence | SupportsOsc | str) → None
unregister(callback: OscCallback) → None

Unregister a callback.

exception OscProtocolAlreadyConnected
exception OscProtocolOffline
find_free_port() → int
format_datagram(datagram: bytes) → str
format_messages(messages: Sequence[OscBundle | OscMessage]) → str

Format a sequence of OSC messages as a string.

Provides a more concise means of verifying OSC contents in the test suite than comparing the messages directly.