This page was generated from examples/osc-communication-examples.ipynb.

import time
import numpy as np

import sc3nb as scn

OSC communication

With the OSC communication module of sc3nb you can directly send and receive OSC packets.

Open Sound Control (OSC) is a networking protocol for sound and is used by SuperCollider to communicate between sclang and scsynth. sc3nb is itself a OSC client and server. This allows sc3nb to send and receive OSC traffic.

For more information on OSC and especially how Supercollider handles OSC packets please refer to the following links:

• Open Sound Control Specification

• Server vs Client SuperCollider Guide

• Server Command Reference

• SuperCollider Synth Server Architecture

sc = scn.startup()

sc3nb serves as OSC server and as client of the SuperCollider server scsynth. You can also communicate with the SuperCollider interpreter sclang via OSC.

You can see the current connection information with sc.server.connection_info()

(sc3nb_ip, sc3nb_port), receivers = sc.server.connection_info()

(sc3nb_ip, sc3nb_port), receivers

If you want to communicate via OSC with another receiver you could add its name via sc.server.add_receiver(name: str, ip: str, port: int), or you can pass a custom receiver when sending OSC

sc.server.add_receiver("sc3nb", sc3nb_ip, sc3nb_port)

sc.server.connection_info()

Sending OSC

You can send OSC with

help(sc.server.send)

Messages

Use the OSCMessage or the python-osc package to build an OscMessage

scn.OSCMessage?

msg = scn.OSCMessage("/s_new", ["s1", -1, 1, 1,])
sc.server.send(msg)

A shortcut for sending Messages is

help(sc.server.msg)

sc.server.msg("/s_new", ["s1", -1, 1, 1,])

a more complex example

for p in [0,2,4,7,5,5,9,7,7,12,11,12,7,4,0,2,4,5,7,9,7,5,4,2,4,0,-1,0,2,-5,-1,2,5,4,2,4]:
    freq = scn.midicps(60+p)  # see helper fns below
    sc.server.msg("/s_new", ["s1", -1, 1, 0, "freq", freq, "dur", 0.5, "num", 1])
    time.sleep(0.15)

Note that the timing is here under python’s control, which is not very precise. The Bundler class allows to do better.

Remarks:

• note that the python code returns immediately and all events remain in scsynth

• note that unfortunately scsynth has a limited buffer for OSC messages, so it is not viable to spawn thousends of events. scsynth will then simply reject OSC messages.

• this sc3-specific problem motivated (and has been solved with) TimedQueue, see below.

Bundles

from sc3nb.osc.osc_communication import Bundler

To send a single or multiple message(s) with a timetag as an OSC Bundle, you can use the Bundler class

• Bundlers allow to specify a timetag and thus let scsynth control the timing, which is much better, if applicable.

• A Bundler can be created as documented here

Bundler?

The prefered way of creating Bundlers for sending to the server is via

help(sc.server.bundler)

This will add the sc.server.latency time to the timetag. By default this is 0.0 but you can set it.

sc.server.latency

sc.server.latency = 0.1
sc.server.latency

A Bundler lets you add Messages and other Bundlers to the Bundler and accepts

• an OSCMessage or Bundler

• an timetag with an OSCMessage or Bundler

• or Bundler arguments like (timetag, msg_addr, msg_params) (timetag, msg_addr) (timetag, msg)

Also see Nesting Bundlers for more details on how Bundlers are nested

help(Bundler.add)

add returns the Bundler for chaining

msg1 = scn.OSCMessage("/s_new", ["s2", -1, 1, 1,])
msg2 = scn.OSCMessage("/n_free", [-1])
sc.server.bundler().add(1.5, msg1).add(1.9, msg2).send() # sound starts in 1.5s

An alternative is the usage of the context manager. This means you can use the with statement for better handling as follows:

with sc.server.bundler() as bundler:
    bundler.add(0.0, msg1)
    bundler.add(0.3, msg2)

Instead of declaring the time explicitly with add you can also use wait.

iterations = 3
with sc.server.bundler() as bundler:
    for i in range(iterations):
        bundler.add(msg1)
        bundler.wait(0.3)
        bundler.add(msg2)
        bundler.wait(0.1)

This adds up the internal time passed of the Bundler

bundler.passed_time

assert bundler.passed_time == iterations * 0.3 + iterations * 0.1, "Internal time seems wrong"

Here are some different styles of coding the same sound with the Bundler features

• server Bundler with add and Bundler Arguments

with sc.server.bundler(send_on_exit=False) as bundler:
    bundler.add(0.0, "/s_new", ["s2", -1, 1, 1,])
    bundler.add(0.3, "/n_free", [-1])

dg1 = bundler.to_raw_osc(0.0)  # we set the time_offset explicitly so all Bundle datagrams are the same
dg1

• Bundler with explict latency set and using add

with Bundler(sc.server.latency, send_on_exit=False) as bundler:
    bundler.add(0.0, "/s_new", ["s2", -1, 1, 1])
    bundler.add(0.3, "/n_free", [-1])

dg2 = bundler.to_raw_osc(0.0)
dg2

• server Bundler with implicit latency and using automatic bundled messages (See Automatic Bundling)

