Tembro/engine/api.py

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

"""
Copyright 2022, 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")

#Standard Library Modules
from typing import List, Set, Dict, Tuple
import atexit

#Template Modules
from template.calfbox import cbox
import template.engine.api #we need direct access to the module to inject data in the provided structures. but we also need the functions directly. next line:
from template.engine.api import *

#Our Modules
from engine.instrument import Instrument


#New callbacks
class ClientCallbacks(Callbacks): #inherits from the templates api callbacks
    def __init__(self):
        super().__init__()

        self.tempCallback = []

        self.rescanSampleDir = []
        self.instrumentListMetadata = []
        self.startLoadingSamples = []
        self.instrumentStatusChanged = []
        self.startLoadingAuditionerInstrument = []
        self.auditionerInstrumentChanged = []
        self.auditionerVolumeChanged = []
        self.instrumentMidiNoteOnActivity = []
        self.instrumentMidiNoteOffActivity = []
        self.instrumentCCActivity = []

    def _tempCallback(self):
        """Just for copy paste during development"""
        export = session.data.export()
        for func in self.tempCallback:
            func(export)
        callbacks._dataChanged()

    def _rescanSampleDir(self):
        """instructs the GUI to forget all cached data and start fresh.
        Pure signal without parameters.

        This only happens on actual, manually instructed rescanning through the api.
        The program start happens without that and just sends data into a prepared but empty GUI."""
        for func in self.rescanSampleDir:
            func()

    def _instrumentListMetadata(self):
        """All libraries with all instruments as meta-data dicts. Hirarchy.
        A dict of dicts.
        Will be sent when instruments got parsed. At least once on program start and maybe
        on a rescan of the sample dir.
        """
        export = session.data.exportMetadata()
        for func in self.instrumentListMetadata:
            func(export)

    def _instrumentStatusChanged(self, libraryId:int, instrumentId:int):
        """For example the current variant changed"""
        lib = session.data.libraries[libraryId]
        export = lib.instruments[instrumentId].exportStatus()
        for func in self.instrumentStatusChanged:
            func(export)
        callbacks._dataChanged()

    def _startLoadingSamples(self, libraryId:int, instrumentId:int):
        """The sample loading of this instrument has started.
        Start flashing an LED or so.

        The end of loading is signaled by a statusChange callback with the current variant included
        """
        key = (libraryId, instrumentId)
        for func in self.startLoadingSamples:
            func(key)

    def _startLoadingAuditionerInstrument(self, libraryId:int, instrumentId:int):
        """The sample loading of the auditioner has started. Start flashing an LED or so. The
        end of loading is signaled by a _auditionerInstrumentChanged callback with the current
        variant included.
        """
        key = (libraryId, instrumentId)
        for func in self.startLoadingAuditionerInstrument:
            func(key)

    def _auditionerInstrumentChanged(self, libraryId:int, instrumentId:int):
        """We just send the key. It is assumed that the receiver already has a key:metadata database.
        We use the metadata of the actual instrument, and not an auditioner copy.

        None for one of the keys means the Auditioner instrument was unloaded.
        """
        if libraryId is None or instrumentId is None:
            export = None
        else:
            export = session.data.libraries[libraryId].instruments[instrumentId].exportMetadata()

        for func in self.auditionerInstrumentChanged:
            func(export)

    def _auditionerVolumeChanged(self):
        export = session.data.auditioner.volume
        for func in self.auditionerVolumeChanged:
            func(export)

    def _instrumentMidiNoteOnActivity(self, idKey, pitch, velocity):
        """This happens for real instruments as well as the auditioner"""
        for func in self.instrumentMidiNoteOnActivity:
            func(idKey, pitch, velocity)

    def _instrumentMidiNoteOffActivity(self, idKey, pitch, velocity):
        for func in self.instrumentMidiNoteOffActivity:
            func(idKey, pitch, velocity)

    def _instrumentCCActivity(self, idKey, ccNumber, value):
        for func in self.instrumentCCActivity:
            func(idKey, ccNumber, value)

#Inject our derived Callbacks into the parent module
template.engine.api.callbacks = ClientCallbacks()
from template.engine.api import callbacks

_templateStartEngine = startEngine

def startEngine(nsmClient, additionalData):
    session.data.parseAndLoadInstrumentLibraries(additionalData["baseSamplePath"], callbacks._instrumentMidiNoteOnActivity, callbacks._instrumentMidiNoteOffActivity, callbacks._instrumentCCActivity  )

    _templateStartEngine(nsmClient) #loads save files or creates empty structure.

    #Send initial Callbacks to create the first GUI state.
    #The order of initial callbacks must not change to avoid GUI problems.
    #For example it is important that the tracks get created first and only then the number of measures
    logger.info("Sending initial callbacks to GUI")

    #In opposite to other LSS programs loading is delayed because load times are so big.
    #Here we go, when everything is already in place and we have callbacks
    if session.data.cachedSerializedDataForStartEngine: #We loaded a save file
        logger.info("We started from a save-file. Restoring saved state now:")
        session.data.loadCachedSerializedData()

    callbacks.instrumentMidiNoteOnActivity.append(_checkForKeySwitch)

    callbacks._instrumentListMetadata() #The big "build the database" callback

    for instrument in session.data.allInstr():
        callbacks._instrumentStatusChanged(*instrument.idKey)

    callbacks._auditionerVolumeChanged()

    #Maybe autoconnect the mixer and auditioner.
    if session.standaloneMode and additionalData["autoconnectMixer"]: #this is only for startup.
        connectMixerToSystemPorts()

    atexit.register(unloadAllInstrumentSamples) #this will handle all python exceptions, but not segfaults of C modules.
    logger.info("Tembro api startEngine complete")


def _instr(idKey:tuple)->Instrument:
    return session.data.libraries[idKey[0]].instruments[idKey[1]]

def connectMixerToSystemPorts():
    session.data.connectMixerToSystemPorts()

def rescanSampleDirectory(newBaseSamplePath):
    """If the user wants to change the sample dir during runtime we use this function
    to rescan. Also useful after the download manager did it's task.

    In opposite to switching a session and reloading samples this function will unload all
    and not automatically reload.

    We do not set the sample dir here ourselves. This function can be used to rescan the existing
    dir again, in case of new downloaded samples.
    """
    logger.info(f"Rescanning sample dir: {newBaseSamplePath}")

    #The auditioner could contain an instrument that got deleted. We unload it to keep it simple.
    session.data.auditioner.unloadInstrument()
    callbacks._auditionerInstrumentChanged(None, None)

    #The next command will unload all deleted instruments and add all newly found instruments
    #However it will not unload and reload instruments that did not change on the surface.
    #In Tembro instruments never musically change through updates.
    #There is no musical danger of keeping an old version alive, even if a newly downloaded .tar
    #contains updates. A user must load/reload these manually or restart the program.
    session.data.parseAndLoadInstrumentLibraries(newBaseSamplePath, session.data.instrumentMidiNoteOnActivity, session.data.instrumentMidiNoteOffActivity, session.data.instrumentCCActivity )

    callbacks._rescanSampleDir() #instructs the GUI to forget all cached data and start fresh.
    callbacks._instrumentListMetadata()  #The big "build the database" callback

    #One round of status updates for all instruments. Still no samples loaded, but we saved the status of each with our own data.
    for instrument in session.data.allInstr():
        callbacks._instrumentStatusChanged(*instrument.idKey)

def loadAllInstrumentSamples():
    """Actually load all instrument samples"""
    logger.info("Loading all instruments.")
    for instrument in session.data.allInstr():
        callbacks._startLoadingSamples(*instrument.idKey)
        instrument.loadSamples()
        callbacks._instrumentStatusChanged(*instrument.idKey)
    callbacks._dataChanged()

def unloadAllInstrumentSamples():
    """Cleanup.
    It turned out relying on cbox closes ports or so too fast for JACK (not confirmed).
    If too many instruments (> ~12) were loaded the program will freeze on quit and freeze
    JACK as well.
    A controlled deactivating of instruments circumvents this.
    #TODO: Fixing jack2 is out of scope for us. If that even is a jack2 error.

    The order of atexit is the reverse order of registering. Since the template stopSession
    is registered before api.startEngine it is safe to rely on atexit.

    The function could also be used from the menu of course.
    """
    logger.info(f"Unloading all instruments.")
    for instrument in session.data.allInstr():
        if instrument.enabled:
            instrument.disable()
            callbacks._instrumentStatusChanged(*instrument.idKey) #Needs to be here to have incremental updates in the gui.
    callbacks._dataChanged()

def loadLibraryInstrumentSamples(libId):
    """Convenience function like loadAllInstrumentSamples or loadInstrumentSamples for the whole
    lib"""
    for instrumentId in session.data.libraries[libId].instruments.keys():
        idKey = (libId, instrumentId)
        loadInstrumentSamples(idKey) #all callbacks are triggered there

def unloadLibraryInstrumentSamples(libId):
    for instrumentId in session.data.libraries[libId].instruments.keys():
        idKey = (libId, instrumentId)
        unloadInstrumentSamples(idKey) #all callbacks are triggered there

def unloadInstrumentSamples(idKey:tuple):
    instrument = _instr(idKey)
    if instrument.enabled:
        instrument.disable()
        callbacks._instrumentStatusChanged(*instrument.idKey)
    callbacks._dataChanged()

def loadInstrumentSamples(idKey:tuple):
    """Load one .sfz from a library."""
    instrument = _instr(idKey)

    callbacks._startLoadingSamples(*instrument.idKey)

    if not instrument.enabled:
        instrument.enable()
    instrument.loadSamples()

    callbacks._instrumentStatusChanged(*instrument.idKey)
    callbacks._dataChanged()

def chooseVariantByIndex(idKey:tuple, variantIndex:int):
    """Choose a variant of an already enabled instrument"""
    instrument = _instr(idKey)
    instrument.chooseVariantByIndex(variantIndex)
    callbacks._instrumentStatusChanged(*instrument.idKey)
    callbacks._dataChanged()

def setInstrumentMixerVolume(idKey:tuple, value:float):
    """From 0 to -21.
    Default is -1.0 """
    instrument = _instr(idKey)
    instrument.mixerLevel = value
    callbacks._instrumentStatusChanged(*instrument.idKey)

def setInstrumentMixerEnabled(idKey:tuple, state:bool):
    instrument = _instr(idKey)
    instrument.setMixerEnabled(state)
    callbacks._instrumentStatusChanged(*instrument.idKey)

def setInstrumentKeySwitch(idKey:tuple, keySwitchMidiPitch:int):
    """Choose a keyswitch of the currently selected variant of this instrument. Keyswitch does not
    exist: The engine will throw a warning if the keyswitch was not in our internal representation,
    but nothing bad will happen otherwise.

    We use the key-midi-pitch directly.
    """
    instrument = _instr(idKey)
    result = instrument.setKeySwitch(keySwitchMidiPitch)
    if result: #could be None
        changed, nowKeySwitchPitch = result
    #Send in any case, no matter if changed or not. Doesn't hurt.
    callbacks._instrumentStatusChanged(*instrument.idKey)

def _checkForKeySwitch(idKey:tuple, pitch:int, velocity:int):
    """We added this ourselves to the note-on midi callback.
    So this gets called for every note-one."""
    instrument = _instr(idKey)

    result = instrument.updateCurrentKeySwitch()
    if result: #could be None
        changed, nowKeySwitchPitch = result
        if changed:
            callbacks._instrumentStatusChanged(*instrument.idKey)

def auditionerInstrument(idKey:tuple):
    """Load an indendepent instance of an instrument into the auditioner port"""
    libraryId, instrumentId = idKey
    callbacks._startLoadingAuditionerInstrument(libraryId, instrumentId)
    originalInstrument = _instr(idKey)

    #It does not matter if the originalInstrument has its samples loaded or not. We just want the path
    if originalInstrument.currentVariant:
        var = originalInstrument.currentVariant
    else:
        var = originalInstrument.defaultVariant

    session.data.auditioner.loadInstrument(idKey, originalInstrument.tarFilePath, originalInstrument.rootPrefixPath, var, originalInstrument.currentKeySwitch)
    callbacks._auditionerInstrumentChanged(libraryId, instrumentId)

def getAvailableAuditionerPorts()->dict:
    """Fetches a new port list each time it is called. No cache."""
    return session.data.auditioner.getAvailablePorts()

def connectAuditionerPort(externalPort:str):
    """externalPort is in the Client:Port JACK format.
    If externalPort evaluates to False it will disconnect any port."""
    session.data.auditioner.connectMidiInputPort(externalPort)

def setAuditionerVolume(value:float):
    """From 0 to -21.
    Default is -1.0 """
    session.data.auditioner.volume = value
    callbacks._auditionerVolumeChanged()


def connectInstrumentPort(idKey:tuple, externalPort:str):
    """externalPort can be empty string, which is disconnect"""
    instrument = _instr(idKey)
    if instrument.enabled:
        instrument.connectMidiInputPort(externalPort)


def sendNoteOnToInstrument(idKey:tuple, midipitch:int):
    """Not Realtime!
    Caller is responsible to shut off the note.
    Sends a fake midi-in callback."""
    v = 90
    #A midi event send to scene is different from an incoming midi event, which we use as callback trigger. We need to fake this one.
    instrument = _instr(idKey)
    if instrument.enabled:
        instrument.scene.send_midi_event(0x90, midipitch, v)
        callbacks._instrumentMidiNoteOnActivity(idKey, midipitch, v)

def sendNoteOffToInstrument(idKey:tuple, midipitch:int):
    """Not Realtime!
    Sends a fake midi-in callback."""
    #callbacks._stepEntryNoteOff(midipitch, 0)
    instrument = _instr(idKey)
    if instrument.enabled:
        instrument.scene.send_midi_event(0x80, midipitch, 0)
        callbacks._instrumentMidiNoteOffActivity(idKey, midipitch, 0)


def ccTrackingState(idKey:tuple):
    """Get the current values of all CCs that have been modified through midi-in so far.
    This also includes values we changed ourselves through sentCCToInstrument"""
    instrument = _instr(idKey)
    if instrument.enabled:
        return instrument.midiProcessor.ccState
    else:
        return None

def sentCCToInstrument(idKey:tuple, ccNumber:int, value:int):
    instrument = _instr(idKey)
    if instrument.enabled:
        instrument.scene.send_midi_event(0xB0, ccNumber, value)
        instrument.midiProcessor.ccState[ccNumber] = value #midi processor doesn't get send_midi_event
    else:
        return None