IRC Sonic pi repl

[general]
duration        = 0
bufferSecs      = 5
reconnect       = yes
realtime        = yes
rtprio          = 3

[input]
device          = pulse # Maybe you just use jack?
sampleRate      = 44100
bitsPerSample   = 16
channel         = 2

[icecast2-0]
bitrateMode     = cbr
format          = vorbis
bitrate         = 128
server          = radio.dot.org.es
port            = 8000
password        = passwordherehahaha
mountPoint      = playground.ogg
name            = Sonic Pi
description     = Sonic Pi REPL
url             = https://radio.dot.org.es
genre           = radio
public          = yes

sox -c1 -n result.wav synth 10 sin 25000 vol 1

mpv -loop result.wav

import collections
import html
import logging
import os
import re
import socket
import sys
import time

from oscpy.client import OSCClient
from oscpy.server import OSCThreadServer

SERVER_OUTPUT = "~/.sonic-pi/log/server-output.log"


logger = logging.getLogger()


class Server:
    """Represents a running instance of Sonic Pi."""

    preamble = '@osc_server||=SonicPi::OSC::UDPServer.new' + \
               '({},use_decoder_cache:true) #__nosave__\n'

    def __init__(self, host, cmd_port, osc_port, send_preamble, verbose):
        self.client_name = 'SONIC_PI_TOOL_PY'
        self.host = host
        self._cmd_port = cmd_port
        self._cached_cmd_port = None
        self.osc_port = osc_port
        # fix for https://github.com/repl-electric/sonic-pi.el/issues/19#issuecomment-345222832
        self.send_preamble = send_preamble
        self._cmd_client = None
        self._osc_client = None

    def get_cmd_port(self):
        return self._cmd_port

    def cmd_client(self):
        if self._cmd_client is None:
            self._cmd_client = OSCClient(self.host, self.get_cmd_port(),
                                         encoding='utf8')
        return self._cmd_client

    def osc_client(self):
        if self._osc_client is None:
            self._osc_client = OSCClient(self.host, self.osc_port,
                                         encoding='utf8')
        return self._osc_client

    def get_preamble(self):
        if self.send_preamble:
            return Server.preamble.format(self.get_cmd_port())
        return ''

    def send_cmd(self, msg, *args):
        client = self.cmd_client()
        logger.info("Sending command to {}:{}: {} {}"
                    .format(self.host, self.get_cmd_port(), msg,
                            ', '.join(repr(v) for v in (self.client_name,) + args)))
        client.send_message(msg, (self.client_name,) + args)

    def send_osc(self, path, args):
        def parse_val(s):
            try:
                return int(s)
            except ValueError:
                pass
            try:
                return float(s)
            except ValueError:
                pass
            if len(s) > 1 and s[0] == '"' and s[-1] == '"':
                return s[1:-1]
            return s

        client = self.osc_client()
        parsed = [parse_val(s) for s in args]
        logger.info("Sending OSC message to {}:{}: {} {}"
                    .format(self.host, self.osc_port, path,
                            ', '.join(repr(v) for v in parsed)))
        client.send_message(path, parsed)

    def check_if_running(self):
        cmd_listening = Server.port_in_use(self.get_cmd_port())
        logger.info("The command port ({}) is {}in use".format(self.get_cmd_port(),
                                                               "" if cmd_listening else "not "))
        osc_listening = Server.port_in_use(self.osc_port)
        logger.info("The OSC port ({}) is {}in use".format(self.osc_port,
                                                           "" if osc_listening else "not "))
        osc_listening = True
        if cmd_listening and osc_listening:
            logger.info("Sonic Pi is running, and listening on port {} for commands and {} for OSC"
                        .format(self.get_cmd_port(), self.osc_port), True)
            return 0
        elif not cmd_listening and not osc_listening:
            logger.info("Sonic Pi is not running", True)
            return 1
        else:
            logger.info("Sonic Pi is not running properly, or there's an issue with the port numbers",
                        True)
            return 2

    def stop_all_jobs(self):
        self.send_cmd('/stop-all-jobs')

    def run_code(self, code):
        self.send_cmd('/run-code', self.get_preamble() + code)

    def start_recording(self):
        self.send_cmd('/start-recording')

    def stop_and_save_recording(self, path):
        self.send_cmd('/stop-recording')
        self.send_cmd('/save-recording', path)

    @staticmethod
    def port_in_use(port):
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
            try:
                sock.bind(('127.0.0.1', port))
            except OSError:
                return True
        return False

    @staticmethod
    def determine_command_port():
        try:
            with open(os.path.expanduser(SERVER_OUTPUT)) as f:
                for line in f:
                    m = re.search('^Listen port: *([0-9]+)', line)
                    if m:
                        return int(m.groups()[0])
        except FileNotFoundError:
            pass

    @staticmethod
    def handle_log_info(style, msg):
        msg = "=> {}".format(msg)
        logger.info(msg)
        logger.info()

    @staticmethod
    def handle_multi_message(run, thread, time, n, *msgs):
        msg = "{{run: {}, time: {}}}".format(run, time)
        logger.info(msg)
        for i in range(n):
            typ, msg = msgs[2 * i: 2 * i + 2]
            for j, line in enumerate(msg.splitlines()):
                if i < n - 1:
                    prefix = "  ├─ " if j == 0 else "  │"
                else:
                    prefix = "  └─ " if j == 0 else "   "
                logger.info(f"{prefix}, {line}, {typ}")
        logger.info()

    @staticmethod
    def handle_runtime_error(run, msg, trace, line_num):
        lines = html.unescape(msg).splitlines()
        prefix = "Runtime Error: "
        for line in lines:
            logger.debug(f"{prefix=} {line=}")
            prefix = ""
        logger.debug(html.unescape(trace))

    @staticmethod
    def handle_syntax_error(run, msg, code, line_num, line_s):
        logger.error("Error: " + html.unescape(msg))
        prefix = "[Line {}]: ".format(line_num) if line_num >= 0 else ""
        logger.error(f"{prefix=}, {code=}")

    def follow_logs(self):
        try:
            server = OSCThreadServer(encoding='utf8')
            server.listen(address='127.0.0.1', port=4558, default=True)
            server.bind('/log/multi_message', self.handle_multi_message)
            server.bind('/multi_message', self.handle_multi_message)
            server.bind('/log/info', self.handle_log_info)
            server.bind('/info', self.handle_log_info)
            server.bind('/error', self.handle_runtime_error)
            server.bind('/syntax_error', self.handle_syntax_error)
            while True:
                time.sleep(1)
        except Exception as e:
            return e


