youfufan
b9573e8644
Same inaSpeechSegmenter engine split into reusable core/ (fftools, segmenter, exporters) + utils/ and a CustomTkinter multi-tab UI. Adds JSON and ffmpeg-script export. Extracted from the stream_tools app. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
506 lines
22 KiB
Python
506 lines
22 KiB
Python
"""
|
|
Tab 2: Song Detection
|
|
Detect singing segments in a video and export EDL/CSV/JSON for DaVinci Resolve.
|
|
"""
|
|
|
|
import os
|
|
import threading
|
|
import tkinter as tk
|
|
from pathlib import Path
|
|
from tkinter import filedialog, messagebox
|
|
|
|
import customtkinter as ctk
|
|
|
|
from core.fftools import find_tool, extract_audio
|
|
from core.segmenter import check_segmenter_available, run_segmentation, merge_singing_segments, get_segment_stats
|
|
from core.exporters import export_edl, export_csv, export_markers_csv, export_json, export_ffmpeg_script
|
|
from utils.formats import format_timecode, fmt_dur, fmt_elapsed
|
|
from ui.theme import BTN_NEUTRAL, ACCENT_PURPLE, ACCENT_GREEN
|
|
|
|
|
|
class DetectTab:
|
|
"""Song detection tab."""
|
|
|
|
VIDEO_EXTS = {".mp4", ".mov", ".mkv", ".avi", ".flv",
|
|
".ts", ".wmv", ".m4v", ".webm", ".mts", ".m2ts"}
|
|
|
|
def __init__(self, parent_frame, app):
|
|
self.frame = parent_frame
|
|
self.app = app
|
|
self.ffmpeg = app.ffmpeg
|
|
|
|
# State
|
|
self.input_path_var = tk.StringVar(value="")
|
|
self.output_dir_var = tk.StringVar(value="")
|
|
self.gap_var = tk.DoubleVar(value=10.0)
|
|
self.min_dur_var = tk.DoubleVar(value=20.0)
|
|
self.padding_var = tk.DoubleVar(value=2.0)
|
|
self.fps_var = tk.DoubleVar(value=29.97)
|
|
self.autocut_var = tk.BooleanVar(value=False)
|
|
|
|
self._running = False
|
|
self._stop_req = False
|
|
|
|
# Results
|
|
self._singing_segments = []
|
|
self._raw_segments = []
|
|
|
|
self._build_ui()
|
|
self._check_deps()
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════
|
|
# UI
|
|
# ══════════════════════════════════════════════════════════════════════════
|
|
|
|
def _build_ui(self):
|
|
scroll = ctk.CTkScrollableFrame(self.frame, corner_radius=12)
|
|
scroll.pack(fill="both", expand=True, padx=4, pady=4)
|
|
|
|
self._build_sec_input(scroll)
|
|
self._build_sec_params(scroll)
|
|
self._build_sec_run(scroll)
|
|
self._build_sec_results(scroll)
|
|
|
|
def _section(self, parent, title):
|
|
f = ctk.CTkFrame(parent, corner_radius=12)
|
|
hdr = ctk.CTkFrame(f, fg_color="transparent")
|
|
hdr.pack(fill="x", padx=14, pady=(10, 0))
|
|
ctk.CTkLabel(hdr, text=title,
|
|
font=ctk.CTkFont(size=13, weight="bold")).pack(side="left")
|
|
ctk.CTkFrame(f, height=1, fg_color=("gray78", "gray32")).pack(
|
|
fill="x", padx=14, pady=(6, 0))
|
|
return f
|
|
|
|
# ── Input ─────────────────────────────────────────────────────────────────
|
|
|
|
def _build_sec_input(self, parent):
|
|
sec = self._section(parent, "1 · Input Video")
|
|
sec.pack(fill="x", pady=(0, 12))
|
|
|
|
body = ctk.CTkFrame(sec, fg_color="transparent")
|
|
body.pack(fill="x", padx=14, pady=(10, 4))
|
|
|
|
r1 = ctk.CTkFrame(body, fg_color="transparent")
|
|
r1.pack(fill="x", pady=(0, 4))
|
|
ctk.CTkLabel(r1, text="Video file:", width=96, anchor="w").pack(side="left")
|
|
ctk.CTkEntry(r1, textvariable=self.input_path_var).pack(side="left", fill="x", expand=True)
|
|
ctk.CTkButton(r1, text="Browse…", width=100, command=self._pick_input,
|
|
**BTN_NEUTRAL).pack(side="left", padx=(8, 0))
|
|
|
|
# "Use Concat Output" button
|
|
self.use_concat_btn = ctk.CTkButton(
|
|
body, text="📎 Use Concat Output", width=200,
|
|
fg_color=ACCENT_PURPLE["fg"], text_color=ACCENT_PURPLE["text"],
|
|
hover_color=ACCENT_PURPLE["hover"],
|
|
command=self._use_concat_output)
|
|
self.use_concat_btn.pack(anchor="w", pady=(4, 4))
|
|
|
|
r2 = ctk.CTkFrame(body, fg_color="transparent")
|
|
r2.pack(fill="x", pady=(4, 0))
|
|
ctk.CTkLabel(r2, text="Output dir:", width=96, anchor="w").pack(side="left")
|
|
ctk.CTkEntry(r2, textvariable=self.output_dir_var).pack(side="left", fill="x", expand=True)
|
|
ctk.CTkButton(r2, text="Browse…", width=100, command=self._pick_output_dir,
|
|
**BTN_NEUTRAL).pack(side="left", padx=(8, 0))
|
|
|
|
ctk.CTkLabel(sec, text="Output defaults to same directory as input video.",
|
|
font=ctk.CTkFont(size=11), text_color=("gray50", "gray55")
|
|
).pack(anchor="w", padx=14, pady=(4, 10))
|
|
|
|
# ── Parameters ────────────────────────────────────────────────────────────
|
|
|
|
def _build_sec_params(self, parent):
|
|
sec = self._section(parent, "2 · Detection Parameters")
|
|
sec.pack(fill="x", pady=(0, 12))
|
|
|
|
body = ctk.CTkFrame(sec, fg_color="transparent")
|
|
body.pack(fill="x", padx=14, pady=(10, 10))
|
|
|
|
params = [
|
|
("Merge gap (s):", self.gap_var, 1, 60, "Max silence between song parts to merge"),
|
|
("Min duration (s):", self.min_dur_var, 5, 120, "Ignore segments shorter than this"),
|
|
("Padding (s):", self.padding_var, 0, 15, "Extra seconds before/after each song"),
|
|
("FPS:", self.fps_var, 23, 60, "Video frame rate for timecodes"),
|
|
]
|
|
|
|
for label_text, var, lo, hi, tooltip in params:
|
|
row = ctk.CTkFrame(body, fg_color="transparent")
|
|
row.pack(fill="x", pady=3)
|
|
|
|
ctk.CTkLabel(row, text=label_text, width=130, anchor="w",
|
|
font=ctk.CTkFont(size=12)).pack(side="left")
|
|
|
|
slider = ctk.CTkSlider(row, from_=lo, to=hi, variable=var,
|
|
width=200, number_of_steps=max(1, hi - lo))
|
|
slider.pack(side="left", padx=(0, 8))
|
|
|
|
val_lbl = ctk.CTkLabel(row, text=f"{var.get():.1f}", width=50, anchor="w",
|
|
font=ctk.CTkFont(size=12, weight="bold"))
|
|
val_lbl.pack(side="left")
|
|
|
|
ctk.CTkLabel(row, text=tooltip, font=ctk.CTkFont(size=11),
|
|
text_color=("gray50", "gray55")).pack(side="left", padx=(10, 0))
|
|
|
|
# Update label on slider move
|
|
var.trace_add("write", lambda *_, v=var, l=val_lbl: l.configure(text=f"{v.get():.1f}"))
|
|
|
|
# Auto-cut checkbox
|
|
ctk.CTkCheckBox(body, text="Generate auto-cut ffmpeg script (no DaVinci needed)",
|
|
variable=self.autocut_var).pack(anchor="w", pady=(8, 0))
|
|
|
|
# ── Run ───────────────────────────────────────────────────────────────────
|
|
|
|
def _build_sec_run(self, parent):
|
|
sec = self._section(parent, "3 · Detect")
|
|
sec.pack(fill="x", pady=(0, 12))
|
|
|
|
body = ctk.CTkFrame(sec, fg_color="transparent")
|
|
body.pack(fill="x", padx=14, pady=(12, 6))
|
|
|
|
btn_row = ctk.CTkFrame(body, fg_color="transparent")
|
|
btn_row.pack(fill="x", pady=(0, 10))
|
|
|
|
self.detect_btn = ctk.CTkButton(
|
|
btn_row, text="🎤 Start Detection", width=230, height=42,
|
|
font=ctk.CTkFont(size=14, weight="bold"),
|
|
fg_color=ACCENT_GREEN["fg"], hover_color=ACCENT_GREEN["hover"],
|
|
command=self.start_detect)
|
|
self.detect_btn.pack(side="left")
|
|
|
|
self.stop_btn = ctk.CTkButton(
|
|
btn_row, text="■ Stop", width=106, height=42,
|
|
fg_color=("gray80", "gray25"), text_color=("gray10", "gray90"),
|
|
hover_color=("#FCA5A5", "#7F1D1D"),
|
|
command=self._stop, state="disabled")
|
|
self.stop_btn.pack(side="left", padx=(12, 0))
|
|
|
|
# Dependency warning
|
|
self.dep_lbl = ctk.CTkLabel(body, text="", font=ctk.CTkFont(size=11),
|
|
text_color="#F97316", wraplength=600, anchor="w", justify="left")
|
|
self.dep_lbl.pack(fill="x", pady=(0, 4))
|
|
|
|
# Progress
|
|
self.progress = ctk.CTkProgressBar(body, height=16, corner_radius=8, mode="indeterminate")
|
|
self.progress.pack(fill="x", pady=(0, 6))
|
|
self.progress.stop()
|
|
self.progress.set(0)
|
|
|
|
self.status_lbl = ctk.CTkLabel(body, text="", anchor="w", font=ctk.CTkFont(size=12))
|
|
self.status_lbl.pack(fill="x")
|
|
|
|
ctk.CTkFrame(sec, height=6, fg_color="transparent").pack()
|
|
|
|
# ── Results ───────────────────────────────────────────────────────────────
|
|
|
|
def _build_sec_results(self, parent):
|
|
sec = self._section(parent, "4 · Results")
|
|
sec.pack(fill="x", pady=(0, 14))
|
|
|
|
self.results_body = ctk.CTkFrame(sec, fg_color="transparent")
|
|
self.results_body.pack(fill="x", padx=14, pady=(10, 10))
|
|
|
|
self.results_lbl = ctk.CTkLabel(
|
|
self.results_body, text="No results yet. Run detection first.",
|
|
font=ctk.CTkFont(size=12), text_color=("gray50", "gray55"))
|
|
self.results_lbl.pack(anchor="w")
|
|
|
|
# Song list (populated after detection)
|
|
self.song_list_frame = ctk.CTkFrame(self.results_body, fg_color="transparent")
|
|
|
|
# Export buttons (hidden until results)
|
|
self.export_frame = ctk.CTkFrame(sec, fg_color="transparent")
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════
|
|
# Logic
|
|
# ══════════════════════════════════════════════════════════════════════════
|
|
|
|
def _check_deps(self):
|
|
available, msg = check_segmenter_available()
|
|
if not available:
|
|
self.dep_lbl.configure(text=f"⚠ {msg}")
|
|
self.detect_btn.configure(state="disabled")
|
|
else:
|
|
self.dep_lbl.configure(text="")
|
|
|
|
if not self.ffmpeg:
|
|
self.dep_lbl.configure(text="⚠ ffmpeg not found. Install ffmpeg and add to PATH.")
|
|
self.detect_btn.configure(state="disabled")
|
|
|
|
def _pick_input(self):
|
|
f = filedialog.askopenfilename(
|
|
title="Select video file",
|
|
filetypes=[("Video files", "*.mp4 *.mov *.mkv *.avi *.flv *.ts *.wmv *.m4v *.webm *.mts"),
|
|
("All files", "*.*")])
|
|
if f:
|
|
self.input_path_var.set(f)
|
|
if not self.output_dir_var.get().strip():
|
|
self.output_dir_var.set(str(Path(f).parent))
|
|
|
|
def _pick_output_dir(self):
|
|
d = filedialog.askdirectory(title="Choose output folder")
|
|
if d:
|
|
self.output_dir_var.set(d)
|
|
|
|
def _use_concat_output(self):
|
|
"""Fill input from the Concat tab's output path."""
|
|
if hasattr(self.app, 'concat_tab'):
|
|
path = self.app.concat_tab.get_last_output_path()
|
|
if path and Path(path).exists():
|
|
self.input_path_var.set(path)
|
|
if not self.output_dir_var.get().strip():
|
|
self.output_dir_var.set(str(Path(path).parent))
|
|
self._set_status(f"Loaded concat output: {Path(path).name}")
|
|
elif path:
|
|
self.input_path_var.set(path)
|
|
if not self.output_dir_var.get().strip():
|
|
self.output_dir_var.set(str(Path(path).parent))
|
|
self._set_status("Concat output path set (file not yet created — run concat first)")
|
|
else:
|
|
messagebox.showinfo("No output", "Set an output path in the Concat tab first.")
|
|
|
|
def _set_status(self, text, error=False):
|
|
color = "#DC2626" if error else ("gray10", "gray90")
|
|
self.status_lbl.configure(text=text, text_color=color)
|
|
|
|
def _stop(self):
|
|
self._stop_req = True
|
|
self._set_status("Stopping…")
|
|
|
|
def start_detect(self):
|
|
"""Start the detection pipeline in a background thread."""
|
|
input_path = self.input_path_var.get().strip()
|
|
if not input_path:
|
|
messagebox.showerror("No input", "Select a video file first.")
|
|
return
|
|
if not os.path.isfile(input_path):
|
|
messagebox.showerror("File not found", f"Cannot find:\n{input_path}")
|
|
return
|
|
|
|
available, msg = check_segmenter_available()
|
|
if not available:
|
|
messagebox.showerror("Missing dependency", msg)
|
|
return
|
|
|
|
self._running = True
|
|
self._stop_req = False
|
|
self.detect_btn.configure(state="disabled")
|
|
self.stop_btn.configure(state="normal")
|
|
self.progress.configure(mode="indeterminate")
|
|
self.progress.start()
|
|
self._set_status("Starting detection…")
|
|
|
|
threading.Thread(target=self._detect_worker, args=(input_path,), daemon=True).start()
|
|
|
|
def _detect_worker(self, input_path):
|
|
output_dir = self.output_dir_var.get().strip() or str(Path(input_path).parent)
|
|
os.makedirs(output_dir, exist_ok=True)
|
|
stem = Path(input_path).stem
|
|
|
|
wav_path = os.path.join(output_dir, f"{stem}_audio.wav")
|
|
|
|
try:
|
|
# Step 1: Extract audio
|
|
def progress_cb(msg):
|
|
self.app.after(0, lambda: self._set_status(msg))
|
|
|
|
extract_audio(input_path, wav_path, self.ffmpeg, progress_cb=progress_cb)
|
|
|
|
if self._stop_req:
|
|
self._cleanup_and_finish(wav_path, "Stopped by user.")
|
|
return
|
|
|
|
# Step 2: Run segmentation
|
|
self.app.after(0, lambda: self._set_status("Running audio segmentation (this may take a while)…"))
|
|
raw_segments = run_segmentation(wav_path, progress_cb=progress_cb)
|
|
|
|
if self._stop_req:
|
|
self._cleanup_and_finish(wav_path, "Stopped by user.")
|
|
return
|
|
|
|
self._raw_segments = raw_segments
|
|
|
|
# Step 3: Merge
|
|
self.app.after(0, lambda: self._set_status("Merging singing segments…"))
|
|
singing = merge_singing_segments(
|
|
raw_segments,
|
|
max_gap=self.gap_var.get(),
|
|
min_duration=self.min_dur_var.get(),
|
|
padding=self.padding_var.get(),
|
|
)
|
|
self._singing_segments = singing
|
|
|
|
# Step 4: Export
|
|
if singing:
|
|
fps = self.fps_var.get()
|
|
source_name = Path(input_path).name
|
|
|
|
edl_path = os.path.join(output_dir, f"{stem}_singing.edl")
|
|
csv_path = os.path.join(output_dir, f"{stem}_singing.csv")
|
|
markers_path = os.path.join(output_dir, f"{stem}_markers.csv")
|
|
json_path = os.path.join(output_dir, f"{stem}_singing.json")
|
|
|
|
export_edl(singing, edl_path, fps=fps,
|
|
title=f"{stem} - Singing", source_filename=source_name)
|
|
export_csv(singing, csv_path)
|
|
export_markers_csv(singing, markers_path, fps=fps)
|
|
export_json(singing, json_path)
|
|
|
|
if self.autocut_var.get():
|
|
export_ffmpeg_script(singing, input_path, output_dir)
|
|
|
|
total_singing = sum(s["duration"] for s in singing)
|
|
result_msg = (f"✓ Found {len(singing)} songs · "
|
|
f"Total singing: {format_timecode(total_singing)} · "
|
|
f"Exported to: {output_dir}")
|
|
|
|
self.app.after(0, lambda: self._show_results(singing, edl_path, output_dir))
|
|
else:
|
|
result_msg = "No singing segments detected. Try lowering min duration or increasing gap."
|
|
|
|
# Cleanup WAV
|
|
if os.path.exists(wav_path):
|
|
os.remove(wav_path)
|
|
|
|
self.app.after(0, lambda: self._set_status(result_msg))
|
|
|
|
except Exception as e:
|
|
self.app.after(0, lambda: self._set_status(f"Error: {e}", error=True))
|
|
if os.path.exists(wav_path):
|
|
try:
|
|
os.remove(wav_path)
|
|
except Exception:
|
|
pass
|
|
|
|
self.app.after(0, self._finish_detect)
|
|
|
|
def _cleanup_and_finish(self, wav_path, msg):
|
|
if os.path.exists(wav_path):
|
|
try:
|
|
os.remove(wav_path)
|
|
except Exception:
|
|
pass
|
|
self.app.after(0, lambda: self._set_status(msg))
|
|
self.app.after(0, self._finish_detect)
|
|
|
|
def _finish_detect(self):
|
|
self._running = False
|
|
self._stop_req = False
|
|
self.detect_btn.configure(state="normal")
|
|
self.stop_btn.configure(state="disabled")
|
|
self.progress.stop()
|
|
self.progress.configure(mode="determinate")
|
|
self.progress.set(1.0 if self._singing_segments else 0)
|
|
|
|
def _show_results(self, segments, edl_path, output_dir):
|
|
"""Display detection results in the Results section."""
|
|
# Clear previous
|
|
for w in self.song_list_frame.winfo_children():
|
|
w.destroy()
|
|
self.export_frame.pack_forget()
|
|
|
|
self.results_lbl.configure(
|
|
text=f"Found {len(segments)} singing segments:",
|
|
text_color=("gray10", "gray90"))
|
|
|
|
self.song_list_frame.pack(fill="x", pady=(8, 0))
|
|
|
|
# Header
|
|
hdr = ctk.CTkFrame(self.song_list_frame, fg_color=("gray88", "gray20"), corner_radius=8)
|
|
hdr.pack(fill="x", pady=(0, 4))
|
|
for text, w in [("#", 40), ("Start", 90), ("End", 90), ("Duration", 80)]:
|
|
ctk.CTkLabel(hdr, text=text, width=w, anchor="center",
|
|
font=ctk.CTkFont(size=11, weight="bold")).pack(side="left", padx=4, pady=4)
|
|
|
|
# Rows
|
|
for seg in segments:
|
|
row = ctk.CTkFrame(self.song_list_frame, corner_radius=6, height=30)
|
|
row.pack(fill="x", pady=2)
|
|
row.pack_propagate(False)
|
|
|
|
vals = [
|
|
(f"Song {seg['index']}", 40),
|
|
(format_timecode(seg["start"]), 90),
|
|
(format_timecode(seg["end"]), 90),
|
|
(f"{seg['duration']:.0f}s", 80),
|
|
]
|
|
for text, w in vals:
|
|
ctk.CTkLabel(row, text=text, width=w, anchor="center",
|
|
font=ctk.CTkFont(size=11)).pack(side="left", padx=4)
|
|
|
|
# Total
|
|
total = sum(s["duration"] for s in segments)
|
|
ctk.CTkLabel(self.song_list_frame,
|
|
text=f"Total singing time: {format_timecode(total)}",
|
|
font=ctk.CTkFont(size=12, weight="bold")).pack(anchor="w", pady=(8, 0))
|
|
|
|
# Export info
|
|
self.export_frame.pack(fill="x", padx=14, pady=(4, 10))
|
|
for w in self.export_frame.winfo_children():
|
|
w.destroy()
|
|
|
|
ctk.CTkLabel(self.export_frame,
|
|
text=f"Files exported to: {output_dir}",
|
|
font=ctk.CTkFont(size=11), text_color=("gray45", "gray55"),
|
|
wraplength=600, anchor="w", justify="left").pack(anchor="w")
|
|
|
|
info_lines = [
|
|
"• _singing.edl → DaVinci Resolve: File → Import → Timeline",
|
|
"• _markers.csv → DaVinci Resolve: Timeline → Import Markers from CSV",
|
|
"• _singing.csv → Human-readable segment list",
|
|
"• _singing.json → Programmatic use",
|
|
]
|
|
if self.autocut_var.get():
|
|
info_lines.append("• _autocut.bat → Run to auto-cut singing segments (no DaVinci needed)")
|
|
|
|
for line in info_lines:
|
|
ctk.CTkLabel(self.export_frame, text=line,
|
|
font=ctk.CTkFont(size=11),
|
|
text_color=("gray50", "gray55"), anchor="w").pack(anchor="w")
|
|
|
|
# ── Public API for Quick Flow ─────────────────────────────────────────────
|
|
|
|
def run_detect_headless(self, input_path: str, output_dir: str,
|
|
gap: float, min_dur: float, padding: float,
|
|
fps: float, autocut: bool,
|
|
progress_cb=None) -> list:
|
|
"""
|
|
Run detection synchronously (called from Quick Flow worker thread).
|
|
Returns list of singing segments.
|
|
"""
|
|
os.makedirs(output_dir, exist_ok=True)
|
|
stem = Path(input_path).stem
|
|
wav_path = os.path.join(output_dir, f"{stem}_audio.wav")
|
|
|
|
try:
|
|
extract_audio(input_path, wav_path, self.ffmpeg, progress_cb=progress_cb)
|
|
|
|
raw_segments = run_segmentation(wav_path, progress_cb=progress_cb)
|
|
|
|
if progress_cb:
|
|
progress_cb("Merging singing segments…")
|
|
|
|
singing = merge_singing_segments(raw_segments,
|
|
max_gap=gap, min_duration=min_dur, padding=padding)
|
|
|
|
if singing:
|
|
source_name = Path(input_path).name
|
|
export_edl(singing, os.path.join(output_dir, f"{stem}_singing.edl"),
|
|
fps=fps, title=f"{stem} - Singing", source_filename=source_name)
|
|
export_csv(singing, os.path.join(output_dir, f"{stem}_singing.csv"))
|
|
export_markers_csv(singing, os.path.join(output_dir, f"{stem}_markers.csv"), fps=fps)
|
|
export_json(singing, os.path.join(output_dir, f"{stem}_singing.json"))
|
|
if autocut:
|
|
export_ffmpeg_script(singing, input_path, output_dir)
|
|
|
|
if os.path.exists(wav_path):
|
|
os.remove(wav_path)
|
|
|
|
return singing
|
|
|
|
except Exception:
|
|
if os.path.exists(wav_path):
|
|
try:
|
|
os.remove(wav_path)
|
|
except Exception:
|
|
pass
|
|
raise
|