<jupyter_start><jupyter_text>library<jupyter_code>import numpy as np
import numpy as np
import numpy
from miditoolkit.midi import parser as mid_parser  
from miditoolkit.midi import containers as ct
from numpy import array, linspace
from sklearn.neighbors import KernelDensity
from matplotlib.pyplot import plot
from scipy.signal import argrelextrema
from scipy.ndimage import gaussian_filter1d
import random
import os<jupyter_output><empty_output><jupyter_text>skyline (helper functions)<jupyter_code>def mergeIntervals(arr):
        # Sorting based on the increasing order 
        # of the start intervals
        arr.sort(key = lambda x: x[0]) 
        # array to hold the merged intervals
        m = []
        s = -10000
        max = -100000
        for i in range(len(arr)):
            a = arr[i]
            if a[0] > max:
                if i != 0:
                    m.append([s,max])
                max = a[1]
                s = a[0]
            else:
                if a[1] >= max:
                    max = a[1]        
        #'max' value gives the last point of 
        # that particular interval
        # 's' gives the starting point of that interval
        # 'm' array contains the list of all merged intervals
        if max != -100000 and [s, max] not in m:
            m.append([s, max])
        return m

def gettop(note,intervals):
    note_interval = [note[4],note[5]]#onset,offset
    overlap_time = 0
    total_time = note[5] - note[4]
    if total_time == 0:
        return 1 #(we do not need this note)
    for interval in intervals:
        maxstart = max(note_interval[0],interval[0])
        minend = min(note_interval[1],interval[1])
        if maxstart < minend:
            overlap_time += minend-maxstart
    return overlap_time/total_time

def skyline(notes): #revised skyline algorithm by Chai, 2000
    #Performed on a single channel
    accepted_notes = []
    notes = sorted(notes, key=lambda x: x[2], reverse=True) #sort by pitch
    intervals = []
    for note in notes:
        if gettop(note,intervals) <=0.5:
            accepted_notes.append(note)
            intervals.append([note[4],note[5]]) #onset,offset
            intervals = mergeIntervals(intervals)
    return sorted(accepted_notes,key=lambda x: (x[4],x[0])) #sort by onset & bar(new)
    
def skyline_reverse(notes): #revised skyline algorithm by Chai, 2000
    #Performed on a single channel
    accepted_notes = []
    notes = sorted(notes, key=lambda x: x[2]) #sort by pitch
    intervals = []
    for note in notes:
        if gettop(note,intervals) <=0.8:
            accepted_notes.append(note)
            intervals.append([note[4],note[5]]) #onset,offset
            intervals = mergeIntervals(intervals)
    return sorted(accepted_notes,key=lambda x: (x[4],x[0])) #sort by onset & bar(new)

def align_token(notes,length):                                   #align the tokens bar by bar
    out=[]
    bar=[]
    bar_count=0
    seen_first=False
    tpb=480
    note_idx=0
    
    while note_idx<len(notes):
        note=notes[note_idx]
        if bar_count*4*tpb<=note[4]<(bar_count+1)*tpb*4:         #within current bar
            bar.append(note[:4])
            if (seen_first==True and note[0]==0):
                print(note,note_idx,notes)
            assert(not(seen_first==True and note[0]==0))         #ASSERT: no two 0(newbar) within the same bar
            if not seen_first :
                seen_first=True
                bar[-1][0]=0
            note_idx+=1
        else:            
            #assert(len(bar)>0)
            if len(bar)>0:                                       # add <ABS> if it is an empty bar
                out.append(bar)
            else:
                out.append([list(ABS)])
            bar=[]
            bar_count+=1
            seen_first=False
    

    if len(bar)>0:                                               # add <ABS> if it is an empty bar
        out.append(bar)
    else:
        out.append([list(ABS)])

    bar=[]
    bar_count+=1
    seen_first=False
    

    assert(bar_count==length)
    return out<jupyter_output><empty_output><jupyter_text>README tokenlize all .mid files inside a foldersource data: PianoMidi_nicely_formatted cd ~/prepare_data/CP python main.py --task skyline --input_dir ../../skyline_data --output_dir ../../skyline_data --name skylineNPY --dict ../../dict/CP_skyline.pkl<jupyter_code>tokens=np.load('./skyline_data/skylineNPY.npy')
