v2: transformers-pipeline detector with CustomTkinter GUI

Switches inference to a transformers pipeline (+pyannote segments), adds overlapping-chunk processing and CSV timecode export for DaVinci.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-01-02 10:00:00 +00:00
parent 678d6d1568
commit 58014dfd09

369
sing_detect.py Normal file → Executable file
View File

@@ -1,108 +1,261 @@
import subprocess import customtkinter as ctk
import os import tkinter as tk
import torch from tkinter import filedialog, messagebox
import librosa import os
import numpy as np import csv
from huggingface_hub import hf_hub_download from datetime import timedelta
import tkinter as tk
from tkinter import filedialog import math
import librosa
# ------------------------ import soundfile as sf
# 1. Extract audio from video
# ------------------------ from transformers import pipeline
def extract_audio(input_video, output_wav): from pyannote.core import Segment
cmd = [
"ffmpeg", "-y", # ------------------------
"-i", input_video, # Time formatting function
"-vn", # no video # ------------------------
"-ac", "1", # mono def seconds_to_timecode(total_seconds):
"-ar", "16000", # 16kHz """
output_wav Convert float seconds -> "HH:MM:SS.mmm" for DaVinci Resolve CSV markers.
] """
subprocess.run(cmd, check=True) td = timedelta(seconds=total_seconds)
hours, remainder = divmod(td.seconds, 3600)
# ------------------------ minutes, secs = divmod(remainder, 60)
# 2. Load pretrained model ms = int(td.microseconds / 1000)
# ------------------------ return f"{hours:02d}:{minutes:02d}:{secs:02d}.{ms:03d}"
def load_model():
repo_id = "matthijs/svd" # HuggingFace singing voice detection model def chunk_audio(audio_path, chunk_length_s=10.0, stride_s=5.0, sr=16000):
filename = hf_hub_download(repo_id=repo_id, filename="model.pt") """
model = torch.jit.load(filename) Loads the entire audio with librosa, then yields overlapping chunks
model.eval() (chunk_data, chunk_start, chunk_end, sr).
return model
- chunk_length_s = length (seconds) of each chunk
# ------------------------ - stride_s = how much we "step back" within each chunk for overlap
# 3. Run inference on audio e.g. chunk_length=10, stride=5 => chunk #0 covers
# ------------------------ 0-10s, chunk #1 covers 5-15s, chunk #2 covers 10-20s, etc.
def detect_singing(audio_path, model, hop_length=512, threshold=0.5, min_duration=10.0): - sr = sampling rate to load the audio
y, sr = librosa.load(audio_path, sr=16000) """
x = torch.tensor(y).float().unsqueeze(0) audio, sr = librosa.load(audio_path, sr=sr)
total_len_s = len(audio) / sr
with torch.no_grad():
pred = model(x).squeeze().numpy() # Step in seconds, between the start points of consecutive chunks
# e.g. for chunk=10, stride=5 => step=10 - 5=5
times = librosa.frames_to_time(np.arange(len(pred)), sr=sr, hop_length=hop_length) # means next chunk starts 5s after the previous chunk start
step = chunk_length_s - stride_s
segments = []
in_segment = False # Edge case: if stride_s >= chunk_length_s, you won't have overlap
seg_start = None if step <= 0:
step = chunk_length_s # no overlap
for t, p in zip(times, pred):
if p > threshold and not in_segment: # Figure out how many chunks needed so we dont exceed total length
in_segment = True # (a bit of math to ensure we handle trailing audio < chunk_length_s)
seg_start = t num_chunks = math.ceil((total_len_s - chunk_length_s) / step) + 1
elif p <= threshold and in_segment: if num_chunks < 1:
in_segment = False num_chunks = 1
seg_end = t
if seg_end - seg_start >= min_duration: for i in range(num_chunks):
segments.append((seg_start, seg_end)) chunk_start = i * step
chunk_end = chunk_start + chunk_length_s
if in_segment:
seg_end = times[-1] # If the chunk goes beyond total_len_s, clamp it
if seg_end - seg_start >= min_duration: if chunk_end > total_len_s:
segments.append((seg_start, seg_end)) chunk_end = total_len_s
return segments # Convert times to sample indexes
start_sample = int(chunk_start * sr)
# ------------------------ end_sample = int(chunk_end * sr)
# 4. Export CSV
# ------------------------ # If our chunk_start >= total_len_s, we can stop
def export_csv(segments, out_file="segments.csv"): if chunk_start >= total_len_s:
with open(out_file, "w", encoding="utf-8") as f: break
f.write("start_time,end_time\n")
for s, e in segments: # Slice the waveform
f.write(f"{s:.2f},{e:.2f}\n") chunk_data = audio[start_sample:end_sample]
print(f"[+] Saved: {out_file}")
yield chunk_data, chunk_start, chunk_end, sr
# ------------------------
# MAIN class MusicMarkerApp(ctk.CTk):
# ------------------------ def __init__(self):
if __name__ == "__main__": super().__init__()
# Tkinter GUI to choose file
root = tk.Tk() self.title("Music Marker Generator")
root.withdraw() self.geometry("500x350")
input_video = filedialog.askopenfilename(
title="Choose your video file", # Variables to hold file paths & threshold
filetypes=[("Video files", "*.mp4 *.mkv *.flv *.mov *.avi"), ("All files", "*.*")] self.audio_file_path = tk.StringVar(value="")
) self.output_csv_path = tk.StringVar(value="")
if not input_video: self.merge_threshold_var = tk.DoubleVar(value=2.0) # default: merge segments within 2 seconds
print("No file selected. Exiting.") self.pipeline = None # We'll load it once on demand
exit()
self._create_widgets()
audio_file = "audio.wav"
def _create_widgets(self):
print(f"[+] Processing: {input_video}") # 1) Frame for selecting audio file
file_frame = ctk.CTkFrame(self)
# Step 1: extract audio file_frame.pack(pady=10, padx=10, fill="x")
extract_audio(input_video, audio_file)
file_label = ctk.CTkLabel(file_frame, text="Audio File:")
# Step 2: load model file_label.pack(side="left", padx=5)
model = load_model()
file_entry = ctk.CTkEntry(file_frame, textvariable=self.audio_file_path, width=250)
# Step 3: run detection file_entry.pack(side="left", padx=5)
segments = detect_singing(audio_file, model)
file_button = ctk.CTkButton(file_frame, text="Browse", command=self._browse_audio_file)
# Step 4: save results file_button.pack(side="left", padx=5)
export_csv(segments, "segments.csv")
# 2) Frame for output CSV
print("[+] Done! Detected singing segments have been saved to segments.csv") output_frame = ctk.CTkFrame(self)
output_frame.pack(pady=10, padx=10, fill="x")
output_label = ctk.CTkLabel(output_frame, text="Output CSV:")
output_label.pack(side="left", padx=5)
output_entry = ctk.CTkEntry(output_frame, textvariable=self.output_csv_path, width=250)
output_entry.pack(side="left", padx=5)
output_button = ctk.CTkButton(output_frame, text="Browse", command=self._browse_output_csv)
output_button.pack(side="left", padx=5)
# 3) Merge Threshold
threshold_frame = ctk.CTkFrame(self)
threshold_frame.pack(pady=10, padx=10, fill="x")
threshold_label = ctk.CTkLabel(threshold_frame, text="Merge Gap (sec):")
threshold_label.pack(side="left", padx=5)
threshold_entry = ctk.CTkEntry(threshold_frame, textvariable=self.merge_threshold_var, width=50)
threshold_entry.pack(side="left", padx=5)
# 4) Run button
run_button = ctk.CTkButton(self, text="Run Music Detection", command=self._run_detection)
run_button.pack(pady=10)
# 5) Status label
self.status_label = ctk.CTkLabel(self, text="", wraplength=400, justify="left")
self.status_label.pack(pady=5)
def _browse_audio_file(self):
file_path = filedialog.askopenfilename(
title="Select Audio File",
filetypes=[("Audio Files", "*.wav *.mp3 *.flac *.m4a *.aac *.ogg *.wma *.aif *.aiff")]
)
if file_path:
self.audio_file_path.set(file_path)
def _browse_output_csv(self):
file_path = filedialog.asksaveasfilename(
title="Select Output CSV",
defaultextension=".csv",
filetypes=[("CSV Files", "*.csv")]
)
if file_path:
self.output_csv_path.set(file_path)
def _run_detection(self):
audio_path = self.audio_file_path.get().strip()
output_csv = self.output_csv_path.get().strip()
merge_threshold = self.merge_threshold_var.get()
if not audio_path or not os.path.isfile(audio_path):
messagebox.showerror("Error", "Please select a valid audio file.")
return
if not output_csv:
messagebox.showerror("Error", "Please specify an output CSV file.")
return
self._set_status("Loading model, please wait...")
# Load pipeline if not loaded yet
if self.pipeline is None:
try:
# Standard audio-classification pipeline
self.pipeline = pipeline(
"audio-classification",
model="MarekCech/GenreVim-Music-Detection-DistilHuBERT"
)
except Exception as e:
messagebox.showerror("Model Error", f"Could not load the model:\n{e}")
return
self._set_status("Chunking audio and running music detection...")
# We'll collect "music" segments from each chunk
music_segments = []
# Manually chunk the audio & classify each chunk
try:
for chunk_data, chunk_start, chunk_end, sr in chunk_audio(
audio_path,
chunk_length_s=10.0,
stride_s=5.0,
sr=16000
):
# The pipeline expects a waveform plus sampling_rate, not a path
# So pass the chunk_data + sr
results = self.pipeline(chunk_data, sampling_rate=sr)
if not results:
continue
# best guess from the pipeline for this chunk
best_guess = max(results, key=lambda x: x["score"])
if best_guess["label"].lower() == "music" and best_guess["score"] > 0.5:
music_segments.append(Segment(chunk_start, chunk_end))
except Exception as e:
messagebox.showerror("Detection Error", f"Error running music detection:\n{e}")
return
# Sort segments by start time
music_segments.sort(key=lambda seg: seg.start)
# Merge close segments based on threshold
merged_segments = []
if not music_segments:
self._set_status("No music segments found. No CSV generated.")
return
else:
current_start = music_segments[0].start
current_end = music_segments[0].end
for seg in music_segments[1:]:
# If the new segment starts within X seconds of the old segment, merge them
if seg.start <= current_end + merge_threshold:
current_end = max(current_end, seg.end)
else:
merged_segments.append(Segment(current_start, current_end))
current_start = seg.start
current_end = seg.end
# finalize last
merged_segments.append(Segment(current_start, current_end))
# Write CSV for DaVinci Resolve
try:
with open(output_csv, "w", newline="", encoding="utf-8") as csvfile:
writer = csv.writer(csvfile)
# DaVinci Resolve CSV marker columns
writer.writerow(["Name", "Start", "End", "Color", "Marker Type"])
for i, seg in enumerate(merged_segments, start=1):
start_tc = seconds_to_timecode(seg.start)
end_tc = seconds_to_timecode(seg.end)
name = f"Music Segment #{i}"
color = "Green"
marker_type = "Comment"
writer.writerow([name, start_tc, end_tc, color, marker_type])
self._set_status(
f"Done! Found {len(merged_segments)} music segments.\n"
f"CSV saved to: {output_csv}"
)
except Exception as e:
messagebox.showerror("File Error", f"Could not write CSV:\n{e}")
def _set_status(self, msg):
self.status_label.configure(text=msg)
self.update_idletasks()
if __name__ == "__main__":
app = MusicMarkerApp()
app.mainloop()