Source code for scamp_extensions.playback.supercollider.sc_lang

"""
Module containing functionality for starting up and communicating with an instance of sclang. (Note that this assumes
that SuperCollider is installed and can be run from the command line.)
"""

#  ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++  #
#  This file is part of SCAMP (Suite for Computer-Assisted Music in Python)                      #
#  Copyright © 2020 Marc Evanstein <marc@marcevanstein.com>.                                     #
#                                                                                                #
#  This program is free software: you can redistribute it and/or modify it under the terms of    #
#  the GNU General Public License as published by the Free Software Foundation, either version   #
#  3 of the License, or (at your option) any later version.                                      #
#                                                                                                #
#  This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;     #
#  without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.     #
#  See the GNU General Public License for more details.                                          #
#                                                                                                #
#  You should have received a copy of the GNU General Public License along with this program.    #
#  If not, see <http://www.gnu.org/licenses/>.                                                   #
#  ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++  #

from subprocess import Popen
import socket
from threading import Event
from pythonosc import dispatcher, osc_server, udp_client
import threading
import inspect
import os
import atexit


module_dir = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))


def _pick_unused_port():
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.bind(('localhost', 0))
    address, port = s.getsockname()
    s.close()
    return port


[docs]class SCLangInstance: """ Object that starts up an instance of sclang as a subprocess, and facilitates communication with that subprocess via OSC. The SCAMP supercollider extensions will be loaded into the sclang library path. """ def __init__(self): self._listening_port = _pick_unused_port() command = ["sclang", "-l", os.path.join(module_dir, "./scamp_sc_config.yaml"), os.path.join(module_dir, "scInit.scd"), str(self._listening_port)] Popen(command, cwd=module_dir) self.port = self.wait_for_response("/supercollider/port") self._client = udp_client.SimpleUDPClient("127.0.0.1", self.port) atexit.register(lambda: self.send_message("/quit", 0))
[docs] def send_message(self, address, value) -> None: """ Sends an OSC message to the running instance of sclang. :param address: the osc address string :param value: the message value """ self._client.send_message(address, value)
[docs] def wait_for_response(self, address) -> str: """ Waits for a response from sclang to be sent to the given address, confirming that we are on the same page and telling us what address to send messages to. :param address: the OSC message address at which to expect the response. """ osc_dispatcher = dispatcher.Dispatcher() response = None response_received = Event() def response_handler(_, message): nonlocal response response = message response_received.set() osc_dispatcher.map(address, response_handler) server = osc_server.ThreadingOSCUDPServer(('127.0.0.1', self._listening_port), osc_dispatcher) threading.Thread(target=server.serve_forever, daemon=True).start() response_received.wait() server.shutdown() return response
[docs] def new_synth_def(self, synth_def_code: str) -> None: r""" Sends the given SynthDef to SuperCollider to compile and add to the server. :param synth_def_code: the sclang code for the SynthDef (i.e. "SynthDef(\nameOFSynth, {[ugen graph function]}"). """ self.send_message("/compile/synth_def", [synth_def_code]) self.wait_for_response("/done_compiling")