3 Commits
v2 ... v5

Author SHA1 Message Date
b9573e8644 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>
2026-01-05 10:00:00 +00:00
98a4d37162 v4: batch processing layer over the CLI engine
Wraps detect_singing to process whole folders of recordings at once: --recursive, --pattern, --output-dir, --auto-cut.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-01-04 10:00:00 +00:00
91df8d3cec v3: inaSpeechSegmenter CLI pipeline
Adopts inaSpeechSegmenter (singing classified as music). Full argparse CLI: gap-merge nearby segments, min-duration filter, EDL + CSV export. Engine that all later versions keep.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-01-03 10:00:00 +00:00
7 changed files with 1016 additions and 261 deletions

188
core/exporters.py Normal file
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),
])

99
core/fftools.py Normal file
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

118
core/segmenter.py Normal file
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

@@ -1,261 +0,0 @@
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 dont 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()

505
ui/tab_detect.py Normal file
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

64
ui/theme.py Normal file
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"),
}

42
utils/formats.py Normal file
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}"