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.

509 lines
20 KiB

#! /usr/bin/env python3
# -*- coding: utf-8 -*-
Copyright 2022, Nils Hilbricht, Germany ( )
This file is part of the Laborejo Software Suite ( )
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
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 <>.
import logging; logger = logging.getLogger(__name__);"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 = []
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 or template api.
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."""
def _historyChanged(self):
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:
def _historySequenceStopped(self):
"""see _historySequenceStarted"""
for func in self.historySequenceStopped:
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:
self._dataChanged() #includes _historyChanged
def _setPlaybackTicks(self):
"""This gets called very very often (~60 times per second).
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:
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:
beat #first index is 1
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:
#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(
self._rememberBarBeatTempo = data.beat #this should be enough inertia to not fire every 100ms
for func in self.barBeatTempo:
elif not data.beat:
for func in self.barBeatTempo:
self._rememberBarBeatTempo = data.beat
clock = str(timedelta(seconds=data.frame / data.frame_rate))
for func in self.clock:
#Live Midi Recording
def _recordingModeChanged(self):
if session.recordingEnabled:
for func in self.recordingModeChanged:
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.
lst = [track.export() for track in]
for func in self.numberOfTracksChanged:
self._dataChanged() #includes _historyChanged
def _metronomeChanged(self):
"""returns a dictionary with meta data such as the mute-state and the track name"""
exportDict =
for func in self.metronomeChanged:
#Sf2 Sampler
def _soundfontChanged(self):
"""User loads a new soundfont or on load. Resets everything."""
exportDict =
if exportDict:
for func in self.soundfontChanged:
def _channelChanged(self, channel):
"""A single channel changed its parameters. The soundfont stays the same."""
exportDict =
for func in self.channelChanged:
func(channel, exportDict)
def _ignoreProgramChangesChanged(self):
#TODO: this only checks the first layer. There might be a second one as well. fine.jpg
state =[0].status().ignore_program_changes
for func in self.ignoreProgramChangesChanged:
def _channelActivity(self, channel):
"""send all note on to the GUI"""
for func in self.channelActivity:
def startEngine(nsmClient, additionalData:dict={}):
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.
""""Starting template api engine")
assert session
assert callbacks
session.nsmClient = nsmClient
session.eventLoop.fastConnect(cbox.get_new_events) #global cbox.get_new_events does not eat dynamic midi port events.
#session.eventLoop.slowConnect(lambda: print(cbox.Transport.status().tempo))
#asession.eventLoop.slowConnect(lambda: print(cbox.Transport.status()))
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."Template api engine started")
def isStandaloneMode():
return session.standaloneMode
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.
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"""
def undo():
"""No callbacks need to be called. Undo is done via a complementary
function, already defined, which has all the callbacks in it."""
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"""
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()
def getMidiInputNameAndUuid():
return #tuple name:str, uuid
#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
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():
#It takes a few ms for playbackStatus to catch up. If you call it here again it will not be updated.
def stop():
def getPlaybackTicks()->int:
return cbox.Transport.status().pos_ppqn
def seek(value):
if value < 0:
value = 0
def toStart():
def playFrom(ticks):
if not playbackStatus():
def playFromStart():
if not playbackStatus():
def toggleRecordingMode():
session.recordingEnabled = not session.recordingEnabled
# Sequencer Metronome
def setMetronome(data, label):, label)
def enableMetronome(value): #has side effects
def isMetronomeEnabled():
def toggleMetronome():
enableMetronome(not #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 =
if success:
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
for layer in #midi in[0] and real time midi thru[1]
assert layer.status().ignore_program_changes == state
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}"), bank, program)
#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
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"):
#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.