Patroneo/engine/api.py

#! /usr/bin/env python3
# -*- coding: utf-8 -*-

#Standard Library Modules
from typing import List, Set, Dict, Tuple

#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 .pattern import NUMBER_OF_STEPS

DEFAULT_FACTOR = 1 #for the GUI.


#New callbacks
class ClientCallbacks(Callbacks): #inherits from the templates api callbacks
    def __init__(self):
        super().__init__()

        self.timeSignatureChanged = []
        self.scoreChanged = []
        self.numberOfMeasuresChanged = []
        self.trackStructureChanged = []
        self.trackMetaDataChanged = []
        self.patternChanged = []
        self.stepChanged = []
        self.removeStep = []
        self.exportCacheChanged = []
        self.subdivisionsChanged = []
        self.quarterNotesPerMinuteChanged = []
        self.loopChanged = []

    def _quarterNotesPerMinuteChanged(self):
        """There is one tempo for the entire song in quarter notes per mintue.
        score.isTransportMaster to False means we do not create our own changes
        and leave everything to the default. Negative values are not possible"""
        if session.data.tempoMap.isTransportMaster:
            export = session.data.tempoMap.getQuarterNotesPerMinute()
        else:
            export = None
        for func in self.quarterNotesPerMinuteChanged:
            func(export)

    def _setPlaybackTicks(self): #Differs from the template because it has subdivisions.
        ppqn = cbox.Transport.status().pos_ppqn * session.data.subdivisions
        status = playbackStatus()
        for func in self.setPlaybackTicks:
            func(ppqn, status)

    def _loopChanged(self, measurenumber, loopStart, loopEnd):
        export = measurenumber
        for func in self.loopChanged:
            func(export)

    def _timeSignatureChanged(self):
        nr = session.data.howManyUnits
        typ = session.data.whatTypeOfUnit
        for func in self.timeSignatureChanged:
            func(nr, typ)

        ##All patterns and tracks need updates:
        for track in session.data.tracks:
            self._patternChanged(track)

        self._subdivisionsChanged() #update subdivisions. We do not include them in the score or track export on purpose.

    def _subdivisionsChanged(self):
        """Subdivisions are tricky, therefore we keep them isolated in their own callback.
        You don't need to redraw anything if you don't want to. One recommendation is to
        draw every n step a little more important (bigger, different color).
        where n = subdivions"""
        export = session.data.subdivisions
        for func in self.subdivisionsChanged:
            func(export)

    def _scoreChanged(self):
        """This includes the time signature as well, but is not send on a timesig change.
        A timesig change needs to update all tracks as playback as well as for the GUI
        so it is its own callback.
        Use this for fast and inexpensive updates like drawing a label or adjusting the
        GUI that shows your measure groups (a label each 8 measures or so)"""
        export = session.data.export()
        for func in self.scoreChanged:
            func(export)

    def _exportCacheChanged(self, track):
        """Send the export cache for GUI caching reasons. Don't react by redrawing immediately!
        This is sent often, redundantly and more than you need.
        Example: If you only show one pattern at the same time use this to cache the data
        in all hidden patterns and redraw only if you change the active pattern.
        You can react to real changes in your active pattern by using patternChanged
        and stepChanged."""
        export = track.export()
        for func in self.exportCacheChanged:
            func(export)

    def _patternChanged(self, track):
        """each track has only one pattern. We can identify the pattern by track and vice versa.
        Don't use this to react to clicks on the pattern editor. Use stepChanged instead and
        keep book of your incremental updates.
        This is used for the whole pattern: timesig changes, invert, clean etc.
        """
        export = track.export()
        self._exportCacheChanged(track)
        for func in self.patternChanged:
            func(export)

    def _stepChanged(self, track, stepDict):
        """A simple GUI will most like not listen to that callback since they
        already changed the step on their side. Only useful for parallel
        views.
        We do not export anything but just sent back the change we received
        as dict message."""
        self._exportCacheChanged(track)
        for func in self.stepChanged:
            func(stepDict)

    def _removeStep(self, track, index, pitch):
        """Opposite of _stepChanged"""
        self._exportCacheChanged(track)
        for func in self.stepChanged:
            func(index, pitch)

    def _trackStructureChanged(self, track):
        """update one track structure. Does not export cbox.
        Also includes transposition """
        export = track.export()
        for func in self.trackStructureChanged:
            func(export)

    def _trackMetaDataChanged(self, track):
        """a low cost function that should not trigger anything costly to redraw
        but some text and simple widgets."""
        export = track.export()
        for func in self.trackMetaDataChanged:
            func(export)

    def _numberOfMeasuresChanged(self):
        export = session.data.export()
        for func in self.numberOfMeasuresChanged:
            func(export)




