Sampled Instrument Player with static and monolithic design. All instruments are built-in.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

395 lines
16 KiB

#! /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 -3.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 -3.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