tokens.shape<jupyter_output><empty_output><jupyter_text>test datatokens=np.array([  ([[0,0,40,0]]+[[1,0,40,0]])*256  ,([[0,0,40,0]]+[[1,0,40,0]])*255+[[4,18,88,66],[2,16,86,64]] ])tokens.shape<jupyter_code>#ref to ./dict/CP_skyline.pkl
PAD=np.array([2,16,86,64])                                                                               # --> padding
EOS=np.array([4,18,88,66])                                                                               # --> End of input segment
ABS=np.array([5,19,89,67])                                                                               #--> empty bar produced by the skyline algo, 
                                                                                                         #    (e.g. skyline pick a long note from the bar ahead)
tokens_by_song=[]                                                                                        #group tokens by song, ragged array in shape:(N,) 
last_idx=0
for idx,page in enumerate(tokens):
    if (page[-1] == PAD).all() or (page[-1] == EOS).all():
        tokens_by_song.append(tokens[last_idx:idx+1])
        last_idx=idx+1
'''
Repackage the tokens:

const: (pick randomly)
   skyline_max_len: 100
   full_max_len: 600 <-- this may need to be adjusted if the data is orchestra score (i guess around 1200?)
   * provided that the full_max_len encoutered so far is around 500 in PianoMidi_nively_formatted dataset

input:
    tokens_by_song:           Ragged integer array in shape(#song, ) 
    
output:
    allsong_skyline_tokens    Integer array in shape(-1,skyline_max_len,4)
    allsong_full_tokens       Integer array in shape(-1,full_max_len,4)
    
description:
    Assumption in this section:
    * Let 0<=N<len(allsong_skyline_tokens)
    * assert(len(allsong_skyline_tokens) == len(allsong_full_tokens))
    
    allsong_skyline_tokens is paired with allsong_full_tokens
        - e.g. allsong_skyline_tokens[N] and allsong_full_tokens[N] represent the same section in the same score
        - IF allsong_skyline_tokens[N] contains 10 bars, allsong_full_tokens[N] will only contain the same (10) bars
          even if it has plenty of space lefts
          
    Two extra tokens:
        - <EOS>[4,18,88,66]: indicate the end of allsong_skyline_tokens[N] or allsong_full_tokens[N]
        - <ABS>[5,19,89,67]: indicate a empty bar
        - <PAD>[2,16,86,64]: indicate a pad

example:
    input:
          [
             [[0,0,20,10],[1,0,10,10],[1,0,90,10],[0,0,50,0]]
          ]
    
    output:
        allsong_skyline_tokens=[
                                    [[0,0,10,10],[1,0,90,10],[5,19,89,67],[4,18,88,66],[2,16,86,64]................. ]
                                ]
        allsong_full_tokens=[
                                   [[0,0,20,10],[1,0,10,10],[1,0,90,10],[0,0,50,0],[4,18,88,66],[2,16,86,64].................] 
                            ]
                            
implementation:
    1. add onset & offset to each token
    2. reuse the skyline&skyline_reverse functions
    3. combine the results and filter out duplicated tokens AS skyline_tokens
    4. align the skyline_tokens and the original tokens bar by bar
        skyine_tokens = remaining notes after skyline algorithm
        full_tokens = all original notes
    3. DO
            DO
            Let N = the maximum number of subsequence bar such that the total #tokens do not exceed "skyline_max_len"
            3.1 extract all skyline_tokens within that N bars into "temp_skyline"
            3.2 extract all tokens within that N bars into "temp_full" with assertion (#token <"full_max_len")
            3.3 add <EOS> to "temp_skyline" and "temp_full"
            3.3 pad both "temp_skyline" and "temp_full"
            3.4 put "temp_skyline" and "temp_full" into "allsong_skyline_tokens" and "allsong_full_tokens"
            3.4 repeat until all tokens inside the song are processed
        repeat until all songs are processed
'''


skyline_max_len=100                                                                                      # parameters
full_max_len=600
temp_skyline=[]
allsong_skyline_tokens=[]
allsong_full_tokens=[]
max_token_len=0