#Inject our derived Callbacks into the parent module
template.engine.api.callbacks = ClientCallbacks()
from template.engine.api import callbacks

_templateStartEngine = startEngine

def startEngine(nsmClient):
    _templateStartEngine(nsmClient)

    session.inLoopMode = None # remember if we are in loop mode or not. Positive value is None or  a tuple with start and end

    #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
    callbacks._numberOfTracksChanged()
    callbacks._timeSignatureChanged()
    callbacks._numberOfMeasuresChanged()
    callbacks._subdivisionsChanged()
    callbacks._quarterNotesPerMinuteChanged()

    session.eventLoop.fastConnect(cbox.call_on_idle)

    for track in session.data.tracks:
        callbacks._trackMetaDataChanged(track) #for colors, scale and simpleNoteNames

    session.data.buildAllTracks()

    updatePlayback()

def toggleLoop():
    """Plays the current measure as loop.
    Current measure is where the playback cursor is

    session.inLoopMode is a tuple (start, end)
    """
    if session.inLoopMode:
        session.data.buildSongDuration() #no parameter removes the loop
        updatePlayback()
        session.inLoopMode = None
        callbacks._loopChanged(None, None, None)
    else:
        now = loopMeasureAroundPpqn=cbox.Transport.status().pos_ppqn
        loopStart, loopEnd = session.data.buildSongDuration(now)

        updatePlayback()
        session.inLoopMode = (loopStart, loopEnd)

        assert loopStart <= now < loopEnd
        if not playbackStatus():
            cbox.Transport.play()

        oneMeasureInTicks = (session.data.howManyUnits * session.data.whatTypeOfUnit) / session.data.subdivisions

        measurenumber, rest = divmod(loopStart, oneMeasureInTicks)
        callbacks._loopChanged(int(measurenumber), loopStart, loopEnd)


def seek(value):
    """override template one, which does not have a loop"""
    if value < 0:
        value = 0
    if session.inLoopMode and not session.inLoopMode[0] <= value < session.inLoopMode[1]: #if you seek outside the loop the loop will be destroyed.
        toggleLoop()
    cbox.Transport.seek_ppqn(value)

##Score
def set_quarterNotesPerMinute(value:int):
    if value is None:
        session.data.tempoMap.isTransportMaster = False #triggers rebuild
    elif value == "on":
        assert not session.data.tempoMap.isTransportMaster
        #keep old bpm value
        session.data.tempoMap.isTransportMaster = True #triggers rebuild
    else:
        assert value > 0
        session.data.tempoMap.setQuarterNotesPerMinute(value)
        session.data.tempoMap.isTransportMaster = True #triggers rebuild
    #Does not need track rebuilding
    updatePlayback()
    callbacks._quarterNotesPerMinuteChanged()


def set_whatTypeOfUnit(ticks):
    if session.data.whatTypeOfUnit == ticks: return
    session.data.whatTypeOfUnit = ticks
    session.data.buildAllTracks()
    updatePlayback()
    callbacks._timeSignatureChanged()

def set_howManyUnits(value):
    if session.data.howManyUnits == value: return
    session.data.howManyUnits = value
    session.data.buildAllTracks()
    updatePlayback()
    callbacks._timeSignatureChanged()

def set_subdivisions(value):
    if session.data.subdivisions == value: return
    session.data.subdivisions = value
    session.data.buildAllTracks()
    updatePlayback()
    callbacks._subdivisionsChanged()

def convert_subdivisions(value, errorHandling):
    """"errorHandling can be fail, delete or merge"""
    if session.data.subdivisions == value: return
    result = session.data.convertSubdivisions(value, errorHandling)
    if result:
        session.data.buildAllTracks()
        updatePlayback()
        callbacks._timeSignatureChanged() #includes subdivisions
        for tr in session.data.tracks:
            callbacks._patternChanged(tr)
    else:
        callbacks._subdivisionsChanged() #to reset the GUI value back to the working one.
    return result

def set_numberOfMeasures(value):
    if session.data.numberOfMeasures == value: return
    session.data.numberOfMeasures = value
    session.data.buildSongDuration()
    updatePlayback()
    callbacks._numberOfMeasuresChanged()
    callbacks._scoreChanged() #redundant but cheap and convenient

def set_measuresPerGroup(value):
    if session.data.measuresPerGroup == value: return
    session.data.measuresPerGroup = value
    #No playback change
    callbacks._scoreChanged()

def changeTrackName(trackId, name):
    track = session.data.trackById(trackId)
    track.sequencerInterface.name = " ".join(name.split())
    callbacks._trackMetaDataChanged(track)