def eval_stdin(server: Server):
    server.run_code(sys.stdin.read())


def eval_file(server: Server, path):
    server.run_code(path.read())


def osc(server: Server, path, args):
    server.send_osc(path, args)

socat UDP-LISTEN:4557,fork,bind=$CONTAINER_IP UDP:127.0.0.1:4557

import re

from sonic_pi import NoteNotFound
from sonic_pi import Server as PiServer
from sonic_pi import convert_to_notes

from IrcBot.bot import Color, IrcBot, Message, utils

sonic_pi_users = {}
sonic_pi_history = {}

@utils.regex_cmd_with_messsage(r"^(.+)$")
def all_msgs(args: re.Match, msg: Message):
    if msg.nick not in sonic_pi_users or msg.message.strip().startswith(PREFIX):
        return
    sonic_pi_users[msg.nick].append(args[1])
def convert(*args):
    transpose = 0
    if args[0].isdigit():
        octave = int(args[0])
        if args[1][0] in ["+", "-"] and args[1][1:].isdigit():
            transpose = int(args[1])
            letters = "".join(args[2:])
        else:
            letters = "".join(args[1:])
    else:
        octave = 4
        if args[0][0] in ["+", "-"] and args[0][1:].isdigit():
            transpose = int(args[0])
            letters = "".join(args[1:])
        else:
            letters = "".join(args)
        logger.debug(f"{octave=} {letters=}")
        return ", ".join(convert_to_notes(letters, octave, transpose))


