From 58014dfd0998711a76f6389a525f36d447e6bf15 Mon Sep 17 00:00:00 2001 From: youfufan Date: Fri, 2 Jan 2026 10:00:00 +0000 Subject: [PATCH] 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 --- sing_detect.py | 369 ++++++++++++++++++++++++++++++++++--------------- 1 file changed, 261 insertions(+), 108 deletions(-) mode change 100644 => 100755 sing_detect.py diff --git a/sing_detect.py b/sing_detect.py old mode 100644 new mode 100755 index 5504563..4b0081f --- a/sing_detect.py +++ b/sing_detect.py @@ -1,108 +1,261 @@ -import subprocess -import os -import torch -import librosa -import numpy as np -from huggingface_hub import hf_hub_download -import tkinter as tk -from tkinter import filedialog - -# ------------------------ -# 1. Extract audio from video -# ------------------------ -def extract_audio(input_video, output_wav): - cmd = [ - "ffmpeg", "-y", - "-i", input_video, - "-vn", # no video - "-ac", "1", # mono - "-ar", "16000", # 16kHz - output_wav - ] - subprocess.run(cmd, check=True) - -# ------------------------ -# 2. Load pretrained model -# ------------------------ -def load_model(): - repo_id = "matthijs/svd" # HuggingFace singing voice detection model - filename = hf_hub_download(repo_id=repo_id, filename="model.pt") - model = torch.jit.load(filename) - model.eval() - return model - -# ------------------------ -# 3. Run inference on audio -# ------------------------ -def detect_singing(audio_path, model, hop_length=512, threshold=0.5, min_duration=10.0): - y, sr = librosa.load(audio_path, sr=16000) - x = torch.tensor(y).float().unsqueeze(0) - - with torch.no_grad(): - pred = model(x).squeeze().numpy() - - times = librosa.frames_to_time(np.arange(len(pred)), sr=sr, hop_length=hop_length) - - segments = [] - in_segment = False - seg_start = None - - for t, p in zip(times, pred): - if p > threshold and not in_segment: - in_segment = True - seg_start = t - elif p <= threshold and in_segment: - in_segment = False - seg_end = t - if seg_end - seg_start >= min_duration: - segments.append((seg_start, seg_end)) - - if in_segment: - seg_end = times[-1] - if seg_end - seg_start >= min_duration: - segments.append((seg_start, seg_end)) - - return segments - -# ------------------------ -# 4. Export CSV -# ------------------------ -def export_csv(segments, out_file="segments.csv"): - with open(out_file, "w", encoding="utf-8") as f: - f.write("start_time,end_time\n") - for s, e in segments: - f.write(f"{s:.2f},{e:.2f}\n") - print(f"[+] Saved: {out_file}") - -# ------------------------ -# MAIN -# ------------------------ -if __name__ == "__main__": - # Tkinter GUI to choose file - root = tk.Tk() - root.withdraw() - input_video = filedialog.askopenfilename( - title="Choose your video file", - filetypes=[("Video files", "*.mp4 *.mkv *.flv *.mov *.avi"), ("All files", "*.*")] - ) - if not input_video: - print("No file selected. Exiting.") - exit() - - audio_file = "audio.wav" - - print(f"[+] Processing: {input_video}") - - # Step 1: extract audio - extract_audio(input_video, audio_file) - - # Step 2: load model - model = load_model() - - # Step 3: run detection - segments = detect_singing(audio_file, model) - - # Step 4: save results - export_csv(segments, "segments.csv") - - print("[+] Done! Detected singing segments have been saved to segments.csv") +import customtkinter as ctk +import tkinter as tk +from tkinter import filedialog, messagebox +import os +import csv +from datetime import timedelta + +import math +import librosa +import soundfile as sf + +from transformers import pipeline +from pyannote.core import Segment + +# ------------------------ +# Time formatting function +# ------------------------ +def seconds_to_timecode(total_seconds): + """ + Convert float seconds -> "HH:MM:SS.mmm" for DaVinci Resolve CSV markers. + """ + td = timedelta(seconds=total_seconds) + hours, remainder = divmod(td.seconds, 3600) + minutes, secs = divmod(remainder, 60) + ms = int(td.microseconds / 1000) + return f"{hours:02d}:{minutes:02d}:{secs:02d}.{ms:03d}" + +def chunk_audio(audio_path, chunk_length_s=10.0, stride_s=5.0, sr=16000): + """ + Loads the entire audio with librosa, then yields overlapping chunks + (chunk_data, chunk_start, chunk_end, sr). + + - chunk_length_s = length (seconds) of each chunk + - stride_s = how much we "step back" within each chunk for overlap + e.g. chunk_length=10, stride=5 => chunk #0 covers + 0-10s, chunk #1 covers 5-15s, chunk #2 covers 10-20s, etc. + - sr = sampling rate to load the audio + """ + audio, sr = librosa.load(audio_path, sr=sr) + total_len_s = len(audio) / sr + + # Step in seconds, between the start points of consecutive chunks + # e.g. for chunk=10, stride=5 => step=10 - 5=5 + # means next chunk starts 5s after the previous chunk start + step = chunk_length_s - stride_s + + # Edge case: if stride_s >= chunk_length_s, you won't have overlap + if step <= 0: + step = chunk_length_s # no overlap + + # Figure out how many chunks needed so we don’t exceed total length + # (a bit of math to ensure we handle trailing audio < chunk_length_s) + num_chunks = math.ceil((total_len_s - chunk_length_s) / step) + 1 + if num_chunks < 1: + num_chunks = 1 + + for i in range(num_chunks): + chunk_start = i * step + chunk_end = chunk_start + chunk_length_s + + # If the chunk goes beyond total_len_s, clamp it + if chunk_end > total_len_s: + chunk_end = total_len_s + + # Convert times to sample indexes + start_sample = int(chunk_start * sr) + end_sample = int(chunk_end * sr) + + # If our chunk_start >= total_len_s, we can stop + if chunk_start >= total_len_s: + break + + # Slice the waveform + chunk_data = audio[start_sample:end_sample] + + yield chunk_data, chunk_start, chunk_end, sr + +class MusicMarkerApp(ctk.CTk): + def __init__(self): + super().__init__() + + self.title("Music Marker Generator") + self.geometry("500x350") + + # Variables to hold file paths & threshold + self.audio_file_path = tk.StringVar(value="") + self.output_csv_path = tk.StringVar(value="") + self.merge_threshold_var = tk.DoubleVar(value=2.0) # default: merge segments within 2 seconds + self.pipeline = None # We'll load it once on demand + + self._create_widgets() + + def _create_widgets(self): + # 1) Frame for selecting audio file + file_frame = ctk.CTkFrame(self) + file_frame.pack(pady=10, padx=10, fill="x") + + file_label = ctk.CTkLabel(file_frame, text="Audio File:") + file_label.pack(side="left", padx=5) + + file_entry = ctk.CTkEntry(file_frame, textvariable=self.audio_file_path, width=250) + file_entry.pack(side="left", padx=5) + + file_button = ctk.CTkButton(file_frame, text="Browse", command=self._browse_audio_file) + file_button.pack(side="left", padx=5) + + # 2) Frame for output 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()