Duplicate from JammyMachina/the-jam-machine-app

Co-authored-by: Halid Bayram <[email protected]>

.gitattributes

@@ -0,0 +1,34 @@
+*.7z filter=lfs diff=lfs merge=lfs -text
+*.arrow filter=lfs diff=lfs merge=lfs -text
+*.bin filter=lfs diff=lfs merge=lfs -text
+*.bz2 filter=lfs diff=lfs merge=lfs -text
+*.ckpt filter=lfs diff=lfs merge=lfs -text
+*.ftz filter=lfs diff=lfs merge=lfs -text
+*.gz filter=lfs diff=lfs merge=lfs -text
+*.h5 filter=lfs diff=lfs merge=lfs -text
+*.joblib filter=lfs diff=lfs merge=lfs -text
+*.lfs.* filter=lfs diff=lfs merge=lfs -text
+*.mlmodel filter=lfs diff=lfs merge=lfs -text
+*.model filter=lfs diff=lfs merge=lfs -text
+*.msgpack filter=lfs diff=lfs merge=lfs -text
+*.npy filter=lfs diff=lfs merge=lfs -text
+*.npz filter=lfs diff=lfs merge=lfs -text
+*.onnx filter=lfs diff=lfs merge=lfs -text
+*.ot filter=lfs diff=lfs merge=lfs -text
+*.parquet filter=lfs diff=lfs merge=lfs -text
+*.pb filter=lfs diff=lfs merge=lfs -text
+*.pickle filter=lfs diff=lfs merge=lfs -text
+*.pkl filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.pth filter=lfs diff=lfs merge=lfs -text
+*.rar filter=lfs diff=lfs merge=lfs -text
+*.safetensors filter=lfs diff=lfs merge=lfs -text
+saved_model/**/* filter=lfs diff=lfs merge=lfs -text
+*.tar.* filter=lfs diff=lfs merge=lfs -text
+*.tflite filter=lfs diff=lfs merge=lfs -text
+*.tgz filter=lfs diff=lfs merge=lfs -text
+*.wasm filter=lfs diff=lfs merge=lfs -text
+*.xz filter=lfs diff=lfs merge=lfs -text
+*.zip filter=lfs diff=lfs merge=lfs -text
+*.zst filter=lfs diff=lfs merge=lfs -text
+*tfevents* filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1 @@
+__pycache__

.vscode/launch.json

@@ -0,0 +1,16 @@
+{
+  // Use IntelliSense to learn about possible attributes.
+  // Hover to view descriptions of existing attributes.
+  // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+  "version": "0.2.0",
+  "configurations": [
+    {
+      "name": "playground.py",
+      "type": "python",
+      "request": "launch",
+      "program": "playground.py",
+      "console": "integratedTerminal",
+      "justMyCode": false
+    }
+  ]
+}

.vscode/settings.json

@@ -0,0 +1,3 @@
+{
+  "python.formatting.provider": "black"
+}

README.md

