diff --git a/v5_stream_tools_app/core/exporters.py b/v5_stream_tools_app/core/exporters.py new file mode 100644 index 0000000..6e22306 --- /dev/null +++ b/v5_stream_tools_app/core/exporters.py @@ -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), + ]) diff --git a/v5_stream_tools_app/core/fftools.py b/v5_stream_tools_app/core/fftools.py new file mode 100644 index 0000000..d3a7ac1 --- /dev/null +++ b/v5_stream_tools_app/core/fftools.py @@ -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 diff --git a/v5_stream_tools_app/core/segmenter.py b/v5_stream_tools_app/core/segmenter.py new file mode 100644 index 0000000..b30bba8 --- /dev/null +++ b/v5_stream_tools_app/core/segmenter.py @@ -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, + } diff --git a/v5_stream_tools_app/ui/tab_detect.py b/v5_stream_tools_app/ui/tab_detect.py new file mode 100644 index 0000000..5b1d9bd --- /dev/null +++ b/v5_stream_tools_app/ui/tab_detect.py @@ -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 diff --git a/v5_stream_tools_app/ui/theme.py b/v5_stream_tools_app/ui/theme.py new file mode 100644 index 0000000..f317f0b --- /dev/null +++ b/v5_stream_tools_app/ui/theme.py @@ -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"), +} diff --git a/v5_stream_tools_app/utils/formats.py b/v5_stream_tools_app/utils/formats.py new file mode 100644 index 0000000..a5edc59 --- /dev/null +++ b/v5_stream_tools_app/utils/formats.py @@ -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}"