def changeTrackColor(trackId, colorInHex):
    """Expects "#rrggbb"""
    track = session.data.trackById(trackId)
    assert len(colorInHex) == 7, colorInHex
    track.color = colorInHex
    callbacks._trackMetaDataChanged(track)

def addTrack(scale=None):
    if scale:
        assert type(scale) == tuple
    session.data.addTrack(scale=scale)
    callbacks._numberOfTracksChanged()

def createSiblingTrack(trackId):
    """Create a new track with scale, color and jack midi out the same as the given track.
    The jack midi out will be independent after creation, but connected to the same instrument
    (if any)"""
    track = session.data.trackById(trackId)
    assert type(track.pattern.scale) == tuple
    newTrack = session.data.addTrack(name=track.sequencerInterface.name, scale=track.pattern.scale, color=track.color, simpleNoteNames=track.pattern.simpleNoteNames) #track name increments itself from "Track A" to "Track B" or "Track 1" -> "Track 2"
    newTrack.pattern.averageVelocity = track.pattern.averageVelocity
    jackConnections = cbox.JackIO.get_connected_ports(track.sequencerInterface.cboxPortName())
    for port in jackConnections:
        cbox.JackIO.port_connect(newTrack.sequencerInterface.cboxPortName(), port)
    #Move new track to neighbour the old one.
    oldIndex = session.data.tracks.index(track)
    newIndex = session.data.tracks.index(newTrack)
    newTrackAgain = session.data.tracks.pop(newIndex)
    assert newTrackAgain is newTrack
    session.data.tracks.insert(oldIndex+1, newTrackAgain)
    callbacks._numberOfTracksChanged()
    return newTrack.export()

def deleteTrack(trackId):
    track = session.data.trackById(trackId)
    session.data.deleteTrack(track)
    if not session.data.tracks: #always keep at least one track
        session.data.addTrack()
    updatePlayback()
    callbacks._numberOfTracksChanged()

def moveTrack(trackId, newIndex):
    """index is 0 based"""
    track = session.data.trackById(trackId)
    oldIndex = session.data.tracks.index(track)
    if not oldIndex == newIndex:
        session.data.tracks.pop(oldIndex)
        session.data.tracks.insert(newIndex, track)
        callbacks._numberOfTracksChanged()

#Track Switches

def setSwitches(trackId, setOfPositions, newBool):
    track = session.data.trackById(trackId)
    if newBool:
        track.structure = track.structure.union(setOfPositions) #add setOfPositions to the existing one
    else:
        track.structure = track.structure.difference(setOfPositions) #remove everything from setOfPositions that is in the existing one, ignore entries from setOfPositions not in existing.
    track.buildTrack()
    updatePlayback()
    callbacks._trackStructureChanged(track)

def setSwitch(trackId, position, newBool):
    track = session.data.trackById(trackId)
    if newBool:
        if position in track.structure: return
        track.structure.add(position)
    else:
        if not position in track.structure: return
        track.structure.remove(position)
    track.buildTrack()
    updatePlayback()
    callbacks._trackStructureChanged(track)
    return True

def trackInvertSwitches(trackId):
    track = session.data.trackById(trackId)
    """
    if track.structure:
        new = set(i for i in range(max(track.structure)))
        track.structure = new.difference(track.structure)
    else:
        track.structure = set(i for i in range(session.data.numberOfMeasures))
    """
    new = set(i for i in range(session.data.numberOfMeasures))
    track.structure = new.difference(track.structure)
    track.buildTrack()
    updatePlayback()
    callbacks._trackStructureChanged(track)

def trackOffAllSwitches(trackId):
    track = session.data.trackById(trackId)
    track.structure = set()
    track.buildTrack()
    updatePlayback()
    callbacks._trackStructureChanged(track)

def trackOnAllSwitches(trackId):
    track = session.data.trackById(trackId)
    track.structure = set(i for i in range(session.data.numberOfMeasures))
    track.buildTrack()
    updatePlayback()
    callbacks._trackStructureChanged(track)

def trackMergeCopyFrom(sourceTrackId, targetTrackId):
    if not sourceTrackId == targetTrackId:
        sourceTrack = session.data.trackById(sourceTrackId)
        targetTrack = session.data.trackById(targetTrackId)
        targetTrack.structure = targetTrack.structure.union(sourceTrack.structure)
        targetTrack.whichPatternsAreScaleTransposed.update(sourceTrack.whichPatternsAreScaleTransposed)
        targetTrack.whichPatternsAreHalftoneTransposed.update(sourceTrack.whichPatternsAreHalftoneTransposed)
        targetTrack.buildTrack()
        updatePlayback()
        callbacks._trackStructureChanged(targetTrack)