@utils.arg_command("pi", "Toggles sonic pi repl", f"{PREFIX}pi [command]- https://sonic-pi.net/tutorial.html")
async def pi(bot: IrcBot, args: re.Match, msg: Message):
    args = utils.m2list(args)
    if args:
        sonic_pi_users[msg.nick] = [" ".join(args)]

    if msg.nick in sonic_pi_users:
        # append to history
        if msg.nick in sonic_pi_history:
            sonic_pi_history[msg.nick].extend(sonic_pi_users[msg.nick])
        else:
            sonic_pi_history[msg.nick] = sonic_pi_users[msg.nick]

        # Apply template
        for i, line in enumerate(deepcopy(sonic_pi_users[msg.nick])):
            for match in re.findall(r"\$\{([^}]+?)\}", line):
                try:
                    replace = convert(*match.split(" "))
                    sonic_pi_users[msg.nick][i] = sonic_pi_users[msg.nick][i].replace(
                        "${" + match + "}", replace)
                    logger.debug(
                        f"Applying template at {i=} {line=} {match=}, {replace=} {sonic_pi_users[msg.nick][i]=}")
                except NoteNotFound as e:
                    await reply(bot, msg, error(f"Could not find note '{e}'"))
                    del sonic_pi_users[msg.nick]
                    return

        await reply(bot, msg, "Your Sonic Pi repl is now off. Sending code to sonic pi...")
        server.run_code("\n".join(sonic_pi_users[msg.nick]))
        del sonic_pi_users[msg.nick]
        return

    sonic_pi_users[msg.nick] = []
    await reply(bot, msg, f"Your Sonic Pi repl is now live at: {SONIC_PI_LIVE_URL}. Type {PREFIX}pi to turn it off and evaluate your code.")


@utils.arg_command("convert", "Convert keyboard characters into sonic pi notes", f"{PREFIX}convert [octave] [±transpose] <letters>  -  Only qwerty layout")
async def cmd_convert(bot: IrcBot, args: re.Match, msg: Message):
    args = utils.m2list(args)
    if not args:
        await reply(bot, msg, error("You need to specify an octave an a string to convert"))
        return
    if args[0].isdigit() and (int(args[0]) < 0 or int(args[0]) > 10):
        await reply(bot, msg, error("The first argument must be between 0 and 10"))
        return

    try:
        res = convert(*args)
        await reply(bot, msg, res)
    except NoteNotFound as e:
        await reply(bot, msg, error(f"Could not find note for '{e}'"))


@utils.arg_command("pstop", "Stops sonic pi audio")
async def stop(bot: IrcBot, args: re.Match, msg: Message):
    server.stop_all_jobs()
    await reply(bot, msg, "Stopping audio...")


@utils.arg_command("paste", "Pastes your sonic pi code and clears your history")
async def pipaste(bot: IrcBot, args: re.Match, msg: Message):
    if msg.nick not in sonic_pi_history:
        await reply(bot, msg, error("You need to turn on your sonic pi repl first. Use {}pi".format(PREFIX)))
        return
    await reply(bot, msg, paste("\n".join(sonic_pi_history[msg.nick])))
    del sonic_pi_history[msg.nick]


@utils.arg_command("read", "Read code from ix.io paste (or any raw text url)")
async def readurl(bot: IrcBot, args: re.Match, msg: Message):
    try:
        server.run_code(read_paste(args[1]))
    except Exception as e:
        await reply(bot, msg, error("Failed to read paste: ") + str(e))
    await reply(bot, msg, "Code has been read and sent!")


# And you might want to define all those constants somewhere
if __name__ == "__main__":
    utils.setLogging(LOG_LEVEL, LOGFILE)
    bot = IrcBot(HOST, PORT, NICK, CHANNELS, PASSWORD, use_ssl=PORT == 6697,
                 dcc_host=DCC_HOST, dcc_ports=DCC_PORTS, dcc_announce_host=DCC_ANNOUNCE_HOST)
    bot.runWithCallback(onconnect)