with sc.server.bundler(send_on_exit=False) as bundler:
    sc.server.msg("/s_new", ["s2", -1, 1, 1,], bundle=True)
    bundler.wait(0.3)
    sc.server.msg("/n_free", [-1], bundle=True)

dg3 = bundler.to_raw_osc(0.0)
dg3

# assert that all created raw OSC datagrams are the same
assert dg1 == dg2 and dg1 == dg3, "The datagrams are not the same"

Note: You can use the Bundler with the Synth and Group classes for easier Message creation. This also removes the burden of managing the IDs for the different commands.

Also make sure to look at the Automatic Bundling Feature which is using the bundled messages (msg(..., bundle=True)

t0 = time.time()
with sc.server.bundler() as bundler:
    for i, r in enumerate(np.random.randn(100)):
        onset = t0 + 3 + r
        freq = 500 + 5 * i
        bundler.add(onset, scn.Synth("s1",
            {"freq": freq, "dur": 1.5, "num": abs(r)+1}, new=False
        ).new(return_msg=True))

Bundler Timestamp

Small numbers (<1e6) are interpreted as times in seconds relative to time.time(), evaluated at the time of sending

sc.server.bundler(0.5, "/s_new", ["s1", -1, 1, 0, "freq", 200, "dur", 1]).send()  # a tone starts in 0.5s
sc.server.bundler(1.0, "/s_new", ["s1", -1, 1, 0, "freq", 300, "dur", 1]).send()  # a tone starts in 1.0s

Attention:

Sending bundles with relative times could lead to unprecise timings. If you care about precision

• use a bundler with multiple messages (if you care about the timings relative to each other in one Bundler)

  • because all relative times of the inner messages are calculated on top of the outermost bundler timetag

• or provide an explict timetag (>1e6) to specify absolute times (see the following examples)

A single Bundler with multiple messages

bundler = sc.server.bundler()
bundler.add(0.5, "/s_new", ["s1", -1, 1, 0, "freq", 200, "dur", 1])
bundler.add(1.0, "/s_new", ["s1", -1, 1, 0, "freq", 300, "dur", 1])
bundler.send()  # second tone starts in 1.0s

using time.time()+timeoffset for absolute times

t0 = time.time()
sc.server.bundler(t0 + 0.5, "/s_new", ["s1", -1, 1, 0, "freq", 200, "dur", 1]).send()  # a tone starts in 0.5s
sc.server.bundler(t0 + 1.0, "/s_new", ["s1", -1, 1, 0, "freq", 300, "dur", 1]).send()  # a tone starts in 1.0s

t0 = time.time()
with sc.server.bundler() as bundler:
    for i, r in enumerate(np.random.randn(100)): # note: 1000 will give: msg too long
        onset = t0 + 3 + r
        freq = 500 + 5 * i
        msg_params = ["s1", -1, 1, 0, "freq", freq, "dur", 1.5, "num", abs(r)+1]
        bundler.add(onset, "/s_new", msg_params)

sc.server.free_all()

Nesting Bundlers

You can nest Bundlers: this will recalculate the time relative to the sending time of the outermost bundler

with sc.server.bundler() as bundler_outer:
    with sc.server.bundler(0.2) as bundler:
        sc.server.msg("/s_new", ["s2", -1, 1, 1,], bundle=True)
        bundler.wait(0.3)
        sc.server.msg("/n_free", [-1], bundle=True)
    bundler_outer.wait(0.8)
    bundler_outer.add(bundler)

bundler_outer

Or you can nest using Bundler.add and get the same results

bundler_outer_add = sc.server.bundler()
bundler_outer_add.add(bundler)
bundler_outer_add.add(0.8, bundler)

assert bundler_outer.to_raw_osc(0.0) == bundler_outer_add.to_raw_osc(0.0), "Bundler contents are not the same"

Notice that using relative timetags with wait will delay the relative timetags aswell.

bundler_outer_add = sc.server.bundler()
bundler_outer_add.wait(1)  # delays both bundles
bundler_outer_add.add(bundler)
bundler_outer_add.add(0.8, bundler)

bundler_outer_add = sc.server.bundler()
bundler_outer_add.add(bundler)
bundler_outer_add.wait(1)  # delays the 2nd bundle
bundler_outer_add.add(0.8, bundler)

This can be helpful in loops where the relative time then can be seen as relative for this iteration

with sc.server.bundler() as nested_bundler_loop:
    for i in range(3):
        nested_bundler_loop.add(0.5, bundler)
        nested_bundler_loop.wait(1)
nested_bundler_loop

Managing IDs

The OSC commands often require IDs for the different OSC commands. These should be manged by the SCServer to ensure not accidentally using wrong IDs.

• See Allocating IDs in the Server guide for more information about correctly using IDs manually.

• Or use Automatic Bundling and let sc3nb do the work for you!

Automatic Bundling

Probably the most convenient way of sending OSC is by using the Automatic Bundling Feature.

This allows you to simply use the SuperCollider Objects Synth and Group in the Context Manager of a Bundler and they will be automatically captured and stored.

with sc.server.bundler() as bundler:
    synth = scn.Synth("s2")
    bundler.wait(0.3)
    synth.set("freq", 1000)
    bundler.wait(0.1)
    synth.free()
synth.wait()

Note that it is important that to only wait on the Synth after the context of the Bundler has been closed.

If you’d call synth.wait() in the Bundler context, it would wait before sending the /s_new Message to the server and then wait forever (or until timeout) for the /n_end notification.

try:
    with sc.server.bundler() as bundler:
        synth = scn.Synth("s2")
        bundler.wait(0.3)
        synth.set("freq", 1000)
        bundler.wait(0.1)
        synth.free()
        synth.wait(timeout=2)  # without a timeout this would hang forever
except RuntimeError as error:
    print(error)

Receiving OSC packets

sc3nb is receiving OSC messages with the help of queues, one AddressQueue for each OSC address for which we want to receive messages.

sc.server.msg_queues

To see more information what messages are sent and received, set the logging level to INFO as demonstrated below.

import logging
logging.basicConfig(level=logging.INFO)
# even more verbose logging is avaible via
# logging.basicConfig(level=logging.DEBUG)

Getting replies

For certain outgoing OSC messages an incoming Message is defined.

This means that on sending such a message sc3nb automatically waits for the incoming message at the corresponding Queue and returns the result.

An example for this is /sync {sync_id} -> /synced {sync_id}

sc.server.msg("/sync", 12345)

See all (outgoing message, incoming message) pairs:

sc.server.reply_addresses

You can get the reply address via

sc.server.get_reply_address("/sync")

or

sc.server.reply_addresses["/sync"]

If we specify await_reply=False the message will be kept in the queue

sc.server.msg("/sync", 1, await_reply=False)

sc.server.msg_queues[sc.server.get_reply_address("/sync")]

sc.server.msg("/sync", 2, await_reply=False)

sc.server.msg_queues[sc.server.reply_addresses["/sync"]]

sc.server.msg_queues["/synced"]

You can see how many values were hold.

sc.server.msg_queues["/synced"].skips

Notice that these hold messages will be skipped.

sc.server.msg("/sync", 3, await_reply=True)

sc.server.msg_queues["/synced"]

Therefore you should retrieve them with get and set skip=False if you care for old values in the queue and dont want them to be skipped.

sc.server.msg("/sync", 42, await_reply=False)
sc.server.msg_queues["/synced"].get(skip=False)

Custom Message Queues

If you want to get additional OSC Messages you need to create a custom MessageQueue

from sc3nb.osc.osc_communication import MessageQueue

help(MessageQueue)

mq = MessageQueue("/test")

sc.server.add_msg_queue(mq)

sc.server.msg("/test", ["Hi!"], receiver="sc3nb")

sc.server.msg_queues["/test"]

sc.server.msg("/test", ["Hello!"], receiver="sc3nb")

sc.server.msg_queues["/test"]

sc.server.msg_queues["/test"].get()

sc.server.msg_queues["/test"].get()

If you want to create a pair of an outgoing message that will receive a certain incomming message you need to specify it via the out_addr arugment of add_msg_queue or you could use the shortcut for this add_msg_pairs

help(sc.server.add_msg_pairs)

sc.server.add_msg_pairs({"/hi": "/hi.reply"})

Let’s use OSCdef in sclang to send us replies.

%%sc
OSCdef.newMatching("say_hi", {|msg, time, addr, recvPort| addr.sendMsg("/hi.reply", "Hello there!")}, '/hi');

sc.server.msg("/hi", receiver="sclang")

There is also the class MessageQueueCollection, which allows to create multiple MessageQueues for a multiple address/subaddresses combination

from sc3nb.osc.osc_communication import MessageQueueCollection

help(MessageQueueCollection)

mqc = MessageQueueCollection("/collect", ["/address1", "/address2"])
sc.server.add_msg_queue_collection(mqc)

mqc = MessageQueueCollection("/auto_collect")
sc.server.add_msg_queue_collection(mqc)

sc.server.reply_addresses

sc.server.msg_queues

%%scv
OSCdef.newMatching("ab", {|msg, time, addr, recvPort| addr.sendMsg('/collect', '/address1', "toast".scramble)}, '/address1');

%%scv
OSCdef.newMatching("ab", {|msg, time, addr, recvPort| addr.sendMsg('/collect', '/address2', "sonification".scramble)}, '/address2');

sc.server.msg("/address1", receiver="sclang")

sc.server.msg("/address2", receiver="sclang")

Examples

Creating an OSC responder and msg to sclang for synthesis

%%sc
OSCdef(\dinger, { | msg, time, addr, recvPort |
    var freq = msg[2];
    {Pulse.ar(freq, 0.04, 0.3)!2 * EnvGen.ar(Env.perc, doneAction:2)}.play()
}, '/ding')

with scn.Bundler(receiver=sc.lang.addr):
    for i in range(5):
        sc.server.msg("/ding", ["freq", 1000-5*i], bundle=True)

for i in range(5):
    sc.server.msg("/ding", ["freq", 1000-5*i], receiver=sc.lang.addr)

%scv OSCdef.freeAll()

sc.exit()