Patroneo/template/engine/sampler_sf2.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")


#Python Standard Lib
import os.path
from os import PathLike
from typing import List, Dict, Union

#Third Party
from template.calfbox import cbox

#Template Modules
from .data import Data
from .input_midi import MidiInput

#Client Modules
from engine.config import * #imports METADATA

class Sampler_sf2(Data):
    """
    Channels are base 1
    Programs and banks are base 0

    We cache the complete list of banks/programs in self.patchlist
    but we always fetch the active patches (16 channels) live from calfbox through
    self.activePatches()
    """
    def __init__(self, parentSession, filePath, activePatches, ignoreProgramChanges, mixer:bool, defaultSoundfont=None):
        super().__init__(parentSession)

        self._unlinkOnSave:List[Union[str, PathLike]] = [] #old sf2 files no longer in use.
        self.filePath = filePath
        self.patchlist:Dict[int,Dict[int, str]] = {} #bank:{program:name}
        self.buffer_ignoreProgramChanges = ignoreProgramChanges

        self.midiInput = MidiInput(session=parentSession, portName="all")
        self.midiInputJackName = f"{cbox.JackIO.status().client_name}:all"
        cbox.JackIO.Metadata.set_port_order(self.midiInputJackName, 0) #first one. The other port order later start at 1 because we loop 1-based for channels.

        #Set libfluidsynth! to 16 output pairs. We prepared 32 jack ports in the session start. "soundfont" is our given name, in the line below. This is a prepared config which will be looked up by add_new_instrument
        cbox.Config.set("instrument:soundfont", "engine", "fluidsynth")
        cbox.Config.set("instrument:soundfont", "output_pairs", 16) #this is not the same as session.py cbox.Config.set("io", "outputs", METADATA["cboxOutputs"]). It  is yet another layer and those two need connections
        #Create the permanent instrument which loads changing sf2
        layer = self.midiInput.scene.add_instrument_layer("soundfont")
        self.instrumentLayer = layer
        self.instrument = layer.get_instrument()
        self.loadSoundfont(filePath, defaultSoundfont)

        #Block incoming program and bank changes per midi.
        #Save and load is done via the instance variable.
        self.midiInput.scene.status().layers[0].set_ignore_program_changes(ignoreProgramChanges)
        assert self._ignoreProgramChanges == ignoreProgramChanges

        #Active patches is a dynamic value. Load the saved ones here, but do not save their state locally
        try:
            self.setActivePatches(activePatches)
        except:
            self._correctActivePatches()

        #Create a dynamic pair of audio output ports and route all stereo pairs to our summing channel.
        #This does _not_ need updating when loading another sf2 or changing instruments.

        assert not self.parentSession.standaloneMode is None
        if self.parentSession.standaloneMode:
            lmixUuid = cbox.JackIO.create_audio_output('left_mix', "#1")  #add "#1" as second parameter for auto-connection to system out 1
            rmixUuid = cbox.JackIO.create_audio_output('right_mix', "#2") #add "#2" as second parameter for auto-connection to system out 2
        else:
            lmixUuid = cbox.JackIO.create_audio_output('left_mix')
            rmixUuid = cbox.JackIO.create_audio_output('right_mix')

        for i in range(16):
            router = cbox.JackIO.create_audio_output_router(lmixUuid, rmixUuid)
            router.set_gain(-3.0)
            self.instrument.get_output_slot(i).rec_wet.attach(router) #output_slot is 0 based and means a pair
            #slot 17 is an error. cbox tells us there is only [1, 16], good.

        #Set port order. It is already correct because alphanumerial order, but explicit is better than implicit
        for channelNumber in range(1,33):
            portname = f"{cbox.JackIO.status().client_name}:out_{channelNumber}"
            try:
                cbox.JackIO.Metadata.set_port_order(portname, channelNumber)
            except Exception as e: #No Jack Meta Data
                logger.error(e)
                break

        #Also sort the mixing channels
        try:
            portname = f"{cbox.JackIO.status().client_name}:left_mix"
            cbox.JackIO.Metadata.set_port_order(portname, 33)
            portname = f"{cbox.JackIO.status().client_name}:right_mix"
            cbox.JackIO.Metadata.set_port_order(portname, 34)
        except Exception as e: #No Jack Meta Data
            logger.error(e)

        #If demanded create 16 routing midi channels, basically a jack port -> midi channel merger.
        #TODO: This is not working. Better leave that to JACK itself and an outside program for now.
        #Also deactivated the pretty name metadata update call in updateChannelMidiInJackMetadaPrettyname. It is called by an api callback and by our "all" function
        if False:
            for midiRouterPortNum in range(1,17):
                portName = f"channel_{str(midiRouterPortNum).zfill(2)}"
                router_cboxMidiPortUid = cbox.JackIO.create_midi_input(portName)
                router_scene = cbox.Document.get_engine().new_scene()
                router_scene.clear()
                router_scene.set_enable_default_song_input(False)
                #router_scene.add_instrument_layer("soundfont") #This creates multiple sf2 instances
                router_midi_layer = router_scene.add_new_midi_layer(router_cboxMidiPortUid) #Create a midi layer for our input port. That layer supports manipulation like transpose or channel routing.
                #cbox.JackIO.route_midi_input(router_cboxMidiPortUid, router_scene.uuid)
                router_midi_layer.set_out_channel(midiRouterPortNum)
                router_midi_layer.set_in_channel(midiRouterPortNum)
                #router_midi_layer.set_enable(False)

                #cbox.JackIO.route_midi_input(router_cboxMidiPortUid, self.midiInput.scene.uuid) #routes directly to the instrument and ignores the layer options like channel forcing
                #cbox.JackIO.route_midi_input(router_cboxMidiPortUid, self.midiInput.cboxMidiPortUid) #runs, but does nothing
                #cbox.JackIO.route_midi_input(router_midi_layer, self.midiInput.cboxMidiPortUid) #does not run
                router_midi_layer.set_external_output(self.midiInput.cboxMidiPortUid)
                #router_midi_layer.set_external_output(self.midiInput.scene.uuid)
                #Port order
                fullPortname = f"{cbox.JackIO.status().client_name}:out_{channelNumber}"
                try:
                    cbox.JackIO.Metadata.set_port_order(fullPortname, midiRouterPortNum)
                except Exception as e: #No Jack Meta Data
                    pass #don't break here.

    @property
    def _ignoreProgramChanges(self):
        try:
            result = self.midiInput.scene.status().layers[0].status().ignore_program_changes
            self.buffer_ignoreProgramChanges = result
        except IndexError: #No layers means no loaded instrument yet. e.g. program start or in between loading.
            result = self.buffer_ignoreProgramChanges
        return result

    def unlinkUnusedSoundfonts(self, stopSession=False):
        if stopSession and self.parentSession.lastSavedData and not self.filePath == self.parentSession.lastSavedData["filePath"]:
            #The user might want to discard the current unsaved setup (self.filePath) by closing without saving.
            #self.filePath holds the current file, not the saved file. We override to not get the actual saved setting deleted on quit.
            self._unlinkOnSave.append(self.filePath)
            self.filePath = self.parentSession.lastSavedData["filePath"]

        for filePath in self._unlinkOnSave:
            if not filePath == self.filePath and os.path.exists(filePath):
                os.unlink(filePath)
        self._unlinkOnSave = []


    def _convertPatches(self, dictionary:dict):
        """Returns a dict of dicts:
            bank:{program:name}"""
        result:Dict[int,Dict[int, str]] = {} #same as self.patchlist
        for value, name in dictionary.items():
            program = value & 127
            bank =  value >> 7
            if not bank in result:
                result[bank] = {}
            result[bank][program] = name
        return result

    def loadSoundfont(self, filePath, defaultSoundfont=None):
        """defaultSoundfont is a special case. The path is not saved"""

        logger.info(f"loading path: \"{filePath}\" defaultSoundfont parameter: {defaultSoundfont}")

        #Remove the old link, if present. We cannot unlink directly in loadSoundfont because it is quite possible that a user will try out another soundfont but decide not to save but close and reopen to get his old soundfont back.
        if self.filePath and os.path.islink(self.filePath):
            self._unlinkOnSave.append(self.filePath)

        #self.midiInput.scene.clear() do not call for the sf2 engine. Will kill our instrument.

        for stereoPair in range(16): #midi channels fluidsynth output    . 0-15
            #print (stereoPair)
            slot = self.instrument.get_output_slot(stereoPair) #slot is 0 based
            slot.set_output(stereoPair+1) #output is 1 based. setOutput routes to cbox jack ports, which are the same order.

        self.patchlist = {}
        try:
            if defaultSoundfont:
                self.instrument.engine.load_soundfont(defaultSoundfont)
                self.filePath = "" #redundant, but not wrong.
            else:
                self.instrument.engine.load_soundfont(filePath)
                self.filePath = filePath
            self.patchlist = self._convertPatches( self.instrument.engine.get_patches() )
            self._correctActivePatches()
            return True, ""
        except Exception as e: #throws a general Exception if not a good file
            #TODO: An error here leads to an undefined program state with no soundfont loaded. Either restore the old state perfectly (but why?) or go into a defined "nothing loaded" state.
            if os.path.exists(filePath) and os.path.islink(filePath):
                os.unlink(filePath) #the file was maybe already linked. undo
            return False, str(e)

    def _correctActivePatches(self):
        """Check if the current choice of bank and program actually exists in the soundfont
        and fall back to an existing value (or error).
        We keep no internal state of active patches but always ask cbox. Therefore we directly
        change cbox here and now."""
        if not self.patchlist:
            self.patchlist = self._convertPatches( self.instrument.engine.get_patches() ) #try again
            if not self.patchlist: raise ValueError("No programs in this sound font")

        for channel, (value, name) in self.instrument.engine.status().patch.items(): #channel is base 1
            program = value & 127
            bank =  value >> 7
            if bank in self.patchlist and program in self.patchlist[bank]:
                continue
            else:
                #just pick one at random
                for youchoose_bank in self.patchlist.keys():
                    for youchoose_program, youchoose_name in self.patchlist[youchoose_bank].items():
                        youchoose_bank, youchoose_program, youchoose_name
                        logger.info(METADATA["name"] + f": Channel {channel} Old bank/program not possible in new soundfont. Switching to first available program.")
                        self.setPatch(channel, youchoose_bank, youchoose_program)
                        break #inner loop. one instrument is enough.



    def activePatches(self):
        """Returns a dict of 16 tuples:
        channel(count from 1):(int bank, int program, str name)

        "Acoustic Piano" is the default name, since that is fluidsynth behaviour.

        Takes the current bank into consideration
        """
        result = {}
        for channel, (value, name) in self.instrument.engine.status().patch.items():
            program = value & 127
            bank =  value >> 7
            result[channel] = (bank,program,name) #the name doesn't matter for the program at all, it is only to get a human readable save file.
        return result

    def setActivePatches(self, activePatchesDict):
        """used for loading.
        receives an activePatches dict and sets all banks and programs accordingly"""
        assert self.patchlist, self.patchlist
        for channel, (bank, program, name) in activePatchesDict.items(): #the name doesn't matter for the program at all, it is only to get a human readable save file.
            self.setPatch(channel, bank, program)

    def setPatch(self, channel, bank, program):
        """An input error happens not often, but can happen if the saved data mismatches the
        soundfont. This happens on manual save file change or soundfont change (version update?)
        between program runs.
        We assume self.patchlist is up to date."""
        #After changing the bank we need to check if the program still exists.
        #If not fall back to an existing program
        if not program in self.patchlist[bank]:
            for i in range(128): #0-127
                if i in self.patchlist[bank]:
                    program = i
                    break
            else:
                return ValueError(f"No Program in Bank {bank}")

        newValue = program | (bank<<7)
        self.instrument.engine.set_patch(channel, newValue) #call to cbox

    def updateChannelAudioJackMetadaPrettyname(self, channel:int):
        """Audio output channel pairs 1-16"""
        bank, program, name = self.activePatches()[channel]  #activePatches returns a dict, not a list. channel is a 1-based key, not a 0-based index.

        chanL = channel*2-1 #channel is 1-based. [1]*2-1 = 1.
        chanR = channel*2

        portnameL = f"{cbox.JackIO.status().client_name}:out_{chanL}"
        portnameR = f"{cbox.JackIO.status().client_name}:out_{chanR}"

        #Use the instrument name as port name: 03-L:Violin
        try:
            cbox.JackIO.Metadata.set_pretty_name(portnameL, f"{str(channel).zfill(2)}-L : {name}")
            cbox.JackIO.Metadata.set_pretty_name(portnameR, f"{str(channel).zfill(2)}-R : {name}")
        except Exception as e: #No Jack Meta Data
            logger.error(e)

    def updateChannelMidiInJackMetadaPrettyname(self, channel:int):
        """Midi in single channels 1-16"""
        return #TODO: Until this works.
        bank, program, name = self.activePatches()[channel]  #activePatches returns a dict, not a list. channel is a 1-based key, not a 0-based index.

        portName = f"channel_{str(channel).zfill(2)}"
        fullPortname = f"{cbox.JackIO.status().client_name}:{portName}"

        #Use the instrument name as port name: 03-L:Violin
        try:
            cbox.JackIO.Metadata.set_pretty_name(fullPortname, f"{str(channel).zfill(2)} : {name}")
        except Exception as e: #No Jack Meta Data
            logger.error(e)


    def updateAllChannelJackMetadaPrettyname(self):
        """Add this to a callback. It can't be run on program startup because sampler_sf2 is
        initiated before cbox creates its audio ports.

        It is advised to use this in a controlled manner. There is no internal check if
        instruments changed and subsequent renaming. Multiple changes in a row are common,
        therefore the place to update is in the API, where the new state is also sent to the UI.
        """
        for channel in range(1,17):
            self.updateChannelAudioJackMetadaPrettyname(channel)
            self.updateChannelMidiInJackMetadaPrettyname(channel)

        """
        for channel, (bank, program, name) in self.activePatches().items():
            chanL = channel*2-1 #channel is 1-based. [1]*2-1 = 1.
            chanR = channel*2

            portnameL = f"{cbox.JackIO.status().client_name}:out_{chanL}"
            portnameR = f"{cbox.JackIO.status().client_name}:out_{chanR}"

            #Use the instrument name as port name: 03-L:Violin
            cbox.JackIO.Metadata.set_pretty_name(portnameL, f"{str(channel).zfill(2)}-L : {name}")
            cbox.JackIO.Metadata.set_pretty_name(portnameR, f"{str(channel).zfill(2)}-R : {name}")
        """


    def getMidiInputNameAndUuid(self):
        """
        Return name and cboxMidiPortUid.
        name is Client:Port JACK format

        Reimplement this in your actual data classes like sf2_sampler and sfz_sampler and
        sequencers. Used by the quick connect midi input widget.
        If double None as return the widget in the GUI might hide and deactivate itself."""
        return self.midiInputJackName, self.midiInput.cboxMidiPortUid



    #Save / Load / Export
    def serialize(self)->dict:
        self.unlinkUnusedSoundfonts() #We do not want them to be in the save file.
        return {
            "filePath" : self.filePath,
            "activePatches" : self.activePatches(),
            "ignoreProgramChanges" : self.midiInput.scene.status().layers[0].status().ignore_program_changes, #includes bank changes
            }

    @classmethod
    def instanceFromSerializedData(cls, parentSession, serializedData):
        """The entry function to create a sampler state from saved data. It is called by the session.

        This functions triggers a tree of other createInstanceFromSerializedData which finally
        returns Sampler_sf2, which gets saved in the session.

        The serializedData is already converted to primitive python types from json,
        but nothing more. Here we create the actual objects."""
        self = cls(parentSession=parentSession,
            #Other parameters or set loaded data different from standard init after creation
            filePath=serializedData["filePath"],
            activePatches=serializedData["activePatches"],
            ignoreProgramChanges=serializedData["ignoreProgramChanges"],
            )
        return self

    def exportChannel(self, channel:int):
        bank, program, name = self.activePatches()[channel]  #activePatches returns a dict, not a list. channel is a 1-based key, not a 0-based index.
        return {
            "channel" : channel,
            "bank": bank,
            "program": program,
            "name": name,
        }

    def export(self)->dict:
        prettyName = os.path.basename(self.filePath)
        if prettyName.endswith(".sf2"):
            prettyName = prettyName[:-4]
        return {
            "filePath":self.filePath,
            "patchlist":self.patchlist,
            "activePatches":self.activePatches(),
            "name": prettyName.title(),
            }