Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| b9573e8644 |
183
batch_detect.py
183
batch_detect.py
@@ -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()
|
||||
188
core/exporters.py
Normal file
188
core/exporters.py
Normal file
@@ -0,0 +1,188 @@
|
||||
"""
|
||||
Export singing segments to various formats:
|
||||
- EDL (DaVinci Resolve timeline import)
|
||||
- CSV (human-readable)
|
||||
- DaVinci Resolve Markers CSV
|
||||
- JSON (programmatic)
|
||||
- ffmpeg concat script (auto-cut)
|
||||
"""
|
||||
|
||||
import csv
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from utils.formats import format_timecode, seconds_to_smpte
|
||||
|
||||
|
||||
def export_edl(segments: list, output_path: str, fps: float = 29.97,
|
||||
title: str = "Singing Segments", source_filename: str = None):
|
||||
"""Export EDL file for DaVinci Resolve."""
|
||||
if source_filename:
|
||||
clip_name = source_filename
|
||||
reel_name_safe = Path(source_filename).stem[:32].replace(" ", "_")
|
||||
else:
|
||||
clip_name = None
|
||||
reel_name_safe = "001"
|
||||
|
||||
rec_offset = 3600.0
|
||||
|
||||
with open(output_path, "w", encoding="utf-8") as f:
|
||||
f.write(f"TITLE: {title}\n")
|
||||
f.write("FCM: NON-DROP FRAME\n\n")
|
||||
|
||||
current_rec_pos = rec_offset
|
||||
for seg in segments:
|
||||
edit_num = f"{seg['index']:03d}"
|
||||
src_in = seconds_to_smpte(seg["start"], fps)
|
||||
src_out = seconds_to_smpte(seg["end"], fps)
|
||||
rec_in = seconds_to_smpte(current_rec_pos, fps)
|
||||
rec_out = seconds_to_smpte(current_rec_pos + seg["duration"], fps)
|
||||
|
||||
f.write(f"{edit_num} {reel_name_safe} V C {src_in} {src_out} {rec_in} {rec_out}\n")
|
||||
|
||||
if clip_name:
|
||||
f.write(f"* FROM CLIP NAME: {clip_name}\n")
|
||||
|
||||
f.write(f"* COMMENT: Song {seg['index']} - Duration {seg['duration']:.0f}s\n\n")
|
||||
current_rec_pos += seg["duration"]
|
||||
|
||||
|
||||
def export_csv(segments: list, output_path: str):
|
||||
"""Export singing segments as CSV."""
|
||||
with open(output_path, "w", newline="", encoding="utf-8") as f:
|
||||
writer = csv.writer(f)
|
||||
writer.writerow(["index", "start_sec", "end_sec", "duration_sec",
|
||||
"start_timecode", "end_timecode"])
|
||||
for seg in segments:
|
||||
writer.writerow([
|
||||
seg["index"],
|
||||
f"{seg['start']:.2f}",
|
||||
f"{seg['end']:.2f}",
|
||||
f"{seg['duration']:.1f}",
|
||||
format_timecode(seg["start"]),
|
||||
format_timecode(seg["end"]),
|
||||
])
|
||||
|
||||
|
||||
def export_markers_csv(segments: list, output_path: str, fps: float = 29.97):
|
||||
"""Export DaVinci Resolve Markers CSV."""
|
||||
with open(output_path, "w", newline="", encoding="utf-8") as f:
|
||||
writer = csv.writer(f)
|
||||
writer.writerow(["#", "Color", "Name", "Start TC", "End TC", "Duration TC", "Notes"])
|
||||
for seg in segments:
|
||||
writer.writerow([
|
||||
seg["index"],
|
||||
"Blue",
|
||||
f"Song {seg['index']}",
|
||||
seconds_to_smpte(seg["start"], fps),
|
||||
seconds_to_smpte(seg["end"], fps),
|
||||
seconds_to_smpte(seg["duration"], fps),
|
||||
f"Duration: {seg['duration']:.0f}s",
|
||||
])
|
||||
|
||||
|
||||
def export_json(segments: list, output_path: str):
|
||||
"""Export as JSON."""
|
||||
with open(output_path, "w", encoding="utf-8") as f:
|
||||
json.dump(segments, f, indent=2, ensure_ascii=False)
|
||||
|
||||
|
||||
def export_ffmpeg_script(segments: list, video_path: str, output_dir: str):
|
||||
"""
|
||||
Export a script that uses ffmpeg to cut and concatenate all
|
||||
singing segments into a single video file.
|
||||
"""
|
||||
video_name = Path(video_path).stem
|
||||
ext = Path(video_path).suffix
|
||||
out_video = str(Path(output_dir) / f"{video_name}_singing_only{ext}")
|
||||
|
||||
is_windows = os.name == "nt" or sys.platform == "win32"
|
||||
script_ext = ".bat" if is_windows else ".sh"
|
||||
script_path = str(Path(output_dir) / f"{video_name}_autocut{script_ext}")
|
||||
concat_list_path = str(Path(output_dir) / f"{video_name}_concat_list.txt")
|
||||
|
||||
lines = []
|
||||
if is_windows:
|
||||
lines.append("@echo off")
|
||||
lines.append("REM Auto-generated ffmpeg script to extract singing segments")
|
||||
lines.append(f'REM Source: {video_path}')
|
||||
lines.append("")
|
||||
else:
|
||||
lines.append("#!/bin/bash")
|
||||
lines.append("# Auto-generated ffmpeg script to extract singing segments")
|
||||
lines.append(f'# Source: {video_path}')
|
||||
lines.append("")
|
||||
|
||||
segment_files = []
|
||||
for seg in segments:
|
||||
seg_file = f"_seg_{seg['index']:03d}{ext}"
|
||||
seg_path = str(Path(output_dir) / seg_file)
|
||||
segment_files.append(seg_file)
|
||||
|
||||
cmd = (f'ffmpeg -y -ss {seg["start"]:.2f} -i "{video_path}" '
|
||||
f'-t {seg["duration"]:.2f} -c copy "{seg_path}"')
|
||||
lines.append(f"echo Extracting Song {seg['index']}...")
|
||||
lines.append(cmd)
|
||||
lines.append("")
|
||||
|
||||
lines.append("echo Creating concat list...")
|
||||
if is_windows:
|
||||
lines.append("(")
|
||||
for sf in segment_files:
|
||||
seg_path = str(Path(output_dir) / sf)
|
||||
lines.append(f" echo file '{seg_path}'")
|
||||
lines.append(f') > "{concat_list_path}"')
|
||||
else:
|
||||
for i, sf in enumerate(segment_files):
|
||||
seg_path = str(Path(output_dir) / sf)
|
||||
op = ">" if i == 0 else ">>"
|
||||
lines.append(f"echo \"file '{seg_path}'\" {op} \"{concat_list_path}\"")
|
||||
|
||||
lines.append("")
|
||||
lines.append("echo Concatenating all segments...")
|
||||
lines.append(f'ffmpeg -y -f concat -safe 0 -i "{concat_list_path}" -c copy "{out_video}"')
|
||||
lines.append("")
|
||||
lines.append("echo Cleaning up temp files...")
|
||||
|
||||
for sf in segment_files:
|
||||
seg_path = str(Path(output_dir) / sf)
|
||||
if is_windows:
|
||||
lines.append(f'del "{seg_path}"')
|
||||
else:
|
||||
lines.append(f'rm -f "{seg_path}"')
|
||||
if is_windows:
|
||||
lines.append(f'del "{concat_list_path}"')
|
||||
else:
|
||||
lines.append(f'rm -f "{concat_list_path}"')
|
||||
|
||||
lines.append("")
|
||||
lines.append(f'echo Done! Output: {out_video}')
|
||||
if is_windows:
|
||||
lines.append("pause")
|
||||
|
||||
with open(script_path, "w", encoding="utf-8") as f:
|
||||
f.write("\n".join(lines))
|
||||
|
||||
if not is_windows:
|
||||
os.chmod(script_path, 0o755)
|
||||
|
||||
return script_path
|
||||
|
||||
|
||||
def export_raw_segments(segments: list, output_path: str):
|
||||
"""Export ALL raw segments (speech/music/noise) as CSV for debugging."""
|
||||
with open(output_path, "w", newline="", encoding="utf-8") as f:
|
||||
writer = csv.writer(f)
|
||||
writer.writerow(["label", "start_sec", "end_sec", "duration_sec",
|
||||
"start_timecode", "end_timecode"])
|
||||
for label, start, end in segments:
|
||||
writer.writerow([
|
||||
label,
|
||||
f"{start:.2f}",
|
||||
f"{end:.2f}",
|
||||
f"{end - start:.1f}",
|
||||
format_timecode(start),
|
||||
format_timecode(end),
|
||||
])
|
||||
99
core/fftools.py
Normal file
99
core/fftools.py
Normal file
@@ -0,0 +1,99 @@
|
||||
"""
|
||||
ffmpeg / ffprobe discovery and probing helpers.
|
||||
Shared by both concat and singing detection features.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# ── Suppress console windows on Windows ──────────────────────────────────────
|
||||
_NO_WINDOW: dict = {}
|
||||
if sys.platform == "win32":
|
||||
_NO_WINDOW["creationflags"] = subprocess.CREATE_NO_WINDOW
|
||||
|
||||
|
||||
def app_dir() -> str:
|
||||
if getattr(sys, "frozen", False):
|
||||
return os.path.dirname(sys.executable)
|
||||
return os.path.dirname(os.path.abspath(__file__))
|
||||
|
||||
|
||||
def find_tool(name: str) -> str | None:
|
||||
"""Locate ffmpeg / ffprobe next to the exe / script, then on PATH."""
|
||||
base = app_dir()
|
||||
for sub in ["_internal", "..", ""]:
|
||||
for variant in [name + ".exe", name]:
|
||||
p = os.path.join(base, sub, variant)
|
||||
if os.path.exists(p):
|
||||
return os.path.abspath(p)
|
||||
return shutil.which(name + ".exe") or shutil.which(name)
|
||||
|
||||
|
||||
def probe_duration(path: str, ffprobe_bin: str | None = None) -> float:
|
||||
"""Get video/audio duration in seconds via ffprobe."""
|
||||
ffprobe = ffprobe_bin or find_tool("ffprobe")
|
||||
if not ffprobe:
|
||||
return 0.0
|
||||
try:
|
||||
r = subprocess.run(
|
||||
[ffprobe, "-v", "error",
|
||||
"-show_entries", "format=duration",
|
||||
"-of", "json", path],
|
||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
||||
text=True, timeout=20,
|
||||
**_NO_WINDOW,
|
||||
)
|
||||
if r.returncode != 0:
|
||||
return 0.0
|
||||
data = json.loads(r.stdout or "{}")
|
||||
return max(0.0, float(data.get("format", {}).get("duration", 0) or 0))
|
||||
except Exception:
|
||||
return 0.0
|
||||
|
||||
|
||||
def extract_audio(video_path: str, output_wav: str,
|
||||
ffmpeg_bin: str | None = None,
|
||||
progress_cb=None) -> str:
|
||||
"""
|
||||
Extract audio from video file as 16kHz mono WAV.
|
||||
Required format for inaSpeechSegmenter.
|
||||
|
||||
progress_cb: optional callable(message: str) for status updates.
|
||||
"""
|
||||
ffmpeg = ffmpeg_bin or find_tool("ffmpeg")
|
||||
if not ffmpeg:
|
||||
raise FileNotFoundError("ffmpeg not found. Install ffmpeg and add to PATH.")
|
||||
|
||||
if progress_cb:
|
||||
progress_cb(f"Extracting audio from: {Path(video_path).name}")
|
||||
|
||||
cmd = [
|
||||
ffmpeg,
|
||||
"-i", video_path,
|
||||
"-vn",
|
||||
"-acodec", "pcm_s16le",
|
||||
"-ar", "16000",
|
||||
"-ac", "1",
|
||||
"-y",
|
||||
output_wav,
|
||||
]
|
||||
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=1800,
|
||||
**_NO_WINDOW,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
raise RuntimeError(f"ffmpeg failed:\n{result.stderr[-500:]}")
|
||||
|
||||
if progress_cb:
|
||||
size_mb = os.path.getsize(output_wav) / (1024 * 1024)
|
||||
progress_cb(f"Audio extracted: {size_mb:.1f} MB")
|
||||
|
||||
return output_wav
|
||||
118
core/segmenter.py
Normal file
118
core/segmenter.py
Normal file
@@ -0,0 +1,118 @@
|
||||
"""
|
||||
Singing segment detection via inaSpeechSegmenter.
|
||||
|
||||
Pipeline:
|
||||
1. Run inaSpeechSegmenter to classify speech/music/noise
|
||||
2. Filter for 'music' segments (singing voice is classified as music)
|
||||
3. Merge nearby segments, apply padding, filter by minimum duration
|
||||
"""
|
||||
|
||||
import time
|
||||
from utils.formats import format_timecode
|
||||
|
||||
|
||||
def check_segmenter_available() -> tuple[bool, str]:
|
||||
"""Check if inaSpeechSegmenter is installed."""
|
||||
try:
|
||||
from inaSpeechSegmenter import Segmenter # noqa: F401
|
||||
return True, ""
|
||||
except ImportError:
|
||||
return False, (
|
||||
"inaSpeechSegmenter is not installed.\n\n"
|
||||
"Install it with:\n"
|
||||
" pip install inaSpeechSegmenter\n\n"
|
||||
"You also need TensorFlow:\n"
|
||||
" pip install tensorflow[and-cuda] (GPU)\n"
|
||||
" pip install tensorflow (CPU)"
|
||||
)
|
||||
|
||||
|
||||
def run_segmentation(wav_path: str, progress_cb=None) -> list:
|
||||
"""
|
||||
Run inaSpeechSegmenter on audio file.
|
||||
Returns list of (label, start_sec, end_sec) tuples.
|
||||
Labels: 'music', 'speech', 'noise', 'noEnergy'
|
||||
"""
|
||||
from inaSpeechSegmenter import Segmenter
|
||||
|
||||
if progress_cb:
|
||||
progress_cb("Loading segmentation model…")
|
||||
|
||||
start_time = time.time()
|
||||
seg = Segmenter(vad_engine='smn', detect_gender=False)
|
||||
|
||||
if progress_cb:
|
||||
progress_cb("Running audio segmentation (this may take a while)…")
|
||||
|
||||
segments = seg(wav_path)
|
||||
elapsed = time.time() - start_time
|
||||
|
||||
if progress_cb:
|
||||
progress_cb(f"Segmentation complete in {elapsed:.1f}s — {len(segments)} raw segments")
|
||||
|
||||
return segments
|
||||
|
||||
|
||||
def merge_singing_segments(
|
||||
segments: list,
|
||||
max_gap: float = 10.0,
|
||||
min_duration: float = 20.0,
|
||||
padding: float = 2.0,
|
||||
) -> list:
|
||||
"""
|
||||
Filter for 'music' segments and merge nearby ones.
|
||||
|
||||
Returns list of dicts:
|
||||
[{"index", "start", "end", "duration", "original_start", "original_end"}, ...]
|
||||
"""
|
||||
music_segs = [(start, end) for label, start, end in segments if label == "music"]
|
||||
|
||||
if not music_segs:
|
||||
return []
|
||||
|
||||
music_segs.sort(key=lambda x: x[0])
|
||||
|
||||
# Merge segments with gaps smaller than max_gap
|
||||
merged = []
|
||||
current_start, current_end = music_segs[0]
|
||||
|
||||
for start, end in music_segs[1:]:
|
||||
if start - current_end <= max_gap:
|
||||
current_end = max(current_end, end)
|
||||
else:
|
||||
merged.append((current_start, current_end))
|
||||
current_start, current_end = start, end
|
||||
merged.append((current_start, current_end))
|
||||
|
||||
# Apply padding and minimum duration filter
|
||||
results = []
|
||||
idx = 1
|
||||
for start, end in merged:
|
||||
duration = end - start
|
||||
if duration >= min_duration:
|
||||
padded_start = max(0, start - padding)
|
||||
padded_end = end + padding
|
||||
results.append({
|
||||
"index": idx,
|
||||
"start": padded_start,
|
||||
"end": padded_end,
|
||||
"duration": padded_end - padded_start,
|
||||
"original_start": start,
|
||||
"original_end": end,
|
||||
})
|
||||
idx += 1
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def get_segment_stats(segments: list) -> dict:
|
||||
"""Get summary statistics from raw segmentation output."""
|
||||
type_counts = {}
|
||||
type_durations = {}
|
||||
for label, start, end in segments:
|
||||
type_counts[label] = type_counts.get(label, 0) + 1
|
||||
type_durations[label] = type_durations.get(label, 0) + (end - start)
|
||||
return {
|
||||
"counts": type_counts,
|
||||
"durations": type_durations,
|
||||
}
|
||||
@@ -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()
|
||||
505
ui/tab_detect.py
Normal file
505
ui/tab_detect.py
Normal file
@@ -0,0 +1,505 @@
|
||||
"""
|
||||
Tab 2: Song Detection
|
||||
Detect singing segments in a video and export EDL/CSV/JSON for DaVinci Resolve.
|
||||
"""
|
||||
|
||||
import os
|
||||
import threading
|
||||
import tkinter as tk
|
||||
from pathlib import Path
|
||||
from tkinter import filedialog, messagebox
|
||||
|
||||
import customtkinter as ctk
|
||||
|
||||
from core.fftools import find_tool, extract_audio
|
||||
from core.segmenter import check_segmenter_available, run_segmentation, merge_singing_segments, get_segment_stats
|
||||
from core.exporters import export_edl, export_csv, export_markers_csv, export_json, export_ffmpeg_script
|
||||
from utils.formats import format_timecode, fmt_dur, fmt_elapsed
|
||||
from ui.theme import BTN_NEUTRAL, ACCENT_PURPLE, ACCENT_GREEN
|
||||
|
||||
|
||||
class DetectTab:
|
||||
"""Song detection tab."""
|
||||
|
||||
VIDEO_EXTS = {".mp4", ".mov", ".mkv", ".avi", ".flv",
|
||||
".ts", ".wmv", ".m4v", ".webm", ".mts", ".m2ts"}
|
||||
|
||||
def __init__(self, parent_frame, app):
|
||||
self.frame = parent_frame
|
||||
self.app = app
|
||||
self.ffmpeg = app.ffmpeg
|
||||
|
||||
# State
|
||||
self.input_path_var = tk.StringVar(value="")
|
||||
self.output_dir_var = tk.StringVar(value="")
|
||||
self.gap_var = tk.DoubleVar(value=10.0)
|
||||
self.min_dur_var = tk.DoubleVar(value=20.0)
|
||||
self.padding_var = tk.DoubleVar(value=2.0)
|
||||
self.fps_var = tk.DoubleVar(value=29.97)
|
||||
self.autocut_var = tk.BooleanVar(value=False)
|
||||
|
||||
self._running = False
|
||||
self._stop_req = False
|
||||
|
||||
# Results
|
||||
self._singing_segments = []
|
||||
self._raw_segments = []
|
||||
|
||||
self._build_ui()
|
||||
self._check_deps()
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════════
|
||||
# UI
|
||||
# ══════════════════════════════════════════════════════════════════════════
|
||||
|
||||
def _build_ui(self):
|
||||
scroll = ctk.CTkScrollableFrame(self.frame, corner_radius=12)
|
||||
scroll.pack(fill="both", expand=True, padx=4, pady=4)
|
||||
|
||||
self._build_sec_input(scroll)
|
||||
self._build_sec_params(scroll)
|
||||
self._build_sec_run(scroll)
|
||||
self._build_sec_results(scroll)
|
||||
|
||||
def _section(self, parent, title):
|
||||
f = ctk.CTkFrame(parent, corner_radius=12)
|
||||
hdr = ctk.CTkFrame(f, fg_color="transparent")
|
||||
hdr.pack(fill="x", padx=14, pady=(10, 0))
|
||||
ctk.CTkLabel(hdr, text=title,
|
||||
font=ctk.CTkFont(size=13, weight="bold")).pack(side="left")
|
||||
ctk.CTkFrame(f, height=1, fg_color=("gray78", "gray32")).pack(
|
||||
fill="x", padx=14, pady=(6, 0))
|
||||
return f
|
||||
|
||||
# ── Input ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def _build_sec_input(self, parent):
|
||||
sec = self._section(parent, "1 · Input Video")
|
||||
sec.pack(fill="x", pady=(0, 12))
|
||||
|
||||
body = ctk.CTkFrame(sec, fg_color="transparent")
|
||||
body.pack(fill="x", padx=14, pady=(10, 4))
|
||||
|
||||
r1 = ctk.CTkFrame(body, fg_color="transparent")
|
||||
r1.pack(fill="x", pady=(0, 4))
|
||||
ctk.CTkLabel(r1, text="Video file:", width=96, anchor="w").pack(side="left")
|
||||
ctk.CTkEntry(r1, textvariable=self.input_path_var).pack(side="left", fill="x", expand=True)
|
||||
ctk.CTkButton(r1, text="Browse…", width=100, command=self._pick_input,
|
||||
**BTN_NEUTRAL).pack(side="left", padx=(8, 0))
|
||||
|
||||
# "Use Concat Output" button
|
||||
self.use_concat_btn = ctk.CTkButton(
|
||||
body, text="📎 Use Concat Output", width=200,
|
||||
fg_color=ACCENT_PURPLE["fg"], text_color=ACCENT_PURPLE["text"],
|
||||
hover_color=ACCENT_PURPLE["hover"],
|
||||
command=self._use_concat_output)
|
||||
self.use_concat_btn.pack(anchor="w", pady=(4, 4))
|
||||
|
||||
r2 = ctk.CTkFrame(body, fg_color="transparent")
|
||||
r2.pack(fill="x", pady=(4, 0))
|
||||
ctk.CTkLabel(r2, text="Output dir:", width=96, anchor="w").pack(side="left")
|
||||
ctk.CTkEntry(r2, textvariable=self.output_dir_var).pack(side="left", fill="x", expand=True)
|
||||
ctk.CTkButton(r2, text="Browse…", width=100, command=self._pick_output_dir,
|
||||
**BTN_NEUTRAL).pack(side="left", padx=(8, 0))
|
||||
|
||||
ctk.CTkLabel(sec, text="Output defaults to same directory as input video.",
|
||||
font=ctk.CTkFont(size=11), text_color=("gray50", "gray55")
|
||||
).pack(anchor="w", padx=14, pady=(4, 10))
|
||||
|
||||
# ── Parameters ────────────────────────────────────────────────────────────
|
||||
|
||||
def _build_sec_params(self, parent):
|
||||
sec = self._section(parent, "2 · Detection Parameters")
|
||||
sec.pack(fill="x", pady=(0, 12))
|
||||
|
||||
body = ctk.CTkFrame(sec, fg_color="transparent")
|
||||
body.pack(fill="x", padx=14, pady=(10, 10))
|
||||
|
||||
params = [
|
||||
("Merge gap (s):", self.gap_var, 1, 60, "Max silence between song parts to merge"),
|
||||
("Min duration (s):", self.min_dur_var, 5, 120, "Ignore segments shorter than this"),
|
||||
("Padding (s):", self.padding_var, 0, 15, "Extra seconds before/after each song"),
|
||||
("FPS:", self.fps_var, 23, 60, "Video frame rate for timecodes"),
|
||||
]
|
||||
|
||||
for label_text, var, lo, hi, tooltip in params:
|
||||
row = ctk.CTkFrame(body, fg_color="transparent")
|
||||
row.pack(fill="x", pady=3)
|
||||
|
||||
ctk.CTkLabel(row, text=label_text, width=130, anchor="w",
|
||||
font=ctk.CTkFont(size=12)).pack(side="left")
|
||||
|
||||
slider = ctk.CTkSlider(row, from_=lo, to=hi, variable=var,
|
||||
width=200, number_of_steps=max(1, hi - lo))
|
||||
slider.pack(side="left", padx=(0, 8))
|
||||
|
||||
val_lbl = ctk.CTkLabel(row, text=f"{var.get():.1f}", width=50, anchor="w",
|
||||
font=ctk.CTkFont(size=12, weight="bold"))
|
||||
val_lbl.pack(side="left")
|
||||
|
||||
ctk.CTkLabel(row, text=tooltip, font=ctk.CTkFont(size=11),
|
||||
text_color=("gray50", "gray55")).pack(side="left", padx=(10, 0))
|
||||
|
||||
# Update label on slider move
|
||||
var.trace_add("write", lambda *_, v=var, l=val_lbl: l.configure(text=f"{v.get():.1f}"))
|
||||
|
||||
# Auto-cut checkbox
|
||||
ctk.CTkCheckBox(body, text="Generate auto-cut ffmpeg script (no DaVinci needed)",
|
||||
variable=self.autocut_var).pack(anchor="w", pady=(8, 0))
|
||||
|
||||
# ── Run ───────────────────────────────────────────────────────────────────
|
||||
|
||||
def _build_sec_run(self, parent):
|
||||
sec = self._section(parent, "3 · Detect")
|
||||
sec.pack(fill="x", pady=(0, 12))
|
||||
|
||||
body = ctk.CTkFrame(sec, fg_color="transparent")
|
||||
body.pack(fill="x", padx=14, pady=(12, 6))
|
||||
|
||||
btn_row = ctk.CTkFrame(body, fg_color="transparent")
|
||||
btn_row.pack(fill="x", pady=(0, 10))
|
||||
|
||||
self.detect_btn = ctk.CTkButton(
|
||||
btn_row, text="🎤 Start Detection", width=230, height=42,
|
||||
font=ctk.CTkFont(size=14, weight="bold"),
|
||||
fg_color=ACCENT_GREEN["fg"], hover_color=ACCENT_GREEN["hover"],
|
||||
command=self.start_detect)
|
||||
self.detect_btn.pack(side="left")
|
||||
|
||||
self.stop_btn = ctk.CTkButton(
|
||||
btn_row, text="■ Stop", width=106, height=42,
|
||||
fg_color=("gray80", "gray25"), text_color=("gray10", "gray90"),
|
||||
hover_color=("#FCA5A5", "#7F1D1D"),
|
||||
command=self._stop, state="disabled")
|
||||
self.stop_btn.pack(side="left", padx=(12, 0))
|
||||
|
||||
# Dependency warning
|
||||
self.dep_lbl = ctk.CTkLabel(body, text="", font=ctk.CTkFont(size=11),
|
||||
text_color="#F97316", wraplength=600, anchor="w", justify="left")
|
||||
self.dep_lbl.pack(fill="x", pady=(0, 4))
|
||||
|
||||
# Progress
|
||||
self.progress = ctk.CTkProgressBar(body, height=16, corner_radius=8, mode="indeterminate")
|
||||
self.progress.pack(fill="x", pady=(0, 6))
|
||||
self.progress.stop()
|
||||
self.progress.set(0)
|
||||
|
||||
self.status_lbl = ctk.CTkLabel(body, text="", anchor="w", font=ctk.CTkFont(size=12))
|
||||
self.status_lbl.pack(fill="x")
|
||||
|
||||
ctk.CTkFrame(sec, height=6, fg_color="transparent").pack()
|
||||
|
||||
# ── Results ───────────────────────────────────────────────────────────────
|
||||
|
||||
def _build_sec_results(self, parent):
|
||||
sec = self._section(parent, "4 · Results")
|
||||
sec.pack(fill="x", pady=(0, 14))
|
||||
|
||||
self.results_body = ctk.CTkFrame(sec, fg_color="transparent")
|
||||
self.results_body.pack(fill="x", padx=14, pady=(10, 10))
|
||||
|
||||
self.results_lbl = ctk.CTkLabel(
|
||||
self.results_body, text="No results yet. Run detection first.",
|
||||
font=ctk.CTkFont(size=12), text_color=("gray50", "gray55"))
|
||||
self.results_lbl.pack(anchor="w")
|
||||
|
||||
# Song list (populated after detection)
|
||||
self.song_list_frame = ctk.CTkFrame(self.results_body, fg_color="transparent")
|
||||
|
||||
# Export buttons (hidden until results)
|
||||
self.export_frame = ctk.CTkFrame(sec, fg_color="transparent")
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════════
|
||||
# Logic
|
||||
# ══════════════════════════════════════════════════════════════════════════
|
||||
|
||||
def _check_deps(self):
|
||||
available, msg = check_segmenter_available()
|
||||
if not available:
|
||||
self.dep_lbl.configure(text=f"⚠ {msg}")
|
||||
self.detect_btn.configure(state="disabled")
|
||||
else:
|
||||
self.dep_lbl.configure(text="")
|
||||
|
||||
if not self.ffmpeg:
|
||||
self.dep_lbl.configure(text="⚠ ffmpeg not found. Install ffmpeg and add to PATH.")
|
||||
self.detect_btn.configure(state="disabled")
|
||||
|
||||
def _pick_input(self):
|
||||
f = filedialog.askopenfilename(
|
||||
title="Select video file",
|
||||
filetypes=[("Video files", "*.mp4 *.mov *.mkv *.avi *.flv *.ts *.wmv *.m4v *.webm *.mts"),
|
||||
("All files", "*.*")])
|
||||
if f:
|
||||
self.input_path_var.set(f)
|
||||
if not self.output_dir_var.get().strip():
|
||||
self.output_dir_var.set(str(Path(f).parent))
|
||||
|
||||
def _pick_output_dir(self):
|
||||
d = filedialog.askdirectory(title="Choose output folder")
|
||||
if d:
|
||||
self.output_dir_var.set(d)
|
||||
|
||||
def _use_concat_output(self):
|
||||
"""Fill input from the Concat tab's output path."""
|
||||
if hasattr(self.app, 'concat_tab'):
|
||||
path = self.app.concat_tab.get_last_output_path()
|
||||
if path and Path(path).exists():
|
||||
self.input_path_var.set(path)
|
||||
if not self.output_dir_var.get().strip():
|
||||
self.output_dir_var.set(str(Path(path).parent))
|
||||
self._set_status(f"Loaded concat output: {Path(path).name}")
|
||||
elif path:
|
||||
self.input_path_var.set(path)
|
||||
if not self.output_dir_var.get().strip():
|
||||
self.output_dir_var.set(str(Path(path).parent))
|
||||
self._set_status("Concat output path set (file not yet created — run concat first)")
|
||||
else:
|
||||
messagebox.showinfo("No output", "Set an output path in the Concat tab first.")
|
||||
|
||||
def _set_status(self, text, error=False):
|
||||
color = "#DC2626" if error else ("gray10", "gray90")
|
||||
self.status_lbl.configure(text=text, text_color=color)
|
||||
|
||||
def _stop(self):
|
||||
self._stop_req = True
|
||||
self._set_status("Stopping…")
|
||||
|
||||
def start_detect(self):
|
||||
"""Start the detection pipeline in a background thread."""
|
||||
input_path = self.input_path_var.get().strip()
|
||||
if not input_path:
|
||||
messagebox.showerror("No input", "Select a video file first.")
|
||||
return
|
||||
if not os.path.isfile(input_path):
|
||||
messagebox.showerror("File not found", f"Cannot find:\n{input_path}")
|
||||
return
|
||||
|
||||
available, msg = check_segmenter_available()
|
||||
if not available:
|
||||
messagebox.showerror("Missing dependency", msg)
|
||||
return
|
||||
|
||||
self._running = True
|
||||
self._stop_req = False
|
||||
self.detect_btn.configure(state="disabled")
|
||||
self.stop_btn.configure(state="normal")
|
||||
self.progress.configure(mode="indeterminate")
|
||||
self.progress.start()
|
||||
self._set_status("Starting detection…")
|
||||
|
||||
threading.Thread(target=self._detect_worker, args=(input_path,), daemon=True).start()
|
||||
|
||||
def _detect_worker(self, input_path):
|
||||
output_dir = self.output_dir_var.get().strip() or str(Path(input_path).parent)
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
stem = Path(input_path).stem
|
||||
|
||||
wav_path = os.path.join(output_dir, f"{stem}_audio.wav")
|
||||
|
||||
try:
|
||||
# Step 1: Extract audio
|
||||
def progress_cb(msg):
|
||||
self.app.after(0, lambda: self._set_status(msg))
|
||||
|
||||
extract_audio(input_path, wav_path, self.ffmpeg, progress_cb=progress_cb)
|
||||
|
||||
if self._stop_req:
|
||||
self._cleanup_and_finish(wav_path, "Stopped by user.")
|
||||
return
|
||||
|
||||
# Step 2: Run segmentation
|
||||
self.app.after(0, lambda: self._set_status("Running audio segmentation (this may take a while)…"))
|
||||
raw_segments = run_segmentation(wav_path, progress_cb=progress_cb)
|
||||
|
||||
if self._stop_req:
|
||||
self._cleanup_and_finish(wav_path, "Stopped by user.")
|
||||
return
|
||||
|
||||
self._raw_segments = raw_segments
|
||||
|
||||
# Step 3: Merge
|
||||
self.app.after(0, lambda: self._set_status("Merging singing segments…"))
|
||||
singing = merge_singing_segments(
|
||||
raw_segments,
|
||||
max_gap=self.gap_var.get(),
|
||||
min_duration=self.min_dur_var.get(),
|
||||
padding=self.padding_var.get(),
|
||||
)
|
||||
self._singing_segments = singing
|
||||
|
||||
# Step 4: Export
|
||||
if singing:
|
||||
fps = self.fps_var.get()
|
||||
source_name = Path(input_path).name
|
||||
|
||||
edl_path = os.path.join(output_dir, f"{stem}_singing.edl")
|
||||
csv_path = os.path.join(output_dir, f"{stem}_singing.csv")
|
||||
markers_path = os.path.join(output_dir, f"{stem}_markers.csv")
|
||||
json_path = os.path.join(output_dir, f"{stem}_singing.json")
|
||||
|
||||
export_edl(singing, edl_path, fps=fps,
|
||||
title=f"{stem} - Singing", source_filename=source_name)
|
||||
export_csv(singing, csv_path)
|
||||
export_markers_csv(singing, markers_path, fps=fps)
|
||||
export_json(singing, json_path)
|
||||
|
||||
if self.autocut_var.get():
|
||||
export_ffmpeg_script(singing, input_path, output_dir)
|
||||
|
||||
total_singing = sum(s["duration"] for s in singing)
|
||||
result_msg = (f"✓ Found {len(singing)} songs · "
|
||||
f"Total singing: {format_timecode(total_singing)} · "
|
||||
f"Exported to: {output_dir}")
|
||||
|
||||
self.app.after(0, lambda: self._show_results(singing, edl_path, output_dir))
|
||||
else:
|
||||
result_msg = "No singing segments detected. Try lowering min duration or increasing gap."
|
||||
|
||||
# Cleanup WAV
|
||||
if os.path.exists(wav_path):
|
||||
os.remove(wav_path)
|
||||
|
||||
self.app.after(0, lambda: self._set_status(result_msg))
|
||||
|
||||
except Exception as e:
|
||||
self.app.after(0, lambda: self._set_status(f"Error: {e}", error=True))
|
||||
if os.path.exists(wav_path):
|
||||
try:
|
||||
os.remove(wav_path)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
self.app.after(0, self._finish_detect)
|
||||
|
||||
def _cleanup_and_finish(self, wav_path, msg):
|
||||
if os.path.exists(wav_path):
|
||||
try:
|
||||
os.remove(wav_path)
|
||||
except Exception:
|
||||
pass
|
||||
self.app.after(0, lambda: self._set_status(msg))
|
||||
self.app.after(0, self._finish_detect)
|
||||
|
||||
def _finish_detect(self):
|
||||
self._running = False
|
||||
self._stop_req = False
|
||||
self.detect_btn.configure(state="normal")
|
||||
self.stop_btn.configure(state="disabled")
|
||||
self.progress.stop()
|
||||
self.progress.configure(mode="determinate")
|
||||
self.progress.set(1.0 if self._singing_segments else 0)
|
||||
|
||||
def _show_results(self, segments, edl_path, output_dir):
|
||||
"""Display detection results in the Results section."""
|
||||
# Clear previous
|
||||
for w in self.song_list_frame.winfo_children():
|
||||
w.destroy()
|
||||
self.export_frame.pack_forget()
|
||||
|
||||
self.results_lbl.configure(
|
||||
text=f"Found {len(segments)} singing segments:",
|
||||
text_color=("gray10", "gray90"))
|
||||
|
||||
self.song_list_frame.pack(fill="x", pady=(8, 0))
|
||||
|
||||
# Header
|
||||
hdr = ctk.CTkFrame(self.song_list_frame, fg_color=("gray88", "gray20"), corner_radius=8)
|
||||
hdr.pack(fill="x", pady=(0, 4))
|
||||
for text, w in [("#", 40), ("Start", 90), ("End", 90), ("Duration", 80)]:
|
||||
ctk.CTkLabel(hdr, text=text, width=w, anchor="center",
|
||||
font=ctk.CTkFont(size=11, weight="bold")).pack(side="left", padx=4, pady=4)
|
||||
|
||||
# Rows
|
||||
for seg in segments:
|
||||
row = ctk.CTkFrame(self.song_list_frame, corner_radius=6, height=30)
|
||||
row.pack(fill="x", pady=2)
|
||||
row.pack_propagate(False)
|
||||
|
||||
vals = [
|
||||
(f"Song {seg['index']}", 40),
|
||||
(format_timecode(seg["start"]), 90),
|
||||
(format_timecode(seg["end"]), 90),
|
||||
(f"{seg['duration']:.0f}s", 80),
|
||||
]
|
||||
for text, w in vals:
|
||||
ctk.CTkLabel(row, text=text, width=w, anchor="center",
|
||||
font=ctk.CTkFont(size=11)).pack(side="left", padx=4)
|
||||
|
||||
# Total
|
||||
total = sum(s["duration"] for s in segments)
|
||||
ctk.CTkLabel(self.song_list_frame,
|
||||
text=f"Total singing time: {format_timecode(total)}",
|
||||
font=ctk.CTkFont(size=12, weight="bold")).pack(anchor="w", pady=(8, 0))
|
||||
|
||||
# Export info
|
||||
self.export_frame.pack(fill="x", padx=14, pady=(4, 10))
|
||||
for w in self.export_frame.winfo_children():
|
||||
w.destroy()
|
||||
|
||||
ctk.CTkLabel(self.export_frame,
|
||||
text=f"Files exported to: {output_dir}",
|
||||
font=ctk.CTkFont(size=11), text_color=("gray45", "gray55"),
|
||||
wraplength=600, anchor="w", justify="left").pack(anchor="w")
|
||||
|
||||
info_lines = [
|
||||
"• _singing.edl → DaVinci Resolve: File → Import → Timeline",
|
||||
"• _markers.csv → DaVinci Resolve: Timeline → Import Markers from CSV",
|
||||
"• _singing.csv → Human-readable segment list",
|
||||
"• _singing.json → Programmatic use",
|
||||
]
|
||||
if self.autocut_var.get():
|
||||
info_lines.append("• _autocut.bat → Run to auto-cut singing segments (no DaVinci needed)")
|
||||
|
||||
for line in info_lines:
|
||||
ctk.CTkLabel(self.export_frame, text=line,
|
||||
font=ctk.CTkFont(size=11),
|
||||
text_color=("gray50", "gray55"), anchor="w").pack(anchor="w")
|
||||
|
||||
# ── Public API for Quick Flow ─────────────────────────────────────────────
|
||||
|
||||
def run_detect_headless(self, input_path: str, output_dir: str,
|
||||
gap: float, min_dur: float, padding: float,
|
||||
fps: float, autocut: bool,
|
||||
progress_cb=None) -> list:
|
||||
"""
|
||||
Run detection synchronously (called from Quick Flow worker thread).
|
||||
Returns list of singing segments.
|
||||
"""
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
stem = Path(input_path).stem
|
||||
wav_path = os.path.join(output_dir, f"{stem}_audio.wav")
|
||||
|
||||
try:
|
||||
extract_audio(input_path, wav_path, self.ffmpeg, progress_cb=progress_cb)
|
||||
|
||||
raw_segments = run_segmentation(wav_path, progress_cb=progress_cb)
|
||||
|
||||
if progress_cb:
|
||||
progress_cb("Merging singing segments…")
|
||||
|
||||
singing = merge_singing_segments(raw_segments,
|
||||
max_gap=gap, min_duration=min_dur, padding=padding)
|
||||
|
||||
if singing:
|
||||
source_name = Path(input_path).name
|
||||
export_edl(singing, os.path.join(output_dir, f"{stem}_singing.edl"),
|
||||
fps=fps, title=f"{stem} - Singing", source_filename=source_name)
|
||||
export_csv(singing, os.path.join(output_dir, f"{stem}_singing.csv"))
|
||||
export_markers_csv(singing, os.path.join(output_dir, f"{stem}_markers.csv"), fps=fps)
|
||||
export_json(singing, os.path.join(output_dir, f"{stem}_singing.json"))
|
||||
if autocut:
|
||||
export_ffmpeg_script(singing, input_path, output_dir)
|
||||
|
||||
if os.path.exists(wav_path):
|
||||
os.remove(wav_path)
|
||||
|
||||
return singing
|
||||
|
||||
except Exception:
|
||||
if os.path.exists(wav_path):
|
||||
try:
|
||||
os.remove(wav_path)
|
||||
except Exception:
|
||||
pass
|
||||
raise
|
||||
64
ui/theme.py
Normal file
64
ui/theme.py
Normal file
@@ -0,0 +1,64 @@
|
||||
"""
|
||||
Theme, DPI, and shared styling constants.
|
||||
"""
|
||||
|
||||
import sys
|
||||
|
||||
# ── DPI awareness (call before any Tk/CTk window) ────────────────────────────
|
||||
def setup_dpi():
|
||||
if sys.platform == "win32":
|
||||
try:
|
||||
import ctypes
|
||||
ctypes.windll.shcore.SetProcessDpiAwareness(2)
|
||||
except Exception:
|
||||
try:
|
||||
import ctypes
|
||||
ctypes.windll.user32.SetProcessDPIAware()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def dpi_scale(widget) -> float:
|
||||
try:
|
||||
return max(1.0, widget.winfo_fpixels("1i") / 96.0)
|
||||
except Exception:
|
||||
return 1.0
|
||||
|
||||
|
||||
# ── Color accents ─────────────────────────────────────────────────────────────
|
||||
# Each feature area has its own accent color scheme: (light_mode, dark_mode)
|
||||
|
||||
ACCENT_BLUE = {
|
||||
"fg": ("#3B82F6", "#2563EB"),
|
||||
"hover": ("#2563EB", "#1D4ED8"),
|
||||
}
|
||||
|
||||
ACCENT_PURPLE = {
|
||||
"fg": ("#EDE9FE", "#3B1F6E"),
|
||||
"text": ("#5B21B6", "#C4B5FD"),
|
||||
"hover": ("#DDD6FE", "#4C2889"),
|
||||
"preview": ("#7C3AED", "#A78BFA"),
|
||||
}
|
||||
|
||||
ACCENT_GREEN = {
|
||||
"fg": ("#10B981", "#059669"),
|
||||
"hover": ("#059669", "#047857"),
|
||||
}
|
||||
|
||||
ACCENT_AMBER = {
|
||||
"fg": ("#F59E0B", "#D97706"),
|
||||
"hover": ("#D97706", "#B45309"),
|
||||
}
|
||||
|
||||
# Neutral button style
|
||||
BTN_NEUTRAL = {
|
||||
"fg_color": ("gray80", "gray25"),
|
||||
"text_color": ("gray10", "gray90"),
|
||||
"hover_color": ("gray70", "gray35"),
|
||||
}
|
||||
|
||||
BTN_SIDE = {
|
||||
"fg_color": ("gray82", "gray22"),
|
||||
"text_color": ("gray10", "gray90"),
|
||||
"hover_color": ("gray72", "gray32"),
|
||||
}
|
||||
42
utils/formats.py
Normal file
42
utils/formats.py
Normal file
@@ -0,0 +1,42 @@
|
||||
"""
|
||||
Timecode & duration formatting utilities.
|
||||
"""
|
||||
|
||||
|
||||
def fmt_dur(seconds: float) -> str:
|
||||
"""Format a duration as H:MM:SS or M:SS for display."""
|
||||
if seconds < 0:
|
||||
return "…"
|
||||
if seconds == 0:
|
||||
return "—"
|
||||
s = int(round(seconds))
|
||||
h, rem = divmod(s, 3600)
|
||||
m, sec = divmod(rem, 60)
|
||||
return f"{h}:{m:02d}:{sec:02d}" if h else f"{m}:{sec:02d}"
|
||||
|
||||
|
||||
def fmt_elapsed(s: float) -> str:
|
||||
if s < 60:
|
||||
return f"{int(s)}s"
|
||||
m, sec = divmod(int(s), 60)
|
||||
return f"{m}m {sec:02d}s"
|
||||
|
||||
|
||||
def format_timecode(seconds: float) -> str:
|
||||
"""Format seconds as HH:MM:SS."""
|
||||
h = int(seconds // 3600)
|
||||
m = int((seconds % 3600) // 60)
|
||||
s = int(seconds % 60)
|
||||
return f"{h:02d}:{m:02d}:{s:02d}"
|
||||
|
||||
|
||||
def seconds_to_smpte(seconds: float, fps: float = 29.97) -> str:
|
||||
"""Convert seconds to SMPTE timecode HH:MM:SS:FF."""
|
||||
total_frames = int(seconds * fps)
|
||||
ff = total_frames % int(round(fps))
|
||||
total_seconds = total_frames // int(round(fps))
|
||||
ss = total_seconds % 60
|
||||
total_minutes = total_seconds // 60
|
||||
mm = total_minutes % 60
|
||||
hh = total_minutes // 60
|
||||
return f"{hh:02d}:{mm:02d}:{ss:02d}:{ff:02d}"
|
||||
Reference in New Issue
Block a user