add quick n dirty version of counter, make consecutive hits configurable

Author: linomp

index.html

@@ -55,6 +55,7 @@ <h4>Blastbeat Detection Visualizer</h4>
     <span id="audioPath"></span> <span id="snareFreq"></span> <span id="bassdrumFreq"></span>
 </div>
 <div id="controls">
+    <div id="seenCount">Blastbeat count: 0</div>
     <div id="currentTime">0:00:000</div>
     <div id="mobile-controls">
         <button id="playPauseBtn">Play/Pause</button>
@@ -90,6 +91,7 @@ <h4>Blastbeat Detection Visualizer</h4>
             document.getElementById('waveform').innerHTML = ""
             document.getElementById('spectrogram').innerHTML = ""
             document.getElementById('currentTime').textContent = "0:00:000"
+            document.getElementById('seenCount').textContent = "Blastbeat count: 0"
         }
 
         const lw = document.getElementById('loadingWaveform')
@@ -123,6 +125,8 @@ <h4>Blastbeat Detection Visualizer</h4>
 
         wavesurfer.load(audioFile)
 
+        let seenCount = 0
+
         wavesurfer.on('ready', () => {
             lw.style.display = 'none'
             document.getElementById('explainer').style.display = 'none'
@@ -145,12 +149,18 @@ <h4>Blastbeat Detection Visualizer</h4>
             const ms = Math.floor((t % 1) * 1000)
             document.getElementById('currentTime').textContent =
                 `${m}:${s.toString().padStart(2, '0')}:${ms.toString().padStart(3, '0')}`
+            const newCount = intervals.filter(b => t > b.start_time).length
+            if (newCount !== seenCount) {
+                seenCount = newCount
+                document.getElementById('seenCount').textContent = `Blastbeat count: ${seenCount}`
+            }
         })
 
         document.getElementById('waveform').addEventListener('wheel', handleWheelZoom, { passive: false })
         document.getElementById('spectrogram').addEventListener('wheel', handleWheelZoom, { passive: false })
     }
 
+
     async function loadZip(zipFile) {
         const zip = await JSZip.loadAsync(zipFile)
         const jsonFile = Object.keys(zip.files).find(n => n.endsWith('.json'))

pipeline.py

@@ -18,6 +18,10 @@ def parse_range(row, start, end):
     )
 
 
+def parse_int(row, key):
+    return int(row[key]) if row.get(key) else None
+
+
 def load_rows(csv_path):
     with open(csv_path, newline="") as f:
         return [
@@ -64,6 +68,7 @@ def load_rows(csv_path):
                 "snare_range": parse_range(
                     row, "snare_drum_range_start", "snare_drum_range_end"
                 ),
+                "min_consecutive_hits": parse_int(row, "min_consecutive_hits"),
             }.items()
             if v is not None
         }

processing.py

@@ -103,20 +103,24 @@ def get_sections_labeled_by_percussion_content_from_audio(
     return results
 
 
-def identify_blastbeats(sections: list[LabeledSection]) -> list[tuple[int, int]]:
-    min_hits = 8  # primitive approach: 4 snares+bass in a series at least
-    start_idx = 0
+def identify_blastbeats(
+    sections: list[LabeledSection], min_hits
+) -> list[tuple[int, int]]:
+    blastbeat_start_idx = 0
     hits = 0
     results = []
 
-    for i, s in enumerate(sections):
-        if s.snare_present and s.bass_drum_present:
+    # Caveman approach: consider it as blast beat if a given number of consecutive labeled sections contain snare & bassdrum
+    for i, section in enumerate(sections):
+        if section.snare_present and section.bass_drum_present:
             if hits == 0:
-                start_idx = i
+                blastbeat_start_idx = i
             hits += 1
         else:
             if hits >= min_hits:
-                results.append((sections[start_idx].start_idx, s.start_idx))
+                results.append(
+                    (sections[blastbeat_start_idx].start_idx, section.start_idx)
+                )
             hits = 0
 
     return results
@@ -177,6 +181,7 @@ def process_song(
     step_size_in_seconds=0.15,
     bass_drum_range=(10, 100),
     snare_range=(170, 600),
+    min_consecutive_hits=8,
 ):
     print("Separating drum track...")
     (time, audio_data, sample_rate), drumtrack_path = extract_drums(file_path)
@@ -203,7 +208,7 @@ def process_song(
         peak_detection_band_width,
         peak_detection_min_area_threshold,
     )
-    blastbeat_intervals = identify_blastbeats(labeled_sections)
+    blastbeat_intervals = identify_blastbeats(labeled_sections, min_consecutive_hits)
 
     print("Exporting result...")
     save_result(
@@ -215,7 +220,7 @@ def process_song(
     import argparse
     import webbrowser
 
-    OPEN_BROWSER_AFTER_PROCESSING = False
+    OPEN_BROWSER_AFTER_PROCESSING = True
 
     parser = argparse.ArgumentParser()
     parser.add_argument("--file", type=str)
@@ -244,6 +249,5 @@ def process_song(
         webbrowser.open(f"file://{Path(__file__).parent.resolve()}/index.html")
 
 # TODO:
-# - experiment with compression of drum track before fft?
 # - investigate false positives in all songs & tune parameters
 # - make min consecutive hits configurable

test_processing.py

@@ -15,7 +15,7 @@ def test_identify_blasts_1():
         LabeledSection(10, 11, False, False),
     ]
     expected = [(2, 10)]
-    actual = identify_blastbeats(input)
+    actual = identify_blastbeats(input, 8)
 
     assert actual == expected
 
@@ -32,6 +32,6 @@ def test_identify_blasts_2():
         LabeledSection(8, 9, True, True),
     ]
     expected = []
-    actual = identify_blastbeats(input)
+    actual = identify_blastbeats(input, 8)
 
     assert actual == expected