Patroneo/template/engine/api.py

#! /usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Copyright 2021, Nils Hilbricht, Germany ( https://www.hilbricht.net )

This file is part of the Laborejo Software Suite ( https://www.laborejo.org )

This application 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/>.
"""


import logging; logger = logging.getLogger(__name__); logger.info("import")


#Python Standard Library
import os.path
from datetime import timedelta

#Third Party Modules
from calfbox import cbox

#Our own template modules
from .session import Session
from .duration import DB, DL, D1, D2, D4, D8, D16, D32, D64, D128, D256, D512, D1024, jackBBTicksToDuration
from ..helper import nothing

class Callbacks(object):
    """GUI methods register themselves here.
    These methods get called by us, the engine.

    None of these methods produce any return value.

    The lists may be unordered.

    We need the lists for audio feedbacks in parallel to GUI updates.
    Or whatever parallel representations we run."""

    def __init__(self):
        self.debugChanged = [] #only for testing and debug.
        self.setPlaybackTicks = []
        self.playbackStatusChanged = []
        self.bbtStatusChanged = []
        self.barBeatTempo = []
        self.clock = []
        self.historyChanged = []
        self.historySequenceStarted = []
        self.historySequenceStopped = []
        self.message = []

        #Live Midi Recording
        self.recordingModeChanged = []

        #Sequencer
        self.numberOfTracksChanged = []
        self.metronomeChanged = []

        #Sf2 Sampler
        self.soundfontChanged = []
        self.channelChanged = []
        self.channelActivity = []
        self.ignoreProgramChangesChanged = []

        #Not callbacks
        self._rememberPlaybackStatus = None #Once set it will be True or False
        self._rememberBBT = None #Will be a dict
        self._rememberBarBeatTempo = None #Specialized case of BBT


    def _dataChanged(self):
        """Only called from within the callbacks.
        This is about data the user cares about. In other words this is the indicator if you need
        to save again.

        Insert, delete edit are real data changes. Cursor movement or playback ticks are not."""
        session.nsmClient.announceSaveStatus(False)
        self._historyChanged()


    def _historyChanged(self):
        """processQueue
        Only called from within the callbacks.

        sends two lists of strings.
        the first is the undoHistory, the last added item is [-1]. We can show that to a user to
        indicate what the next undo will do.

        the second is redoHistory, same as undo: [-1] shows the next redo action."""
        undoHistory, redoHistory = session.history.asList()
        for func in self.historyChanged:
            func(undoHistory, redoHistory)

    def _historySequenceStarted(self):
        """sends a signal when a sequence of high level api functions will be executed next.
        Also valid for their undo sequences.
        A GUI has the chance to disable immediate drawing, e.g. qt Graphics Scene could stop
        scene updates and allow all callbacks to come in first.
        historySequenceStopped will be sent when the sequence is over and a GUI could reactivate
        drawing to have the buffered changes take effect.
        This signal is automatically sent by the history sequence context"""
        for func in self.historySequenceStarted:
            func()

    def _historySequenceStopped(self):
        """see _historySequenceStarted"""
        for func in self.historySequenceStopped:
            func()

    def _message(self, title, text):
        """Send a message of any kind to get displayed.
        Enables an api function to display a message via the GUI.
        Does _not_ support translations, therefore ued for errors mostly"""
        for func in self.message:
            func(title, text)

    def _debugChanged(self):
        for func in self.debugChanged:
            func()
        self._dataChanged() #includes _historyChanged


    def _setPlaybackTicks(self):
        """This gets called very very often. Any connected function needs to watch closely
        for performance issues"""
        ppqn = cbox.Transport.status().pos_ppqn
        status = playbackStatus()
        for func in self.setPlaybackTicks:
            func(ppqn, status)

    def _playbackStatusChanged(self):
        """Returns a bool if the playback is running.
        Under rare circumstances it may send the same status in a row, which means
        you actually need to check the result and not only toggle as a response.

        This callback cannot be called manually. Instead it will be called automatically to make
        it possible to react to external jack transport changes.

        This is deprecated. Append to _checkPlaybackStatusAndSendSignal which is checked by the
        event loop.
        """
        raise NotImplementedError("this function was deprecated. use _checkPlaybackStatusAndSendSignal")

        pass #only keep for the docstring and to keep the pattern.

    def _checkPlaybackStatusAndSendSignal(self):
        """Added to the event loop.
        We don'T have a jack callback to inform us of this so we drive our own polling system
        which in turn triggers our own callback, when needed."""
        status = playbackStatus()
        if not self._rememberPlaybackStatus == status:
            self._rememberPlaybackStatus = status
            for func in self.playbackStatusChanged:
                func(status)

    def _checkBBTAndSendSignal(self):
        """Added to the event loop.
        We don'T have a jack callback to inform us of this so we drive our own polling system
        which in turn triggers our own callback, when needed.

        We are interested in:
            bar
            beat  #first index is 1
            tick
            bar_start_tick
            beats_per_bar [4.0]
            beat_type [4.0]
            ticks_per_beat [960.0] #JACK ticks, not cbox.
            beats_per_minute [120.0]

            int bar is the current bar.
            int beat current beat-within-bar
            int tick current tick-within-beat
            double bar_start_tick  number of ticks that have elapsed between frame 0 and the first beat of the current measure.
        """
        data = cbox.JackIO.jack_transport_position() #this includes a lot of everchanging data. If no jack-master client set /bar and the others they will simply not be in the list
        t = (data.beats_per_bar, data.ticks_per_beat)
        if not self._rememberBBT == t: #new situation, but not just frame position update
            self._rememberBBT = t
            export = {}
            if data.beats_per_bar:
                offset = (data.beat-1) * data.ticks_per_beat + data.tick #if timing is good this is the same as data.tick because beat is 1.
                offset = jackBBTicksToDuration(data.beat_type, offset, data.ticks_per_beat)
                export["nominator"] = data.beats_per_bar
                export["denominator"] = jackBBTicksToDuration(data.beat_type, data.ticks_per_beat, data.ticks_per_beat) #the middle one is the changing one we are interested in
                export["measureInTicks"] = export["nominator"] * export["denominator"]
                export["offsetToMeasureBeginning"] = offset
                #export["tickposition"] = cbox.Transport.status().pos_ppqn #this is a different position than our current one because it takes a few cycles and ticks to calculate
                export["tickposition"] = cbox.Transport.samples_to_ppqn(data.frame)
            for func in self.bbtStatusChanged:
                func(export)

        #Send bar beats tempo, for displays
        #TODO: broken
        """
        bbtExport = {}
        if data.beat and not self._rememberBarBeatTempo == data.beat:

            bbtExport["timesig"] = f"{int(data.beats_per_bar)}/{int(data.beat_type)}" #for displays
            bbtExport["beat"] = data.beat #index from 1
            bbtExport["tempo"] = int(data.beats_per_minute)
            bbtExport["bar"] = int(data.bar)
            self._rememberBarBeatTempo = data.beat #this should be enough inertia to not fire every 100ms
            for func in self.barBeatTempo:
                func(bbtExport)
        elif not data.beat:
            for func in self.barBeatTempo:
                func(bbtExport)
            self._rememberBarBeatTempo = data.beat
        """

        clock = str(timedelta(seconds=data.frame / data.frame_rate))
        for func in self.clock:
            func(clock)


    #Live Midi Recording
    def _recordingModeChanged(self):
        if session.recordingEnabled:
            session.nsmClient.changeLabel("Recording")
        else:
            session.nsmClient.changeLabel("")
        for func in self.recordingModeChanged:
            func(session.recordingEnabled)

    #Sequencer
    def _numberOfTracksChanged(self):
        """New track, delete track, reorder
        Sent the current track order as list of ids, combined with their structure.
        This is also used when tracks get created or deleted, also on initial load.
        This also includes the pattern.
        """
        session.data.updateJackMetadataSorting()
        lst = [track.export() for track in session.data.tracks]
        for func in self.numberOfTracksChanged:
            func(lst)

    def _metronomeChanged(self):
        """returns a dictionary with meta data such as the mute-state and the track name"""
        exportDict = session.data.metronome.export()
        for func in self.metronomeChanged:
            func(exportDict)

    #Sf2 Sampler
    def _soundfontChanged(self):
        """User loads a new soundfont or on load. Resets everything."""
        exportDict = session.data.export()
        session.data.updateAllChannelJackMetadaPrettyname()
        session.nsmClient.changeLabel(exportDict["name"])
        if exportDict:
            for func in self.soundfontChanged:
                func(exportDict)

    def _channelChanged(self, channel):
        """A single channel changed its parameters. The soundfont stays the same."""
        exportDict = session.data.exportChannel(channel)
        session.data.updateChannelAudioJackMetadaPrettyname(channel)
        session.data.updateChannelMidiInJackMetadaPrettyname(channel)
        for func in self.channelChanged:
            func(channel, exportDict)

    def _ignoreProgramChangesChanged(self):
        state = session.data.midiInput.scene.status().layers[0].status().ignore_program_changes
        for func in self.ignoreProgramChangesChanged:
            func(state)

    def _channelActivity(self, channel):
        """send all note on to the GUI"""
        for func in self.channelActivity:
            func(channel)

def startEngine(nsmClient):
    """
    This function gets called after initializing the GUI, calfbox
    and loading saved data from a file.

    It gets called by client applications before their own startEngine.

    Stopping the engine is done via pythons atexit in the session.
    """
    logger.info("Starting template api engine")
    assert session
    assert callbacks

    session.nsmClient = nsmClient
    session.eventLoop.fastConnect(callbacks._checkPlaybackStatusAndSendSignal)
    session.eventLoop.fastConnect(callbacks._setPlaybackTicks)
    session.eventLoop.fastConnect(cbox.get_new_events) #global cbox.get_new_events does not eat dynamic midi port events.
    session.eventLoop.slowConnect(callbacks._checkBBTAndSendSignal)
    #session.eventLoop.slowConnect(lambda: print(cbox.Transport.status().tempo))
    #asession.eventLoop.slowConnect(lambda: print(cbox.Transport.status()))

    cbox.Document.get_song().update_playback()

    callbacks._recordingModeChanged() #recording mode is in the save file.
    callbacks._historyChanged() #send initial undo status to the GUI, which will probably deactivate its undo/redo menu because it is empty.
    logger.info("Template api engine started")


def _deprecated_updatePlayback():
    """The only place in the program to update the cbox playback besides startEngine.
    We only need to update it after a user action, which always goes through the api.
    Even if triggered through a midi in or other command.
    Hence there is no need to update playback in the session or directly from the GUI."""
    #if session.nsmClient.cachedSaveStatus = False: #dirty  #TODO: wait for cbox optimisations. The right place to cache and check if an update is necessary is in cbox, not here.
    cbox.Document.get_song().update_playback()

def save():
    """Saves the file in place. This is mostly here for psychological reasons. Users like to hit
    Ctrl+S from muscle memory.
    But it can also be used if we run with fake NSM. In any case, it does not accept paths"""
    session.nsmClient.serverSendSaveToSelf()

def undo():
    """No callbacks need to be called. Undo is done via a complementary
    function, already defined, which has all the callbacks in it."""
    session.history.undo()
    callbacks._dataChanged()  #this is in most of the functions already but high-level scripts that use the history.sequence context do not trigger this and get no redo without.

def redo():
    """revert undo if nothing new has happened so far.
    see undo"""
    session.history.redo()
    callbacks._dataChanged()  #this is in most of the functions already but high-level scripts that use the history.sequence context do not trigger this and get no redo without.


def getUndoLists():
    #undoHistory, redoHistory = session.history.asList()
    return session.history.asList()


#Calfbox Sequencer Controls
def playbackStatus()->bool:
    #status = "[Running]" if cbox.Transport.status().playing else "[Stopped]" #it is not that simple.
    cboxStatus = cbox.Transport.status().playing
    if cboxStatus == 1:
        #status = "[Running]"
        return True
    elif cboxStatus == 0:
        #status = "[Stopped]"
        return False
    elif cboxStatus == 2:
        #status = "[Stopping]"
        return False
    elif cboxStatus  is None:
        #status = "[Uninitialized]"
        return False
    elif cboxStatus == "":
        #running with cbox dummy module
        return False
    else:
        raise ValueError("Unknown playback status: {}".format(cboxStatus))

def playPause():
    """There are no internal callback to start and stop playback.
    The api, or the session, do not call that.
    Playback can be started externally via jack transport.
    We use the jack transport callbacks instead and trigger our own callbacks directly from them,
    in the callback class above"""
    if playbackStatus():
        cbox.Transport.stop()
    else:
        cbox.Transport.play()
    #It takes a few ms for playbackStatus to catch up. If you call it here again it will not be updated.

def getPlaybackTicks()->int:
    return cbox.Transport.status().pos_ppqn

def seek(value):
    if value < 0:
        value = 0
    cbox.Transport.seek_ppqn(value)

def toStart():
    seek(0)

def playFrom(ticks):
    seek(ticks)
    if not playbackStatus():
        cbox.Transport.play()

def playFromStart():
    toStart()
    if not playbackStatus():
        cbox.Transport.play()

def toggleRecordingMode():
    session.recordingEnabled = not session.recordingEnabled
    callbacks._recordingModeChanged()

# Sequencer Metronome
def setMetronome(data, label):
    session.data.metronome.generate(data, label)
    callbacks._metronomeChanged()

def enableMetronome(value):
    session.data.metronome.setEnabled(value) #has side effects
    callbacks._metronomeChanged()

def isMetronomeEnabled():
    return session.data.metronome.enabled

def toggleMetronome():
    enableMetronome(not session.data.metronome.enabled) #handles callback etc.


#Sf2 Sampler
def loadSoundfont(filePath):
    """User callable function. Load from saved state is done directly in the session with callbacks
    in startEngine
    The filePath MUST be in our session dir.
    """
    filePathInOurSession = os.path.commonprefix([filePath, session.nsmClient.ourPath]) == session.nsmClient.ourPath
    if not filePathInOurSession:
        raise Exception("api loadSoundfont tried to load .sf2 from outside session dir. Forbidden")

    success, errormessage = session.data.loadSoundfont(filePath)

    if success:
        callbacks._soundfontChanged()
        session.history.clear()
        callbacks._historyChanged()
        callbacks._dataChanged()
    else:
        callbacks._message("Load Soundfont Error", errormessage)

    return success

def setIgnoreProgramAndBankChanges(state):
    state = bool(state)
    #there is no session wrapper function. we use cbox directly. Save file and callbacks will fetch the current value on its own
    session.data.midiInput.scene.status().layers[0].set_ignore_program_changes(state)
    assert session.data.midiInput.scene.status().layers[0].status().ignore_program_changes == state
    callbacks._ignoreProgramChangesChanged()
    callbacks._dataChanged()


def setPatch(channel, bank, program):
    if not 1 <= channel <= 16:
        raise ValueError (f"Channel must be a number between 1 and 16. Yours: {channel}")
    #Bank is split into CC0 and CC32. That makes it a 14bit value (2**14 or 128 * 128) = 16384
    if not 0 <= bank <= 16384:
        raise ValueError (f"Program must be a number between 0 and 16384. Yours: {bank}")
    if not 0 <= program <= 127:
        raise ValueError (f"Program must be a number between 0 and 127. Yours: {program}")

    session.data.setPatch(channel, bank, program)
    callbacks._channelChanged(channel)
    callbacks._dataChanged()

#Debug, Test and Template Functions
class TestValues(object):
    value = 0

def history_test_change():
    """
    We simulate a function that gets its value from context.
    Here it is random, but it may be the cursor position in a real program."""
    from random import randint
    value = 0
    while value == 0:
        value = randint(-10,10)

    if value > 0:
        session.history.setterWithUndo(TestValues, "value", TestValues.value + value, "Increase Value", callback=callbacks._debugChanged) #callback includes dataChanged which inlucdes historyChanged
    else:
        session.history.setterWithUndo(TestValues, "value", TestValues.value + value, "Decrease Value", callback=callbacks._debugChanged) #callback includes dataChanged which inlucdes historyChanged

def history_test_undoSequence():
    with session.history.sequence("Change Value Multiple Times"):
        history_test_change()
        history_test_change()
        history_test_change()
        history_test_change()
    callbacks._historyChanged()


#Module Level Data
callbacks = Callbacks() #This needs to be defined before startEngine() so a GUI can register its callbacks. The UI will then startEngine and wait to reveice the initial round of callbacks
session = Session()

session.history.apiCallback_historySequenceStarted = callbacks._historySequenceStarted
session.history.apiCallback_historySequenceStopped = callbacks._historySequenceStopped

#Import complete. Now the parent module, like a gui, will call startEngine() and provide an event loop.