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

#Standard Library Modules

#Template Modules
from template.calfbox import cbox
from template.engine.data import Data as TemplateData
import template.engine.sequencer
from template.engine.sequencer import SequencerInterface #group tracks
from template.engine.duration import D4
from template.engine.pitch import simpleNoteNames

#Our modules
from .track import Track


class Data(template.engine.sequencer.Score):
    """There must always be a Data class in a file main.py.
    Simply inheriting from engine.data.Data is easiest.

    You need to match the init parameters of your parent class. They vary from class to class
    of course. Simply copy and paste them from your Data parent class


    Pattern is our measure. Since Patroneo is not a metrical program we use the simple
    traditional time signatures.
    """

    def __init__(self, parentSession):
        super().__init__(parentSession)
        self.howManyUnits = 8
        self.whatTypeOfUnit = D4
        self.numberOfMeasures = 64
        self.measuresPerGroup = 8 # meta data, has no effect on playback.
        self.subdivisions = 1
        self.lastUsedNotenames = simpleNoteNames["English"] #The default value for new tracks/patterns. Changed each time the user picks a new representation via api.setNoteNames . noteNames are saved with the patterns.
        self.loopMeasureFactor = 1 #when looping how many at once?
        self.swing = 0 #-0.5 to 0.5  See pattern.buildPattern docstring.

        self.globalOffsetMeasures = 0  #relative to the current measure length. Parsed every export.
        self.globalOffsetTicks = 0 #absolute
        self.cachedOffsetInTicks = 0

        #Create three tracks with their first pattern activated, so 'play' after startup already produces sounding notes. This is less confusing for a new user.
        #self.tracks is created in the template.
        self._emptyFile = True
        self.addTrack(name="Melody A", color="#ffff00")
        self.addTrack(name="Chords A", color="#0055ff")
        self.addTrack(name="Bass A", color="#00ff00")
        self.addTrack(name="Drums A", color="#ff5500")
        self.addTrack(name="Drums B", color="#ff5500")

        self.tracks[0].group = "Melody"
        self.tracks[1].group = "Chords"
        self.tracks[2].group = "Bass"
        self.tracks[3].group = "Drums"
        self.tracks[4].group = "Drums"

        self.tracks[0].structure=set((0,))
        self.tracks[1].structure=set((0,))
        self.tracks[2].structure=set((0,))
        self.tracks[3].structure=set((0,))

        self.tracks[1].pattern.scale = (60, 59, 57, 55, 53, 52, 50, 48) #Middle base notes, C-Major
        self.tracks[2].pattern.scale = (48, 47, 45, 43, 41, 40, 38, 36) #Low base notes, C-Major
        self.tracks[3].pattern.simpleNoteNames = simpleNoteNames["Drums GM"]
        self.tracks[3].pattern.scale = (49, 53, 50, 45, 42, 39, 38, 36) #A pretty good starter drum set
        self.tracks[4].pattern.simpleNoteNames = simpleNoteNames["Drums GM"]
        self.tracks[4].pattern.scale = (49, 53, 50, 45, 42, 39, 38, 36) #A pretty good starter drum set


        self._processAfterInit()

    def _processAfterInit(self):
        self._lastLoopStart = 0 #ticks. not saved
        self.parentData = self # a trick to get SequencerInterface to play nicely with our groups

        #After load each track has it's own groupless Sequencerinterface.
        #self.groups itself is not saved.
        #Restore groups:
        logger.info("Restoring saved groups")
        self.groups = {} #NameString:SequencerInterface . Updated through function self.setGroup.
        assert self.tracks #at least 1
        self._duringFileLoadGroupIndicator = True
        for track in self.tracks:
            self.setGroup(track, track.group, buildAllTracksAfterwards=False) #buildAllTracks is handled by the API initial callback/load engineStart
        self.sortTracks()
        self._duringFileLoadGroupIndicator = False

    def setLanguageForEmptyFile(self, language):
        """Only use in api.startEngine. Overrides the empty.
        language is the summarized language string from Qt, like "German" which covers
        de_DE, _AT and _CH. If another GUI is used in the future it needs to send these as well."""
        if self._emptyFile:
            if language in simpleNoteNames:
                logger.info("Setting language for empty / new file to " + language)
                self.lastUsedNotenames = simpleNoteNames[language]
                self.tracks[0].pattern.simpleNoteNames = simpleNoteNames[language]
                self.tracks[1].pattern.simpleNoteNames = simpleNoteNames[language]
                self.tracks[2].pattern.simpleNoteNames = simpleNoteNames[language]
                #not drums.

    def cacheOffsetInTicks(self):
        oneMeasureInTicks = (self.howManyUnits * self.whatTypeOfUnit) / self.subdivisions
        oneMeasureInTicks = int(oneMeasureInTicks)
        self.cachedOffsetInTicks = oneMeasureInTicks * self.globalOffsetMeasures + self.globalOffsetTicks
        return self.cachedOffsetInTicks

    def addTrack(self, name="", scale=None, color=None, simpleNoteNames=None):
        """Overrides the simpler template version"""
        track = Track(parentData=self, name=name, scale=scale, color=color, simpleNoteNames=simpleNoteNames)
        self.tracks.append(track)
        #self.sortTracks() no sortTracks here. This is used during file load! The api takes care of sortTracks
        return track

    def convertSubdivisions(self, value, errorHandling):
        """Not only setting the subdivisions but also trying to scale existing notes up or down
        proportionally. But only if possible."""

        assert errorHandling in ("fail", "delete", "merge")

        scaleFactor = value / self.subdivisions
        inverseScaleFactor = self.subdivisions / value

        #the easiest case. New value is bigger and a multiple of the old one. 1->everything, 2->4.
        #We do need not check if the old notes have a place in the new grid because there are more new places than before
        if int(scaleFactor) == scaleFactor:
            assert int(scaleFactor * self.howManyUnits) == scaleFactor * self.howManyUnits, (scaleFactor, self.howManyUnits)
            self.howManyUnits = int(scaleFactor * self.howManyUnits)
            for track in self.tracks:
                for step in track.pattern.data:
                    step["index"] = int(scaleFactor * step["index"])
                    step["factor"] = scaleFactor * step["factor"]

        #Possible case, but needs checking.
        elif int(inverseScaleFactor) == inverseScaleFactor:
            #The following assertion is simply not correct. scaleFactor 0.5  and howManyUnits 11  means with 11 steps per measure you go from group 4 to 2, or 2 to 1.  With "delete if fail". normal operation
            #assert int(scaleFactor * self.howManyUnits) == scaleFactor * self.howManyUnits, (scaleFactor, self.howManyUnits)

            #Test, if in "fail" mode
            if errorHandling == "fail":
                if not int(scaleFactor * self.howManyUnits) == scaleFactor * self.howManyUnits:
                    return False
                for track in self.tracks:
                    for step in track.pattern.data:
                        if not int(scaleFactor * step["index"]) == scaleFactor * step["index"]: #yes, not inverse.
                            return False

            #Then apply
            todelete = []
            self.howManyUnits = int(scaleFactor * self.howManyUnits)
            for track in self.tracks:
                for step in track.pattern.data:
                    if errorHandling == "delete" and not int(scaleFactor * step["index"]) == scaleFactor * step["index"]:
                        todelete.append(step)
                    else: # if error handling was "merge" then impossible conversions will lead to step positions that can't be undone by restoring the old subdivision value.
                        step["index"] = int(scaleFactor * step["index"]) #yes, not inverse.
                        step["factor"] = scaleFactor * step["factor"]
                track.pattern.data = [d for d in track.pattern.data if not d in todelete]

        else: #anything involving a 3.
            #test if a conversion is possible. It is possible if you could first convert to 1 manually and then back up to the target number.
            #Or in other words: if only the main positions are set as steps.
            if errorHandling == "fail":
                if not int(scaleFactor * self.howManyUnits) == scaleFactor * self.howManyUnits:
                    return False
                for track in self.tracks:
                    for step in track.pattern.data:
                        if step["index"] % self.subdivisions: #not on a main position.
                            return False
            #Test without error. Go!
            self.howManyUnits = int(scaleFactor * self.howManyUnits)
            todelete = []
            for track in self.tracks:
                for step in track.pattern.data:
                    if errorHandling == "delete" and not int(scaleFactor * step["index"]) == scaleFactor * step["index"]:
                        todelete.append(step)
                    step["index"] = int(scaleFactor * step["index"]) #yes, not inverse.
                    step["factor"] = scaleFactor * step["factor"]
                track.pattern.data = [d for d in track.pattern.data if not d in todelete]

        self.subdivisions = value
        return True

    def _isGroupVisible(self, groupName:str):
        for track in self.tracks:
            if track.group == groupName:
                return track.visible
        return None #not a group

    def setGroup(self, track, groupName:str, buildAllTracksAfterwards=True):
        """Assigns a track to a group.

        This is the only function that changes track.group, be it to a group or empty string
        for "standalone track".
        We assume that groupName is a sanitized string. The api processes it as:
        groupName = ''.join(ch for ch in groupName if ch.isalnum())

        Group will be created if it does not exist yet,
        otherwise the track will be attached to the existing group.

        Track loses its existing standalone jack midi port OR its existing group connection
        and is assigned to the new one.

        Groups without connected tracks will be cleaned and their jack ports closed.
        Tracks that lose their group to empty string will get a new jack port.

        In the end we call buildAllTracks(), if not overriden by the parameter flag.
        If you do several track-groups in a row (e.g. on file load) you don't need to call that
        everytime. Once after you're done is enough.
        """
        #Simplest case first. If this is during load and this was a standalone track don't try to do anything group related
        if self._duringFileLoadGroupIndicator and track.group == "":
            return

        if (not self._duringFileLoadGroupIndicator) and track.group == groupName: #no change. During file load these are the same. see secondinit
            return

        #Convert ex-group to standalone track
        elif groupName == "" and not self._duringFileLoadGroupIndicator:
            self.groups[track.group].deleteSubtrack(id(track))
            track.group = ""
            track.visible = True
            #we never deleted the tracks standalone sequencerInterface. Reactivate.
            logger.info(f"Re-activiating standalone SequencerInterface for track {track.sequencerInterface.name}")
            track.sequencerInterface.recreateThroughUndo()

        #Add track to new or existing group
        else:
            if groupName in self.groups:
                #We need to base track visibility on the other group members. But we cannot use the inner position in the group for that because the first one might already be the new track.
                groupVisibility = self._isGroupVisible(groupName)  #checking one track is enough. They are all the same, once the file is loaded.
                assert not groupVisibility is None
            else:
                logger.info(f"Creating new group SequencerInterface for {groupName}")
                if not self._duringFileLoadGroupIndicator: #During file load tracks have groups already.
                    assert not groupName in [track.group for track in self.tracks if track.group], ([track.group for track in self.tracks]) #no track should have that group yet
                self.groups[groupName] = SequencerInterface(parentTrack=self, name=groupName) #we are not really a parentTrack, but it works.
                groupVisibility = True #First member of a new group, everything stays visible. It already is, but this makes it easier to set self.visible below.

            if track.group == "" or self._duringFileLoadGroupIndicator: #Change from standalone to group
                logger.info(f"Deactiviating standalone SequencerInterface for track {track.sequencerInterface.name}")
                track.sequencerInterface.prepareForDeletion()  #we do NOT delete the sequencerInterface. That has more data, such as the name!
            else: #track was already in another group. But could also be file load!
                logger.info(f"Changing group for track {track.sequencerInterface.name}")
                if not self._duringFileLoadGroupIndicator:
                    self.groups[track.group].deleteSubtrack(id(track))
                #next export (called at bottom of this function) will use the new subtrack already

            self.groups[groupName].setSubtrack(id(track), blobs=[(bytes(), 0, 0)])
            track.group = groupName #this switches the tracks midi generation to the subtrack based on its own id
            if (not self._duringFileLoadGroupIndicator):
                track.visible = groupVisibility

        if not self._duringFileLoadGroupIndicator:
            #Now that all tracks are set parse them all again to check if we have groups without tracks that need to get removed.
            logger.info(f"Removing groups without tracks")
            leftOverGroups = set(self.groups.keys())
            for track in self.tracks:
                if track.group and track.group in leftOverGroups:
                    leftOverGroups.remove(track.group)

            for loGroup in leftOverGroups:
                logger.info(f"Group {loGroup} has no tracks left and will be deleted now.")
                self.groups[loGroup].prepareForDeletion() #SequencerInterface.prepareForDeletion()
                del self.groups[loGroup]

            self.sortTracks()

            if buildAllTracksAfterwards:
                #TODO: this could be optimized. Currently everytime you delete a track, even an empty one, all tracks are re-built.
                #However, it is not *that* important. Everything works, good enough.
                self.buildAllTracks()

    def sortTracks(self):
        """Called by api.move and self.setGroup and all functions that either move tracks or
        manipulate groups.
        All tracks of a group must be neighbours in self.tracks.
        The track order within a group is choice to the user.

        We assume that self.groups is up to date.
        self.groups = {} #NameString:SequencerInterface . Updated through function self.setGroup.
        """
        logger.info(f"Sorting tracks and groups")
        tempGroups = {}
        listOfLists = [] #holds mutable instances of the same lists in tempGroups.values(). Standalone tracks are a list of len==1
        tempGroupOrder = [] #just group names. We use it for a test below.

        #Gather data
        for track in self.tracks:
            if track.group:
                assert track.group in self.groups
                if not track.group in tempGroupOrder:
                    tempGroupOrder.append(track.group)
                    sublist = []
                    tempGroups[track.group] = sublist #sub-order per group
                    listOfLists.append(sublist) #mutable, same instances.
                    assert tempGroups[track.group] is listOfLists[-1]

                tempGroups[track.group].append(track)
            else: #standalone track.
                #insert as group of one into listOfLists so we a sorted list, with a compatible format, later
                l = []
                l.append(track)
                track.visible = True # all standalone tracks are visible
                listOfLists.append(l)

        #Assemble new track order
        self._cachedTrackAndGroupOrderForJackMetadata = {}
        counter = 0
        newTracks = []
        for grouplist in listOfLists:
            for track in grouplist:
                newTracks.append(track)

                #Cache jack metadata port order
                if track.group:
                    portname = self.groups[track.group].cboxPortName()
                    if not portname in self._cachedTrackAndGroupOrderForJackMetadata: #did we already encounter this group?
                        self._cachedTrackAndGroupOrderForJackMetadata[portname] = counter
                        counter += 1
                else:
                    portname = track.sequencerInterface.cboxPortName()
                    self._cachedTrackAndGroupOrderForJackMetadata[portname] = counter
                    counter += 1

        assert len(newTracks) == len(self.tracks), (newTracks, self.tracks)
        assert set(newTracks) == set(self.tracks), (newTracks, self.tracks)
        self.tracks = newTracks


        #Test for data corruption. Not 100%, but combined with testing track.group in self.groups above it is reasonably good.
        assert len(tempGroupOrder) == len(tempGroups.keys()) == len(self.groups.keys()), (tempGroupOrder, tempGroups, self.groups)

    #Override template function to include groups
    def updateJackMetadataSorting(self):
        """Add this to you  "tracksChanged" or "numberOfTracksChanged" callback.
        Tell cbox to reorder the tracks by metadata. Deleted ports are automatically removed by JACK.

        It is advised to use this in a controlled manner. There is no Score-internal check if
        self.tracks changed and subsequent sorting. Multiple track changes in a row are common,
        therefore the place to update jack order is in the API, where the new track order is also
        sent to the UI.

        We also check if the track is 'deactivated' by probing track.cboxMidiOutUuid.
        Patroneo uses prepareForDeletion to deactive the tracks standalone track but keeps the
        interface around for later use.
        """
        #order = {portName:index for index, portName in enumerate(track.sequencerInterface.cboxPortName() for track in self.tracks if track.sequencerInterface.cboxMidiOutUuid)}
        order = self._cachedTrackAndGroupOrderForJackMetadata
        try:
            cbox.JackIO.Metadata.set_all_port_order(order)
        except Exception as e: #No Jack Meta Data or Error with ports.
            logger.error(e)


    def buildAllTracks(self, buildSongDuration=False):
        """Includes all patterns.

        buildSongDuration is True at least once in the programs life time, on startup.
        If True it will reset the loop.  The api calls buildSongDuration directly when it sets
        the loop.
        """
        self.cacheOffsetInTicks()

        for track in self.tracks:
            track.pattern.buildExportCache()
            track.buildTrack()
        if buildSongDuration:
            self.buildSongDuration()

    def buildSongDuration(self, loopMeasureAroundPpqn=None):
        """Loop does not reset automatically. We keep it until explicitely changed.
        If we do not have a loop the song duration is already maxTrackDuration, no update needed.

        An optional loopMeasureFactor can loop more than one measure. This is especially useful is
        track multiplicators are used. This is a saved context value in self.
        """
        self.cacheOffsetInTicks()
        oneMeasureInTicks = (self.howManyUnits * self.whatTypeOfUnit) / self.subdivisions
        oneMeasureInTicks = int(oneMeasureInTicks)
        maxTrackDuration = self.numberOfMeasures * oneMeasureInTicks + self.cachedOffsetInTicks
        if loopMeasureAroundPpqn is None: #could be 0
            cbox.Document.get_song().set_loop(maxTrackDuration, maxTrackDuration) #set playback length for the entire score. Why is the first value not zero? That would create an actual loop from the start to end. We want the song to play only once. The cbox way of doing that is to set the loop range to zero at the end of the track. Zero length is stop.
        else:
            loopMeasure =  int(loopMeasureAroundPpqn / oneMeasureInTicks) #0 based
            start = loopMeasure * oneMeasureInTicks
            end = start + oneMeasureInTicks * self.loopMeasureFactor
            start = start + self.cachedOffsetInTicks
            end = end + self.cachedOffsetInTicks
            cbox.Document.get_song().set_loop(start, end) #set playback length for the entire score. Why is the first value not zero? That would create an actual loop from the start to end. We want the song to play only once. The cbox way of doing that is to set the loop range to zero at the end of the track. Zero length is stop.
            return start, end

    #Save / Load / Export

    def serialize(self)->dict:
        dictionary = super().serialize()
        dictionary.update( { #update in place
            "howManyUnits" : self.howManyUnits,
            "whatTypeOfUnit" : self.whatTypeOfUnit,
            "numberOfMeasures" : self.numberOfMeasures,
            "measuresPerGroup" : self.measuresPerGroup,
            "subdivisions" : self.subdivisions,
            "lastUsedNotenames" : self.lastUsedNotenames,
            "loopMeasureFactor" : self.loopMeasureFactor,
            "swing" : self.swing,
            "globalOffsetMeasures" : self.globalOffsetMeasures,
            "globalOffsetTicks" : self.globalOffsetTicks,
        })
        return dictionary

    @classmethod
    def instanceFromSerializedData(cls, parentSession, serializedData):
        self = cls.__new__(cls)
        self.howManyUnits = serializedData["howManyUnits"]
        self.whatTypeOfUnit = serializedData["whatTypeOfUnit"]
        self.numberOfMeasures = serializedData["numberOfMeasures"]
        self.measuresPerGroup = serializedData["measuresPerGroup"]
        self.subdivisions = serializedData["subdivisions"]
        self.lastUsedNotenames = serializedData["lastUsedNotenames"]
        #v2.0
        if "loopMeasureFactor" in serializedData:
            self.loopMeasureFactor = serializedData["loopMeasureFactor"]
        else:
            self.loopMeasureFactor = 1
        if "swing" in serializedData:
            self.swing = serializedData["swing"]
        else:
            self.swing = 0

        #v2.1
        if "globalOffsetMeasures" in serializedData:
            self.globalOffsetMeasures = serializedData["globalOffsetMeasures"]
        else:
            self.globalOffsetMeasures = 0
        if "globalOffsetTicks" in serializedData:
            self.globalOffsetTicks = serializedData["globalOffsetTicks"]
        else:
            self.globalOffsetTicks = 0

        self._emptyFile = False


        #Tracks depend on the rest of the data already in place because they create a cache on creation.
        super().copyFromSerializedData(parentSession, serializedData, self) #Tracks, parentSession and tempoMap

        self._processAfterInit()

        return self

    def export(self):
        return {
            "numberOfTracks" : len(self.tracks),
            "howManyUnits" : self.howManyUnits,
            "whatTypeOfUnit" : self.whatTypeOfUnit,
            "numberOfMeasures" : self.numberOfMeasures,
            "measuresPerGroup" : self.measuresPerGroup,
            "subdivisions" : self.subdivisions,
            "loopMeasureFactor" : self.loopMeasureFactor,
            "isTransportMaster" : self.tempoMap.export()["isTransportMaster"],
            "swing" : self.swing,
            "globalOffsetMeasures" : self.globalOffsetMeasures,
            "globalOffsetTicks" : self.globalOffsetTicks,
            }