sc3nb.osc.osc_communication

OSC communication

Classes and functions to communicate with SuperCollider using the Open Sound Control (OSC) protocol over UDP

Module Contents

Function List

get_max_udp_packet_size	Get the max UDP packet size by trial and error
split_into_max_size
convert_to_sc3nb_osc	Get binary OSC representation

Class List

OSCMessage	Class for creating messages to send over OSC
Bundler	Class for creating OSCBundles and bundling of messages
MessageHandler	Base class for Message Handling
MessageQueue	Queue to retrieve OSC messages send to the corresponding OSC address
MessageQueueCollection	A collection of MessageQueues that are all sent to one and the same first address.
OSCCommunication	Class to send and receive OSC messages and bundles.

Content

sc3nb.osc.osc_communication._LOGGER

sc3nb.osc.osc_communication.get_max_udp_packet_size()

Get the max UDP packet size by trial and error

sc3nb.osc.osc_communication.split_into_max_size(bundler: Bundler, max_dgram_size) → List[pythonosc.osc_bundle.OscBundle]

class sc3nb.osc.osc_communication.OSCMessage(msg_address: str, msg_parameters: Optional[Union[Sequence, Any]] = None)

Class for creating messages to send over OSC

Parameters:

msg_addressstr

OSC message address

msg_parametersOptional[Union[Sequence]], optional

OSC message parameters, by default None

Overview:

to_pythonosc	Return python-osc OscMessage
_build_message	Builds pythonsosc OSC message.
__repr__	Return repr(self).

to_pythonosc() → pythonosc.osc_message.OscMessage

Return python-osc OscMessage

static _build_message(msg_address: str, msg_parameters: Optional[Union[Sequence, Any]] = None) → pythonosc.osc_message.OscMessage

Builds pythonsosc OSC message.

Parameters:

msg_addressstr

SuperCollider address.

msg_parameterslist, optional

List of parameters to add to message.

Returns:

OscMessage

Message ready to be sent.

__repr__() → str

Return repr(self).

class sc3nb.osc.osc_communication.Bundler(timetag: float = 0, msg: Optional[Union[OSCMessage, str]] = None, msg_params: Optional[Sequence[Any]] = None, *, server: Optional[OSCCommunication] = None, receiver: Optional[Union[str, Tuple[str, int]]] = None, send_on_exit: bool = True)

Class for creating OSCBundles and bundling of messages

Create a Bundler

Parameters:

timetagfloat, optional

Starting time at which bundle content should be executed. If timetag > 1e6 it is interpreted as POSIX time. If timetag <= 1e6 it is assumed to be relative value in seconds and is added to time.time(), by default 0, i.e. ‘now’.

msgOSCMessage or str, optional

OSCMessage or message address, by default None

msg_paramssequence of any type, optional

Parameters for the message, by default None

serverOSCCommunication, optional

OSC server, by default None

receiverUnion[str, Tuple[str, int]], optional

Where to send the bundle, by default send to default receiver of server

send_on_exitbool, optional

Whether the bundle is sent when using as context manager, by default True

Overview:

wait	Add time to internal time
add	Add content to this Bundler.
messages	Generate a dict with all messages in this Bundler.
send	Send this Bundler.
to_raw_osc	Create a raw OSC Bundle from this bundler.
to_pythonosc	Build this bundle.
_calc_timetag
__deepcopy__
__enter__
__exit__
__repr__	Return repr(self).

wait(time_passed: float) → None

Add time to internal time

Parameters:

time_passedfloat

How much seconds should be passed.

add(*args) → Bundler

Add content to this Bundler.

Parameters:

argsaccepts an OSCMessage or Bundler

or a timetag with an OSCMessage or Bundler or Bundler arguments like

(timetag, msg_addr, msg_params) (timetag, msg_addr) (timetag, msg)

Returns:

Bundler

self for chaining

messages(start_time: Optional[float] = 0.0, delay: Optional[float] = None) → Dict[float, List[OSCMessage]]

Generate a dict with all messages in this Bundler.

They dict key is the time tag of the messages.

Parameters:

start_timeOptional[float], optional

start time when using relative timing, by default 0.0

Returns:

Dict[float, List[OSCMessage]]

dict containg all OSCMessages

send(server: Optional[OSCCommunication] = None, receiver: Tuple[str, int] = None, bundle: bool = True)

Send this Bundler.

Parameters:

serverOSCCommunication, optional

Server instance for sending the bundle. If None it will use the server from init or try to use sc3nb.SC.get_default().server, by default None

receiverTuple[str, int], optional

Address (ip, port) to send to, if None it will send the bundle to the default receiver of the Bundler