for song in tokens_by_song:     
    
    current_bar=-1                                                                                       #add onset &offset on each token
    tpb=480
    token_with_on_off_set=[]
    skyline_tokens=[]
    full_tokens=[]
    for page in song:
        for token in page:
            if not((token==PAD).all() or (token==EOS).all()):
                if token[0]==0:
                    current_bar+=1
                temp=list(token)
                temp.append(int(current_bar*4*tpb+token[1]*tpb/4))                                        #onset
                temp.append(int(current_bar*4*tpb+token[1]*tpb/4+(token[3]+1)*tpb/8))                     #offset
                token_with_on_off_set.append(temp)
                
    
    
    total_bar=current_bar+1                                                                                #skyline
    org=align_token(token_with_on_off_set,total_bar)
    sl=skyline(token_with_on_off_set)+skyline_reverse(token_with_on_off_set)
    
    sl = [tuple(x) for x in sl]                                                                            #remove duplication
    sl = list(dict.fromkeys(sl))
    sl = [list(x) for x in sl]
    sl=sorted(sl,key=lambda x: (x[4],x[0])) #sort by onset & bar(new)
    sl=align_token(sl,total_bar)
    
    
    current_bar=0
    temp_skyline=[]
    temp_full=[]
    while current_bar<total_bar:
        while current_bar<total_bar and len(temp_skyline)+len(sl[current_bar])<skyline_max_len:
            temp_skyline+=sl[current_bar]
            temp_full+=org[current_bar]
            current_bar+=1
        assert(0<len(temp_skyline)<skyline_max_len and 0<len(temp_full)<full_max_len )                     #at least it shld has the ABS token
        
        temp_skyline.append(EOS)                                                                           #add EOS
        temp_full.append(EOS)
        temp_skyline=np.array(temp_skyline).reshape(-1,4)
        temp_full=np.array(temp_full).reshape(-1,4)
        
        while len(temp_skyline)<skyline_max_len:                                                           #add PAD       
            temp_skyline=np.vstack((temp_skyline,PAD))
        if len(temp_full)>max_token_len:
            max_token_len=len(temp_full)
        while len(temp_full)<full_max_len:
            temp_full=np.vstack((temp_full,PAD))
        skyline_tokens.append(temp_skyline)
        full_tokens.append(temp_full)
        temp_skyline=[]
        temp_full=[]    
    assert(len(allsong_skyline_tokens)==len(allsong_full_tokens))    
    
    for batch in skyline_tokens:                                                                            #stack all pages together
        allsong_skyline_tokens.append(batch)
    for batch in full_tokens:
        allsong_full_tokens.append(batch)
allsong_skyline_tokens=np.array(allsong_skyline_tokens)
allsong_full_tokens=np.array(allsong_full_tokens)        
assert(allsong_skyline_tokens.shape[0]==allsong_full_tokens.shape[0])
max_token_len
allsong_skyline_tokens.shape,allsong_full_tokens.shape
with open('skyline_data/skyline_tokens.npy', 'wb') as f1:
    np.save(f1, allsong_skyline_tokens)
with open('skyline_data/full_tokens.npy', 'wb') as f2:
    np.save(f2, allsong_full_tokens)<jupyter_output><empty_output><jupyter_text>Testing (reconstruction)<jupyter_code>'''
 simply to reconstruct a .mid file from any page inside skyline_tokens & full_tokens
'''

allsong_skyline_tokens=np.load('skyline_data/skyline_tokens.npy')
allsong_full_tokens=np.load('skyline_data/full_tokens.npy')
first_skyline = allsong_skyline_tokens[1]
first_full = allsong_full_tokens[1]
first_skyline.shape, first_full.shape
def token2mid(page,out_path):
    out = mid_parser.MidiFile()
    out.ticks_per_beat = 480
    out.instruments = [ct.Instrument(program=0,is_drum=False,name='reduction')]
    current_beat=-1
    for idx,token in enumerate(page):
        if (token==EOS).all():
            break
        assert((token!=PAD).all())
        if token[0]==0 or (token==ABS).all():
            current_beat+=1
        if (token!=ABS).all():
            n=token
            out.instruments[0].notes.append(ct.Note(start=int(current_beat*4*out.ticks_per_beat+n[1]*out.ticks_per_beat/4),
                                                        end=int(current_beat*4*out.ticks_per_beat+n[1]*out.ticks_per_beat/4+(n[3]+1)*(out.ticks_per_beat/8)),
                                                        pitch=n[2]+22,
                                                        velocity=90))
    out.dump(out_path)
token2mid(first_skyline,'skylineFromToken.mid')
token2mid(first_full,'fullFromToken.mid')<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10659670164917541, 'fraction_numerical': 0.027661169415292355, 'mean_word_length': 1.9719313878369347, 'pattern_counts': {'":': 0, '<': 46, '<?xml version=': 0, '>': 36, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 54, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '203697', 'license': 'MIT', 'n_tokens_mistral': 4282, 'n_tokens_neox': 4046, 'n_words': 880, 'path': '.ipynb_checkpoints/skyline+tokenlize-checkpoint.ipynb', 'repo_name': 'tokaho/MIDI-BERT', 'size': '13340'}