Adapt files other than sound_process to accomodate changes

compare_plot.py

@@ -8,10 +8,9 @@ def compare_plot():
     beatmap = sl.Beatmap.from_path(filename)
     timing = beatmap.timing_points[0]
     bpm = timing.bpm
-    offset = timing.offset.total_seconds() * 1000
+    offset = timing.offset
-    data = sound_process.process_song(beatmap.audio_filename, bpm, offset0=offset, n_iter_2=48, divisor=4)
+    timings, amplitudes = sound_process.process_song(beatmap.audio_filename, bpm, offset=offset, n_iter_2=-1)
-    
+    timings = [x.total_seconds() for x in timings]
-    timings, amplitudes, freqs = [x[0].total_seconds() for x in data], [x[1] for x in data], [x[2] for x in data]
 
     original_times = [x.time.total_seconds() for x in beatmap.hit_objects(spinners=False) if x.time.total_seconds() <= timings[len(timings) - 1]]
 

main.py

@@ -10,15 +10,13 @@ def main():
     beatmap = sl.Beatmap.from_path(filename)
     timing = beatmap.timing_points[0]
     bpm = timing.bpm
-    offset = timing.offset.total_seconds() * 1000
+    offset = timing.offset
     print(beatmap.audio_filename)
 
-    data = sound_process.process_song(beatmap.audio_filename, int(bpm), offset0=offset, n_iter_2=-1)
+    timings, amplitudes = sound_process.process_song(beatmap.audio_filename, bpm, offset=offset, n_iter_2=-1)
     # NOTE : remove n_iter_2 to map the whole music
-    timings, amplitudes, freqs = [x[0] for x in data], [x[1] for x in data], [x[2] for x in data]
 
     beatmap._hit_objects = place.greedy(bpm, offset, timings, amplitudes)
-    beatmap.display_name = "TIPE's Extra"
     #beatmap._hit_objects = [sl.Slider(sl.Position(0, 0), timedelta(milliseconds=3), timedelta(milliseconds=130), 0, sl.curve.Linear([sl.Position(0, 0), sl.Position(100, 100)], 100), 100, 2, 1, 1, 1, timing.ms_per_beat, [], [],)]
     beatmap.write_path("rewrite.osu")
     

sound_process.py

@@ -15,6 +15,8 @@ from pathlib import Path
 from time import sleep
 from datetime import timedelta
 
+WORKING_SAMPLE_RATE = 1000
+
 print("Starting...\n")
 
 def filter_n_percent_serial(song_name, offset, n_iter, step, threshold):
@@ -28,6 +30,8 @@ def filter_n_percent_serial(song_name, offset, n_iter, step, threshold):
     filter data associated with song_name to keep only the highest threshold% values
     """
 
+    offset = offset.total_seconds()
+
     subprocess.run(["ffmpeg", "-ss", str(offset), "-t", str(offset+step*n_iter), "-i", song_name, "crop.wav"]) 
 
     sample_rate, global_data = wavfile.read('crop.wav')
@@ -94,12 +98,12 @@ def get_freq(song_name, offset, step, songlen, data, display=False):
     """
     for a given list of amplitudes, returns the corresponding peak frequencies
     """
+    offset = offset.total_seconds()
     fft_list = []
     times = []
     current_time = offset
     k = 0
-
+    subprocess.run(["ffmpeg", "-ss", str(offset), "-t", str(offset+songlen),"-i",song_name, "crop.wav"]) 
-    subprocess.run(["ffmpeg", "-ss", str(offset), "-t", str(offset+songlen), "-i", song_name, "crop.wav"]) 
 
     sample_rate, global_data = wavfile.read("crop.wav")
     #blit = int(len(global_data) / len(data))
@@ -118,7 +122,7 @@ def get_freq(song_name, offset, step, songlen, data, display=False):
 
     for s in range(len(data)):
         if(data[s] != 0):
-            pff = scipy.fft.rfft(global_data[int(s*len(global_data)/len(data)):int(44100*step+int(s*len(global_data)/len(data)))])
+            pff = scipy.fft.rfft(global_data[int(s*len(global_data)/len(data)):int(WORKING_SAMPLE_RATE*step+int(s*len(global_data)/len(data)))])
             
             mx = max(np.abs(pff))
             for id in range(len(pff)):
@@ -167,6 +171,8 @@ def void_freq(song_name, offset, songlen, increment, minfreq, maxfreq, upperthr,
     write : bool (should be set to True)
     output_file : technical
     """
