osc¶

Tools for sending, receiving and handling OSC messages.

class AsyncOscProtocol¶

Bases: DatagramProtocol, OscProtocol

activate_healthcheck() → None¶

async connect(ip_address: str, port: int, *, healthcheck: HealthCheck | None = None)¶

connection_lost(exc)¶

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)¶

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)¶: Called when some datagram is received.

error_received(exc)¶

Called when a send or receive operation raises an OSError.

(Other than BlockingIOError or InterruptedError.)

register(pattern: Sequence[str | float], procedure: Callable[[OscMessage], None], *, failure_pattern: Sequence[str | float] | None = None, once: bool = False) → OscCallback¶

send(message)¶

unregister(callback: OscCallback)¶

class Capture(osc_protocol)¶

Bases: object

__enter__()¶

__exit__(exc_type, exc_value, traceback)¶

__iter__()¶

__len__()¶

filtered(sent=True, received=True, status=True) → list[OscBundle | OscMessage]¶

property received_messages¶

property sent_messages¶

class CaptureEntry(timestamp, label, message)¶: Bases: NamedTuple

class HealthCheck(request_pattern: list[str], response_pattern: list[str], callback: collections.abc.Callable, active: bool = True, timeout: float = 1.0, backoff_factor: float = 1.5, max_attempts: int = 5)¶: Bases: object

class OscBundle(timestamp=None, contents=None)¶

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(
    contents=(
        OscMessage('/one', 1),
        OscMessage('/two', 2),
    ),
    timestamp=1401557034.5,
)

>>> 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(
            contents=(
                OscMessage('/one', 1),
                OscMessage('/two', 2),
            ),
            timestamp=1401557034.5,
        ),
        OscMessage('/three', 3),
    ),
)

>>> print(outer_bundle)
size 96
 23 62 75 6e  64 6c 65 00  00 00 00 00  00 00 00 01   |#bundle.........|
 00 00 00 38  23 62 75 6e  64 6c 65 00  d7 34 8e aa   |...8#bundle..4..|
 80 00 00 00  00 00 00 10  2f 6f 6e 65  00 00 00 00   |......../one....|
 2c 69 00 00  00 00 00 01  00 00 00 10  2f 74 77 6f   |,i........../two|
 00 00 00 00  2c 69 00 00  00 00 00 02  00 00 00 10   |....,i..........|
 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(
            contents=(
                OscMessage('/one', 1),
                OscMessage('/two', 2),
            ),
            timestamp=1401557034.5,
        ),
        OscMessage('/three', 3),
    ),
)

>>> decoded_bundle == outer_bundle
True

classmethod from_datagram(datagram)¶

classmethod partition(messages, timestamp=None)¶

to_datagram(realtime=True) → bytes¶

to_list()¶

class OscCallback(pattern, procedure, failure_pattern, once)¶: Bases: NamedTuple

class OscMessage(address, *contents)¶

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
 2f 66 6f 6f  00 00 00 00  2c 69 66 62  5b 73 73 5b   |/foo....,ifb[ss[|
 73 73 5d 5d  00 00 00 00  00 00 00 01  40 20 00 00   |ss]]........@ ..|
 00 00 00 3c  23 62 75 6e  64 6c 65 00  00 00 00 00   |...<#bundle.....|
 00 00 00 01  00 00 00 14  2f 62 61 72  00 00 00 00   |......../bar....|
 2c 73 66 00  62 61 7a 00  40 40 00 00  00 00 00 10   |,sf.baz.@@......|
 2f 66 66 66  66 00 00 00  2c 46 54 4e  00 00 00 00   |/ffff...,FTN....|
 61 00 00 00  62 00 00 00  63 00 00 00  64 00 00 00   |a...b...c...d...|

classmethod from_datagram(datagram)¶

to_datagram() → bytes¶

to_list()¶

class OscProtocol¶

Bases: object

abstract activate_healthcheck() → None¶

capture() → Capture¶

disconnect() → None¶

abstract register(pattern: Sequence[str | float], procedure: Callable[[OscMessage], None], *, failure_pattern: Sequence[str | float] | None = None, once: bool = False) → OscCallback¶

abstract send(message) → None¶

abstract unregister(callback: OscCallback)¶

class ThreadedOscHandler(request, client_address, server)¶

Bases: BaseRequestHandler

handle()¶

class ThreadedOscProtocol¶

Bases: OscProtocol

activate_healthcheck() → None¶

connect(ip_address: str, port: int, *, healthcheck: HealthCheck | None = None)¶

register(pattern: Sequence[str | float], procedure: Callable[[OscMessage], None], *, failure_pattern: Sequence[str | float] | None = None, once: bool = False) → OscCallback¶: Register a callback.

send(message) → None¶

unregister(callback: OscCallback) → None¶: Unregister a callback.

class ThreadedOscServer(server_address, RequestHandlerClass, bind_and_activate=True)¶

Bases: UDPServer

service_actions()¶

Called by the serve_forever() loop.

May be overridden by a subclass / Mixin to implement any code that needs to be run during the loop.

verify_request(request, client_address)¶

Verify the request. May be overridden.

Return True if we should proceed with this request.

exception OscProtocolAlreadyConnected¶

exception OscProtocolOffline¶

find_free_port()¶

format_datagram(datagram)¶