bundlebool, optional

If True this is allowed to be bundled, by default True

Raises:

RuntimeError

When no server could be found.

to_raw_osc(start_time: Optional[float] = None, delay: Optional[float] = None) → bytes

Create a raw OSC Bundle from this bundler.

Parameters:

start_timeOptional[float], optional

used as start time when using relative timing, by default time.time()

delay: float, optinal

used to delay the timing.

Returns:

OscBundle

bundle instance for sending

to_pythonosc(start_time: Optional[float] = None, delay: Optional[float] = None) → pythonosc.osc_bundle.OscBundle

Build this bundle.

Parameters:

start_timeOptional[float], optional

used as start time when using relative timing, by default time.time()

delay: float, optinal

used to delay the timing.

Returns:

OscBundle

bundle instance for sending

_calc_timetag(start_time: Optional[float])

__deepcopy__(memo) → Bundler

__enter__()

__exit__(exc_type, exc_value, exc_traceback)

__repr__() → str

Return repr(self).

sc3nb.osc.osc_communication.convert_to_sc3nb_osc(data: Union[OSCMessage, Bundler, pythonosc.osc_message.OscMessage, pythonosc.osc_bundle.OscBundle, bytes]) → Union[OSCMessage, Bundler]

Get binary OSC representation

Parameters:

packageUnion[OscMessage, Bundler, OscBundle]

OSC Package object

Returns:

bytes

raw OSC binary representation of OSC Package

Raises:

ValueError

If package is not supported

class sc3nb.osc.osc_communication.MessageHandler

Bases: abc.ABC

Base class for Message Handling

Overview:

put	Add message to MessageHandler

abstract put(address: str, *args) → None

Add message to MessageHandler

Parameters:

addressstr

Message address

class sc3nb.osc.osc_communication.MessageQueue(address: str, preprocess: Optional[Callable] = None)

Bases: MessageHandler

Queue to retrieve OSC messages send to the corresponding OSC address

Create a new AddressQueue

Parameters:

addressstr

OSC address for this queue

preprocessfunction, optional

function that will be applied to the value before they are enqueued

(Default value = None)

Overview:

put	Add a message to MessageQueue
skipped	Skipp one queue value
get	Returns a value from the queue
show	Print the content of the queue.
_repr_pretty_

put(address: str, *args) → None

Add a message to MessageQueue

Parameters:

addressstr

message address

skipped()

Skipp one queue value

get(timeout: float = 5, skip: bool = True) → Any

Returns a value from the queue

Parameters:

timeoutint, optional

Time in seconds that will be waited on the queue, by default 5

skipbool, optional

If True the queue will skip as many values as skips, by default True

Returns:

obj

value from queue

Raises:

Empty

If the queue has no value

show() → None

Print the content of the queue.

_repr_pretty_(printer, cycle) → None

class sc3nb.osc.osc_communication.MessageQueueCollection(address: str, sub_addrs: Optional[Sequence[str]] = None)

Bases: MessageHandler

A collection of MessageQueues that are all sent to one and the same first address.

Create a collection of MessageQueues under the same first address

Parameters:

addressstr

first message address that is the same for all MessageQueues

sub_addrsOptional[Sequence[str]], optional

secound message addresses with seperate queues, by default None Additional MessageQueues will be created on demand.

Overview:

put	Add a message to the corresponding MessageQueue
__contains__
__getitem__

put(address: str, *args) → None

Add a message to the corresponding MessageQueue

Parameters:

addressstr

first message address

__contains__(item) → bool

__getitem__(key)

exception sc3nb.osc.osc_communication.OSCCommunicationError(message, send_message)

Bases: Exception

Exception for OSCCommunication errors.

Initialize self. See help(type(self)) for accurate signature.

class sc3nb.osc.osc_communication.OSCCommunication(server_ip: str, server_port: int, default_receiver_ip: str, default_receiver_port: int)

Class to send and receive OSC messages and bundles.

Create an OSC communication server

Parameters:

server_ipstr

IP address to use for this server

server_portint

port to use for this server

default_receiver_ipstr

IP address used for sending by default

default_receiver_portint

port used for sending by default

Overview:

add_msg_pairs	Add the provided pairs for message receiving.
add_msg_queue	Add a MessageQueue to this servers dispatcher
add_msg_queue_collection	Add a MessageQueueCollection
_check_sender
lookup_receiver	Reverse lookup the address of a specific receiver
connection_info	Get information about the known addresses
add_receiver	Adds a receiver with the specified address.
send	Sends OSC packet
_handle_outgoing_message
get_reply_address	Get the corresponding reply address for the given address
msg	Creates and sends OSC message over UDP.
bundler	Generate a Bundler.
quit	Shuts down the sc3nb OSC server