@@ -0,0 +1,26 @@
+---
+title: The Jam Machine
+emoji: 🎶
+colorFrom: darkblue
+colorTo: black
+sdk: gradio
+sdk_version: 3.13.1
+python_version: 3.10.6
+app_file: playground.py
+pinned: true
+duplicated_from: JammyMachina/the-jam-machine-app
+---
+
+[Presentation](pitch.com/public/417162a8-88b0-4472-a651-c66bb89428be)
+## Contributors:
+### Jean Simonnet:
+- [Github](https://github.com/misnaej)
+- [Linkedin](https://www.linkedin.com/in/jeansimonnet/)
+    
+### Louis Demetz:
+- [Github](https://github.com/louis-demetz)
+- [Linkedin](https://www.linkedin.com/in/ldemetz/)
+    
+### Halid Bayram:
+- [Github](https://github.com/m41w4r3exe)
+- [Linkedin](https://www.linkedin.com/in/halid-bayram-6b9ba861/)

constants.py

@@ -0,0 +1,121 @@
+# fmt: off
+# Instrument mapping and mapping functions
+INSTRUMENT_CLASSES = [
+    {"name": "Piano", "program_range": range(0, 8), "family_number": 0},
+    {"name": "Chromatic Percussion", "program_range": range(8, 16), "family_number": 1},
+    {"name": "Organ", "program_range": range(16, 24), "family_number": 2},
+    {"name": "Guitar", "program_range": range(24, 32), "family_number": 3},
+    {"name": "Bass", "program_range": range(32, 40), "family_number": 4},
+    {"name": "Strings", "program_range": range(40, 48), "family_number": 5},
+    {"name": "Ensemble", "program_range": range(48, 56), "family_number": 6},
+    {"name": "Brass", "program_range": range(56, 64), "family_number": 7},
+    {"name": "Reed", "program_range": range(64, 72), "family_number": 8},
+    {"name": "Pipe", "program_range": range(72, 80), "family_number": 9},
+    {"name": "Synth Lead", "program_range": range(80, 88), "family_number": 10},
+    {"name": "Synth Pad", "program_range": range(88, 96), "family_number": 11},
+    {"name": "Synth Effects", "program_range": range(96, 104), "family_number": 12},
+    {"name": "Ethnic", "program_range": range(104, 112), "family_number": 13},
+    {"name": "Percussive", "program_range": range(112, 120), "family_number": 14},
+    {"name": "Sound Effects", "program_range": range(120, 128), "family_number": 15,},
+]
+# fmt: on
+
+# Instrument mapping for decodiing our midi sequence into midi instruments of our choice
+INSTRUMENT_TRANSFER_CLASSES = [
+    {
+        "name": "Piano",
+        "program_range": [4],
+        "family_number": 0,
+        "transfer_to": "Electric Piano 1",
+    },
+    {
+        "name": "Chromatic Percussion",
+        "program_range": [11],
+        "family_number": 1,
+        "transfer_to": "Vibraphone",
+    },
+    {
+        "name": "Organ",
+        "program_range": [17],
+        "family_number": 2,
+        "transfer_to": "Percussive Organ",
+    },
+    {
+        "name": "Guitar",
+        "program_range": [80],
+        "family_number": 3,
+        "transfer_to": "Synth Lead Square",
+    },
+    {
+        "name": "Bass",
+        "program_range": [38],
+        "family_number": 4,
+        "transfer_to": "Synth Bass 1",
+    },
+    {
+        "name": "Strings",
+        "program_range": [50],
+        "family_number": 5,
+        "transfer_to": "Synth Strings 1",
+    },
+    {
+        "name": "Ensemble",
+        "program_range": [51],
+        "family_number": 6,
+        "transfer_to": "Synth Strings 2",
+    },
+    {
+        "name": "Brass",
+        "program_range": [63],
+        "family_number": 7,
+        "transfer_to": "Synth Brass 1",
+    },
+    {
+        "name": "Reed",
+        "program_range": [64],
+        "family_number": 8,
+        "transfer_to": "Synth Brass 2",
+    },
+    {
+        "name": "Pipe",
+        "program_range": [82],
+        "family_number": 9,
+        "transfer_to": "Synth Lead Calliope",
+    },
+    {
+        "name": "Synth Lead",
+        "program_range": [81],  # Synth Lead Sawtooth
+        "family_number": 10,
+        "transfer_to": "Synth Lead Sawtooth",
+    },
+    {
+        "name": "Synth Pad",
+        "program_range": range(88, 96),
+        "family_number": 11,
+        "transfer_to": "Synth Pad",
+    },
+    {
+        "name": "Synth Effects",
+        "program_range": range(96, 104),
+        "family_number": 12,
+        "transfer_to": "Synth Effects",
+    },
+    {
+        "name": "Ethnic",
+        "program_range": range(104, 112),
+        "family_number": 13,
+        "transfer_to": "Ethnic",
+    },
+    {
+        "name": "Percussive",
+        "program_range": range(112, 120),
+        "family_number": 14,
+        "transfer_to": "Percussive",
+    },
+    {
+        "name": "Sound Effects",
+        "program_range": range(120, 128),
+        "family_number": 15,
+        "transfer_to": "Sound Effects",
+    },
+]

decoder.py

@@ -0,0 +1,197 @@
+from utils import *
+from familizer import Familizer
+from miditok import Event
+
+
+class TextDecoder:
+    """Decodes text into:
+    1- List of events
+    2- Then converts these events to midi file via MidiTok and miditoolkit
+
+    :param tokenizer: from MidiTok
+
+    Usage with write_to_midi method:
+        args: text(String) example ->  PIECE_START TRACK_START INST=25 DENSITY=2 BAR_START NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50...BAR_END TRACK_END
+        returns: midi file from miditoolkit
+    """
+
+    def __init__(self, tokenizer, familized=True):
+        self.tokenizer = tokenizer
+        self.familized = familized
+
+    def decode(self, text):
+        r"""converts from text to instrument events
+        Args:
+            text (String): example ->  PIECE_START TRACK_START INST=25 DENSITY=2 BAR_START NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50...BAR_END TRACK_END
+
+        Returns:
+            Dict{inst_id: List[Events]}: List of events of Notes with velocities, aggregated Timeshifts, for each instrument
+        """
+        piece_events = self.text_to_events(text)
+        inst_events = self.piece_to_inst_events(piece_events)
+        events = self.add_timeshifts_for_empty_bars(inst_events)
+        events = self.aggregate_timeshifts(events)
+        events = self.add_velocity(events)
+        return events
+
+    def tokenize(self, events):
+        r"""converts from events to MidiTok tokens
+        Args:
+            events (Dict{inst_id: List[Events]}): List of events for each instrument
+
+        Returns:
+            List[List[Events]]: List of tokens for each instrument
+        """
+        tokens = []
+        for inst in events.keys():
+            tokens.append(self.tokenizer.events_to_tokens(events[inst]))
+        return tokens
+
+    def get_midi(self, text, filename=None):
+        r"""converts from text to midi
+        Args:
+            text (String): example ->  PIECE_START TRACK_START INST=25 DENSITY=2 BAR_START NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50...BAR_END TRACK_END
+
+        Returns:
+            miditoolkit midi: Returns and writes to midi
+        """
+        events = self.decode(text)
+        tokens = self.tokenize(events)
+        instruments = self.get_instruments_tuple(events)
+        midi = self.tokenizer.tokens_to_midi(tokens, instruments)
+
+        if filename is not None:
+            midi.dump(f"{filename}")
+            print(f"midi file written: {filename}")
+
+        return midi
+
+    @staticmethod
+    def text_to_events(text):
+        events = []
+        for word in text.split(" "):
+            # TODO: Handle bar and track values with a counter
+            _event = word.split("=")
+            value = _event[1] if len(_event) > 1 else None
+            event = get_event(_event[0], value)
+            if event:
+                events.append(event)
+        return events
+
+    @staticmethod
+    def piece_to_inst_events(piece_events):
+        """Converts piece events of 8 bars to instrument events for entire song
+
+        Args:
+            piece_events (List[Events]): List of events of Notes, Timeshifts, Bars, Tracks
+
+        Returns:
+            Dict{inst_id: List[Events]}: List of events for each instrument
+
+        """
+        inst_events = {}
+        current_instrument = -1
+        for event in piece_events:
+            if event.type == "Instrument":
+                current_instrument = event.value
+                if current_instrument not in inst_events:
+                    inst_events[current_instrument] = []
+            elif current_instrument != -1:
+                inst_events[current_instrument].append(event)
+        return inst_events
+
+    @staticmethod
+    def add_timeshifts_for_empty_bars(inst_events):
+        """Adds time shift events instead of consecutive [BAR_START BAR_END] events"""
+        new_inst_events = {}
+        for inst, events in inst_events.items():
+            new_inst_events[inst] = []
+            for index, event in enumerate(events):
+                if event.type == "Bar-End" or event.type == "Bar-Start":
+                    if events[index - 1].type == "Bar-Start":
+                        new_inst_events[inst].append(Event("Time-Shift", "4.0.8"))
+                else:
+                    new_inst_events[inst].append(event)
+        return new_inst_events
+
+    @staticmethod
+    def add_timeshifts(beat_values1, beat_values2):
+        """Adds two beat values
+
+        Args:
+            beat_values1 (String): like 0.3.8
+            beat_values2 (String): like 1.7.8
+
+        Returns:
+            beat_str (String): added beats like 2.2.8 for example values
+        """
+        value1 = to_base10(beat_values1)
+        value2 = to_base10(beat_values2)
+        return to_beat_str(value1 + value2)
+
+    def aggregate_timeshifts(self, events):
+        """Aggregates consecutive time shift events bigger than a bar
+        -> like Timeshift 4.0.8
+
+        Args:
+            events (_type_): _description_
+
+        Returns:
+            _type_: _description_
+        """
+        new_events = {}
+        for inst, events in events.items():
+            inst_events = []
+            for i, event in enumerate(events):
+                if (
+                    event.type == "Time-Shift"
+                    and len(inst_events) > 0
+                    and inst_events[-1].type == "Time-Shift"
+                ):
+                    inst_events[-1].value = self.add_timeshifts(
+                        inst_events[-1].value, event.value
+                    )
+                else:
+                    inst_events.append(event)
+            new_events[inst] = inst_events
+        return new_events
+
+    @staticmethod
+    def add_velocity(events):
+        """Adds default velocity 99 to note events since they are removed from text, needed to generate midi"""
+        new_events = {}
+        for inst, events in events.items():
+            inst_events = []
+            for event in events:
+                inst_events.append(event)
+                if event.type == "Note-On":
+                    inst_events.append(Event("Velocity", 99))
+            new_events[inst] = inst_events
+        return new_events
+
+    def get_instruments_tuple(self, events):
+        """Returns instruments tuple for midi generation"""
+        instruments = []
+        for inst in events.keys():
+            is_drum = 0
+            if inst == "DRUMS":
+                inst = 0
+                is_drum = 1
+            if self.familized:
+                inst = Familizer(arbitrary=True).get_program_number(int(inst))
+            instruments.append((int(inst), is_drum))
+        return tuple(instruments)
+
+
+if __name__ == "__main__":
+
+    filename = "midi/generated/misnaej/the-jam-machine-elec-famil/20221209_175750"
+    encoded_json = readFromFile(
+        f"{filename}.json",
+        True,
+    )
+    encoded_text = encoded_json["sequence"]
+    # encoded_text = "PIECE_START TRACK_START INST=25 DENSITY=2 BAR_START NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=66 NOTE_ON=62 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_ON=66 NOTE_ON=62 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_ON=64 NOTE_ON=60 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=64 NOTE_ON=60 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_ON=64 NOTE_ON=60 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_ON=66 NOTE_ON=62 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_ON=64 NOTE_ON=60 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=66 NOTE_ON=62 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=67 NOTE_ON=64 TIME_DELTA=1 NOTE_OFF=67 NOTE_OFF=64 BAR_END BAR_START NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=66 NOTE_ON=62 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=64 NOTE_ON=62 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=62 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=64 NOTE_ON=60 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_ON=66 NOTE_ON=62 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_ON=64 NOTE_ON=60 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=66 NOTE_ON=62 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_ON=64 NOTE_ON=60 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 BAR_END BAR_START NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=66 NOTE_ON=62 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=64 NOTE_ON=60 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=64 NOTE_ON=60 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_OFF=50 NOTE_ON=66 NOTE_ON=62 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_ON=64 NOTE_ON=60 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_OFF=50 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=64 NOTE_ON=60 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_OFF=50 NOTE_ON=66 NOTE_ON=62 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_OFF=50 NOTE_ON=67 NOTE_ON=64 TIME_DELTA=1 NOTE_OFF=67 NOTE_OFF=64 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 BAR_END BAR_START NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=66 NOTE_ON=62 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=64 NOTE_ON=60 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_OFF=50 NOTE_ON=66 NOTE_ON=62 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_OFF=50 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=64 NOTE_ON=60 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_OFF=50 NOTE_ON=66 NOTE_ON=62 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_ON=64 NOTE_ON=60 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_OFF=50 NOTE_ON=66 NOTE_ON=62 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_OFF=50 NOTE_ON=67 NOTE_ON=64 TIME_DELTA=1 NOTE_OFF=67 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=64 NOTE_ON=60 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_ON=64 NOTE_ON=60 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_ON=64 NOTE_ON=60 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 BAR_END BAR_START NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=66 NOTE_ON=62 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_OFF=50 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=66 NOTE_ON=62 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=64 NOTE_ON=60 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_OFF=50 NOTE_ON=64 NOTE_ON=60 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_OFF=50 NOTE_ON=66 NOTE_ON=62 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_OFF=50 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=69 NOTE_ON=64 NOTE_ON=60 TIME_DELTA=1 NOTE_OFF=69 NOTE_OFF=64 NOTE_OFF=60 NOTE_ON=64 NOTE_ON=60 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_OFF=50 NOTE_ON=57 TIME_DELTA=1 NOTE_OFF=57 NOTE_ON=56 TIME_DELTA=1 NOTE_OFF=56 NOTE_ON=64 NOTE_ON=60 NOTE_ON=55 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_OFF=55 BAR_END BAR_START NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=66 NOTE_ON=62 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=64 NOTE_ON=60 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_ON=66 NOTE_ON=62 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_OFF=50 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=64 NOTE_ON=60 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_OFF=50 NOTE_ON=64 NOTE_ON=60 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_OFF=50 NOTE_ON=66 NOTE_ON=62 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_OFF=50 NOTE_ON=66 NOTE_ON=62 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=66 NOTE_OFF=62 NOTE_OFF=50 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=67 NOTE_ON=64 TIME_DELTA=1 NOTE_OFF=67 NOTE_OFF=64 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=50 NOTE_ON=64 NOTE_ON=60 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=64 NOTE_OFF=60 NOTE_OFF=50 NOTE_ON=59 NOTE_ON=55 NOTE_ON=50 TIME_DELTA=1 NOTE_OFF=59 NOTE_OFF=50 NOTE_OFF=55 NOTE_OFF=50 BAR_END BAR_START BAR_END TRACK_END"
+
+    miditok = get_miditok()
+    TextDecoder(miditok).get_midi(encoded_text, filename=filename)

familizer.py

@@ -0,0 +1,137 @@
+import random
+from joblib import Parallel, delayed
+from pathlib import Path
+from constants import INSTRUMENT_CLASSES, INSTRUMENT_TRANSFER_CLASSES
+from utils import get_files, timeit, FileCompressor
+
+
+class Familizer:
+    def __init__(self, n_jobs=-1, arbitrary=False):
+        self.n_jobs = n_jobs
+        self.reverse_family(arbitrary)
+
+    def get_family_number(self, program_number):
+        """
+        Given a MIDI instrument number, return its associated instrument family number.
+        """
+        for instrument_class in INSTRUMENT_CLASSES:
+            if program_number in instrument_class["program_range"]:
+                return instrument_class["family_number"]
+
+    def reverse_family(self, arbitrary):
+        """
+        Create a dictionary of family numbers to randomly assigned program numbers.
+        This is used to reverse the family number tokens back to program number tokens.
+        """
+
+        if arbitrary is True:
+            int_class = INSTRUMENT_TRANSFER_CLASSES
+        else:
+            int_class = INSTRUMENT_CLASSES
+
+        self.reference_programs = {}
+        for family in int_class:
+            self.reference_programs[family["family_number"]] = random.choice(
+                family["program_range"]
+            )
+
+    def get_program_number(self, family_number):
+        """
+        Given given a family number return a random program number in the respective program_range.
+        This is the reverse operation of get_family_number.
+        """
+        assert family_number in self.reference_programs
+        return self.reference_programs[family_number]
+
+    # Replace instruments in text files
+    def replace_instrument_token(self, token):
+        """
+        Given a MIDI program number in a word token, replace it with the family or program
+        number token depending on the operation.
+        e.g. INST=86 -> INST=10
+        """
+        inst_number = int(token.split("=")[1])
+        if self.operation == "family":
+            return "INST=" + str(self.get_family_number(inst_number))
+        elif self.operation == "program":
+            return "INST=" + str(self.get_program_number(inst_number))
+
+    def replace_instrument_in_text(self, text):
+        """Given a text piece, replace all instrument tokens with family number tokens."""
+        return " ".join(
+            [
+                self.replace_instrument_token(token)
+                if token.startswith("INST=") and not token == "INST=DRUMS"
+                else token
+                for token in text.split(" ")
+            ]
+        )
+
+    def replace_instruments_in_file(self, file):
+        """Given a text file, replace all instrument tokens with family number tokens."""
+        text = file.read_text()
+        file.write_text(self.replace_instrument_in_text(text))
+
+    @timeit
+    def replace_instruments(self):
+        """
+        Given a directory of text files:
+        Replace all instrument tokens with family number tokens.
+        """
+        files = get_files(self.output_directory, extension="txt")
+        Parallel(n_jobs=self.n_jobs)(
+            delayed(self.replace_instruments_in_file)(file) for file in files
+        )
+
+    def replace_tokens(self, input_directory, output_directory, operation):
+        """
+        Given a directory and an operation, perform the operation on all text files in the directory.
+        operation can be either 'family' or 'program'.
+        """
+        self.input_directory = input_directory
+        self.output_directory = output_directory
+        self.operation = operation
+
+        # Uncompress files, replace tokens, compress files
+        fc = FileCompressor(self.input_directory, self.output_directory, self.n_jobs)
+        fc.unzip()
+        self.replace_instruments()
+        fc.zip()
+        print(self.operation + " complete.")
+
+    def to_family(self, input_directory, output_directory):
+        """
+        Given a directory containing zip files, replace all instrument tokens with
+        family number tokens. The output is a directory of zip files.
+        """
+        self.replace_tokens(input_directory, output_directory, "family")
+
+    def to_program(self, input_directory, output_directory):
+        """
+        Given a directory containing zip files, replace all instrument tokens with
+        program number tokens. The output is a directory of zip files.
+        """
+        self.replace_tokens(input_directory, output_directory, "program")
+
+
+if __name__ == "__main__":
+
+    # Choose number of jobs for parallel processing
+    n_jobs = -1
+
+    # Instantiate Familizer
+    familizer = Familizer(n_jobs)
+
+    # Choose directory to process for program
+    input_directory = Path("midi/dataset/first_selection/validate").resolve()  # fmt: skip
+    output_directory = input_directory / "family"
+
+    # familize files
+    familizer.to_family(input_directory, output_directory)
+
+    # Choose directory to process for family
+    # input_directory = Path("../data/music_picks/encoded_samples/validate/family").resolve()  # fmt: skip
+    # output_directory = input_directory.parent / "program"
+
+    # # programize files
+    # familizer.to_program(input_directory, output_directory)

generate.py

@@ -0,0 +1,486 @@
+from generation_utils import *
+from utils import WriteTextMidiToFile, get_miditok
+from load import LoadModel
+from decoder import TextDecoder
+from playback import get_music
+
+
+class GenerateMidiText:
+    """Generating music with Class
+
+    LOGIC:
+
+    FOR GENERATING FROM SCRATCH:
+    - self.generate_one_new_track()
+    it calls
+        - self.generate_until_track_end()
+
+    FOR GENERATING NEW BARS:
+    - self.generate_one_more_bar()
+    it calls
+        - self.process_prompt_for_next_bar()
+        - self.generate_until_track_end()"""
+
+    def __init__(self, model, tokenizer, piece_by_track=[]):
+        self.model = model
+        self.tokenizer = tokenizer
+        # default initialization
+        self.initialize_default_parameters()
+        self.initialize_dictionaries(piece_by_track)
+
+    """Setters"""
+
+    def initialize_default_parameters(self):
+        self.set_device()
+        self.set_attention_length()
+        self.generate_until = "TRACK_END"
+        self.set_force_sequence_lenth()
+        self.set_nb_bars_generated()
+        self.set_improvisation_level(0)
+
+    def initialize_dictionaries(self, piece_by_track):
+        self.piece_by_track = piece_by_track
+
+    def set_device(self, device="cpu"):
+        self.device = ("cpu",)
+
+    def set_attention_length(self):
+        self.max_length = self.model.config.n_positions
+        print(
+            f"Attention length set to {self.max_length} -> 'model.config.n_positions'"
+        )
+
+    def set_force_sequence_lenth(self, force_sequence_length=True):
+        self.force_sequence_length = force_sequence_length
+
+    def set_improvisation_level(self, improvisation_value):
+        self.no_repeat_ngram_size = improvisation_value
+        print("--------------------")
+        print(f"no_repeat_ngram_size set to {improvisation_value}")
+        print("--------------------")
+
+    def reset_temperatures(self, track_id, temperature):
+        self.piece_by_track[track_id]["temperature"] = temperature
+
+    def set_nb_bars_generated(self, n_bars=8):  # default is a 8 bar model
+        self.model_n_bar = n_bars
+
+    """ Generation Tools - Dictionnaries """
+
+    def initiate_track_dict(self, instr, density, temperature):
+        label = len(self.piece_by_track)
+        self.piece_by_track.append(
+            {
+                "label": f"track_{label}",
+                "instrument": instr,
+                "density": density,
+                "temperature": temperature,
+                "bars": [],
+            }
+        )
+
+    def update_track_dict__add_bars(self, bars, track_id):
+        """Add bars to the track dictionnary"""
+        for bar in self.striping_track_ends(bars).split("BAR_START "):
+            if bar == "":  # happens is there is one bar only
+                continue
+            else:
+                if "TRACK_START" in bar:
+                    self.piece_by_track[track_id]["bars"].append(bar)
+                else:
+                    self.piece_by_track[track_id]["bars"].append("BAR_START " + bar)
+
+    def get_all_instr_bars(self, track_id):
+        return self.piece_by_track[track_id]["bars"]
+
+    def striping_track_ends(self, text):
+        if "TRACK_END" in text:
+            # first get rid of extra space if any
+            # then gets rid of "TRACK_END"
+            text = text.rstrip(" ").rstrip("TRACK_END")
+        return text
+
+    def get_last_generated_track(self, full_piece):
+        track = (
+            "TRACK_START "
+            + self.striping_track_ends(full_piece.split("TRACK_START ")[-1])
+            + "TRACK_END "
+        )  # forcing the space after track and
+        return track
+
+    def get_selected_track_as_text(self, track_id):
+        text = ""
+        for bar in self.piece_by_track[track_id]["bars"]:
+            text += bar
+        text += "TRACK_END "
+        return text
+
+    @staticmethod
+    def get_newly_generated_text(input_prompt, full_piece):
+        return full_piece[len(input_prompt) :]
+
+    def get_whole_piece_from_bar_dict(self):
+        text = "PIECE_START "
+        for track_id, _ in enumerate(self.piece_by_track):
+            text += self.get_selected_track_as_text(track_id)
+        return text
+
+    def delete_one_track(self, track):  # TO BE TESTED
+        self.piece_by_track.pop(track)
+
+    # def update_piece_dict__add_track(self, track_id, track):
+    #     self.piece_dict[track_id] = track
+
+    # def update_all_dictionnaries__add_track(self, track):
+    # self.update_piece_dict__add_track(track_id, track)
+
+    """Basic generation tools"""
+
+    def tokenize_input_prompt(self, input_prompt, verbose=True):
+        """Tokenizing prompt
+
+        Args:
+        - input_prompt (str): prompt to tokenize
+
+        Returns:
+        - input_prompt_ids (torch.tensor): tokenized prompt
+        """
+        if verbose:
+            print("Tokenizing input_prompt...")
+
+        return self.tokenizer.encode(input_prompt, return_tensors="pt")
+
+    def generate_sequence_of_token_ids(
+        self,
+        input_prompt_ids,
+        temperature,
+        verbose=True,
+    ):
+        """
+        generate a sequence of token ids based on input_prompt_ids
+        The sequence length depends on the trained model (self.model_n_bar)
+        """
+        generated_ids = self.model.generate(
+            input_prompt_ids,
+            max_length=self.max_length,
+            do_sample=True,
+            temperature=temperature,
+            no_repeat_ngram_size=self.no_repeat_ngram_size,  # default = 0
+            eos_token_id=self.tokenizer.encode(self.generate_until)[0],  # good
+        )
+
+        if verbose:
+            print("Generating a token_id sequence...")
+
+        return generated_ids
+
+    def convert_ids_to_text(self, generated_ids, verbose=True):
+        """converts the token_ids to text"""
+        generated_text = self.tokenizer.decode(generated_ids[0])
+        if verbose:
+            print("Converting token sequence to MidiText...")
+        return generated_text
+
+    def generate_until_track_end(
+        self,
+        input_prompt="PIECE_START ",
+        instrument=None,
+        density=None,
+        temperature=None,
+        verbose=True,
+        expected_length=None,
+    ):
+
+        """generate until the TRACK_END token is reached
+        full_piece = input_prompt + generated"""
+        if expected_length is None:
+            expected_length = self.model_n_bar
+
+        if instrument is not None:
+            input_prompt = f"{input_prompt}TRACK_START INST={str(instrument)} "
+            if density is not None:
+                input_prompt = f"{input_prompt}DENSITY={str(density)} "
+
+        if instrument is None and density is not None:
+            print("Density cannot be defined without an input_prompt instrument #TOFIX")
+
+        if temperature is None:
+            ValueError("Temperature must be defined")
+
+        if verbose:
+            print("--------------------")
+            print(
+                f"Generating {instrument} - Density {density} - temperature {temperature}"
+            )
+        bar_count_checks = False
+        failed = 0
+        while not bar_count_checks:  # regenerate until right length
+            input_prompt_ids = self.tokenize_input_prompt(input_prompt, verbose=verbose)
+            generated_tokens = self.generate_sequence_of_token_ids(
+                input_prompt_ids, temperature, verbose=verbose
+            )
+            full_piece = self.convert_ids_to_text(generated_tokens, verbose=verbose)
+            generated = self.get_newly_generated_text(input_prompt, full_piece)
+            # bar_count_checks
+            bar_count_checks, bar_count = bar_count_check(generated, expected_length)
+
+            if not self.force_sequence_length:
+                # set bar_count_checks to true to exist the while loop
+                bar_count_checks = True
+
+            if not bar_count_checks and self.force_sequence_length:
+                # if the generated sequence is not the expected length
+                if failed > -1:  # deactivated for speed
+                    full_piece, bar_count_checks = forcing_bar_count(
+                        input_prompt,
+                        generated,
+                        bar_count,
+                        expected_length,
+                    )
+                else:
+                    print('"--- Wrong length - Regenerating ---')
+            if not bar_count_checks:
+                failed += 1
+                if failed > 2:
+                    bar_count_checks = True  # TOFIX exit the while loop
+
+        return full_piece
+
+    def generate_one_new_track(
+        self,
+        instrument,
+        density,
+        temperature,
+        input_prompt="PIECE_START ",
+    ):
+        self.initiate_track_dict(instrument, density, temperature)
+        full_piece = self.generate_until_track_end(
+            input_prompt=input_prompt,
+            instrument=instrument,
+            density=density,
+            temperature=temperature,
+        )
+
+        track = self.get_last_generated_track(full_piece)
+        self.update_track_dict__add_bars(track, -1)
+        full_piece = self.get_whole_piece_from_bar_dict()
+        return full_piece
+
+    """ Piece generation - Basics """
+
+    def generate_piece(self, instrument_list, density_list, temperature_list):
+        """generate a sequence with mutiple tracks
+        - inst_list sets the list of instruments of the order of generation
+        - density is paired with inst_list
+        Each track/intrument is generated on a prompt which contains the previously generated track/instrument
+        This means that the first instrument is generated with less bias than the next one, and so on.
+
+        'generated_piece' keeps track of the entire piece
+        'generated_piece' is returned by self.generate_until_track_end
+        # it is returned by self.generate_until_track_end"""
+
+        generated_piece = "PIECE_START "
+        for instrument, density, temperature in zip(
+            instrument_list, density_list, temperature_list
+        ):
+            generated_piece = self.generate_one_new_track(
+                instrument,
+                density,
+                temperature,
+                input_prompt=generated_piece,
+            )
+
+        # generated_piece = self.get_whole_piece_from_bar_dict()
+        self.check_the_piece_for_errors()
+        return generated_piece
+
+    """ Piece generation - Extra Bars """
+
+    @staticmethod
+    def process_prompt_for_next_bar(self, track_idx):
+        """Processing the prompt for the model to generate one more bar only.
+        The prompt containts:
+                if not the first bar: the previous, already processed, bars of the track
+                the bar initialization (ex: "TRACK_START INST=DRUMS DENSITY=2 ")
+                the last (self.model_n_bar)-1 bars of the track
+        Args:
+            track_idx (int): the index of the track to be processed
+
+        Returns:
+            the processed prompt for generating the next bar
+        """
+        track = self.piece_by_track[track_idx]
+        # for bars which are not the bar to prolong
+        pre_promt = "PIECE_START "
+        for i, othertrack in enumerate(self.piece_by_track):
+            if i != track_idx:
+                len_diff = len(othertrack["bars"]) - len(track["bars"])
+                if len_diff > 0:
+                    # if other bars are longer, it mean that this one should catch up
+                    pre_promt += othertrack["bars"][0]
+                    for bar in track["bars"][-self.model_n_bar :]:
+                        pre_promt += bar
+                    pre_promt += "TRACK_END "
+                elif False:  # len_diff <= 0: # THIS GENERATES EMPTINESS
+                    # adding an empty bars at the end of the other tracks if they have not been processed yet
+                    pre_promt += othertracks["bars"][0]
+                    for bar in track["bars"][-(self.model_n_bar - 1) :]:
+                        pre_promt += bar
+                    for _ in range(abs(len_diff) + 1):
+                        pre_promt += "BAR_START BAR_END "
+                    pre_promt += "TRACK_END "
+
+        # for the bar to prolong
+        # initialization e.g TRACK_START INST=DRUMS DENSITY=2
+        processed_prompt = track["bars"][0]
+        for bar in track["bars"][-(self.model_n_bar - 1) :]:
+            # adding the "last" bars of the track
+            processed_prompt += bar
+
+        processed_prompt += "BAR_START "
+        print(
+            f"--- prompt length = {len((pre_promt + processed_prompt).split(' '))} ---"
+        )
+        return pre_promt + processed_prompt
+
+    def generate_one_more_bar(self, i):
+        """Generate one more bar from the input_prompt"""
+        processed_prompt = self.process_prompt_for_next_bar(self, i)
+        prompt_plus_bar = self.generate_until_track_end(
+            input_prompt=processed_prompt,
+            temperature=self.piece_by_track[i]["temperature"],
+            expected_length=1,
+            verbose=False,
+        )
+        added_bar = self.get_newly_generated_bar(prompt_plus_bar)
+        self.update_track_dict__add_bars(added_bar, i)
+
+    def get_newly_generated_bar(self, prompt_plus_bar):
+        return "BAR_START " + self.striping_track_ends(
+            prompt_plus_bar.split("BAR_START ")[-1]
+        )
+
+    def generate_n_more_bars(self, n_bars, only_this_track=None, verbose=True):
+        """Generate n more bars from the input_prompt"""
+        if only_this_track is None:
+            only_this_track
+
+        print(f"================== ")
+        print(f"Adding {n_bars} more bars to the piece ")
+        for bar_id in range(n_bars):
+            print(f"----- added bar #{bar_id+1} --")
+            for i, track in enumerate(self.piece_by_track):
+                if only_this_track is None or i == only_this_track:
+                    print(f"--------- {track['label']}")
+                    self.generate_one_more_bar(i)
+        self.check_the_piece_for_errors()
+
+    def check_the_piece_for_errors(self, piece: str = None):
+
+        if piece is None:
+            piece = generate_midi.get_whole_piece_from_bar_dict()
+        errors = []
+        errors.append(
+            [
+                (token, id)
+                for id, token in enumerate(piece.split(" "))
+                if token not in self.tokenizer.vocab or token == "UNK"
+            ]
+        )
+        if len(errors) > 0:
+            # print(piece)
+            for er in errors:
+                er
+                print(f"Token not found in the piece at {er[0][1]}: {er[0][0]}")
+                print(piece.split(" ")[er[0][1] - 5 : er[0][1] + 5])
+
+
+if __name__ == "__main__":
+
+    # worker
+    DEVICE = "cpu"
+
+    # define generation parameters
+    N_FILES_TO_GENERATE = 2
+    Temperatures_to_try = [0.7]
+
+    USE_FAMILIZED_MODEL = True
+    force_sequence_length = True
+
+    if USE_FAMILIZED_MODEL:
+        # model_repo = "misnaej/the-jam-machine-elec-famil"
+        # model_repo = "misnaej/the-jam-machine-elec-famil-ft32"
+
+        # model_repo = "JammyMachina/elec-gmusic-familized-model-13-12__17-35-53"
+        # n_bar_generated = 8
+
+        model_repo = "JammyMachina/improved_4bars-mdl"
+        n_bar_generated = 4
+        instrument_promt_list = ["4", "DRUMS", "3"]
+        # DRUMS = drums, 0 = piano, 1 = chromatic percussion, 2 = organ, 3 = guitar, 4 = bass, 5 = strings, 6 = ensemble, 7 = brass, 8 = reed, 9 = pipe, 10 = synth lead, 11 = synth pad, 12 = synth effects, 13 = ethnic, 14 = percussive, 15 = sound effects
+        density_list = [3, 2, 2]
+        # temperature_list = [0.7, 0.7, 0.75]
+    else:
+        model_repo = "misnaej/the-jam-machine"
+        instrument_promt_list = ["30"]  # , "DRUMS", "0"]
+        density_list = [3]  # , 2, 3]
+        # temperature_list = [0.7, 0.5, 0.75]
+        pass
+
+    # define generation directory
+    generated_sequence_files_path = define_generation_dir(model_repo)
+
+    # load model and tokenizer
+    model, tokenizer = LoadModel(
+        model_repo, from_huggingface=True
+    ).load_model_and_tokenizer()
+
+    # does the prompt make sense
+    check_if_prompt_inst_in_tokenizer_vocab(tokenizer, instrument_promt_list)
+
+    for temperature in Temperatures_to_try:
+        print(f"================= TEMPERATURE {temperature} =======================")
+        for _ in range(N_FILES_TO_GENERATE):
+            print(f"========================================")
+            # 1 - instantiate
+            generate_midi = GenerateMidiText(model, tokenizer)
+            # 0 - set the n_bar for this model
+            generate_midi.set_nb_bars_generated(n_bars=n_bar_generated)
+            # 1 - defines the instruments, densities and temperatures
+            # 2- generate the first 8 bars for each instrument
+            generate_midi.set_improvisation_level(30)
+            generate_midi.generate_piece(
+                instrument_promt_list,
+                density_list,
+                [temperature for _ in density_list],
+            )
+            # 3 - force the model to improvise
+            # generate_midi.set_improvisation_level(20)
+            # 4 - generate the next 4 bars for each instrument
+            # generate_midi.generate_n_more_bars(n_bar_generated)
+            # 5 - lower the improvisation level
+            generate_midi.generated_piece = (
+                generate_midi.get_whole_piece_from_bar_dict()
+            )
+
+            # print the generated sequence in terminal
+            print("=========================================")
+            print(generate_midi.generated_piece)
+            print("=========================================")
+
+            # write to JSON file
+            filename = WriteTextMidiToFile(
+                generate_midi,
+                generated_sequence_files_path,
+            ).text_midi_to_file()
+
+            # decode the sequence to MIDI """
+            decode_tokenizer = get_miditok()
+            TextDecoder(decode_tokenizer, USE_FAMILIZED_MODEL).get_midi(
+                generate_midi.generated_piece, filename=filename.split(".")[0] + ".mid"
+            )
+            inst_midi, mixed_audio = get_music(filename.split(".")[0] + ".mid")
+            max_time = get_max_time(inst_midi)
+            plot_piano_roll(inst_midi)
+
+            print("Et voilà! Your MIDI file is ready! GO JAM!")

generation_utils.py

@@ -0,0 +1,161 @@
+import os
+import numpy as np
+import matplotlib.pyplot as plt
+import matplotlib
+
+from constants import INSTRUMENT_CLASSES
+from playback import get_music, show_piano_roll
+
+# matplotlib settings
+matplotlib.use("Agg")  # for server
+matplotlib.rcParams["xtick.major.size"] = 0
+matplotlib.rcParams["ytick.major.size"] = 0
+matplotlib.rcParams["axes.facecolor"] = "none"
+matplotlib.rcParams["axes.edgecolor"] = "grey"
+
+
+def define_generation_dir(model_repo_path):
+    #### to remove later ####
+    if model_repo_path == "models/model_2048_fake_wholedataset":
+        model_repo_path = "misnaej/the-jam-machine"
+    #### to remove later ####
+    generated_sequence_files_path = f"midi/generated/{model_repo_path}"
+    if not os.path.exists(generated_sequence_files_path):
+        os.makedirs(generated_sequence_files_path)
+    return generated_sequence_files_path
+
+
+def bar_count_check(sequence, n_bars):
+    """check if the sequence contains the right number of bars"""
+    sequence = sequence.split(" ")
+    # find occurences of "BAR_END" in a "sequence"
+    # I don't check for "BAR_START" because it is not always included in "sequence"
+    # e.g. BAR_START is included the prompt when generating one more bar
+    bar_count = 0
+    for seq in sequence:
+        if seq == "BAR_END":
+            bar_count += 1
+    bar_count_matches = bar_count == n_bars
+    if not bar_count_matches:
+        print(f"Bar count is {bar_count} - but should be {n_bars}")
+    return bar_count_matches, bar_count
+
+
+def print_inst_classes(INSTRUMENT_CLASSES):
+    """Print the instrument classes"""
+    for classe in INSTRUMENT_CLASSES:
+        print(f"{classe}")
+
+
+def check_if_prompt_inst_in_tokenizer_vocab(tokenizer, inst_prompt_list):
+    """Check if the prompt instrument are in the tokenizer vocab"""
+    for inst in inst_prompt_list:
+        if f"INST={inst}" not in tokenizer.vocab:
+            instruments_in_dataset = np.sort(
+                [tok.split("=")[-1] for tok in tokenizer.vocab if "INST" in tok]
+            )
+            print_inst_classes(INSTRUMENT_CLASSES)
+            raise ValueError(
+                f"""The instrument {inst} is not in the tokenizer vocabulary. 
+                Available Instruments: {instruments_in_dataset}"""
+            )
+
+
+def forcing_bar_count(input_prompt, generated, bar_count, expected_length):
+    """Forcing the generated sequence to have the expected length
+    expected_length and bar_count refers to the length of newly_generated_only (without input prompt)"""
+
+    if bar_count - expected_length > 0:  # Cut the sequence if too long
+        full_piece = ""
+        splited = generated.split("BAR_END ")
+        for count, spl in enumerate(splited):
+            if count < expected_length:
+                full_piece += spl + "BAR_END "
+
+        full_piece += "TRACK_END "
+        full_piece = input_prompt + full_piece
+        print(f"Generated sequence trunkated at {expected_length} bars")
+        bar_count_checks = True
+
+    elif bar_count - expected_length < 0:  # Do nothing it the sequence if too short
+        full_piece = input_prompt + generated
+        bar_count_checks = False
+        print(f"--- Generated sequence is too short - Force Regeration ---")
+
+    return full_piece, bar_count_checks
+
+
+def get_max_time(inst_midi):
+    max_time = 0
+    for inst in inst_midi.instruments:
+        max_time = max(max_time, inst.get_end_time())
+    return max_time
+
+
+def plot_piano_roll(inst_midi):
+    piano_roll_fig = plt.figure(figsize=(25, 3 * len(inst_midi.instruments)))
+    piano_roll_fig.tight_layout()
+    piano_roll_fig.patch.set_alpha(0)
+    inst_count = 0
+    beats_per_bar = 4
+    sec_per_beat = 0.5
+    next_beat = max(inst_midi.get_beats()) + np.diff(inst_midi.get_beats())[0]
+    bars_time = np.append(inst_midi.get_beats(), (next_beat))[::beats_per_bar].astype(
+        int
+    )
+    for inst in inst_midi.instruments:
+        # hardcoded for now
+        if inst.name == "Drums":
+            color = "purple"
+        elif inst.name == "Synth Bass 1":
+            color = "orange"
+        else:
+            color = "green"
+
+        inst_count += 1
+        plt.subplot(len(inst_midi.instruments), 1, inst_count)
+
+        for bar in bars_time:
+            plt.axvline(bar, color="grey", linewidth=0.5)
+        octaves = np.arange(0, 128, 12)
+        for octave in octaves:
+            plt.axhline(octave, color="grey", linewidth=0.5)
+        plt.yticks(octaves, visible=False)
+
+        p_midi_note_list = inst.notes
+        note_time = []
+        note_pitch = []
+        for note in p_midi_note_list:
+            note_time.append([note.start, note.end])
+            note_pitch.append([note.pitch, note.pitch])
+        note_pitch = np.array(note_pitch)
+        note_time = np.array(note_time)
+
+        plt.plot(
+            note_time.T,
+            note_pitch.T,
+            color=color,
+            linewidth=4,
+            solid_capstyle="butt",
+        )
+        plt.ylim(0, 128)
+        xticks = np.array(bars_time)[:-1]
+        plt.tight_layout()
+        plt.xlim(min(bars_time), max(bars_time))
+        plt.ylim(max([note_pitch.min() - 5, 0]), note_pitch.max() + 5)
+        plt.xticks(
+            xticks + 0.5 * beats_per_bar * sec_per_beat,
+            labels=xticks.argsort() + 1,
+            visible=False,
+        )
+        plt.text(
+            0.2,
+            note_pitch.max() + 4,
+            inst.name,
+            fontsize=20,
+            color=color,
+            horizontalalignment="left",
+            verticalalignment="top",
+        )
+
+    return piano_roll_fig

load.py

@@ -0,0 +1,63 @@
+from transformers import GPT2LMHeadModel
+from transformers import PreTrainedTokenizerFast
+import os
+import torch
+
+
+class LoadModel:
+    """
+    Example usage:
+
+    # if loading model and tokenizer from Huggingface
+    model_repo = "misnaej/the-jam-machine"
+    model, tokenizer = LoadModel(
+        model_repo, from_huggingface=True
+    ).load_model_and_tokenizer()
+
+    # if loading model and tokenizer from a local folder
+    model_path = "models/model_2048_wholedataset"
+    model, tokenizer = LoadModel(
+        model_path, from_huggingface=False
+    ).load_model_and_tokenizer()
+
+    """
+
+    def __init__(self, path, from_huggingface=True, device="cpu", revision=None):
+        # path is either a relative path on a local/remote machine or a model repo on HuggingFace
+        if not from_huggingface:
+            if not os.path.exists(path):
+                print(path)
+                raise Exception("Model path does not exist")
+        self.from_huggingface = from_huggingface
+        self.path = path
+        self.device = device
+        self.revision = revision
+        if torch.cuda.is_available():
+            self.device = "cuda"
+
+    def load_model_and_tokenizer(self):
+        model = self.load_model()
+        tokenizer = self.load_tokenizer()
+
+        return model, tokenizer
+
+    def load_model(self):
+        if self.revision is None:
+            model = GPT2LMHeadModel.from_pretrained(self.path)  # .to(self.device)
+        else:
+            model = GPT2LMHeadModel.from_pretrained(
+                self.path, revision=self.revision
+            )  # .to(self.device)
+
+        return model
+
+    def load_tokenizer(self):
+        if self.from_huggingface:
+            pass
+        else:
+            if not os.path.exists(f"{self.path}/tokenizer.json"):
+                raise Exception(
+                    f"There is no 'tokenizer.json'file in the defined {self.path}"
+                )
+        tokenizer = PreTrainedTokenizerFast.from_pretrained(self.path)
+        return tokenizer

packages.txt

@@ -0,0 +1 @@
+fluidsynth

playback.py

@@ -0,0 +1,35 @@
+import matplotlib.pyplot as plt
+import librosa.display
+from pretty_midi import PrettyMIDI
+
+
+# Note: these functions are meant to be played within an interactive Python shell
+# Please refer to the synth.ipynb for an example of how to use them
+
+
+def get_music(midi_file):
+    """
+    Load a midi file and return the PrettyMIDI object and the audio signal
+    """
+    music = PrettyMIDI(midi_file=midi_file)
+    waveform = music.fluidsynth()
+    return music, waveform
+
+
+def show_piano_roll(music_notes, fs=100):
+    """
+    Show the piano roll of a music piece, with all instruments squashed onto a single 128xN matrix
+    :param music_notes: PrettyMIDI object
+    :param fs: sampling frequency
+    """
+    # get the piano roll
+    piano_roll = music_notes.get_piano_roll(fs)
+    print("Piano roll shape: {}".format(piano_roll.shape))
+
+    # plot the piano roll
+    plt.figure(figsize=(12, 4))
+    librosa.display.specshow(piano_roll, sr=100, x_axis="time", y_axis="cqt_note")
+    plt.colorbar()
+    plt.title("Piano roll")
+    plt.tight_layout()
+    plt.show()

playground.py

@@ -0,0 +1,195 @@
+import matplotlib.pyplot as plt
+import gradio as gr
+from load import LoadModel
+from generate import GenerateMidiText
+from constants import INSTRUMENT_CLASSES, INSTRUMENT_TRANSFER_CLASSES
+from decoder import TextDecoder
+from utils import get_miditok, index_has_substring
+from playback import get_music
+from matplotlib import pylab
+import sys
+import matplotlib
+from generation_utils import plot_piano_roll
+import numpy as np
+
+matplotlib.use("Agg")
+
+sys.modules["pylab"] = pylab
+
+model_repo = "JammyMachina/elec-gmusic-familized-model-13-12__17-35-53"
+n_bar_generated = 8
+# model_repo = "JammyMachina/improved_4bars-mdl"
+# n_bar_generated = 4
+
+model, tokenizer = LoadModel(
+    model_repo,
+    from_huggingface=True,
+).load_model_and_tokenizer()
+
+miditok = get_miditok()
+decoder = TextDecoder(miditok)
+
+
+def define_prompt(state, genesis):
+    if len(state) == 0:
+        input_prompt = "PIECE_START "
+    else:
+        input_prompt = genesis.get_whole_piece_from_bar_dict()
+    return input_prompt
+
+
+def generator(
+    label,
+    regenerate,
+    temp,
+    density,
+    instrument,
+    state,
+    piece_by_track,
+    add_bars=False,
+    add_bar_count=1,
+):
+
+    genesis = GenerateMidiText(model, tokenizer, piece_by_track)
+    track = {"label": label}
+    inst = next(
+        (
+            inst
+            for inst in INSTRUMENT_TRANSFER_CLASSES
+            if inst["transfer_to"] == instrument
+        ),
+        {"family_number": "DRUMS"},
+    )["family_number"]
+
+    inst_index = -1  # default to last generated
+    if state != []:
+        for index, instrum in enumerate(state):
+            if instrum["label"] == track["label"]:
+                inst_index = index  # changing if exists
+
+    # Generate
+    if not add_bars:
+        # Regenerate
+        if regenerate:
+            state.pop(inst_index)
+            genesis.delete_one_track(inst_index)
+
+            generated_text = (
+                genesis.get_whole_piece_from_bar_dict()
+            )  # maybe not useful here
+            inst_index = -1  # reset to last generated
+
+        # NEW TRACK
+        input_prompt = define_prompt(state, genesis)
+        generated_text = genesis.generate_one_new_track(
+            inst, density, temp, input_prompt=input_prompt
+        )
+
+        regenerate = True  # set generate to true
+    else:
+        # NEW BARS
+        genesis.generate_n_more_bars(add_bar_count)  # for all instruments
+        generated_text = genesis.get_whole_piece_from_bar_dict()
+
+    decoder.get_midi(generated_text, "mixed.mid")
+    mixed_inst_midi, mixed_audio = get_music("mixed.mid")
+
+    inst_text = genesis.get_selected_track_as_text(inst_index)
+    inst_midi_name = f"{instrument}.mid"
+    decoder.get_midi(inst_text, inst_midi_name)
+    _, inst_audio = get_music(inst_midi_name)
+    piano_roll = plot_piano_roll(mixed_inst_midi)
+    track["text"] = inst_text
+    state.append(track)
+
+    return (
+        inst_text,
+        (44100, inst_audio),
+        piano_roll,
+        state,
+        (44100, mixed_audio),
+        regenerate,
+        genesis.piece_by_track,
+    )
+
+
+def instrument_row(default_inst, row_id):
+    with gr.Row():
+        row = gr.Variable(row_id)
+        with gr.Column(scale=1, min_width=100):
+            inst = gr.Dropdown(
+                sorted([inst["transfer_to"] for inst in INSTRUMENT_TRANSFER_CLASSES])
+                + ["Drums"],
+                value=default_inst,
+                label="Instrument",
+            )
+            temp = gr.Dropdown(
+                [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1],
+                value=0.7,
+                label="Creativity",
+            )
+            density = gr.Dropdown([1, 2, 3], value=3, label="Note Density")
+
+        with gr.Column(scale=3):
+            output_txt = gr.Textbox(
+                label="output", lines=10, max_lines=10, show_label=False
+            )
+        with gr.Column(scale=1, min_width=100):
+            inst_audio = gr.Audio(label="TRACK Audio", show_label=True)
+            regenerate = gr.Checkbox(value=False, label="Regenerate", visible=False)
+            # add_bars = gr.Checkbox(value=False, label="Add Bars")
+            # add_bar_count = gr.Dropdown([1, 2, 4, 8], value=1, label="Add Bars")
+            gen_btn = gr.Button("Generate")
+            gen_btn.click(
+                fn=generator,
+                inputs=[row, regenerate, temp, density, inst, state, piece_by_track],
+                outputs=[
+                    output_txt,
+                    inst_audio,
+                    piano_roll,
+                    state,
+                    mixed_audio,
+                    regenerate,
+                    piece_by_track,
+                ],
+            )
+
+
+with gr.Blocks() as demo:
+    piece_by_track = gr.State([])
+    state = gr.State([])
+    title = gr.Markdown(
+        """ # Demo-App of The-Jam-Machine
+    A Generative AI trained on text transcription of MIDI music """
+    )
+    track1_md = gr.Markdown(""" ## Mixed Audio and Piano Roll """)
+    mixed_audio = gr.Audio(label="Mixed Audio")
+    piano_roll = gr.Plot(label="Piano Roll", show_label=False)
+    description = gr.Markdown(
+        """
+        For each **TRACK**, choose your **instrument** along with **creativity** (temperature) and **note density**. Then, hit the **Generate** Button!
+        You can have a look at the generated text; but most importantly, check the **piano roll** and listen to the TRACK audio!
+        If you don't like the track, hit the generate button to regenerate it! Generate more tracks and listen to the **mixed audio**!
+        """
+    )
+    track1_md = gr.Markdown(""" ## TRACK 1 """)
+    instrument_row("Drums", 0)
+    track1_md = gr.Markdown(""" ## TRACK 2 """)
+    instrument_row("Synth Bass 1", 1)
+    track1_md = gr.Markdown(""" ## TRACK 2 """)
+    instrument_row("Synth Lead Square", 2)
+    # instrument_row("Piano")
+
+demo.launch(debug=True)
+
+"""
+TODO: reset button
+TODO: add a button to save the generated midi
+TODO: add improvise button
+TODO: set values for temperature as it is done for density
+TODO: Add bar should be now set for the whole piece - regenerrate should regenerate the added bars only on all instruments
+TODO: row height to fix
+
+TODO: reset state of tick boxes after used maybe (regenerate, add bars) ; 
+TODO: block regenerate if add bar on
+"""

requirements.txt

@@ -0,0 +1,17 @@
+gradio
+matplotlib
+matplotlib
+numpy
+joblib
+pathlib
+transformers
+miditok == 1.3.2
+librosa
+pretty_midi
+pydub
+scipy
+datetime
+torch
+torchvision
+pyFluidSynth
+accelerate

utils.py

@@ -0,0 +1,246 @@
+from datetime import datetime
+from miditok import Event, MIDILike
+import os
+import json
+from time import perf_counter
+from joblib import Parallel, delayed
+from zipfile import ZipFile, ZIP_DEFLATED
+from scipy.io.wavfile import write
+import numpy as np
+from pydub import AudioSegment
+import shutil
+
+
+def writeToFile(path, content):
+    if type(content) is dict:
+        with open(f"{path}", "w") as json_file:
+            json.dump(content, json_file)
+    else:
+        if type(content) is not str:
+            content = str(content)
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "w") as f:
+            f.write(content)
+
+
+# Function to read from text from txt file:
+def readFromFile(path, isJSON=False):
+    with open(path, "r") as f:
+        if isJSON:
+            return json.load(f)
+        else:
+            return f.read()
+
+
+def chain(input, funcs, *params):
+    res = input
+    for func in funcs:
+        try:
+            res = func(res, *params)
+        except TypeError:
+            res = func(res)
+    return res
+
+
+def to_beat_str(value, beat_res=8):
+    values = [
+        int(int(value * beat_res) / beat_res),
+        int(int(value * beat_res) % beat_res),
+        beat_res,
+    ]
+    return ".".join(map(str, values))
+
+
+def to_base10(beat_str):
+    integer, decimal, base = split_dots(beat_str)
+    return integer + decimal / base
+
+
+def split_dots(value):
+    return list(map(int, value.split(".")))
+
+
+def compute_list_average(l):
+    return sum(l) / len(l)
+
+
+def get_datetime():
+    return datetime.now().strftime("%Y%m%d_%H%M%S")
+
+
+def get_text(event):
+    match event.type:
+        case "Piece-Start":
+            return "PIECE_START "
+        case "Track-Start":
+            return "TRACK_START "
+        case "Track-End":
+            return "TRACK_END "
+        case "Instrument":
+            return f"INST={event.value} "
+        case "Bar-Start":
+            return "BAR_START "
+        case "Bar-End":
+            return "BAR_END "
+        case "Time-Shift":
+            return f"TIME_SHIFT={event.value} "
+        case "Note-On":
+            return f"NOTE_ON={event.value} "
+        case "Note-Off":
+            return f"NOTE_OFF={event.value} "
+        case _:
+            return ""
+
+
+def get_event(text, value=None):
+    match text:
+        case "PIECE_START":
+            return Event("Piece-Start", value)
+        case "TRACK_START":
+            return None
+        case "TRACK_END":
+            return None
+        case "INST":
+            return Event("Instrument", value)
+        case "BAR_START":
+            return Event("Bar-Start", value)
+        case "BAR_END":
+            return Event("Bar-End", value)
+        case "TIME_SHIFT":
+            return Event("Time-Shift", value)
+        case "TIME_DELTA":
+            return Event("Time-Shift", to_beat_str(int(value) / 4))
+        case "NOTE_ON":
+            return Event("Note-On", value)
+        case "NOTE_OFF":
+            return Event("Note-Off", value)
+        case _:
+            return None
+
+
+# TODO: Make this singleton
+def get_miditok():
+    pitch_range = range(0, 140)  # was (21, 109)
+    beat_res = {(0, 400): 8}
+    return MIDILike(pitch_range, beat_res)
+
+
+class WriteTextMidiToFile:  # utils saving to file
+    def __init__(self, generate_midi, output_path):
+        self.generated_midi = generate_midi.generated_piece
+        self.output_path = output_path
+        self.hyperparameter_and_bars = generate_midi.piece_by_track
+
+    def hashing_seq(self):
+        self.current_time = get_datetime()
+        self.output_path_filename = f"{self.output_path}/{self.current_time}.json"
+
+    def wrapping_seq_hyperparameters_in_dict(self):
+        # assert type(self.generated_midi) is str, "error: generate_midi must be a string"
+        # assert (
+        #     type(self.hyperparameter_dict) is dict
+        # ), "error: feature_dict must be a dictionnary"
+        return {
+            "generate_midi": self.generated_midi,
+            "hyperparameters_and_bars": self.hyperparameter_and_bars,
+        }
+
+    def text_midi_to_file(self):
+        self.hashing_seq()
+        output_dict = self.wrapping_seq_hyperparameters_in_dict()
+        print(f"Token generate_midi written: {self.output_path_filename}")
+        writeToFile(self.output_path_filename, output_dict)
+        return self.output_path_filename
+
+
+def get_files(directory, extension, recursive=False):
+    """
+    Given a directory, get a list of the file paths of all files matching the
+    specified file extension.
+    directory: the directory to search as a Path object
+    extension: the file extension to match as a string
+    recursive: whether to search recursively in the directory or not
+    """
+    if recursive:
+        return list(directory.rglob(f"*.{extension}"))
+    else:
+        return list(directory.glob(f"*.{extension}"))
+
+
+def timeit(func):
+    def wrapper(*args, **kwargs):
+        start = perf_counter()
+        result = func(*args, **kwargs)
+        end = perf_counter()
+        print(f"{func.__name__} took {end - start:.2f} seconds to run.")
+        return result
+
+    return wrapper
+
+
+class FileCompressor:
+    def __init__(self, input_directory, output_directory, n_jobs=-1):
+        self.input_directory = input_directory
+        self.output_directory = output_directory
+        self.n_jobs = n_jobs
+
+    # File compression and decompression
+    def unzip_file(self, file):
+        """uncompress single zip file"""
+        with ZipFile(file, "r") as zip_ref:
+            zip_ref.extractall(self.output_directory)
+
+    def zip_file(self, file):
+        """compress a single text file to a new zip file and delete the original"""
+        output_file = self.output_directory / (file.stem + ".zip")
+        with ZipFile(output_file, "w") as zip_ref:
+            zip_ref.write(file, arcname=file.name, compress_type=ZIP_DEFLATED)
+            file.unlink()
+
+    @timeit
+    def unzip(self):
+        """uncompress all zip files in folder"""
+        files = get_files(self.input_directory, extension="zip")
+        Parallel(n_jobs=self.n_jobs)(delayed(self.unzip_file)(file) for file in files)
+
+    @timeit
+    def zip(self):
+        """compress all text files in folder to new zip files and remove the text files"""
+        files = get_files(self.output_directory, extension="txt")
+        Parallel(n_jobs=self.n_jobs)(delayed(self.zip_file)(file) for file in files)
+
+
+def load_jsonl(filepath):
+    """Load a jsonl file"""
+    with open(filepath, "r") as f:
+        data = [json.loads(line) for line in f]
+    return data
+
+
+def write_mp3(waveform, output_path, bitrate="92k"):
+    """
+    Write a waveform to an mp3 file.
+    output_path: Path object for the output mp3 file
+    waveform: numpy array of the waveform
+    bitrate: bitrate of the mp3 file (64k, 92k, 128k, 256k, 312k)
+    """
+    # write the wav file
+    wav_path = output_path.with_suffix(".wav")
+    write(wav_path, 44100, waveform.astype(np.float32))
+    # compress the wav file as mp3
+    AudioSegment.from_wav(wav_path).export(output_path, format="mp3", bitrate=bitrate)
+    # remove the wav file
+    wav_path.unlink()
+
+
+def copy_file(input_file, output_dir):
+    """Copy an input file to the output_dir"""
+    output_file = output_dir / input_file.name
+    shutil.copy(input_file, output_file)
+
+
+def index_has_substring(list, substring):
+    for i, s in enumerate(list):
+        if substring in s:
+            return i
+    return -1