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


#Third Party Modules
from calfbox import cbox

#Template Modules
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.instrumentListMetadata = []
        self.startLoadingSamples = []
        self.instrumentStatusChanged = []
        self.startLoadingAuditionerInstrument = []
        self.auditionerInstrumentChanged = []
        self.auditionerVolumeChanged = []
        self.instrumentMidiNoteOnActivity = []

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


    def _instrumentListMetadata(self):
        """All libraries with all instruments as meta-data dicts. Hirarchy.
        A dict of dicts.
        Will be sent once on program start. Does not wait until actual samples are loaded
        but as soon as all data is parsed."""
        export = session.data.exportMetadata()
        for func in self.instrumentListMetadata:
            func(export)

    def _instrumentStatusChanged(self, libraryId:int, instrumentId:int):
        """For example the current variant changed"""
        export = session.data.libraries[libraryId].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"""
        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, libraryId:int, instrumentId:int):
        key = (libraryId, instrumentId)
        for func in self.instrumentMidiNoteOnActivity:
            func(key)

#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"])

    _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()

    #Inject the _instrumentMidiNoteOnActivity callback into session.data for access.
    session.data.instrumentMidiNoteOnActivity = callbacks._instrumentMidiNoteOnActivity

    callbacks.instrumentMidiNoteOnActivity.append(_checkForKeySwitch)

    callbacks._instrumentListMetadata() #Relatively quick. Happens only once in the program
    #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 Instrument.allInstruments.values():
        callbacks._instrumentStatusChanged(*instrument.idKey)

    callbacks._auditionerVolumeChanged()

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

def loadAllInstrumentSamples():
    """Actually load all instrument samples"""
    logger.info(f"Loading all instruments.")
    for instrument in Instrument.allInstruments.values():
        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 Instrument.allInstruments.values():
        if instrument.enabled:
            instrument.disable()
            callbacks._instrumentStatusChanged(*instrument.idKey)
    callbacks._dataChanged() #Needs to be here to have incremental updates in the gui.

def unloadInstrumentSamples(idkey:tuple):
    instrument = Instrument.allInstruments[idkey]
    instrument.disable()
    callbacks._instrumentStatusChanged(*instrument.idKey)
    callbacks._dataChanged()

def loadInstrumentSamples(idkey:tuple):
    """Load one .sfz from a library."""
    instrument = Instrument.allInstruments[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"""
    libraryId, instrumentId = idkey
    instrument = Instrument.allInstruments[idkey]
    instrument.chooseVariantByIndex(variantIndex)
    callbacks._instrumentStatusChanged(*instrument.idKey)
    callbacks._dataChanged()

def setInstrumentMixerVolume(idkey:tuple, value:float):
    """From 0 to -21.
    Default is -3.0 """
    instrument = Instrument.allInstruments[idkey]
    instrument.mixerLevel = value
    callbacks._instrumentStatusChanged(*instrument.idKey)

def setInstrumentMixerEnabled(idkey:tuple, state:bool):
    instrument = Instrument.allInstruments[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 = Instrument.allInstruments[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):
    """We added this ourselves to the note-on midi callback.
    So this gets called for every note-one."""
    instrument = Instrument.allInstruments[idkey]
    changed, nowKeySwitchPitch = instrument.updateCurrentKeySwitch()
    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 = Instrument.allInstruments[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(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 -3.0 """
    session.data.auditioner.volume = value
    callbacks._auditionerVolumeChanged()