Carla/source/modules/carla_native/resources/externalui.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# DISTRHO Plugin Toolkit (DPT)
# Copyright (C) 2012-2013 Filipe Coelho <falktx@falktx.com>
#
# Permission to use, copy, modify, and/or distribute this software for any purpose with
# or without fee is hereby granted, provided that the above copyright notice and this
# permission notice appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH REGARD
# TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN
# NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL
# DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER
# IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
# CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

# -----------------------------------------------------------------------
# Imports

from os import fdopen, O_NONBLOCK
from fcntl import fcntl, F_GETFL, F_SETFL
from sys import argv

# -----------------------------------------------------------------------
# External UI

class ExternalUI(object):
    def __init__(self):
        object.__init__(self)

        self.fPipeRecv     = None
        self.fPipeSend     = None
        self.fQuitReceived = False

        self.fSampleRate = float(argv[1])
        self.fUiName     = argv[2]
        self.fPipeRecvFd = int(argv[3])
        self.fPipeSendFd = int(argv[4])

        fcntl(self.fPipeRecvFd, F_SETFL, fcntl(self.fPipeRecvFd, F_GETFL) | O_NONBLOCK)

        self.fPipeRecv = fdopen(self.fPipeRecvFd, 'r')
        self.fPipeSend = fdopen(self.fPipeSendFd, 'w')

        # send empty line (just newline char)
        self.send([""])

    # -------------------------------------------------------------------
    # Host DSP State

    def d_getSampleRate(self):
        return self.fSampleRate

    def d_editParameter(self, index, started):
        self.send(["editParam", index, started])

    def d_setParameterValue(self, index, value):
        self.send(["control", index, value])

    def d_setState(self, key, value):
        self.send(["configure", key, value])

    def d_sendNote(self, onOff, channel, note, velocity):
        self.send(["note", onOff, channel, note, velocity])

    # -------------------------------------------------------------------
    # DSP Callbacks

    def d_parameterChanged(self, index, value):
        return

    def d_programChanged(self, index):
        return

    def d_stateChanged(self, key, value):
        return

    def d_noteReceived(self, onOff, channel, note, velocity):
        return

    # -------------------------------------------------------------------
    # ExternalUI Callbacks

    def d_uiShow(self):
        return

    def d_uiHide(self):
        return

    def d_uiQuit(self):
        return

    def d_uiTitleChanged(self, uiTitle):
        return

    # -------------------------------------------------------------------
    # Public methods

    def closeExternalUI(self):
        if not self.fQuitReceived:
            self.send(["exiting"])

        if self.fPipeRecv is not None:
          self.fPipeRecv.close()
          self.fPipeRecv = None

        if self.fPipeSend is not None:
          self.fPipeSend.close()
          self.fPipeSend = None

    def idleExternalUI(self):
        if self.fPipeRecv is None:
            return False

        try:
            msg = self.fPipeRecv.readline().strip()
        except IOError:
            return False

        if msg == "":
            return True

        if msg == "control":
            index = int(self.fPipeRecv.readline())
            value = float(self.fPipeRecv.readline())
            self.d_parameterChanged(index, value)

        elif msg == "program":
            index = int(self.fPipeRecv.readline())
            self.d_programChanged(index)

        elif msg == "configure":
            key   = self.fPipeRecv.readline().strip().replace("\r", "\n")
            value = self.fPipeRecv.readline().strip().replace("\r", "\n")
            self.d_stateChanged(key, value)

        elif msg == "note":
            onOff    = bool(self.fPipeRecv.readline().strip() == "true")
            channel  = int(self.fPipeRecv.readline())
            note     = int(self.fPipeRecv.readline())
            velocity = int(self.fPipeRecv.readline())
            self.d_noteReceived(onOff, channel, note, velocity)

        elif msg == "show":
            self.d_uiShow()

        elif msg == "hide":
            self.d_uiHide()

        elif msg == "quit":
            self.fQuitReceived = True
            self.d_uiQuit()

        elif msg == "uiTitle":
            uiTitle = self.fPipeRecv.readline().strip().replace("\r", "\n")
            self.d_uiTitleChanged(uiTitle)

        else:
            print("unknown message: \"" + msg + "\"")

        return True

    # -------------------------------------------------------------------
    # Internal stuff

    def send(self, lines):
        if self.fPipeSend is None:
            return

        for line in lines:
            if isinstance(line, str):
                line2 = line.replace("\n", "\r")
            else:
                if isinstance(line, bool):
                    line2 = "true" if line else "false"
                elif isinstance(line, int):
                    line2 = "%i" % line
                elif isinstance(line, float):
                    line2 = "%.10f" % line
                else:
                    return

            self.fPipeSend.write(line2 + "\n")
            self.fPipeSend.flush()