diff --git a/batch_detect.py b/batch_detect.py deleted file mode 100644 index 1ed4a3e..0000000 --- a/batch_detect.py +++ /dev/null @@ -1,183 +0,0 @@ -#!/usr/bin/env python3 -""" -Batch Singing Detector -======================= -Process multiple stream recordings at once. -Point it at a folder of recordings and it will process each one. - -Usage: - python batch_detect.py "D:\\streams" - python batch_detect.py "D:\\streams" --output-dir "D:\\singing_edits" --auto-cut - python batch_detect.py "D:\\streams" --pattern "*.mp4" --recursive -""" - -import argparse -import os -import sys -import time -import glob -from pathlib import Path - - -def find_videos(input_dir: str, pattern: str = "*", recursive: bool = False, - extensions: tuple = (".mp4", ".mkv", ".flv", ".ts", ".avi", ".mov", ".webm")) -> list: - """Find all video files in directory.""" - videos = [] - - if recursive: - for ext in extensions: - videos.extend(glob.glob(os.path.join(input_dir, "**", f"*{ext}"), recursive=True)) - else: - for ext in extensions: - videos.extend(glob.glob(os.path.join(input_dir, f"*{ext}"))) - - # Filter by pattern if specified - if pattern != "*": - videos = [v for v in videos if Path(v).match(pattern)] - - # Sort by modification time (newest first) - videos.sort(key=lambda x: os.path.getmtime(x), reverse=True) - - return videos - - -def main(): - parser = argparse.ArgumentParser( - description="Batch process multiple stream recordings for singing detection.", - ) - - parser.add_argument("input_dir", help="Directory containing stream recordings") - parser.add_argument("--output-dir", "-o", default=None, - help="Output directory (default: same as input)") - parser.add_argument("--pattern", default="*", - help="Filename pattern to match (default: all video files)") - parser.add_argument("--recursive", "-r", action="store_true", - help="Search subdirectories recursively") - parser.add_argument("--gap", "-g", type=float, default=10.0, - help="Max gap to merge singing segments (default: 10s)") - parser.add_argument("--min-duration", "-m", type=float, default=20.0, - help="Minimum segment duration (default: 20s)") - parser.add_argument("--padding", "-p", type=float, default=2.0, - help="Padding before/after segments (default: 2s)") - parser.add_argument("--fps", type=float, default=29.97, - help="Frame rate (default: 29.97)") - parser.add_argument("--auto-cut", action="store_true", - help="Generate auto-cut ffmpeg scripts") - parser.add_argument("--skip-existing", action="store_true", - help="Skip files that already have output .edl files") - parser.add_argument("--limit", type=int, default=0, - help="Process only N files (0 = all)") - - args = parser.parse_args() - - if not os.path.isdir(args.input_dir): - print(f"[ERROR] Directory not found: {args.input_dir}") - sys.exit(1) - - # Find videos - videos = find_videos(args.input_dir, args.pattern, args.recursive) - - if not videos: - print(f"[INFO] No video files found in: {args.input_dir}") - sys.exit(0) - - # Filter already-processed files - if args.skip_existing: - output_dir = args.output_dir or args.input_dir - unprocessed = [] - for v in videos: - stem = Path(v).stem - edl = os.path.join(output_dir, f"{stem}_singing.edl") - if not os.path.exists(edl): - unprocessed.append(v) - else: - print(f"[SKIP] Already processed: {Path(v).name}") - videos = unprocessed - - # Apply limit - if args.limit > 0: - videos = videos[:args.limit] - - print(f"\n{'='*60}") - print(f" Batch Singing Detector") - print(f"{'='*60}") - print(f" Input: {args.input_dir}") - print(f" Videos: {len(videos)} to process") - print(f"{'='*60}\n") - - # Process each video - # Import the main detect_singing module - sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) - from detect_singing import ( - extract_audio, run_segmentation, merge_singing_segments, - export_edl, export_csv, export_davinci_markers_csv, - export_json, export_ffmpeg_concat, format_timecode, - ) - - results = [] - for i, video_path in enumerate(videos, 1): - video_name = Path(video_path).name - print(f"\n{'─'*60}") - print(f" [{i}/{len(videos)}] Processing: {video_name}") - print(f"{'─'*60}") - - try: - output_dir = args.output_dir or os.path.dirname(video_path) - os.makedirs(output_dir, exist_ok=True) - - stem = Path(video_path).stem - wav_path = os.path.join(output_dir, f"{stem}_audio.wav") - - # Step 1-3 - extract_audio(video_path, wav_path) - raw_segments = run_segmentation(wav_path) - singing = merge_singing_segments( - raw_segments, - max_gap=args.gap, - min_duration=args.min_duration, - padding=args.padding, - ) - - # Step 4: Export - if singing: - edl = os.path.join(output_dir, f"{stem}_singing.edl") - csv_f = os.path.join(output_dir, f"{stem}_singing.csv") - markers = os.path.join(output_dir, f"{stem}_markers.csv") - json_f = os.path.join(output_dir, f"{stem}_singing.json") - - export_edl(singing, edl, fps=args.fps, - source_filename=Path(video_path).name) - export_csv(singing, csv_f) - export_davinci_markers_csv(singing, markers, fps=args.fps) - export_json(singing, json_f) - - if args.auto_cut: - export_ffmpeg_concat(singing, video_path, json_f) - - total_dur = sum(s["duration"] for s in singing) - results.append((video_name, len(singing), total_dur, "OK")) - else: - results.append((video_name, 0, 0, "No songs found")) - - # Cleanup WAV - if os.path.exists(wav_path): - os.remove(wav_path) - - except Exception as e: - print(f" [ERROR] {e}") - results.append((video_name, 0, 0, f"Error: {e}")) - - # Summary - print(f"\n\n{'='*60}") - print(f" BATCH COMPLETE") - print(f"{'='*60}") - print(f" {'Video':<40s} {'Songs':>5s} {'Singing Time':>12s} Status") - print(f" {'─'*40} {'─'*5} {'─'*12} {'─'*15}") - for name, count, dur, status in results: - name_short = name[:38] + ".." if len(name) > 40 else name - print(f" {name_short:<40s} {count:>5d} {format_timecode(dur):>12s} {status}") - print(f"{'='*60}") - - -if __name__ == "__main__": - main() diff --git a/core/exporters.py b/core/exporters.py new file mode 100644 index 0000000..6e22306 --- /dev/null +++ b/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/core/fftools.py b/core/fftools.py new file mode 100644 index 0000000..d3a7ac1 --- /dev/null +++ b/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/core/segmenter.py b/core/segmenter.py new file mode 100644 index 0000000..b30bba8 --- /dev/null +++ b/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/detect_singing.py b/detect_singing.py deleted file mode 100644 index e8dbef4..0000000 --- a/detect_singing.py +++ /dev/null @@ -1,598 +0,0 @@ -#!/usr/bin/env python3 -""" -Singing Segment Detector for Live Stream Recordings -===================================================== -Detects singing segments in long stream recordings and exports -DaVinci Resolve-compatible EDL markers for fast video editing. - -Pipeline: - 1. Extract audio from video (ffmpeg → 16kHz mono WAV) - 2. Run inaSpeechSegmenter to classify speech/music/noise - 3. Merge nearby "music" segments (singing) with configurable gap - 4. Export to EDL (Edit Decision List) for DaVinci Resolve import - 5. Optionally export CSV for review - -Usage: - python detect_singing.py "path/to/stream_recording.mp4" - python detect_singing.py "path/to/stream_recording.mp4" --gap 15 --min-duration 30 - python detect_singing.py "path/to/stream_recording.mp4" --output-dir "D:/singing_edits" -""" - -import argparse -import csv -import json -import os -import subprocess -import sys -import tempfile -import time -from pathlib import Path - - -# ────────────────────────────────────────────── -# Step 1: Extract audio from video using ffmpeg -# ────────────────────────────────────────────── -def extract_audio(video_path: str, output_wav: str, ffmpeg_bin: str = "ffmpeg") -> str: - """Extract audio from video file as 16kHz mono WAV (required by inaSpeechSegmenter).""" - print(f"\n[Step 1/4] Extracting audio from: {video_path}") - print(f" Output WAV: {output_wav}") - - cmd = [ - ffmpeg_bin, - "-i", video_path, - "-vn", # no video - "-acodec", "pcm_s16le", # 16-bit PCM - "-ar", "16000", # 16kHz sample rate - "-ac", "1", # mono - "-y", # overwrite - output_wav, - ] - - try: - result = subprocess.run( - cmd, - capture_output=True, - text=True, - timeout=1800, # 30 min timeout for very long files - ) - if result.returncode != 0: - print(f" [ERROR] ffmpeg failed:\n{result.stderr[-500:]}") - sys.exit(1) - except FileNotFoundError: - print(f" [ERROR] ffmpeg not found at '{ffmpeg_bin}'.") - print(f" Make sure ffmpeg is installed and in your PATH.") - sys.exit(1) - - size_mb = os.path.getsize(output_wav) / (1024 * 1024) - print(f" Audio extracted: {size_mb:.1f} MB") - return output_wav - - -# ────────────────────────────────────────────── -# Step 2: Run inaSpeechSegmenter -# ────────────────────────────────────────────── -def run_segmentation(wav_path: str) -> list: - """ - Run inaSpeechSegmenter on audio file. - Returns list of (label, start_sec, end_sec) tuples. - Labels: 'music', 'speech', 'male', 'female', 'noise', 'noEnergy' - - NOTE: inaSpeechSegmenter classifies singing voice as 'music'. - """ - print(f"\n[Step 2/4] Running audio segmentation (this may take a while)...") - - try: - from inaSpeechSegmenter import Segmenter - except ImportError: - print(" [ERROR] inaSpeechSegmenter is not installed.") - print(" Install it with: pip install inaSpeechSegmenter") - sys.exit(1) - - start_time = time.time() - - # vad_engine='smn' → speech/music/noise detection - # detect_gender=False → faster, we don't need gender info - seg = Segmenter(vad_engine='smn', detect_gender=False) - segments = seg(wav_path) - - elapsed = time.time() - start_time - print(f" Segmentation complete in {elapsed:.1f}s") - print(f" Total segments found: {len(segments)}") - - # Count segment types - 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) - - for label in sorted(type_counts.keys()): - count = type_counts[label] - dur = type_durations[label] - print(f" {label:>10s}: {count:4d} segments, {format_timecode(dur)} total") - - return segments - - -# ────────────────────────────────────────────── -# Step 3: Filter & merge singing (music) 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 (which includes singing) and merge - nearby segments that are likely the same song. - - Args: - segments: Raw segmentation output [(label, start, end), ...] - max_gap: Max gap (seconds) between music segments to merge them. - Stream singers often have brief pauses, audience interaction - between verses, etc. Default 10s works well. - min_duration: Minimum duration (seconds) for a merged segment to be kept. - Filters out short music stings, sound effects, etc. - A typical song is 2-5 minutes, so 20s is a safe minimum. - padding: Seconds to add before/after each segment for clean cuts. - - Returns: - List of dicts: [{"start": float, "end": float, "duration": float, "index": int}, ...] - """ - print(f"\n[Step 3/4] Merging singing segments (gap={max_gap}s, min={min_duration}s, pad={padding}s)") - - # Extract only music segments - music_segs = [(start, end) for label, start, end in segments if label == "music"] - - if not music_segs: - print(" No music segments found!") - return [] - - # Sort by start time (should already be sorted, but just in case) - 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: - # Extend current segment - current_end = max(current_end, end) - else: - # Save current segment and start new one - 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 - - print(f" Found {len(results)} singing segments after merge+filter:") - for seg in results: - print(f" Song {seg['index']:2d}: " - f"{format_timecode(seg['start'])} → {format_timecode(seg['end'])} " - f"({seg['duration']:.0f}s)") - - return results - - -# ────────────────────────────────────────────── -# Step 4: Export formats -# ────────────────────────────────────────────── - -def export_edl(segments: list, output_path: str, fps: float = 29.97, - title: str = "Singing Segments", source_filename: str = None): - """ - Export an EDL (Edit Decision List) file for DaVinci Resolve. - - DaVinci Resolve import: File → Import → Timeline → select the .edl file. - The EDL creates cut points at each singing segment's in/out points. - - Args: - source_filename: Original video filename (e.g. "stream_2026-04-01.mp4"). - Used as reel name + FROM CLIP NAME so DaVinci Resolve - matches the EDL to the correct clip in the Media Pool. - Without this, Resolve picks a random clip when multiple - videos are loaded in the same project. - """ - print(f"\n[Step 4/4] Exporting EDL: {output_path}") - - # DaVinci Resolve uses both the reel name column AND the "* FROM CLIP NAME:" - # comment to identify which media clip an EDL edit refers to. - # - # Reel name: traditionally 8 chars max, but Resolve accepts longer names. - # We use the filename without extension, truncated to a safe length. - # FROM CLIP NAME: Resolve's preferred matching method. Must be the exact - # filename (with extension) as it appears in the Media Pool. - - if source_filename: - clip_name = source_filename # full filename with extension - reel_name = Path(source_filename).stem[:32] # stem, truncated for safety - # Replace spaces with underscores in reel name (some EDL parsers choke on spaces) - reel_name_safe = reel_name.replace(" ", "_") - else: - clip_name = None - reel_name_safe = "001" - - # Calculate record timecodes: sequential placement on the output timeline - # Each segment is placed one after another, starting at 01:00:00:00 - rec_offset = 3600.0 # start at 01:00:00:00 - - 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_timecode(seg["start"], fps) - src_out = seconds_to_timecode(seg["end"], fps) - rec_in = seconds_to_timecode(current_rec_pos, fps) - rec_out = seconds_to_timecode(current_rec_pos + seg["duration"], fps) - - # EDL format: edit# reel_name track_type transition src_in src_out rec_in rec_out - f.write(f"{edit_num} {reel_name_safe} V C {src_in} {src_out} {rec_in} {rec_out}\n") - - # "* FROM CLIP NAME:" is the key line DaVinci uses for media matching. - # It must exactly match the clip name shown in the Media Pool. - 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"] - - print(f" EDL file written with {len(segments)} edits") - if clip_name: - print(f" Source clip name: {clip_name}") - - -def export_csv(segments: list, output_path: str): - """Export singing segments as CSV for review or further processing.""" - 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"]), - ]) - print(f" CSV file written: {output_path}") - - -def export_davinci_markers_csv(segments: list, output_path: str, fps: float = 29.97): - """ - Export a CSV file that can be used with DaVinci Resolve's - 'Import Markers from CSV' function (Resolve 18+). - - Format: #, Color, Name, Start TC, End TC, Duration TC, Notes - """ - with open(output_path, "w", newline="", encoding="utf-8") as f: - writer = csv.writer(f) - # DaVinci Resolve Marker CSV header - writer.writerow(["#", "Color", "Name", "Start TC", "End TC", "Duration TC", "Notes"]) - for seg in segments: - start_tc = seconds_to_timecode(seg["start"], fps) - end_tc = seconds_to_timecode(seg["end"], fps) - dur_tc = seconds_to_timecode(seg["duration"], fps) - writer.writerow([ - seg["index"], - "Blue", - f"Song {seg['index']}", - start_tc, - end_tc, - dur_tc, - f"Duration: {seg['duration']:.0f}s", - ]) - print(f" DaVinci markers CSV written: {output_path}") - - -def export_json(segments: list, output_path: str): - """Export as JSON for programmatic use.""" - with open(output_path, "w", encoding="utf-8") as f: - json.dump(segments, f, indent=2, ensure_ascii=False) - print(f" JSON file written: {output_path}") - - -def export_ffmpeg_concat(segments: list, video_path: str, output_path: str): - """ - Export a .bat/.sh script that uses ffmpeg to directly cut and - concatenate all singing segments into a single video file. - This is the fully automated option — no DaVinci needed. - """ - video_name = Path(video_path).stem - ext = Path(video_path).suffix - out_video = str(Path(output_path).parent / 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_path).with_suffix(script_ext)) - - 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("") - - # Create a concat file list - concat_list_path = str(Path(output_path).parent / f"{video_name}_concat_list.txt") - - # Generate individual segment extraction commands - segment_files = [] - for seg in segments: - seg_file = f"_seg_{seg['index']:03d}{ext}" - seg_path = str(Path(output_path).parent / seg_file) - segment_files.append(seg_file) - - start = seg["start"] - duration = seg["duration"] - - cmd = (f'ffmpeg -y -ss {start:.2f} -i "{video_path}" ' - f'-t {duration:.2f} -c copy "{seg_path}"') - lines.append(f"echo Extracting Song {seg['index']}...") - lines.append(cmd) - lines.append("") - - # Generate concat list file - lines.append(f"echo Creating concat list...") - if is_windows: - lines.append(f'(') - for sf in segment_files: - seg_path = str(Path(output_path).parent / 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_path).parent / sf) - op = ">" if i == 0 else ">>" - lines.append(f"echo \"file '{seg_path}'\" {op} \"{concat_list_path}\"") - - lines.append("") - lines.append(f"echo Concatenating all segments...") - lines.append(f'ffmpeg -y -f concat -safe 0 -i "{concat_list_path}" -c copy "{out_video}"') - lines.append("") - - # Cleanup temp segment files - lines.append("echo Cleaning up temp files...") - for sf in segment_files: - seg_path = str(Path(output_path).parent / 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) - - print(f" FFmpeg concat script written: {script_path}") - print(f" Run it to auto-generate: {out_video}") - - -# ────────────────────────────────────────────── -# Export raw segmentation for debugging -# ────────────────────────────────────────────── -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), - ]) - print(f" Raw segments CSV written: {output_path}") - - -# ────────────────────────────────────────────── -# Utility functions -# ────────────────────────────────────────────── -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_timecode(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}" - - -# ────────────────────────────────────────────── -# Main -# ────────────────────────────────────────────── -def main(): - parser = argparse.ArgumentParser( - description="Detect singing segments in stream recordings and export DaVinci Resolve markers.", - formatter_class=argparse.RawDescriptionHelpFormatter, - epilog=""" -Examples: - python detect_singing.py "D:\\streams\\2026-04-04.mp4" - python detect_singing.py "D:\\streams\\2026-04-04.mp4" --gap 15 --min-duration 30 - python detect_singing.py "D:\\streams\\2026-04-04.mp4" --output-dir "D:\\singing_edits" - python detect_singing.py "D:\\streams\\2026-04-04.mp4" --auto-cut - -Tips: - --gap 10 Merge music segments with <=10s gap (default). Increase if the singer - often pauses >10s between verses in the same song. - --min-duration 20 Ignore segments shorter than 20s (default). Decrease if - the singer does very short songs or covers. - --padding 2 Add 2s padding before/after each segment (default). Helps catch - the very start/end of songs. - --auto-cut Generate an ffmpeg script to automatically cut and concat all - singing segments into a single video file. - --fps 29.97 Frame rate for timecode calculation (default: 29.97 for NTSC). - Use 25 for PAL or 23.976 for film. - """, - ) - - parser.add_argument("video", help="Path to the stream recording video file") - parser.add_argument("--output-dir", "-o", default=None, - help="Output directory (default: same directory as input video)") - parser.add_argument("--gap", "-g", type=float, default=10.0, - help="Max gap (seconds) to merge nearby singing segments (default: 10)") - parser.add_argument("--min-duration", "-m", type=float, default=20.0, - help="Minimum singing segment duration in seconds (default: 20)") - parser.add_argument("--padding", "-p", type=float, default=2.0, - help="Padding (seconds) before/after each segment (default: 2)") - parser.add_argument("--fps", type=float, default=29.97, - help="Video frame rate for timecode export (default: 29.97)") - parser.add_argument("--keep-wav", action="store_true", - help="Keep the extracted WAV file (default: delete after processing)") - parser.add_argument("--auto-cut", action="store_true", - help="Generate an ffmpeg script to auto-cut and concat singing segments") - parser.add_argument("--raw-segments", action="store_true", - help="Also export all raw segments (speech/music/noise) as CSV") - parser.add_argument("--ffmpeg", default="ffmpeg", - help="Path to ffmpeg binary (default: 'ffmpeg' from PATH)") - - args = parser.parse_args() - - # Validate input - video_path = os.path.abspath(args.video) - if not os.path.isfile(video_path): - print(f"[ERROR] Video file not found: {video_path}") - sys.exit(1) - - # Set output directory - if args.output_dir: - output_dir = os.path.abspath(args.output_dir) - os.makedirs(output_dir, exist_ok=True) - else: - output_dir = os.path.dirname(video_path) - - video_stem = Path(video_path).stem - - print("=" * 60) - print(" Singing Segment Detector") - print("=" * 60) - print(f" Input: {video_path}") - print(f" Output dir: {output_dir}") - print(f" Settings: gap={args.gap}s, min={args.min_duration}s, pad={args.padding}s") - print(f" FPS: {args.fps}") - - # Step 1: Extract audio - wav_path = os.path.join(output_dir, f"{video_stem}_audio.wav") - extract_audio(video_path, wav_path, ffmpeg_bin=args.ffmpeg) - - # Step 2: Run segmentation - raw_segments = run_segmentation(wav_path) - - # Step 3: Merge singing segments - singing_segments = merge_singing_segments( - raw_segments, - max_gap=args.gap, - min_duration=args.min_duration, - padding=args.padding, - ) - - if not singing_segments: - print("\n[RESULT] No singing segments detected. Try lowering --min-duration or increasing --gap.") - # Cleanup WAV - if not args.keep_wav and os.path.exists(wav_path): - os.remove(wav_path) - sys.exit(0) - - # Step 4: Export results - edl_path = os.path.join(output_dir, f"{video_stem}_singing.edl") - csv_path = os.path.join(output_dir, f"{video_stem}_singing.csv") - markers_path = os.path.join(output_dir, f"{video_stem}_markers.csv") - json_path = os.path.join(output_dir, f"{video_stem}_singing.json") - - export_edl(singing_segments, edl_path, fps=args.fps, - title=f"{video_stem} - Singing", - source_filename=Path(video_path).name) - export_csv(singing_segments, csv_path) - export_davinci_markers_csv(singing_segments, markers_path, fps=args.fps) - export_json(singing_segments, json_path) - - if args.auto_cut: - export_ffmpeg_concat(singing_segments, video_path, json_path) - - if args.raw_segments: - raw_csv_path = os.path.join(output_dir, f"{video_stem}_raw_segments.csv") - export_raw_segments(raw_segments, raw_csv_path) - - # Cleanup WAV unless --keep-wav - if not args.keep_wav and os.path.exists(wav_path): - os.remove(wav_path) - print(f"\n Temp WAV file deleted.") - - # Summary - total_singing = sum(s["duration"] for s in singing_segments) - print("\n" + "=" * 60) - print(" DONE!") - print("=" * 60) - print(f" Songs detected: {len(singing_segments)}") - print(f" Total singing time: {format_timecode(total_singing)}") - print(f" EDL file: {edl_path}") - print(f" DaVinci markers CSV: {markers_path}") - print(f" Segments CSV: {csv_path}") - print(f" Segments JSON: {json_path}") - if args.auto_cut: - script_ext = ".bat" if (os.name == "nt" or sys.platform == "win32") else ".sh" - print(f" Auto-cut script: {json_path.replace('.json', script_ext)}") - print() - print(" To import into DaVinci Resolve:") - print(" Option A: File → Import → Timeline → select the .edl file") - print(" Option B: Timeline menu → Import Markers from CSV → select _markers.csv") - print("=" * 60) - - -if __name__ == "__main__": - main() diff --git a/ui/tab_detect.py b/ui/tab_detect.py new file mode 100644 index 0000000..5b1d9bd --- /dev/null +++ b/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/ui/theme.py b/ui/theme.py new file mode 100644 index 0000000..f317f0b --- /dev/null +++ b/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/utils/formats.py b/utils/formats.py new file mode 100644 index 0000000..a5edc59 --- /dev/null +++ b/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}"