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>

batch_detect.py

@@ -0,0 +1,183 @@
+#!/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()

detect_singing.py

@@ -0,0 +1,598 @@
+#!/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()

sing_detect.py

@@ -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 don’t 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()