v5: refactor into modular stream_tools desktop app

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>
This commit is contained in:
2026-01-05 10:00:00 +00:00
parent b70402932e
commit 878be5fe4a
6 changed files with 1016 additions and 0 deletions

View File

@@ -0,0 +1,188 @@
"""
Export singing segments to various formats:
- EDL (DaVinci Resolve timeline import)
- CSV (human-readable)
- DaVinci Resolve Markers CSV
- JSON (programmatic)
- ffmpeg concat script (auto-cut)
"""
import csv
import json
import os
import sys
from pathlib import Path
from utils.formats import format_timecode, seconds_to_smpte
def export_edl(segments: list, output_path: str, fps: float = 29.97,
title: str = "Singing Segments", source_filename: str = None):
"""Export EDL file for DaVinci Resolve."""
if source_filename:
clip_name = source_filename
reel_name_safe = Path(source_filename).stem[:32].replace(" ", "_")
else:
clip_name = None
reel_name_safe = "001"
rec_offset = 3600.0
with open(output_path, "w", encoding="utf-8") as f:
f.write(f"TITLE: {title}\n")
f.write("FCM: NON-DROP FRAME\n\n")
current_rec_pos = rec_offset
for seg in segments:
edit_num = f"{seg['index']:03d}"
src_in = seconds_to_smpte(seg["start"], fps)
src_out = seconds_to_smpte(seg["end"], fps)
rec_in = seconds_to_smpte(current_rec_pos, fps)
rec_out = seconds_to_smpte(current_rec_pos + seg["duration"], fps)
f.write(f"{edit_num} {reel_name_safe} V C {src_in} {src_out} {rec_in} {rec_out}\n")
if clip_name:
f.write(f"* FROM CLIP NAME: {clip_name}\n")
f.write(f"* COMMENT: Song {seg['index']} - Duration {seg['duration']:.0f}s\n\n")
current_rec_pos += seg["duration"]
def export_csv(segments: list, output_path: str):
"""Export singing segments as CSV."""
with open(output_path, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(["index", "start_sec", "end_sec", "duration_sec",
"start_timecode", "end_timecode"])
for seg in segments:
writer.writerow([
seg["index"],
f"{seg['start']:.2f}",
f"{seg['end']:.2f}",
f"{seg['duration']:.1f}",
format_timecode(seg["start"]),
format_timecode(seg["end"]),
])
def export_markers_csv(segments: list, output_path: str, fps: float = 29.97):
"""Export DaVinci Resolve Markers CSV."""
with open(output_path, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(["#", "Color", "Name", "Start TC", "End TC", "Duration TC", "Notes"])
for seg in segments:
writer.writerow([
seg["index"],
"Blue",
f"Song {seg['index']}",
seconds_to_smpte(seg["start"], fps),
seconds_to_smpte(seg["end"], fps),
seconds_to_smpte(seg["duration"], fps),
f"Duration: {seg['duration']:.0f}s",
])
def export_json(segments: list, output_path: str):
"""Export as JSON."""
with open(output_path, "w", encoding="utf-8") as f:
json.dump(segments, f, indent=2, ensure_ascii=False)
def export_ffmpeg_script(segments: list, video_path: str, output_dir: str):
"""
Export a script that uses ffmpeg to cut and concatenate all
singing segments into a single video file.
"""
video_name = Path(video_path).stem
ext = Path(video_path).suffix
out_video = str(Path(output_dir) / f"{video_name}_singing_only{ext}")
is_windows = os.name == "nt" or sys.platform == "win32"
script_ext = ".bat" if is_windows else ".sh"
script_path = str(Path(output_dir) / f"{video_name}_autocut{script_ext}")
concat_list_path = str(Path(output_dir) / f"{video_name}_concat_list.txt")
lines = []
if is_windows:
lines.append("@echo off")
lines.append("REM Auto-generated ffmpeg script to extract singing segments")
lines.append(f'REM Source: {video_path}')
lines.append("")
else:
lines.append("#!/bin/bash")
lines.append("# Auto-generated ffmpeg script to extract singing segments")
lines.append(f'# Source: {video_path}')
lines.append("")
segment_files = []
for seg in segments:
seg_file = f"_seg_{seg['index']:03d}{ext}"
seg_path = str(Path(output_dir) / seg_file)
segment_files.append(seg_file)
cmd = (f'ffmpeg -y -ss {seg["start"]:.2f} -i "{video_path}" '
f'-t {seg["duration"]:.2f} -c copy "{seg_path}"')
lines.append(f"echo Extracting Song {seg['index']}...")
lines.append(cmd)
lines.append("")
lines.append("echo Creating concat list...")
if is_windows:
lines.append("(")
for sf in segment_files:
seg_path = str(Path(output_dir) / sf)
lines.append(f" echo file '{seg_path}'")
lines.append(f') > "{concat_list_path}"')
else:
for i, sf in enumerate(segment_files):
seg_path = str(Path(output_dir) / sf)
op = ">" if i == 0 else ">>"
lines.append(f"echo \"file '{seg_path}'\" {op} \"{concat_list_path}\"")
lines.append("")
lines.append("echo Concatenating all segments...")
lines.append(f'ffmpeg -y -f concat -safe 0 -i "{concat_list_path}" -c copy "{out_video}"')
lines.append("")
lines.append("echo Cleaning up temp files...")
for sf in segment_files:
seg_path = str(Path(output_dir) / sf)
if is_windows:
lines.append(f'del "{seg_path}"')
else:
lines.append(f'rm -f "{seg_path}"')
if is_windows:
lines.append(f'del "{concat_list_path}"')
else:
lines.append(f'rm -f "{concat_list_path}"')
lines.append("")
lines.append(f'echo Done! Output: {out_video}')
if is_windows:
lines.append("pause")
with open(script_path, "w", encoding="utf-8") as f:
f.write("\n".join(lines))
if not is_windows:
os.chmod(script_path, 0o755)
return script_path
def export_raw_segments(segments: list, output_path: str):
"""Export ALL raw segments (speech/music/noise) as CSV for debugging."""
with open(output_path, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(["label", "start_sec", "end_sec", "duration_sec",
"start_timecode", "end_timecode"])
for label, start, end in segments:
writer.writerow([
label,
f"{start:.2f}",
f"{end:.2f}",
f"{end - start:.1f}",
format_timecode(start),
format_timecode(end),
])

View File

@@ -0,0 +1,99 @@
"""
ffmpeg / ffprobe discovery and probing helpers.
Shared by both concat and singing detection features.
"""
import json
import os
import shutil
import subprocess
import sys
from pathlib import Path
# ── Suppress console windows on Windows ──────────────────────────────────────
_NO_WINDOW: dict = {}
if sys.platform == "win32":
_NO_WINDOW["creationflags"] = subprocess.CREATE_NO_WINDOW
def app_dir() -> str:
if getattr(sys, "frozen", False):
return os.path.dirname(sys.executable)
return os.path.dirname(os.path.abspath(__file__))
def find_tool(name: str) -> str | None:
"""Locate ffmpeg / ffprobe next to the exe / script, then on PATH."""
base = app_dir()
for sub in ["_internal", "..", ""]:
for variant in [name + ".exe", name]:
p = os.path.join(base, sub, variant)
if os.path.exists(p):
return os.path.abspath(p)
return shutil.which(name + ".exe") or shutil.which(name)
def probe_duration(path: str, ffprobe_bin: str | None = None) -> float:
"""Get video/audio duration in seconds via ffprobe."""
ffprobe = ffprobe_bin or find_tool("ffprobe")
if not ffprobe:
return 0.0
try:
r = subprocess.run(
[ffprobe, "-v", "error",
"-show_entries", "format=duration",
"-of", "json", path],
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
text=True, timeout=20,
**_NO_WINDOW,
)
if r.returncode != 0:
return 0.0
data = json.loads(r.stdout or "{}")
return max(0.0, float(data.get("format", {}).get("duration", 0) or 0))
except Exception:
return 0.0
def extract_audio(video_path: str, output_wav: str,
ffmpeg_bin: str | None = None,
progress_cb=None) -> str:
"""
Extract audio from video file as 16kHz mono WAV.
Required format for inaSpeechSegmenter.
progress_cb: optional callable(message: str) for status updates.
"""
ffmpeg = ffmpeg_bin or find_tool("ffmpeg")
if not ffmpeg:
raise FileNotFoundError("ffmpeg not found. Install ffmpeg and add to PATH.")
if progress_cb:
progress_cb(f"Extracting audio from: {Path(video_path).name}")
cmd = [
ffmpeg,
"-i", video_path,
"-vn",
"-acodec", "pcm_s16le",
"-ar", "16000",
"-ac", "1",
"-y",
output_wav,
]
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=1800,
**_NO_WINDOW,
)
if result.returncode != 0:
raise RuntimeError(f"ffmpeg failed:\n{result.stderr[-500:]}")
if progress_cb:
size_mb = os.path.getsize(output_wav) / (1024 * 1024)
progress_cb(f"Audio extracted: {size_mb:.1f} MB")
return output_wav

View File

@@ -0,0 +1,118 @@
"""
Singing segment detection via inaSpeechSegmenter.
Pipeline:
1. Run inaSpeechSegmenter to classify speech/music/noise
2. Filter for 'music' segments (singing voice is classified as music)
3. Merge nearby segments, apply padding, filter by minimum duration
"""
import time
from utils.formats import format_timecode
def check_segmenter_available() -> tuple[bool, str]:
"""Check if inaSpeechSegmenter is installed."""
try:
from inaSpeechSegmenter import Segmenter # noqa: F401
return True, ""
except ImportError:
return False, (
"inaSpeechSegmenter is not installed.\n\n"
"Install it with:\n"
" pip install inaSpeechSegmenter\n\n"
"You also need TensorFlow:\n"
" pip install tensorflow[and-cuda] (GPU)\n"
" pip install tensorflow (CPU)"
)
def run_segmentation(wav_path: str, progress_cb=None) -> list:
"""
Run inaSpeechSegmenter on audio file.
Returns list of (label, start_sec, end_sec) tuples.
Labels: 'music', 'speech', 'noise', 'noEnergy'
"""
from inaSpeechSegmenter import Segmenter
if progress_cb:
progress_cb("Loading segmentation model…")
start_time = time.time()
seg = Segmenter(vad_engine='smn', detect_gender=False)
if progress_cb:
progress_cb("Running audio segmentation (this may take a while)…")
segments = seg(wav_path)
elapsed = time.time() - start_time
if progress_cb:
progress_cb(f"Segmentation complete in {elapsed:.1f}s — {len(segments)} raw segments")
return segments
def merge_singing_segments(
segments: list,
max_gap: float = 10.0,
min_duration: float = 20.0,
padding: float = 2.0,
) -> list:
"""
Filter for 'music' segments and merge nearby ones.
Returns list of dicts:
[{"index", "start", "end", "duration", "original_start", "original_end"}, ...]
"""
music_segs = [(start, end) for label, start, end in segments if label == "music"]
if not music_segs:
return []
music_segs.sort(key=lambda x: x[0])
# Merge segments with gaps smaller than max_gap
merged = []
current_start, current_end = music_segs[0]
for start, end in music_segs[1:]:
if start - current_end <= max_gap:
current_end = max(current_end, end)
else:
merged.append((current_start, current_end))
current_start, current_end = start, end
merged.append((current_start, current_end))
# Apply padding and minimum duration filter
results = []
idx = 1
for start, end in merged:
duration = end - start
if duration >= min_duration:
padded_start = max(0, start - padding)
padded_end = end + padding
results.append({
"index": idx,
"start": padded_start,
"end": padded_end,
"duration": padded_end - padded_start,
"original_start": start,
"original_end": end,
})
idx += 1
return results
def get_segment_stats(segments: list) -> dict:
"""Get summary statistics from raw segmentation output."""
type_counts = {}
type_durations = {}
for label, start, end in segments:
type_counts[label] = type_counts.get(label, 0) + 1
type_durations[label] = type_durations.get(label, 0) + (end - start)
return {
"counts": type_counts,
"durations": type_durations,
}

View File

@@ -0,0 +1,505 @@
"""
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

View File

@@ -0,0 +1,64 @@
"""
Theme, DPI, and shared styling constants.
"""
import sys
# ── DPI awareness (call before any Tk/CTk window) ────────────────────────────
def setup_dpi():
if sys.platform == "win32":
try:
import ctypes
ctypes.windll.shcore.SetProcessDpiAwareness(2)
except Exception:
try:
import ctypes
ctypes.windll.user32.SetProcessDPIAware()
except Exception:
pass
def dpi_scale(widget) -> float:
try:
return max(1.0, widget.winfo_fpixels("1i") / 96.0)
except Exception:
return 1.0
# ── Color accents ─────────────────────────────────────────────────────────────
# Each feature area has its own accent color scheme: (light_mode, dark_mode)
ACCENT_BLUE = {
"fg": ("#3B82F6", "#2563EB"),
"hover": ("#2563EB", "#1D4ED8"),
}
ACCENT_PURPLE = {
"fg": ("#EDE9FE", "#3B1F6E"),
"text": ("#5B21B6", "#C4B5FD"),
"hover": ("#DDD6FE", "#4C2889"),
"preview": ("#7C3AED", "#A78BFA"),
}
ACCENT_GREEN = {
"fg": ("#10B981", "#059669"),
"hover": ("#059669", "#047857"),
}
ACCENT_AMBER = {
"fg": ("#F59E0B", "#D97706"),
"hover": ("#D97706", "#B45309"),
}
# Neutral button style
BTN_NEUTRAL = {
"fg_color": ("gray80", "gray25"),
"text_color": ("gray10", "gray90"),
"hover_color": ("gray70", "gray35"),
}
BTN_SIDE = {
"fg_color": ("gray82", "gray22"),
"text_color": ("gray10", "gray90"),
"hover_color": ("gray72", "gray32"),
}

View File

@@ -0,0 +1,42 @@
"""
Timecode & duration formatting utilities.
"""
def fmt_dur(seconds: float) -> str:
"""Format a duration as H:MM:SS or M:SS for display."""
if seconds < 0:
return ""
if seconds == 0:
return ""
s = int(round(seconds))
h, rem = divmod(s, 3600)
m, sec = divmod(rem, 60)
return f"{h}:{m:02d}:{sec:02d}" if h else f"{m}:{sec:02d}"
def fmt_elapsed(s: float) -> str:
if s < 60:
return f"{int(s)}s"
m, sec = divmod(int(s), 60)
return f"{m}m {sec:02d}s"
def format_timecode(seconds: float) -> str:
"""Format seconds as HH:MM:SS."""
h = int(seconds // 3600)
m = int((seconds % 3600) // 60)
s = int(seconds % 60)
return f"{h:02d}:{m:02d}:{s:02d}"
def seconds_to_smpte(seconds: float, fps: float = 29.97) -> str:
"""Convert seconds to SMPTE timecode HH:MM:SS:FF."""
total_frames = int(seconds * fps)
ff = total_frames % int(round(fps))
total_seconds = total_frames // int(round(fps))
ss = total_seconds % 60
total_minutes = total_seconds // 60
mm = total_minutes % 60
hh = total_minutes // 60
return f"{hh:02d}:{mm:02d}:{ss:02d}:{ff:02d}"