def setSwitchScaleTranspose(trackId, position, transpose):
    """Scale transposition is flipped. lower value means higher pitch"""
    track = session.data.trackById(trackId)
    track.whichPatternsAreScaleTransposed[position] = transpose
    track.buildTrack()
    updatePlayback()
    callbacks._trackStructureChanged(track)
    return True

def setSwitchHalftoneTranspose(trackId, position, transpose):
    """Halftone transposition is not flipped. Higher value means higher pitch"""
    track = session.data.trackById(trackId)
    track.whichPatternsAreHalftoneTransposed[position] = transpose
    track.buildTrack()
    updatePlayback()
    callbacks._trackStructureChanged(track)
    return True

def insertSilence(howMany, beforeMeasureNumber):
    """Insert empty measures into all tracks"""
    for track in session.data.tracks:
        track.structure = set( (switch + howMany if switch >= beforeMeasureNumber else switch) for switch in track.structure )
        track.whichPatternsAreScaleTransposed = { (k+howMany if k >= beforeMeasureNumber else k):v for k,v in track.whichPatternsAreScaleTransposed.items() }
        track.whichPatternsAreHalftoneTransposed = { (k+howMany if k >= beforeMeasureNumber else k):v for k,v in track.whichPatternsAreHalftoneTransposed.items() }
        callbacks._trackStructureChanged(track)
    session.data.buildAllTracks()
    updatePlayback()

def deleteSwitches(howMany, fromMeasureNumber):
    for track in session.data.tracks:
        new_structure = set()

        for switch in track.structure:
            if switch < fromMeasureNumber:
                new_structure.add(switch)
            elif switch >= fromMeasureNumber+howMany: #like a text editor let gravitate left into the hole left by the deleted range
                new_structure.add(switch-howMany)
            #else: #discard all in range to delete
        track.structure = new_structure

        new_scaleTransposed = dict()
        for k,v in track.whichPatternsAreScaleTransposed.items():
            if k < fromMeasureNumber:
                new_scaleTransposed[k] = v
            elif k >= fromMeasureNumber+howMany: #like a text editor let gravitate left into the hole left by the deleted range
                new_scaleTransposed[k-howMany] = v
            #else: #discard all in range to delete
        track.whichPatternsAreScaleTransposed = new_scaleTransposed

        new_halftoneTransposed = dict()
        for k,v in track.whichPatternsAreHalftoneTransposed.items():
            if k < fromMeasureNumber:
                new_halftoneTransposed[k] = v
            elif k >= fromMeasureNumber+howMany: #like a text editor let gravitate left into the hole left by the deleted range
                new_halftoneTransposed[k-howMany] = v
            #else: #discard all in range to delete
        track.whichPatternsAreHalftoneTransposed = new_halftoneTransposed

        callbacks._trackStructureChanged(track)
    session.data.buildAllTracks()
    updatePlayback()

#Pattern Steps
def setPattern(trackId, patternList):
    """Change the whole pattern, send a callback with the whole pattern"""
    track = session.data.trackById(trackId)
    track.pattern.data = patternList
    track.pattern.buildExportCache()
    track.buildTrack()
    callbacks._patternChanged(track)

def getAverageVelocity(trackId):
    """If a GUI wants to add a new note and choose a sensible velocity it can use this function"""
    return session.data.trackById(trackId).pattern.averageVelocity

def setStep(trackId, stepExportDict):
    """This is an atomic operation that only sets one switch and
    only sends that switch back via callback. A simple GUI
    will most like not listen to that callback since they
    already changed the step on their side. Only useful for parallel
    views."""
    track = session.data.trackById(trackId)
    oldNote = track.pattern.stepByIndexAndPitch(index=stepExportDict["index"], pitch=stepExportDict["pitch"])
    if oldNote:  #modify existing note
        oldNoteIndex = track.pattern.data.index(oldNote)
        track.pattern.data.remove(oldNote)
        track.pattern.data.insert(oldNoteIndex, stepExportDict) #for what its worth, insert at the old place. It doesn't really matter though.
    else: #new note
        track.pattern.data.append(stepExportDict)

    track.pattern.buildExportCache()
    track.buildTrack()
    updatePlayback()
    callbacks._stepChanged(track, stepExportDict)

def removeStep(trackId, index, pitch):
    """Reverse of setStep"""
    track = session.data.trackById(trackId)
    oldNote = track.pattern.stepByIndexAndPitch(index, pitch)
    track.pattern.data.remove(oldNote)
    track.pattern.buildExportCache()
    track.buildTrack()
    updatePlayback()
    callbacks._removeStep(track, index, pitch)

