#! /usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Copyright 2021 , 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 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 = [ ]
#Sequencer
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 . """
session . nsmClient . announceSaveStatus ( False )
self . _historyChanged ( )
def _historyChanged ( self ) :
""" processQueue
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 :
func ( )
def _historySequenceStopped ( self ) :
""" see _historySequenceStarted """
for func in self . historySequenceStopped :
func ( )
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 :
func ( )
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 :
func ( status )
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 :
bar
beat #first index is 1
tick
bar_start_tick
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 :
func ( export )
#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 ( data . bar )
self . _rememberBarBeatTempo = data . beat #this should be enough inertia to not fire every 100ms
for func in self . barBeatTempo :
func ( bbtExport )
elif not data . beat :
for func in self . barBeatTempo :
func ( bbtExport )
self . _rememberBarBeatTempo = data . beat
"""
clock = str ( timedelta ( seconds = data . frame / data . frame_rate ) )
for func in self . clock :
func ( clock )
#Live Midi Recording
def _recordingModeChanged ( self ) :
if session . recordingEnabled :
session . nsmClient . changeLabel ( " Recording " )
else :
session . nsmClient . changeLabel ( " " )
for func in self . recordingModeChanged :
func ( session . recordingEnabled )
#Sequencer
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 .
"""
session . data . updateJackMetadataSorting ( )
lst = [ track . export ( ) for track in session . data . tracks ]
for func in self . numberOfTracksChanged :
func ( lst )
self . _dataChanged ( ) #includes _historyChanged
def _metronomeChanged ( self ) :
""" returns a dictionary with meta data such as the mute-state and the track name """
exportDict = session . data . metronome . export ( )
for func in self . metronomeChanged :
func ( exportDict )
#Sf2 Sampler
def _soundfontChanged ( self ) :
""" User loads a new soundfont or on load. Resets everything. """
exportDict = session . data . export ( )
session . data . updateAllChannelJackMetadaPrettyname ( )
session . nsmClient . changeLabel ( exportDict [ " name " ] )
if exportDict :
for func in self . soundfontChanged :
func ( exportDict )
def _channelChanged ( self , channel ) :
""" A single channel changed its parameters. The soundfont stays the same. """
exportDict = session . data . exportChannel ( channel )
session . data . updateChannelAudioJackMetadaPrettyname ( channel )
session . data . updateChannelMidiInJackMetadaPrettyname ( channel )
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 = session . data . midiInput . scene . status ( ) . layers [ 0 ] . status ( ) . ignore_program_changes
for func in self . ignoreProgramChangesChanged :
func ( state )
def _channelActivity ( self , channel ) :
""" send all note on to the GUI """
for func in self . channelActivity :
func ( channel )
def startEngine ( nsmClient ) :
"""
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 .
"""
logger . info ( " Starting template api engine " )
assert session
assert callbacks
session . nsmClient = nsmClient
session . eventLoop . fastConnect ( callbacks . _checkPlaybackStatusAndSendSignal )
session . eventLoop . fastConnect ( callbacks . _setPlaybackTicks )
session . eventLoop . fastConnect ( cbox . get_new_events ) #global cbox.get_new_events does not eat dynamic midi port events.
session . eventLoop . slowConnect ( callbacks . _checkBBTAndSendSignal )
#session.eventLoop.slowConnect(lambda: print(cbox.Transport.status().tempo))
#asession.eventLoop.slowConnect(lambda: print(cbox.Transport.status()))
cbox . Document . get_song ( ) . update_playback ( )
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.
logger . info ( " 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.
cbox . Document . get_song ( ) . update_playback ( )
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 """
session . nsmClient . serverSendSaveToSelf ( )
def undo ( ) :
""" No callbacks need to be called. Undo is done via a complementary
function , already defined , which has all the callbacks in it . """
session . history . 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 redo ( ) :
""" revert undo if nothing new has happened so far.
see undo """
session . history . redo ( )
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 session . data . getMidiInputNameAndUuid ( ) #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
else :
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 ( ) :
cbox . Transport . stop ( )
else :
cbox . Transport . play ( )
#It takes a few ms for playbackStatus to catch up. If you call it here again it will not be updated.
def getPlaybackTicks ( ) - > int :
return cbox . Transport . status ( ) . pos_ppqn
def seek ( value ) :
if value < 0 :
value = 0
cbox . Transport . seek_ppqn ( value )
def toStart ( ) :
seek ( 0 )
def playFrom ( ticks ) :
seek ( ticks )
if not playbackStatus ( ) :
cbox . Transport . play ( )
def playFromStart ( ) :
toStart ( )
if not playbackStatus ( ) :
cbox . Transport . play ( )
def toggleRecordingMode ( ) :
session . recordingEnabled = not session . recordingEnabled
callbacks . _recordingModeChanged ( )
# Sequencer Metronome
def setMetronome ( data , label ) :
session . data . metronome . generate ( data , label )
callbacks . _metronomeChanged ( )
def enableMetronome ( value ) :
session . data . metronome . setEnabled ( value ) #has side effects
callbacks . _metronomeChanged ( )
def isMetronomeEnabled ( ) :
return session . data . metronome . enabled
def toggleMetronome ( ) :
enableMetronome ( not session . data . metronome . enabled ) #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 = session . data . loadSoundfont ( filePath )
if success :
callbacks . _soundfontChanged ( )
session . history . clear ( )
callbacks . _historyChanged ( )
callbacks . _dataChanged ( )
else :
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 session . data . midiInput . scene . status ( ) . layers : #midi in[0] and real time midi thru[1]
layer . set_ignore_program_changes ( state )
assert layer . status ( ) . ignore_program_changes == state
callbacks . _ignoreProgramChangesChanged ( )
callbacks . _dataChanged ( )
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 } " )
session . data . setPatch ( channel , bank , program )
callbacks . _channelChanged ( channel )
callbacks . _dataChanged ( )
#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
else :
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 " ) :
history_test_change ( )
history_test_change ( )
history_test_change ( )
history_test_change ( )
callbacks . _historyChanged ( )
#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.