+    offset = offset.total_seconds()
+
     fft_list = []
     times = []
     current_time = offset
@@ -257,7 +263,7 @@ def void_freq(song_name, offset, songlen, increment, minfreq, maxfreq, upperthr,
             res[i] = np.int16(32767*res[i]/mx)
 
         res = np.array(res)
-        wavfile.write(output_file, 44100, res)
+        wavfile.write(output_file, WORKING_SAMPLE_RATE, res)
 
     #plt.plot(np.abs(pfreq[:len(fft_list[0])]), np.abs(fft_list[0]))
     #plt.grid()
@@ -418,9 +424,10 @@ def snap2(data, sample_rate, bpm, first_offset=0, div=4, show=False, adjust=Fals
 
     
     if(show):
-        new2 = [0.9 if new[i] != 0 else 0 for i in range(len(new))]
 
         t = [j/1000+first_offset for j in range(len(new))]
+        scatter_t = [t[i] for i in range(len(new)) if new[i] != 0]
+        scatter_chosen_rhythm = [0.9 for i in range(len(new)) if new[i] != 0 ]
         beats_1 = [0 for j in range(len(new))]
         beats_2 = [0 for k in range(len(new))]
         beats_4 = [0 for l in range(len(new))]
@@ -436,40 +443,62 @@ def snap2(data, sample_rate, bpm, first_offset=0, div=4, show=False, adjust=Fals
             beats_4[int(1000*current_t)] = 0.9
 
             if(k % 2 == 0):
-                beats_2[int(1000*current_t)] = 0.92
+                beats_2[int(1000*current_t)] = 0.902
 
             if(k % 4 == 0):
-                beats_1[int(1000*current_t)] = 0.94
+                beats_1[int(1000*current_t)] = 0.91
+
+            if(k % 16 == 0):
+                beats_1[int(1000*current_t)] = 0.915
 
             k += 1
             current_t = first_offset + k*60/(bpm*div)
 
-        plt.plot(t, new2, "bo")
+        points = plt.scatter(scatter_t, scatter_chosen_rhythm, marker="o", label="Detected Rhythm")
-        plt.plot(t, beats_4, "r-")
+        div1_plot, = plt.plot(t, beats_4, "b-", label="1/4")
-        plt.plot(t, beats_2, "y-")
+        div2_plot, = plt.plot(t, beats_2, "r-", label="1/2")
-        plt.plot(t, beats_1, "g-")
+        div3_plot, = plt.plot(t, beats_1, "black", label="1/1")
         plt.xlabel("Time (s)")
-        plt.ylabel("Amplitude")
+        #plt.ylabel("Amplitude")
+        plt.legend(handles=[points, div1_plot, div2_plot, div3_plot])
         plt.grid()
         plt.show()
 
     return new
 
-def convert_to_wav(song_name:str, output_file="audio.wav") -> str:
+def convert_song(song_name:str, output_file="audio.wav") -> str:
     """
-    Converts the song to .wav, only if it's not already in wave format.
+    Converts the song to .wav AND lower its sample rate to 1000.
-    Currently relies on file extension.
     Returns: the song_name that should be used afterwards.
     """
-    extension = Path(song_name).suffix
+    subprocess.run(["ffmpeg", "-y", "-i", song_name, "-ar", "1000", output_file]) 
-    match extension:
-        case ".mp3" | ".ogg":
-            print("Converting to .wav...")
-            subprocess.run(["ffmpeg", "-y", "-i", song_name, output_file]) 
     return output_file
-    return song_name
 
-def process_song(filename, bpm, offset0=0, div_len_factor=1, n_iter_2=-1, threshold=0.5, divisor=4):
+def quantify(time: timedelta, bpm, offset, snapping):
+    """
+    Input: timedelta, bpm, offset, and snapping divisor (2 for 1/2, etc...)
+    Returns a timedelta that is properly timed to the map.
+    """
+    offset_ms = offset.total_seconds() / 1000
+    time_ms = time.total_seconds() * 1000
+    time_spacing = (60000/bpm)/snapping
+    beats_away = round((time_ms - offset_ms)/time_spacing)
+    new_time = timedelta(milliseconds=time_spacing*beats_away + offset_ms)
+    return new_time
+
+def quantify_all(amplitudes_ugly, bpm, offset_ms, divisor):
+    n = len(amplitudes_ugly)
+    covered = [False] * n
+    times = []
+    amplitudes = []
+    for i in range(n):
+        if amplitudes_ugly[i] != 0 and not covered[i]:
+            times.append(quantify(timedelta(milliseconds=i), bpm, offset_ms, divisor))
+            amplitudes.append(amplitudes_ugly[i])
+            covered[i] = True
+    return times, amplitudes
+
+def process_song(filename, bpm, offset=timedelta(milliseconds=0), div_len_factor=1, n_iter_2=-1, threshold=0.5, divisor=4):
     """
     filename : string (name of the song)
     offset : int [+] (song mapping will start from this time in seconds, default is 0)
@@ -480,9 +509,7 @@ def process_song(filename, bpm, offset0=0, div_len_factor=1, n_iter_2=-1, thresh
     divisor : int [+] (beat divisor used to snap the notes, default is 4)
     """
 
-    filename = convert_to_wav(filename)
+    filename = convert_song(filename)
-
-    offset = offset0/1000
 
     div_len = div_len_factor*60/bpm-0.01
 
@@ -490,18 +517,19 @@ def process_song(filename, bpm, offset0=0, div_len_factor=1, n_iter_2=-1, thresh
     song_len = get_songlen(filename)
 
     if(n_iter == -1):
-        n_iter = int((song_len-offset/1000)/div_len)-1
+        n_iter = floor((song_len-offset.total_seconds())/div_len)-1
 
     filtered_name = f"{filename}_trimmed.wav"
 
-    void_freq(filename, offset, min(song_len, offset+div_len*(n_iter+1)+0.01), 4*60/bpm, minfreq=0, maxfreq=220, upperthr=5000, ampthr=60, ampfreq = 1200, ampval = 5.0, leniency = 0.005, write=True, linear=False, output_file=filtered_name)
+    void_freq(filename, offset, min(song_len, offset.total_seconds()+div_len*(n_iter+1)+0.01), 4*60/bpm, minfreq=0, maxfreq=220, upperthr=5000, ampthr=60, ampfreq = 1200, ampval = 5.0, leniency = 0.005, write=True, linear=False, output_file=filtered_name)
     #void_freq(filename, offset, offset+div_len*(n_iter+1)+0.01, 4*60/bpm, minfreq=0, maxfreq=330, upperthr=2500, ampthr=60, ampfreq = 1200, ampval = 1/2000, leniency = 0.0, write=True, linear=True, output_file=filtered_name)
-    datares = filter_n_percent_serial(filtered_name, offset, n_iter, div_len, threshold) 
+    amplitudes_ugly = filter_n_percent_serial(filtered_name, offset, n_iter, div_len, threshold) 
-    #datares = snap(datares, 44100, bpm, 4, True)
+    #datares = snap(datares, WORKING_SAMPLE_RATE, bpm, 4, True)
-    datares = snap2(datares, 44100, bpm, first_offset=offset, div=divisor, show=True, adjust=True)
+    times, amplitudes = quantify_all(amplitudes_ugly, bpm, offset, divisor)
-    frequencies = get_freq(filtered_name, offset, div_len, div_len*n_iter, datares, True)
+    #frequencies = get_freq(filtered_name, offset, div_len, div_len*n_iter, datares, True)
     Path(f"{filename}_trimmed.wav").unlink()
-    return convert_tuple(datares, frequencies)
+    #return convert_tuple(datares, frequencies)
+    return times, amplitudes
 
 def main():
     data = process_song("tetris_4.wav", 160, n_iter_2=48, threshold=100)
@@ -559,7 +587,7 @@ if(False):
     #t, f, Zxx = fct("deltamax.wav", 9.992, 0.032, 20, 3000, False)
     #t, f, Zxx = fct("da^9.wav", 8.463, 0.032, 20, 5000, False)
     t, f, Zxx = fct("13.  Cosmic Mind.wav", 0, 0.032, 20, 5000, False)
-    #t, f, Zxx = fct("Furioso Melodia 44100.wav", 4, 0.032, 8, 3000, False)
+    #t, f, Zxx = fct("Furioso Melodia WORKING_SAMPLE_RATE.wav", 4, 0.032, 8, 3000, False)
     #t, f, Zxx = fct("changing.wav", 0, 0.05, 3.9, 5000, False)
     #fct("worlds_end_3.wav", 75, (60/178)/4, 75+2, 2500)
 
@@ -570,7 +598,7 @@ if(False):
     (t, data) = peaks("worlds_end_3.wav", 74.582, 6, False, 0.9)
     #(t, data) = peaks("da^9.wav", 8.463, 301.924 - 8.463, False, 0.95)
     #(t, data) = peaks("deltamax.wav", 8.463, 30101.924 - 8.463, False, 0.92)
-    da = find_bpm(t, 44100, data, 100, 200, 1, 10)
+    da = find_bpm(t, WORKING_SAMPLE_RATE, data, 100, 200, 1, 10)
     print("BPM data is", da)'''
 
     #data = [-1 for i in range(int(x))]
@@ -772,6 +800,7 @@ def extract_peaks_v2(song_data, sample_rate, offset, display, threshold, seglen)
     return (t, song_data)
 
 def peaks(song_name, offset, length, display, thr):
+    offset = offset.total_seconds()
     subprocess.run(["ffmpeg", "-ss", str(offset), "-t", str(length), "-i", song_name, "crop.wav"]) 
 
     sample_rate, audio_data = wavfile.read('crop.wav')
@@ -780,7 +809,7 @@ def peaks(song_name, offset, length, display, thr):
     subprocess.run(["rm", "crop.wav"])
 
     #return extract_peaks(audio_data, sample_rate, offset, display, thr)
-    return extract_peaks_v2(audio_data, sample_rate, offset, display, thr, 44100*2)
+    return extract_peaks_v2(audio_data, sample_rate, offset, display, thr, WORKING_SAMPLE_RATE*2)
 
 def find_bpm(sample_rate, data, minbpm, maxbpm, step, width):
     optimal = minbpm
@@ -982,6 +1011,8 @@ def filter_n_percent(song_name, offset, length, threshold, reduce, show):
     # threshold is in ]0, 100]
     # filter data associated with song_name to keep only the highest threshold% values
 
+    offset = offset.total_seconds()
+
     subprocess.run(["ffmpeg", "-ss", str(offset), "-t", str(length), "-i", song_name, "crop.wav"]) 
 
     sample_rate, song_data = wavfile.read('crop.wav')
@@ -990,7 +1021,7 @@ def filter_n_percent(song_name, offset, length, threshold, reduce, show):
     subprocess.run(["rm", "crop.wav"])
 
     if(reduce):
-        (song_data,e) = to_ms(song_data, 44100, 1)
+        (song_data,e) = to_ms(song_data, WORKING_SAMPLE_RATE, 1)
         sample_rate = 1000
 
     mx = max(song_data)