def setScale(trackId, scale):
    """Expects a scale list or tuple from lowest index to highest.
    Actual pitches don't matter."""
    track = session.data.trackById(trackId)
    track.pattern.scale = scale
    track.pattern.buildExportCache()
    track.buildTrack()
    updatePlayback()
    callbacks._trackMetaDataChanged(track)

def setSimpleNoteNames(trackId, simpleNoteNames):
    """note names is a list of strings with length 128. One name for each midi note.
    It is saved to file"""
    track = session.data.trackById(trackId)
    track.pattern.simpleNoteNames = simpleNoteNames
    callbacks._trackMetaDataChanged(track)

def transposeHalftoneSteps(trackId, steps):
    track = session.data.trackById(trackId)
    track.pattern.scale = [midipitch+steps for midipitch in track.pattern.scale]
    track.pattern.buildExportCache()
    track.buildTrack()
    updatePlayback()
    callbacks._trackMetaDataChanged(track)

def patternInvertSteps(trackId):
    track = session.data.trackById(trackId)
    track.pattern.invert()
    track.pattern.buildExportCache()
    track.buildTrack()
    updatePlayback()
    callbacks._patternChanged(track)

def patternOnAllSteps(trackId):
    track = session.data.trackById(trackId)
    track.pattern.fill()
    track.pattern.buildExportCache()
    track.buildTrack()
    updatePlayback()
    callbacks._patternChanged(track)

def patternOffAllSteps(trackId):
    track = session.data.trackById(trackId)
    track.pattern.empty()
    track.pattern.buildExportCache()
    track.buildTrack()
    updatePlayback()
    callbacks._patternChanged(track)


major = [0, 2, 4, 5, 7, 9, 11, 12] #this if sorted by pitch, lowest to highest. Patroneo works in reverse order to accomodate the row/column approach of a grid. We reverse in setScaleToKeyword
schemesDict = {
    #this if sorted by pitch, lowest to highest. Patroneo works in reverse order to accomodate the row/column approach of a grid. We reverse in setScaleToKeyword
    "Major": [0,0,0,0,0,0,0,0],
    "Minor": [0,0,-1,0,0,-1,-1,0],
    "Dorian": [0,0,-1,0,0,0,-1,0],
    "Phrygian": [0,-1,-1,0,0,-1,-1,0],
    "Lydian": [0,0,0,+1,0,0,0,0],
    "Mixolydian": [0,0,0,0,0,0,-1,0],
    "Locrian": [0,-1,-1,0,-1,-1,-1,0],
    "Blues": [0,-2,-1,0,-1,-2,-1,0],
    "Hollywood": [0,0,0,0,0,-1,-1,0], #The "Hollywood"-Scale. Stargate, Lord of the Rings etc.
    "Chromatic": [0,-1,-2,-2,-3,-4,-5,-5], #not a complete octave, but that is how it goes.
}
major.reverse()
for l in schemesDict.values():
    l.reverse()

schemes = [
   "Major",
    "Minor",
    "Dorian",
    "Phrygian",
    "Lydian",
    "Mixolydian",
    "Locrian",
    "Blues",
    "Hollywood",
    "Chromatic",
]

def setScaleToKeyword(trackId, keyword):
    track = session.data.trackById(trackId)

    rememberRootNote = track.pattern.scale[-1] #no matter if this is the lowest or not%
    scale = [x + y for x, y in zip(major, schemesDict[keyword])]
    difference = rememberRootNote - scale[-1]
    result = [midipitch+difference for midipitch in scale]
    track.pattern.scale = result
    track.pattern.buildExportCache()
    track.buildTrack()
    updatePlayback()
    callbacks._trackMetaDataChanged(track)

def changePatternVelocity(trackId, steps):
    track = session.data.trackById(trackId)
    for note in track.pattern.data:
        new = note["velocity"] + steps
        note["velocity"] = min(max(new,0), 127)

    track.pattern.buildExportCache()
    track.buildTrack()
    updatePlayback()
    callbacks._patternChanged(track)


#Other functions

def noteOn(trackId, row):
    track = session.data.trackById(trackId)
    midipitch = track.pattern.scale[row]
    cbox.send_midi_event(0x90, midipitch, track.pattern.averageVelocity, output=track.sequencerInterface.cboxMidiOutUuid)

def noteOff(trackId, row):
    track = session.data.trackById(trackId)
    midipitch = track.pattern.scale[row]
    cbox.send_midi_event(0x80, midipitch, track.pattern.averageVelocity, output=track.sequencerInterface.cboxMidiOutUuid)