<commit_before># -*- encoding: utf-8 -*-
from __future__ import print_function
import signal
import subprocess
import time


class Server(object):
    r'''An scsynth server proxy.

    ::

        >>> from supriya import servertools
        >>> server = servertools.Server.get_default_server()
        >>> server.boot()
        <Server: udp://127.0.0.1:57751, 8i8o>

    ::

        >>> server.quit()
        <Server: offline>

    '''

    ### CLASS VARIABLES ###

    __slots__ = (
        '_audio_bus_allocator',
        '_audio_busses',
        '_audio_input_bus',
        '_audio_output_bus',
        '_buffer_allocator',
        '_buffers',
        '_control_bus_allocator',
        '_control_busses',
        '_default_group',
        '_ip_address',
        '_is_running',
        '_node_id_allocator',
        '_nodes',
        '_osc_controller',
        '_osc_dispatcher',
        '_port',
        '_response_manager',
        '_root_node',
        '_server_options',
        '_server_process',
        '_server_status',
        '_status_watcher',
        '_sync_id',
        '_synthdefs',
        )

    _default_server = None

    _servers = {}

    ### CONSTRUCTOR ###

    def __new__(cls, ip_address='127.0.0.1', port=57751):
        key = (ip_address, port)
        if key not in cls._servers:
            instance = object.__new__(cls)
            instance.__init__(
                ip_address=ip_address,
                port=port,
                )
            cls._servers[key] = instance
        return cls._servers[key]

    ### INITIALIZER ###

    def __init__(self, ip_address='127.0.0.1', port=57751):
        from supriya.tools import osctools
        from supriya.tools import responsetools

        if hasattr(self, 'is_running') and self.is_running:
            return

        ### NET ADDRESS ###

        self._ip_address = ip_address
        self._port = port

        ### OSC MESSAGING ###

        self._osc_dispatcher = osctools.OscDispatcher()
        self._osc_controller = osctools.OscController(server=self)
        self._response_manager = responsetools.ResponseManager()
        for callback in (
            responsetools.BufferResponseCallback(self),
            responsetools.ControlBusResponseCallback(self),
            responsetools.NodeResponseCallback(self),
            responsetools.SynthDefResponseCallback(self),
            ):
            self.register_osc_callback(callback)

        ### ALLOCATORS ###

        self._audio_bus_allocator = None
        self._buffer_allocator = None
        self._control_bus_allocator = None
        self._node_id_allocator = None

        ### SERVER PROCESS ###

        self._is_running = False
        self._server_options = None
        self._server_process = None
        self._server_status = None
        self._status_watcher = None

        ### PROXIES ###

        self._audio_input_bus = None
        self._audio_output_bus = None
        self._default_group = None
        self._root_node = None

        ### PROXY MAPPINGS ###

        self._audio_busses = {}
        self._control_busses = {}
        self._buffers = {}
        self._nodes = {}
        self._synthdefs = {}

    ### SPECIAL METHODS ###

    def __contains__(self, expr):
        from supriya.tools import servertools
        from supriya.tools import synthdeftools
        if not isinstance(expr, servertools.ServerObjectProxy):
            return False
        elif expr.server is not self:
            return False
        elif isinstance(expr, servertools.Node):
            node_id = expr.node_id
            if node_id in self._nodes and self._nodes[node_id] is expr:
                return True
        elif isinstance(expr, synthdeftools.SynthDef):
            name = expr.actual_name
            if name in self._synthdefs and self._synthdefs[name] == expr:
                return True
        return False

    def __repr__(self):
        if not self.is_running:
            return '<Server: offline>'
        string = '<Server: {protocol}://{ip}:{port}, '
        string += '{inputs}i{outputs}o>'
        return string.format(
            protocol=self.server_options.protocol,
            ip=self.ip_address,
            port=self.port,
            inputs=self.server_options.input_bus_channel_count,
            outputs=self.server_options.output_bus_channel_count,
            )

    ### PRIVATE METHODS ###

    def _setup(self):
        self._setup_allocators(self.server_options)
        self._setup_proxies()
        self._setup_notifications()
        self._setup_status_watcher()

    def _setup_allocators(self, server_options):
        from supriya.tools import servertools
        self._audio_bus_allocator = servertools.BlockAllocator(
            heap_maximum=server_options.audio_bus_channel_count,
            heap_minimum=server_options.first_private_bus_id,
            )
        self._buffer_allocator = servertools.BlockAllocator(
            heap_maximum=server_options.buffer_count,
            )
        self._control_bus_allocator = servertools.BlockAllocator(
            heap_maximum=server_options.control_bus_channel_count,
            )
        self._node_id_allocator = servertools.NodeIdAllocator(
            initial_node_id=server_options.initial_node_id,
            )
        self._sync_id = 0

    def _setup_notifications(self):
        from supriya.tools import servertools
        notify_message = servertools.CommandManager.make_notify_message(1)
        with servertools.WaitForServer(
            address_pattern='/done',
            argument_template=('/notify', 0),
            server=self,
            ):
            self.send_message(notify_message)

    def _setup_proxies(self):
        from supriya.tools import servertools
        self._audio_input_bus = servertools.AudioInputBus(self)
        self._audio_output_bus = servertools.AudioOutputBus(self)
        self._root_node = servertools.RootNode(server=self)
        self._nodes[0] = self._root_node
        default_group = servertools.Group()
        default_group.allocate(
            add_action=servertools.AddAction.ADD_TO_HEAD,
            node_id_is_permanent=True,
            target_node=self.root_node,
            )
        self._default_group = default_group

    def _setup_status_watcher(self):
        from supriya.tools import servertools
        self._server_status = None
        self._status_watcher = servertools.StatusWatcher(self)
        self._status_watcher.start()

    def _teardown(self):
        self._teardown_proxies()
        self._teardown_allocators()
        self._teardown_status_watcher()

    def _teardown_allocators(self):
        self._audio_bus_allocator = None
        self._buffer_allocator = None
        self._control_bus_allocator = None
        self._node_id_allocator = None
        self._sync_id = 0

    def _teardown_proxies(self):
        for x in tuple(self._audio_busses.values()):
            x.free()
        for x in tuple(self._buffers.values()):
            x.free()
        for x in tuple(self._control_busses.values()):
            x.free()
        for x in tuple(self._nodes.values()):
            x.free()
        self._default_group = None
        self._root_node = None
        self._audio_input_bus = None
        self._audio_output_bus = None

    def _teardown_status_watcher(self):
        self._status_watcher.active = False
        self._status_watcher = None
        self._server_status = None

    ### PUBLIC METHODS ###

    def boot(
        self,
        server_options=None,
        ):
        from supriya.tools import servertools
        from supriya.tools import systemtools
        if self.is_running:
            return
        server_options = server_options or servertools.ServerOptions()
        options_string = server_options.as_options_string(self.port)
        command = 'scsynth {}'.format(options_string)
        server_process = subprocess.Popen(
            command.split(),
            bufsize=0,
            stderr=subprocess.PIPE,
            stdout=subprocess.PIPE,
            )

        error = b'Exception in World_OpenUDP: bind: Address already in use'
        success = b'SuperCollider 3 server ready.'
        reader = systemtools.NonBlockingStreamReader(server_process.stdout)
        with reader:
            while True:
                time.sleep(0.01)
                line = reader.readline(timeout=0.01)
                if line is None:
                    continue
                elif error in line:
                    raise Exception(error)
                elif success in line:
                    break

        self._is_running = True
        self._server_options = server_options
        self._server_process = server_process
        self._setup()
        self.sync()
        return self

    @staticmethod
    def get_default_server():
        if Server._default_server is None:
            Server._default_server = Server()
        return Server._default_server

    def quit(self):
        from supriya.tools import servertools
        if not self.is_running:
            return
        wait = servertools.WaitForServer(
            address_pattern='/(done|fail)',
            argument_template=('/quit',),
            server=self,
            )
        with wait:
            self.send_message('/quit')
        self._is_running = False
        self._server_process.send_signal(signal.SIGINT)
        self._server_process.kill()
        self._teardown()
        return self

    def register_osc_callback(self, osc_callback):
        self._osc_dispatcher.register_osc_callback(osc_callback)

    def send_message(self, message):
        if not message or not self.is_running:
            return
        self._osc_controller.send(message)

    def sync(self, sync_id=None):<commit_msg>Every Server now registers its quit() method with atexit.<commit_after># -*- encoding: utf-8 -*-