add_msg_pairs(msg_pairs: Dict[str, str]) → None

Add the provided pairs for message receiving.

Parameters:

msg_pairsdict[str, str], optional

dict containing user specified message pairs. {msg_addr: reply_addr}

add_msg_queue(msg_queue: MessageQueue, out_addr: Optional[str] = None) → None

Add a MessageQueue to this servers dispatcher

Parameters:

msg_queueMessageQueue

new MessageQueue

out_addrOptional[str], optional

The outgoing message address that belongs to this MessageQeue, by default None

add_msg_queue_collection(msg_queue_collection: MessageQueueCollection) → None

Add a MessageQueueCollection

Parameters:

msg_queue_collectionMessageQueueCollection

MessageQueueCollection to be added

_check_sender(sender: Tuple[str, int]) → Union[str, Tuple[str, int]]

lookup_receiver(receiver: Union[str, Tuple[str, int]]) → Tuple[str, int]

Reverse lookup the address of a specific receiver

Parameters:

receiverstr

Receiver name.

Returns:

Tuple[str, int]

Receiver address (ip, port)

Raises:

KeyError

If receiver is unknown.

ValueError

If the type of the receiver argument is wrong.

connection_info(print_info: bool = True) → Tuple[Tuple[str, int], Dict[Tuple[str, int], str]]

Get information about the known addresses

Parameters:

print_infobool, optional

If True print connection information

(Default value = True)

Returns:

tuple

containing the address of this sc3nb OSC Server and known receivers addresses in a dict with their names as values

add_receiver(name: str, ip_address: str, port: int)

Adds a receiver with the specified address.

Parameters:

namestr

Name of receiver.

ip_addressstr

IP address of receiver (e.g. “127.0.0.1”)

portint

Port of the receiver

send(package: Union[OSCMessage, Bundler], *, receiver: Optional[Union[str, Tuple[str, int]]] = None, bundle: bool = False, await_reply: bool = True, timeout: float = 5) → Any

Sends OSC packet

Parameters:

packageOSCMessage or Bundler

Object with dgram attribute.

receiverstr or Tuple[str, int], optional

Where to send the packet, by default send to default receiver

bundlebool, optional

If True it is allowed to bundle the package with bundling, by default False.

await_replybool, optional

If True ask for reply from the server and return it, otherwise send the message and return None directly, by default True. If the package is bundled None will be returned.

timeoutint, optional

timeout in seconds for reply, by default 5

Returns:

None or reply

None if no reply was received or awaited else reply.

Raises:

ValueError

When the provided package is not supported.

OSCCommunicationError

When the handling of a package fails.

_handle_outgoing_message(message: OSCMessage, receiver_address: Tuple[str, int], await_reply: bool, timeout: float) → Any

get_reply_address(msg_address: str) → Optional[str]

Get the corresponding reply address for the given address

Parameters:

msg_addressstr

outgoing message address

Returns:

str or None

Corresponding reply address if available

msg(msg_addr: str, msg_params: Optional[Sequence] = None, *, bundle: bool = False, receiver: Optional[Tuple[str, int]] = None, await_reply: bool = True, timeout: float = 5) → Optional[Any]

Creates and sends OSC message over UDP.

Parameters:

msg_addrstr

SuperCollider address of the OSC message

msg_paramsOptional[Sequence], optional

List of paramters of the OSC message, by default None

bundlebool, optional

If True it is allowed to bundle the content with bundling, by default False

receivertuple[str, int], optional

(IP address, port) to send the message, by default send to default receiver

await_replybool, optional

If True send message and wait for reply otherwise send the message and return directly, by default True

timeoutfloat, optional

timeout in seconds for reply, by default 5

Returns:

obj

reply if await_reply and there is a reply for this

bundler(timetag: float = 0, msg: Optional[Union[OSCMessage, str]] = None, msg_params: Optional[Sequence[Any]] = None, send_on_exit: bool = True) → Bundler

Generate a Bundler.

This allows the user to easly add messages/bundles and send it.

Parameters:

timetagint

Time at which bundle content should be executed. If timetag <= 1e6 it is added to time.time().

msgOSCMessage or str, optional

OSCMessage or message address, by default None

msg_paramssequence of any type, optional

Parameters for the message, by default None

send_on_exitbool, optional

Whether the bundle is sent when using as context manager, by default True

Returns:

Bundler

bundler for OSC bundling.

quit() → None

Shuts down the sc3nb OSC server