from __future__ import print_function
import atexit
import signal
import subprocess
import time


class Server(object):
    r'''An scsynth server proxy.

    ::

        >>> from supriya import servertools
        >>> server = servertools.Server.get_default_server()
        >>> server.boot()
        <Server: udp://127.0.0.1:57751, 8i8o>

    ::

        >>> server.quit()
        <Server: offline>

    '''

    ### CLASS VARIABLES ###

    __slots__ = (
        '_audio_bus_allocator',
        '_audio_busses',
        '_audio_input_bus',
        '_audio_output_bus',
        '_buffer_allocator',
        '_buffers',
        '_control_bus_allocator',
        '_control_busses',
        '_default_group',
        '_ip_address',
        '_is_running',
        '_node_id_allocator',
        '_nodes',
        '_osc_controller',
        '_osc_dispatcher',
        '_port',
        '_response_manager',
        '_root_node',
        '_server_options',
        '_server_process',
        '_server_status',
        '_status_watcher',
        '_sync_id',
        '_synthdefs',
        )

    _default_server = None

    _servers = {}

    ### CONSTRUCTOR ###

    def __new__(cls, ip_address='127.0.0.1', port=57751):
        key = (ip_address, port)
        if key not in cls._servers:
            instance = object.__new__(cls)
            instance.__init__(
                ip_address=ip_address,
                port=port,
                )
            cls._servers[key] = instance
        return cls._servers[key]

    ### INITIALIZER ###

    def __init__(self, ip_address='127.0.0.1', port=57751):
        from supriya.tools import osctools
        from supriya.tools import responsetools

        if hasattr(self, 'is_running') and self.is_running:
            return

        ### NET ADDRESS ###

        self._ip_address = ip_address
        self._port = port

        ### OSC MESSAGING ###

        self._osc_dispatcher = osctools.OscDispatcher()
        self._osc_controller = osctools.OscController(server=self)
        self._response_manager = responsetools.ResponseManager()
        for callback in (
            responsetools.BufferResponseCallback(self),
            responsetools.ControlBusResponseCallback(self),
            responsetools.NodeResponseCallback(self),
            responsetools.SynthDefResponseCallback(self),
            ):
            self.register_osc_callback(callback)

        ### ALLOCATORS ###

        self._audio_bus_allocator = None
        self._buffer_allocator = None
        self._control_bus_allocator = None
        self._node_id_allocator = None

        ### SERVER PROCESS ###

        self._is_running = False
        self._server_options = None
        self._server_process = None
        self._server_status = None
        self._status_watcher = None

        ### PROXIES ###

        self._audio_input_bus = None
        self._audio_output_bus = None
        self._default_group = None
        self._root_node = None

        ### PROXY MAPPINGS ###

        self._audio_busses = {}
        self._control_busses = {}
        self._buffers = {}
        self._nodes = {}
        self._synthdefs = {}

        ### REGISTER WITH ATEXIT ###

        atexit.register(self.quit)

    ### SPECIAL METHODS ###

    def __contains__(self, expr):
        from supriya.tools import servertools
        from supriya.tools import synthdeftools
        if not isinstance(expr, servertools.ServerObjectProxy):
            return False
        elif expr.server is not self:
            return False
        elif isinstance(expr, servertools.Node):
            node_id = expr.node_id
            if node_id in self._nodes and self._nodes[node_id] is expr:
                return True
        elif isinstance(expr, synthdeftools.SynthDef):
            name = expr.actual_name
            if name in self._synthdefs and self._synthdefs[name] == expr:
                return True
        return False

    def __repr__(self):
        if not self.is_running:
            return '<Server: offline>'
        string = '<Server: {protocol}://{ip}:{port}, '
        string += '{inputs}i{outputs}o>'
        return string.format(
            protocol=self.server_options.protocol,
            ip=self.ip_address,
            port=self.port,
            inputs=self.server_options.input_bus_channel_count,
            outputs=self.server_options.output_bus_channel_count,
            )

    ### PRIVATE METHODS ###

    def _setup(self):
        self._setup_allocators(self.server_options)
        self._setup_proxies()
        self._setup_notifications()
        self._setup_status_watcher()

    def _setup_allocators(self, server_options):
        from supriya.tools import servertools
        self._audio_bus_allocator = servertools.BlockAllocator(
            heap_maximum=server_options.audio_bus_channel_count,
            heap_minimum=server_options.first_private_bus_id,
            )
        self._buffer_allocator = servertools.BlockAllocator(
            heap_maximum=server_options.buffer_count,
            )
        self._control_bus_allocator = servertools.BlockAllocator(
            heap_maximum=server_options.control_bus_channel_count,
            )
        self._node_id_allocator = servertools.NodeIdAllocator(
            initial_node_id=server_options.initial_node_id,
            )
        self._sync_id = 0

    def _setup_notifications(self):
        from supriya.tools import servertools
        notify_message = servertools.CommandManager.make_notify_message(1)
        with servertools.WaitForServer(
            address_pattern='/done',
            argument_template=('/notify', 0),
            server=self,
            ):
            self.send_message(notify_message)

    def _setup_proxies(self):
        from supriya.tools import servertools
        self._audio_input_bus = servertools.AudioInputBus(self)
        self._audio_output_bus = servertools.AudioOutputBus(self)
        self._root_node = servertools.RootNode(server=self)
        self._nodes[0] = self._root_node
        default_group = servertools.Group()
        default_group.allocate(
            add_action=servertools.AddAction.ADD_TO_HEAD,
            node_id_is_permanent=True,
            target_node=self.root_node,
            )
        self._default_group = default_group

    def _setup_status_watcher(self):
        from supriya.tools import servertools
        self._server_status = None
        self._status_watcher = servertools.StatusWatcher(self)
        self._status_watcher.start()

    def _teardown(self):
        self._teardown_proxies()
        self._teardown_allocators()
        self._teardown_status_watcher()

    def _teardown_allocators(self):
        self._audio_bus_allocator = None
        self._buffer_allocator = None
        self._control_bus_allocator = None
        self._node_id_allocator = None
        self._sync_id = 0

    def _teardown_proxies(self):
        for x in tuple(self._audio_busses.values()):
            x.free()
        for x in tuple(self._buffers.values()):
            x.free()
        for x in tuple(self._control_busses.values()):
            x.free()
        for x in tuple(self._nodes.values()):
            x.free()
        self._default_group = None
        self._root_node = None
        self._audio_input_bus = None
        self._audio_output_bus = None

    def _teardown_status_watcher(self):
        self._status_watcher.active = False
        self._status_watcher = None
        self._server_status = None

    ### PUBLIC METHODS ###

    def boot(
        self,
        server_options=None,
        ):
        from supriya.tools import servertools
        from supriya.tools import systemtools
        if self.is_running:
            return
        server_options = server_options or servertools.ServerOptions()
        options_string = server_options.as_options_string(self.port)
        command = 'scsynth {}'.format(options_string)
        server_process = subprocess.Popen(
            command.split(),
            bufsize=0,
            stderr=subprocess.PIPE,
            stdout=subprocess.PIPE,
            )

        error = b'Exception in World_OpenUDP: bind: Address already in use'
        success = b'SuperCollider 3 server ready.'
        reader = systemtools.NonBlockingStreamReader(server_process.stdout)
        with reader:
            while True:
                time.sleep(0.01)
                line = reader.readline(timeout=0.01)
                if line is None:
                    continue
                elif error in line:
                    raise Exception(error)
                elif success in line:
                    break

        self._is_running = True
        self._server_options = server_options
        self._server_process = server_process
        self._setup()
        self.sync()
        return self

    @staticmethod
    def get_default_server():
        if Server._default_server is None:
            Server._default_server = Server()
        return Server._default_server

    def quit(self):
        from supriya.tools import servertools
        if not self.is_running:
            return
        wait = servertools.WaitForServer(
            address_pattern='/(done|fail)',
            argument_template=('/quit',),
            server=self,
            )
        with wait:
            self.send_message('/quit')
        self._is_running = False
        self._server_process.send_signal(signal.SIGINT)
        self._server_process.kill()
        self._teardown()
        return self

    def register_osc_callback(self, osc_callback):
        self._osc_dispatcher.register_osc_callback(osc_callback)

    def send_message(self, message):
        from supriya.tools import osctools
        if not message or not self.is_running:
            return
        if isinstance(message, osctools.OscMessage) and message.address != 2:
            print('SEND:', message)
        self._osc_controller.send(message)

    def sync(self, sync_id=None):
{'fraction_non_alphanumeric': 0.08760888172800967, 'fraction_numerical': 0.004984643270731584, 'mean_word_length': 2.0538130381303814, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 6, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5919172', 'n_tokens_mistral': 5726, 'n_tokens_neox': 5536, 'n_words': 1039}