7 Commits
Author SHA1 Message Date
youfuandClaude Opus 4.8 89dd68fe15 docs: add lineage README mapping v1→v6 evolution
Overview table, engine evolution notes, and per-version caveats.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-01-06 11:00:00 +00:00
youfuandClaude Opus 4.8 1c4c45499f v6: Tauri (Rust + TS) rewrite, Python as ML sidecar
Cross-platform Stream Studio rewrite. UI/ffmpeg/exports move to Rust+TypeScript; Python shrinks to a thin sidecar that runs inaSpeechSegmenter and emits line-delimited JSON. Detection slice extracted from stream-studio.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-01-06 10:00:00 +00:00
youfuandClaude Opus 4.8 878be5fe4a v5: refactor into modular stream_tools desktop app
Same inaSpeechSegmenter engine split into reusable core/ (fftools, segmenter, exporters) + utils/ and a CustomTkinter multi-tab UI. Adds JSON and ffmpeg-script export. Extracted from the stream_tools app.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-01-05 10:00:00 +00:00
youfuandClaude Opus 4.8 b70402932e v4: batch processing layer over the CLI engine
Wraps detect_singing to process whole folders of recordings at once: --recursive, --pattern, --output-dir, --auto-cut.

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

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-01-03 10:00:00 +00:00
youfuandClaude Opus 4.8 36e5473faf v2: transformers-pipeline detector with CustomTkinter GUI
Switches inference to a transformers pipeline (+pyannote segments), adds overlapping-chunk processing and CSV timecode export for DaVinci.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-01-02 10:00:00 +00:00
youfuandClaude Opus 4.8 24812ee204 v1: svd prototype — single-file singing detector
HuggingFace matthijs/svd (torch.jit) model + ffmpeg audio extract + tkinter file picker. Proves the concept in one script.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-01-01 10:00:00 +00:00
17 changed files with 3242 additions and 0 deletions
+33
View File
@@ -0,0 +1,33 @@
# sing_detect
Consolidated history of my singing/music-segment detection tools, gathered from
across several projects and ranked by **technical & functional evolution** (not
by file timestamp). The goal of every version is the same:
> Take a long live-stream recording, find the segments where someone is singing,
> and export markers (CSV / EDL / JSON) for fast editing in DaVinci Resolve.
Each version lives in its own folder and was committed as a separate step so the
git history reads as the evolution story.
| Version | Folder | Engine | Interface | Leap |
|---------|--------|--------|-----------|------|
| v1 | `v1_svd_prototype/` | HuggingFace `matthijs/svd` (torch.jit) | tkinter file picker | First prototype — single file, proves the idea |
| v2 | `v2_transformers_gui/` | `transformers` pipeline + pyannote | CustomTkinter GUI | Overlapping-chunk inference, CSV timecode export |
| v3 | `v3_inaseg_cli/` | inaSpeechSegmenter | CLI (argparse) | Robust pipeline: gap-merge, min-duration, EDL+CSV |
| v4 | `v4_inaseg_batch/` | inaSpeechSegmenter | CLI batch | Process whole folders, `--recursive`, `--auto-cut` |
| v5 | `v5_stream_tools_app/` | inaSpeechSegmenter | CustomTkinter multi-tab app | Refactored into modular `core/` engine + UI tab, JSON + ffmpeg-script export |
| v6 | `v6_stream_studio_tauri/` | inaSpeechSegmenter (Python sidecar) | Tauri (Rust + TS) | Cross-platform rewrite; Python reduced to a thin ML sidecar emitting line-JSON |
## Engine evolution
`svd model` → `transformers pipeline` → **`inaSpeechSegmenter`** (settled from v3 on,
which classifies singing voice as "music").
## Notes per version
- **v4** includes a copy of `detect_singing.py` because `batch_detect.py` imports it at runtime.
- **v5** keeps the original package layout (`core/`, `ui/`, `utils/`) so the relative imports resolve; it was extracted from the larger `stream_tools` app.
- **v6** is the detection slice only (`sidecar_detect.py`, `detect.rs`, `detect.ts`), extracted from the `stream-studio` Tauri app — `detect.ts` still references the app's `../dom`, `../ui`, `../api` modules.
## Excluded
`song_detection_cn.py` (an ACRCloud-style fingerprint *song-identification* API client)
was **not** included — it identifies/names a track rather than detecting singing segments.
+108
View File
@@ -0,0 +1,108 @@
import subprocess
import os
import torch
import librosa
import numpy as np
from huggingface_hub import hf_hub_download
import tkinter as tk
from tkinter import filedialog
# ------------------------
# 1. Extract audio from video
# ------------------------
def extract_audio(input_video, output_wav):
cmd = [
"ffmpeg", "-y",
"-i", input_video,
"-vn", # no video
"-ac", "1", # mono
"-ar", "16000", # 16kHz
output_wav
]
subprocess.run(cmd, check=True)
# ------------------------
# 2. Load pretrained model
# ------------------------
def load_model():
repo_id = "matthijs/svd" # HuggingFace singing voice detection model
filename = hf_hub_download(repo_id=repo_id, filename="model.pt")
model = torch.jit.load(filename)
model.eval()
return model
# ------------------------
# 3. Run inference on audio
# ------------------------
def detect_singing(audio_path, model, hop_length=512, threshold=0.5, min_duration=10.0):
y, sr = librosa.load(audio_path, sr=16000)
x = torch.tensor(y).float().unsqueeze(0)
with torch.no_grad():
pred = model(x).squeeze().numpy()
times = librosa.frames_to_time(np.arange(len(pred)), sr=sr, hop_length=hop_length)
segments = []
in_segment = False
seg_start = None
for t, p in zip(times, pred):
if p > threshold and not in_segment:
in_segment = True
seg_start = t
elif p <= threshold and in_segment:
in_segment = False
seg_end = t
if seg_end - seg_start >= min_duration:
segments.append((seg_start, seg_end))
if in_segment:
seg_end = times[-1]
if seg_end - seg_start >= min_duration:
segments.append((seg_start, seg_end))
return segments
# ------------------------
# 4. Export CSV
# ------------------------
def export_csv(segments, out_file="segments.csv"):
with open(out_file, "w", encoding="utf-8") as f:
f.write("start_time,end_time\n")
for s, e in segments:
f.write(f"{s:.2f},{e:.2f}\n")
print(f"[+] Saved: {out_file}")
# ------------------------
# MAIN
# ------------------------
if __name__ == "__main__":
# Tkinter GUI to choose file
root = tk.Tk()
root.withdraw()
input_video = filedialog.askopenfilename(
title="Choose your video file",
filetypes=[("Video files", "*.mp4 *.mkv *.flv *.mov *.avi"), ("All files", "*.*")]
)
if not input_video:
print("No file selected. Exiting.")
exit()
audio_file = "audio.wav"
print(f"[+] Processing: {input_video}")
# Step 1: extract audio
extract_audio(input_video, audio_file)
# Step 2: load model
model = load_model()
# Step 3: run detection
segments = detect_singing(audio_file, model)
# Step 4: save results
export_csv(segments, "segments.csv")
print("[+] Done! Detected singing segments have been saved to segments.csv")
+185
View File
@@ -0,0 +1,185 @@
# Singing Segment Detector
Automatically detect singing segments in live stream recordings and export timeline markers for DaVinci Resolve.
## What It Does
```
4hr stream recording (.mp4/.mkv/.flv)
│
▼ [ffmpeg: extract 16kHz mono audio]
│
▼ [inaSpeechSegmenter: classify speech/music/noise]
│
▼ [merge nearby music segments, filter short ones]
│
├──→ .edl (import as DaVinci Resolve timeline)
├──→ _markers.csv (import as DaVinci Resolve markers)
├──→ .csv (human-readable segment list)
├──→ .json (programmatic use)
└──→ .bat (optional: auto-cut with ffmpeg, no editing needed)
```
**Key insight**: `inaSpeechSegmenter` classifies **singing voice as "music"**. So in a typical singing stream, talking = "speech", singing = "music". We filter for "music" segments and merge nearby ones (a song might have brief pauses between verses).
## Setup (Windows with conda)
### Prerequisites
- **conda** (Anaconda or Miniconda)
- **ffmpeg** in your PATH
- **NVIDIA GPU** recommended (RTX 2080 works great) but CPU also works
### Step 1: Create conda environment
```powershell
conda create -n singing-detector python=3.11 -y
conda activate singing-detector
```
### Step 2: Install PyTorch with CUDA (for GPU acceleration)
```powershell
# For RTX 2080 (CUDA 12.x)
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121
```
### Step 3: Install TensorFlow (required by inaSpeechSegmenter)
```powershell
pip install tensorflow[and-cuda]
```
Or CPU-only (slower but works):
```powershell
pip install tensorflow
```
### Step 4: Install inaSpeechSegmenter and dependencies
```powershell
pip install inaSpeechSegmenter
```
### Step 5: Verify installation
```powershell
python -c "from inaSpeechSegmenter import Segmenter; print('OK!')"
```
## Usage
### Single file
```powershell
conda activate singing-detector
# Basic usage
python detect_singing.py "D:\streams\2026-04-04_stream.mp4"
# Custom settings
python detect_singing.py "D:\streams\2026-04-04_stream.mp4" --gap 15 --min-duration 30
# Output to a specific directory
python detect_singing.py "D:\streams\2026-04-04_stream.mp4" -o "D:\singing_edits"
# Also generate an auto-cut ffmpeg script (no DaVinci needed)
python detect_singing.py "D:\streams\2026-04-04_stream.mp4" --auto-cut
# For 4K F-log footage at 29.97fps (your XT3 settings)
python detect_singing.py "D:\streams\2026-04-04_stream.mp4" --fps 29.97
```
### Batch processing
```powershell
# Process all videos in a folder
python batch_detect.py "D:\streams"
# Process new files only (skip already-done ones)
python batch_detect.py "D:\streams" --skip-existing
# Process and auto-generate singing-only videos
python batch_detect.py "D:\streams" --auto-cut -o "D:\singing_edits"
# Only process .flv files
python batch_detect.py "D:\streams" --pattern "*.flv"
```
## Importing Results into DaVinci Resolve
### Option A: Import EDL as Timeline (recommended)
1. Open your project in DaVinci Resolve
2. Import the original stream video into your Media Pool
3. Go to **File → Import → Timeline**
4. Select the `_singing.edl` file
5. DaVinci will create a new timeline with only the singing segments
### Option B: Import Markers CSV
1. Create a timeline from your stream video
2. Right-click on the timeline → **Timelines → Import Markers from CSV**
3. Select the `_markers.csv` file
4. Blue markers will appear at each singing segment start/end
### Option C: Auto-Cut (no DaVinci needed)
If you used `--auto-cut`, just run the generated `.bat` script:
```powershell
D:\singing_edits\2026-04-04_stream_singing.bat
```
This uses ffmpeg to directly cut and concatenate all singing segments into a single `_singing_only.mp4` file. Fastest option, but no manual review.
## Tuning Parameters
| Parameter | Default | Description |
|-----------|---------|-------------|
| `--gap` | 10s | Max gap between music segments to merge. Increase if singer pauses >10s between verses. |
| `--min-duration` | 20s | Minimum segment length. Lower if singer does very short songs. |
| `--padding` | 2s | Extra seconds before/after each segment. Helps catch song intro/outro. |
| `--fps` | 29.97 | Frame rate for timecode. Use 25 for PAL. |
### Recommended settings for typical singing streams
- **Singer with lots of chatting between songs**: `--gap 10 --min-duration 30`
- **Singer with minimal breaks**: `--gap 5 --min-duration 20`
- **Singer who does short covers/snippets**: `--gap 8 --min-duration 15`
- **Conservative (catch everything)**: `--gap 20 --min-duration 15 --padding 5`
## Troubleshooting
### "Singing voice detected as speech"
This can happen if the singer talks over background music. Try `--gap 15` to merge nearby segments.
### "Too many false positives (non-singing music detected)"
If there's background music during chatting, increase `--min-duration 45` to only keep longer segments (full songs).
### "Segments cut off the beginning/end of songs"
Increase `--padding 5` to add more buffer.
### GPU not being used
Check that TensorFlow detects your GPU:
```python
python -c "import tensorflow as tf; print(tf.config.list_physical_devices('GPU'))"
```
### Processing is very slow
- A 4-hour stream typically takes 5-15 minutes on GPU, 30-60 minutes on CPU.
- Make sure you're using GPU-enabled TensorFlow.
- Close other GPU-heavy applications during processing.
## Output Files
For input `stream_2026-04-04.mp4`, you get:
| File | Purpose |
|------|---------|
| `stream_2026-04-04_singing.edl` | DaVinci Resolve timeline import |
| `stream_2026-04-04_markers.csv` | DaVinci Resolve marker import |
| `stream_2026-04-04_singing.csv` | Human-readable segment list |
| `stream_2026-04-04_singing.json` | Programmatic use |
| `stream_2026-04-04_singing.bat` | Auto-cut script (with `--auto-cut`) |
| `stream_2026-04-04_raw_segments.csv` | All segments for debugging (with `--raw-segments`) |
+598
View File
@@ -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()
+56
View File
@@ -0,0 +1,56 @@
@echo off
REM ============================================
REM Singing Detector - Quick Setup (Windows)
REM ============================================
REM Run this in Anaconda Prompt / PowerShell
REM Prerequisites: conda, ffmpeg in PATH
REM ============================================
echo.
echo ============================================
echo Singing Segment Detector - Setup
echo ============================================
echo.
REM Step 1: Create conda environment
echo [1/4] Creating conda environment (python 3.11)...
call conda create -n singing-detector python=3.11 -y
if %ERRORLEVEL% neq 0 (
echo ERROR: Failed to create conda environment.
pause
exit /b 1
)
REM Activate environment
call conda activate singing-detector
REM Step 2: Install TensorFlow with GPU support
echo.
echo [2/4] Installing TensorFlow (GPU)...
pip install tensorflow[and-cuda]
if %ERRORLEVEL% neq 0 (
echo WARNING: GPU TensorFlow install failed, trying CPU version...
pip install tensorflow
)
REM Step 3: Install inaSpeechSegmenter
echo.
echo [3/4] Installing inaSpeechSegmenter...
pip install inaSpeechSegmenter
REM Step 4: Verify
echo.
echo [4/4] Verifying installation...
python -c "from inaSpeechSegmenter import Segmenter; print('inaSpeechSegmenter OK!')"
python -c "import tensorflow as tf; gpus = tf.config.list_physical_devices('GPU'); print(f'TensorFlow GPUs: {gpus}' if gpus else 'TensorFlow: CPU only')"
echo.
echo ============================================
echo Setup complete!
echo ============================================
echo.
echo Usage:
echo conda activate singing-detector
echo python detect_singing.py "path\to\stream.mp4"
echo.
pause
+183
View File
@@ -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()
+598
View File
@@ -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()
+188
View File
@@ -0,0 +1,188 @@
"""
Export singing segments to various formats:
- EDL (DaVinci Resolve timeline import)
- CSV (human-readable)
- DaVinci Resolve Markers CSV
- JSON (programmatic)
- ffmpeg concat script (auto-cut)
"""
import csv
import json
import os
import sys
from pathlib import Path
from utils.formats import format_timecode, seconds_to_smpte
def export_edl(segments: list, output_path: str, fps: float = 29.97,
title: str = "Singing Segments", source_filename: str = None):
"""Export EDL file for DaVinci Resolve."""
if source_filename:
clip_name = source_filename
reel_name_safe = Path(source_filename).stem[:32].replace(" ", "_")
else:
clip_name = None
reel_name_safe = "001"
rec_offset = 3600.0
with open(output_path, "w", encoding="utf-8") as f:
f.write(f"TITLE: {title}\n")
f.write("FCM: NON-DROP FRAME\n\n")
current_rec_pos = rec_offset
for seg in segments:
edit_num = f"{seg['index']:03d}"
src_in = seconds_to_smpte(seg["start"], fps)
src_out = seconds_to_smpte(seg["end"], fps)
rec_in = seconds_to_smpte(current_rec_pos, fps)
rec_out = seconds_to_smpte(current_rec_pos + seg["duration"], fps)
f.write(f"{edit_num} {reel_name_safe} V C {src_in} {src_out} {rec_in} {rec_out}\n")
if clip_name:
f.write(f"* FROM CLIP NAME: {clip_name}\n")
f.write(f"* COMMENT: Song {seg['index']} - Duration {seg['duration']:.0f}s\n\n")
current_rec_pos += seg["duration"]
def export_csv(segments: list, output_path: str):
"""Export singing segments as CSV."""
with open(output_path, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(["index", "start_sec", "end_sec", "duration_sec",
"start_timecode", "end_timecode"])
for seg in segments:
writer.writerow([
seg["index"],
f"{seg['start']:.2f}",
f"{seg['end']:.2f}",
f"{seg['duration']:.1f}",
format_timecode(seg["start"]),
format_timecode(seg["end"]),
])
def export_markers_csv(segments: list, output_path: str, fps: float = 29.97):
"""Export DaVinci Resolve Markers CSV."""
with open(output_path, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(["#", "Color", "Name", "Start TC", "End TC", "Duration TC", "Notes"])
for seg in segments:
writer.writerow([
seg["index"],
"Blue",
f"Song {seg['index']}",
seconds_to_smpte(seg["start"], fps),
seconds_to_smpte(seg["end"], fps),
seconds_to_smpte(seg["duration"], fps),
f"Duration: {seg['duration']:.0f}s",
])
def export_json(segments: list, output_path: str):
"""Export as JSON."""
with open(output_path, "w", encoding="utf-8") as f:
json.dump(segments, f, indent=2, ensure_ascii=False)
def export_ffmpeg_script(segments: list, video_path: str, output_dir: str):
"""
Export a script that uses ffmpeg to cut and concatenate all
singing segments into a single video file.
"""
video_name = Path(video_path).stem
ext = Path(video_path).suffix
out_video = str(Path(output_dir) / f"{video_name}_singing_only{ext}")
is_windows = os.name == "nt" or sys.platform == "win32"
script_ext = ".bat" if is_windows else ".sh"
script_path = str(Path(output_dir) / f"{video_name}_autocut{script_ext}")
concat_list_path = str(Path(output_dir) / f"{video_name}_concat_list.txt")
lines = []
if is_windows:
lines.append("@echo off")
lines.append("REM Auto-generated ffmpeg script to extract singing segments")
lines.append(f'REM Source: {video_path}')
lines.append("")
else:
lines.append("#!/bin/bash")
lines.append("# Auto-generated ffmpeg script to extract singing segments")
lines.append(f'# Source: {video_path}')
lines.append("")
segment_files = []
for seg in segments:
seg_file = f"_seg_{seg['index']:03d}{ext}"
seg_path = str(Path(output_dir) / seg_file)
segment_files.append(seg_file)
cmd = (f'ffmpeg -y -ss {seg["start"]:.2f} -i "{video_path}" '
f'-t {seg["duration"]:.2f} -c copy "{seg_path}"')
lines.append(f"echo Extracting Song {seg['index']}...")
lines.append(cmd)
lines.append("")
lines.append("echo Creating concat list...")
if is_windows:
lines.append("(")
for sf in segment_files:
seg_path = str(Path(output_dir) / sf)
lines.append(f" echo file '{seg_path}'")
lines.append(f') > "{concat_list_path}"')
else:
for i, sf in enumerate(segment_files):
seg_path = str(Path(output_dir) / sf)
op = ">" if i == 0 else ">>"
lines.append(f"echo \"file '{seg_path}'\" {op} \"{concat_list_path}\"")
lines.append("")
lines.append("echo Concatenating all segments...")
lines.append(f'ffmpeg -y -f concat -safe 0 -i "{concat_list_path}" -c copy "{out_video}"')
lines.append("")
lines.append("echo Cleaning up temp files...")
for sf in segment_files:
seg_path = str(Path(output_dir) / sf)
if is_windows:
lines.append(f'del "{seg_path}"')
else:
lines.append(f'rm -f "{seg_path}"')
if is_windows:
lines.append(f'del "{concat_list_path}"')
else:
lines.append(f'rm -f "{concat_list_path}"')
lines.append("")
lines.append(f'echo Done! Output: {out_video}')
if is_windows:
lines.append("pause")
with open(script_path, "w", encoding="utf-8") as f:
f.write("\n".join(lines))
if not is_windows:
os.chmod(script_path, 0o755)
return script_path
def export_raw_segments(segments: list, output_path: str):
"""Export ALL raw segments (speech/music/noise) as CSV for debugging."""
with open(output_path, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(["label", "start_sec", "end_sec", "duration_sec",
"start_timecode", "end_timecode"])
for label, start, end in segments:
writer.writerow([
label,
f"{start:.2f}",
f"{end:.2f}",
f"{end - start:.1f}",
format_timecode(start),
format_timecode(end),
])
+99
View File
@@ -0,0 +1,99 @@
"""
ffmpeg / ffprobe discovery and probing helpers.
Shared by both concat and singing detection features.
"""
import json
import os
import shutil
import subprocess
import sys
from pathlib import Path
# ── Suppress console windows on Windows ──────────────────────────────────────
_NO_WINDOW: dict = {}
if sys.platform == "win32":
_NO_WINDOW["creationflags"] = subprocess.CREATE_NO_WINDOW
def app_dir() -> str:
if getattr(sys, "frozen", False):
return os.path.dirname(sys.executable)
return os.path.dirname(os.path.abspath(__file__))
def find_tool(name: str) -> str | None:
"""Locate ffmpeg / ffprobe next to the exe / script, then on PATH."""
base = app_dir()
for sub in ["_internal", "..", ""]:
for variant in [name + ".exe", name]:
p = os.path.join(base, sub, variant)
if os.path.exists(p):
return os.path.abspath(p)
return shutil.which(name + ".exe") or shutil.which(name)
def probe_duration(path: str, ffprobe_bin: str | None = None) -> float:
"""Get video/audio duration in seconds via ffprobe."""
ffprobe = ffprobe_bin or find_tool("ffprobe")
if not ffprobe:
return 0.0
try:
r = subprocess.run(
[ffprobe, "-v", "error",
"-show_entries", "format=duration",
"-of", "json", path],
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
text=True, timeout=20,
**_NO_WINDOW,
)
if r.returncode != 0:
return 0.0
data = json.loads(r.stdout or "{}")
return max(0.0, float(data.get("format", {}).get("duration", 0) or 0))
except Exception:
return 0.0
def extract_audio(video_path: str, output_wav: str,
ffmpeg_bin: str | None = None,
progress_cb=None) -> str:
"""
Extract audio from video file as 16kHz mono WAV.
Required format for inaSpeechSegmenter.
progress_cb: optional callable(message: str) for status updates.
"""
ffmpeg = ffmpeg_bin or find_tool("ffmpeg")
if not ffmpeg:
raise FileNotFoundError("ffmpeg not found. Install ffmpeg and add to PATH.")
if progress_cb:
progress_cb(f"Extracting audio from: {Path(video_path).name}")
cmd = [
ffmpeg,
"-i", video_path,
"-vn",
"-acodec", "pcm_s16le",
"-ar", "16000",
"-ac", "1",
"-y",
output_wav,
]
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=1800,
**_NO_WINDOW,
)
if result.returncode != 0:
raise RuntimeError(f"ffmpeg failed:\n{result.stderr[-500:]}")
if progress_cb:
size_mb = os.path.getsize(output_wav) / (1024 * 1024)
progress_cb(f"Audio extracted: {size_mb:.1f} MB")
return output_wav
+118
View File
@@ -0,0 +1,118 @@
"""
Singing segment detection via inaSpeechSegmenter.
Pipeline:
1. Run inaSpeechSegmenter to classify speech/music/noise
2. Filter for 'music' segments (singing voice is classified as music)
3. Merge nearby segments, apply padding, filter by minimum duration
"""
import time
from utils.formats import format_timecode
def check_segmenter_available() -> tuple[bool, str]:
"""Check if inaSpeechSegmenter is installed."""
try:
from inaSpeechSegmenter import Segmenter # noqa: F401
return True, ""
except ImportError:
return False, (
"inaSpeechSegmenter is not installed.\n\n"
"Install it with:\n"
" pip install inaSpeechSegmenter\n\n"
"You also need TensorFlow:\n"
" pip install tensorflow[and-cuda] (GPU)\n"
" pip install tensorflow (CPU)"
)
def run_segmentation(wav_path: str, progress_cb=None) -> list:
"""
Run inaSpeechSegmenter on audio file.
Returns list of (label, start_sec, end_sec) tuples.
Labels: 'music', 'speech', 'noise', 'noEnergy'
"""
from inaSpeechSegmenter import Segmenter
if progress_cb:
progress_cb("Loading segmentation model…")
start_time = time.time()
seg = Segmenter(vad_engine='smn', detect_gender=False)
if progress_cb:
progress_cb("Running audio segmentation (this may take a while)…")
segments = seg(wav_path)
elapsed = time.time() - start_time
if progress_cb:
progress_cb(f"Segmentation complete in {elapsed:.1f}s — {len(segments)} raw segments")
return segments
def merge_singing_segments(
segments: list,
max_gap: float = 10.0,
min_duration: float = 20.0,
padding: float = 2.0,
) -> list:
"""
Filter for 'music' segments and merge nearby ones.
Returns list of dicts:
[{"index", "start", "end", "duration", "original_start", "original_end"}, ...]
"""
music_segs = [(start, end) for label, start, end in segments if label == "music"]
if not music_segs:
return []
music_segs.sort(key=lambda x: x[0])
# Merge segments with gaps smaller than max_gap
merged = []
current_start, current_end = music_segs[0]
for start, end in music_segs[1:]:
if start - current_end <= max_gap:
current_end = max(current_end, end)
else:
merged.append((current_start, current_end))
current_start, current_end = start, end
merged.append((current_start, current_end))
# Apply padding and minimum duration filter
results = []
idx = 1
for start, end in merged:
duration = end - start
if duration >= min_duration:
padded_start = max(0, start - padding)
padded_end = end + padding
results.append({
"index": idx,
"start": padded_start,
"end": padded_end,
"duration": padded_end - padded_start,
"original_start": start,
"original_end": end,
})
idx += 1
return results
def get_segment_stats(segments: list) -> dict:
"""Get summary statistics from raw segmentation output."""
type_counts = {}
type_durations = {}
for label, start, end in segments:
type_counts[label] = type_counts.get(label, 0) + 1
type_durations[label] = type_durations.get(label, 0) + (end - start)
return {
"counts": type_counts,
"durations": type_durations,
}
+505
View File
@@ -0,0 +1,505 @@
"""
Tab 2: Song Detection
Detect singing segments in a video and export EDL/CSV/JSON for DaVinci Resolve.
"""
import os
import threading
import tkinter as tk
from pathlib import Path
from tkinter import filedialog, messagebox
import customtkinter as ctk
from core.fftools import find_tool, extract_audio
from core.segmenter import check_segmenter_available, run_segmentation, merge_singing_segments, get_segment_stats
from core.exporters import export_edl, export_csv, export_markers_csv, export_json, export_ffmpeg_script
from utils.formats import format_timecode, fmt_dur, fmt_elapsed
from ui.theme import BTN_NEUTRAL, ACCENT_PURPLE, ACCENT_GREEN
class DetectTab:
"""Song detection tab."""
VIDEO_EXTS = {".mp4", ".mov", ".mkv", ".avi", ".flv",
".ts", ".wmv", ".m4v", ".webm", ".mts", ".m2ts"}
def __init__(self, parent_frame, app):
self.frame = parent_frame
self.app = app
self.ffmpeg = app.ffmpeg
# State
self.input_path_var = tk.StringVar(value="")
self.output_dir_var = tk.StringVar(value="")
self.gap_var = tk.DoubleVar(value=10.0)
self.min_dur_var = tk.DoubleVar(value=20.0)
self.padding_var = tk.DoubleVar(value=2.0)
self.fps_var = tk.DoubleVar(value=29.97)
self.autocut_var = tk.BooleanVar(value=False)
self._running = False
self._stop_req = False
# Results
self._singing_segments = []
self._raw_segments = []
self._build_ui()
self._check_deps()
# ══════════════════════════════════════════════════════════════════════════
# UI
# ══════════════════════════════════════════════════════════════════════════
def _build_ui(self):
scroll = ctk.CTkScrollableFrame(self.frame, corner_radius=12)
scroll.pack(fill="both", expand=True, padx=4, pady=4)
self._build_sec_input(scroll)
self._build_sec_params(scroll)
self._build_sec_run(scroll)
self._build_sec_results(scroll)
def _section(self, parent, title):
f = ctk.CTkFrame(parent, corner_radius=12)
hdr = ctk.CTkFrame(f, fg_color="transparent")
hdr.pack(fill="x", padx=14, pady=(10, 0))
ctk.CTkLabel(hdr, text=title,
font=ctk.CTkFont(size=13, weight="bold")).pack(side="left")
ctk.CTkFrame(f, height=1, fg_color=("gray78", "gray32")).pack(
fill="x", padx=14, pady=(6, 0))
return f
# ── Input ─────────────────────────────────────────────────────────────────
def _build_sec_input(self, parent):
sec = self._section(parent, "1 · Input Video")
sec.pack(fill="x", pady=(0, 12))
body = ctk.CTkFrame(sec, fg_color="transparent")
body.pack(fill="x", padx=14, pady=(10, 4))
r1 = ctk.CTkFrame(body, fg_color="transparent")
r1.pack(fill="x", pady=(0, 4))
ctk.CTkLabel(r1, text="Video file:", width=96, anchor="w").pack(side="left")
ctk.CTkEntry(r1, textvariable=self.input_path_var).pack(side="left", fill="x", expand=True)
ctk.CTkButton(r1, text="Browse…", width=100, command=self._pick_input,
**BTN_NEUTRAL).pack(side="left", padx=(8, 0))
# "Use Concat Output" button
self.use_concat_btn = ctk.CTkButton(
body, text="📎 Use Concat Output", width=200,
fg_color=ACCENT_PURPLE["fg"], text_color=ACCENT_PURPLE["text"],
hover_color=ACCENT_PURPLE["hover"],
command=self._use_concat_output)
self.use_concat_btn.pack(anchor="w", pady=(4, 4))
r2 = ctk.CTkFrame(body, fg_color="transparent")
r2.pack(fill="x", pady=(4, 0))
ctk.CTkLabel(r2, text="Output dir:", width=96, anchor="w").pack(side="left")
ctk.CTkEntry(r2, textvariable=self.output_dir_var).pack(side="left", fill="x", expand=True)
ctk.CTkButton(r2, text="Browse…", width=100, command=self._pick_output_dir,
**BTN_NEUTRAL).pack(side="left", padx=(8, 0))
ctk.CTkLabel(sec, text="Output defaults to same directory as input video.",
font=ctk.CTkFont(size=11), text_color=("gray50", "gray55")
).pack(anchor="w", padx=14, pady=(4, 10))
# ── Parameters ────────────────────────────────────────────────────────────
def _build_sec_params(self, parent):
sec = self._section(parent, "2 · Detection Parameters")
sec.pack(fill="x", pady=(0, 12))
body = ctk.CTkFrame(sec, fg_color="transparent")
body.pack(fill="x", padx=14, pady=(10, 10))
params = [
("Merge gap (s):", self.gap_var, 1, 60, "Max silence between song parts to merge"),
("Min duration (s):", self.min_dur_var, 5, 120, "Ignore segments shorter than this"),
("Padding (s):", self.padding_var, 0, 15, "Extra seconds before/after each song"),
("FPS:", self.fps_var, 23, 60, "Video frame rate for timecodes"),
]
for label_text, var, lo, hi, tooltip in params:
row = ctk.CTkFrame(body, fg_color="transparent")
row.pack(fill="x", pady=3)
ctk.CTkLabel(row, text=label_text, width=130, anchor="w",
font=ctk.CTkFont(size=12)).pack(side="left")
slider = ctk.CTkSlider(row, from_=lo, to=hi, variable=var,
width=200, number_of_steps=max(1, hi - lo))
slider.pack(side="left", padx=(0, 8))
val_lbl = ctk.CTkLabel(row, text=f"{var.get():.1f}", width=50, anchor="w",
font=ctk.CTkFont(size=12, weight="bold"))
val_lbl.pack(side="left")
ctk.CTkLabel(row, text=tooltip, font=ctk.CTkFont(size=11),
text_color=("gray50", "gray55")).pack(side="left", padx=(10, 0))
# Update label on slider move
var.trace_add("write", lambda *_, v=var, l=val_lbl: l.configure(text=f"{v.get():.1f}"))
# Auto-cut checkbox
ctk.CTkCheckBox(body, text="Generate auto-cut ffmpeg script (no DaVinci needed)",
variable=self.autocut_var).pack(anchor="w", pady=(8, 0))
# ── Run ───────────────────────────────────────────────────────────────────
def _build_sec_run(self, parent):
sec = self._section(parent, "3 · Detect")
sec.pack(fill="x", pady=(0, 12))
body = ctk.CTkFrame(sec, fg_color="transparent")
body.pack(fill="x", padx=14, pady=(12, 6))
btn_row = ctk.CTkFrame(body, fg_color="transparent")
btn_row.pack(fill="x", pady=(0, 10))
self.detect_btn = ctk.CTkButton(
btn_row, text="🎤 Start Detection", width=230, height=42,
font=ctk.CTkFont(size=14, weight="bold"),
fg_color=ACCENT_GREEN["fg"], hover_color=ACCENT_GREEN["hover"],
command=self.start_detect)
self.detect_btn.pack(side="left")
self.stop_btn = ctk.CTkButton(
btn_row, text="■ Stop", width=106, height=42,
fg_color=("gray80", "gray25"), text_color=("gray10", "gray90"),
hover_color=("#FCA5A5", "#7F1D1D"),
command=self._stop, state="disabled")
self.stop_btn.pack(side="left", padx=(12, 0))
# Dependency warning
self.dep_lbl = ctk.CTkLabel(body, text="", font=ctk.CTkFont(size=11),
text_color="#F97316", wraplength=600, anchor="w", justify="left")
self.dep_lbl.pack(fill="x", pady=(0, 4))
# Progress
self.progress = ctk.CTkProgressBar(body, height=16, corner_radius=8, mode="indeterminate")
self.progress.pack(fill="x", pady=(0, 6))
self.progress.stop()
self.progress.set(0)
self.status_lbl = ctk.CTkLabel(body, text="", anchor="w", font=ctk.CTkFont(size=12))
self.status_lbl.pack(fill="x")
ctk.CTkFrame(sec, height=6, fg_color="transparent").pack()
# ── Results ───────────────────────────────────────────────────────────────
def _build_sec_results(self, parent):
sec = self._section(parent, "4 · Results")
sec.pack(fill="x", pady=(0, 14))
self.results_body = ctk.CTkFrame(sec, fg_color="transparent")
self.results_body.pack(fill="x", padx=14, pady=(10, 10))
self.results_lbl = ctk.CTkLabel(
self.results_body, text="No results yet. Run detection first.",
font=ctk.CTkFont(size=12), text_color=("gray50", "gray55"))
self.results_lbl.pack(anchor="w")
# Song list (populated after detection)
self.song_list_frame = ctk.CTkFrame(self.results_body, fg_color="transparent")
# Export buttons (hidden until results)
self.export_frame = ctk.CTkFrame(sec, fg_color="transparent")
# ══════════════════════════════════════════════════════════════════════════
# Logic
# ══════════════════════════════════════════════════════════════════════════
def _check_deps(self):
available, msg = check_segmenter_available()
if not available:
self.dep_lbl.configure(text=f"⚠ {msg}")
self.detect_btn.configure(state="disabled")
else:
self.dep_lbl.configure(text="")
if not self.ffmpeg:
self.dep_lbl.configure(text="⚠ ffmpeg not found. Install ffmpeg and add to PATH.")
self.detect_btn.configure(state="disabled")
def _pick_input(self):
f = filedialog.askopenfilename(
title="Select video file",
filetypes=[("Video files", "*.mp4 *.mov *.mkv *.avi *.flv *.ts *.wmv *.m4v *.webm *.mts"),
("All files", "*.*")])
if f:
self.input_path_var.set(f)
if not self.output_dir_var.get().strip():
self.output_dir_var.set(str(Path(f).parent))
def _pick_output_dir(self):
d = filedialog.askdirectory(title="Choose output folder")
if d:
self.output_dir_var.set(d)
def _use_concat_output(self):
"""Fill input from the Concat tab's output path."""
if hasattr(self.app, 'concat_tab'):
path = self.app.concat_tab.get_last_output_path()
if path and Path(path).exists():
self.input_path_var.set(path)
if not self.output_dir_var.get().strip():
self.output_dir_var.set(str(Path(path).parent))
self._set_status(f"Loaded concat output: {Path(path).name}")
elif path:
self.input_path_var.set(path)
if not self.output_dir_var.get().strip():
self.output_dir_var.set(str(Path(path).parent))
self._set_status("Concat output path set (file not yet created — run concat first)")
else:
messagebox.showinfo("No output", "Set an output path in the Concat tab first.")
def _set_status(self, text, error=False):
color = "#DC2626" if error else ("gray10", "gray90")
self.status_lbl.configure(text=text, text_color=color)
def _stop(self):
self._stop_req = True
self._set_status("Stopping…")
def start_detect(self):
"""Start the detection pipeline in a background thread."""
input_path = self.input_path_var.get().strip()
if not input_path:
messagebox.showerror("No input", "Select a video file first.")
return
if not os.path.isfile(input_path):
messagebox.showerror("File not found", f"Cannot find:\n{input_path}")
return
available, msg = check_segmenter_available()
if not available:
messagebox.showerror("Missing dependency", msg)
return
self._running = True
self._stop_req = False
self.detect_btn.configure(state="disabled")
self.stop_btn.configure(state="normal")
self.progress.configure(mode="indeterminate")
self.progress.start()
self._set_status("Starting detection…")
threading.Thread(target=self._detect_worker, args=(input_path,), daemon=True).start()
def _detect_worker(self, input_path):
output_dir = self.output_dir_var.get().strip() or str(Path(input_path).parent)
os.makedirs(output_dir, exist_ok=True)
stem = Path(input_path).stem
wav_path = os.path.join(output_dir, f"{stem}_audio.wav")
try:
# Step 1: Extract audio
def progress_cb(msg):
self.app.after(0, lambda: self._set_status(msg))
extract_audio(input_path, wav_path, self.ffmpeg, progress_cb=progress_cb)
if self._stop_req:
self._cleanup_and_finish(wav_path, "Stopped by user.")
return
# Step 2: Run segmentation
self.app.after(0, lambda: self._set_status("Running audio segmentation (this may take a while)…"))
raw_segments = run_segmentation(wav_path, progress_cb=progress_cb)
if self._stop_req:
self._cleanup_and_finish(wav_path, "Stopped by user.")
return
self._raw_segments = raw_segments
# Step 3: Merge
self.app.after(0, lambda: self._set_status("Merging singing segments…"))
singing = merge_singing_segments(
raw_segments,
max_gap=self.gap_var.get(),
min_duration=self.min_dur_var.get(),
padding=self.padding_var.get(),
)
self._singing_segments = singing
# Step 4: Export
if singing:
fps = self.fps_var.get()
source_name = Path(input_path).name
edl_path = os.path.join(output_dir, f"{stem}_singing.edl")
csv_path = os.path.join(output_dir, f"{stem}_singing.csv")
markers_path = os.path.join(output_dir, f"{stem}_markers.csv")
json_path = os.path.join(output_dir, f"{stem}_singing.json")
export_edl(singing, edl_path, fps=fps,
title=f"{stem} - Singing", source_filename=source_name)
export_csv(singing, csv_path)
export_markers_csv(singing, markers_path, fps=fps)
export_json(singing, json_path)
if self.autocut_var.get():
export_ffmpeg_script(singing, input_path, output_dir)
total_singing = sum(s["duration"] for s in singing)
result_msg = (f"✓ Found {len(singing)} songs · "
f"Total singing: {format_timecode(total_singing)} · "
f"Exported to: {output_dir}")
self.app.after(0, lambda: self._show_results(singing, edl_path, output_dir))
else:
result_msg = "No singing segments detected. Try lowering min duration or increasing gap."
# Cleanup WAV
if os.path.exists(wav_path):
os.remove(wav_path)
self.app.after(0, lambda: self._set_status(result_msg))
except Exception as e:
self.app.after(0, lambda: self._set_status(f"Error: {e}", error=True))
if os.path.exists(wav_path):
try:
os.remove(wav_path)
except Exception:
pass
self.app.after(0, self._finish_detect)
def _cleanup_and_finish(self, wav_path, msg):
if os.path.exists(wav_path):
try:
os.remove(wav_path)
except Exception:
pass
self.app.after(0, lambda: self._set_status(msg))
self.app.after(0, self._finish_detect)
def _finish_detect(self):
self._running = False
self._stop_req = False
self.detect_btn.configure(state="normal")
self.stop_btn.configure(state="disabled")
self.progress.stop()
self.progress.configure(mode="determinate")
self.progress.set(1.0 if self._singing_segments else 0)
def _show_results(self, segments, edl_path, output_dir):
"""Display detection results in the Results section."""
# Clear previous
for w in self.song_list_frame.winfo_children():
w.destroy()
self.export_frame.pack_forget()
self.results_lbl.configure(
text=f"Found {len(segments)} singing segments:",
text_color=("gray10", "gray90"))
self.song_list_frame.pack(fill="x", pady=(8, 0))
# Header
hdr = ctk.CTkFrame(self.song_list_frame, fg_color=("gray88", "gray20"), corner_radius=8)
hdr.pack(fill="x", pady=(0, 4))
for text, w in [("#", 40), ("Start", 90), ("End", 90), ("Duration", 80)]:
ctk.CTkLabel(hdr, text=text, width=w, anchor="center",
font=ctk.CTkFont(size=11, weight="bold")).pack(side="left", padx=4, pady=4)
# Rows
for seg in segments:
row = ctk.CTkFrame(self.song_list_frame, corner_radius=6, height=30)
row.pack(fill="x", pady=2)
row.pack_propagate(False)
vals = [
(f"Song {seg['index']}", 40),
(format_timecode(seg["start"]), 90),
(format_timecode(seg["end"]), 90),
(f"{seg['duration']:.0f}s", 80),
]
for text, w in vals:
ctk.CTkLabel(row, text=text, width=w, anchor="center",
font=ctk.CTkFont(size=11)).pack(side="left", padx=4)
# Total
total = sum(s["duration"] for s in segments)
ctk.CTkLabel(self.song_list_frame,
text=f"Total singing time: {format_timecode(total)}",
font=ctk.CTkFont(size=12, weight="bold")).pack(anchor="w", pady=(8, 0))
# Export info
self.export_frame.pack(fill="x", padx=14, pady=(4, 10))
for w in self.export_frame.winfo_children():
w.destroy()
ctk.CTkLabel(self.export_frame,
text=f"Files exported to: {output_dir}",
font=ctk.CTkFont(size=11), text_color=("gray45", "gray55"),
wraplength=600, anchor="w", justify="left").pack(anchor="w")
info_lines = [
"• _singing.edl → DaVinci Resolve: File → Import → Timeline",
"• _markers.csv → DaVinci Resolve: Timeline → Import Markers from CSV",
"• _singing.csv → Human-readable segment list",
"• _singing.json → Programmatic use",
]
if self.autocut_var.get():
info_lines.append("• _autocut.bat → Run to auto-cut singing segments (no DaVinci needed)")
for line in info_lines:
ctk.CTkLabel(self.export_frame, text=line,
font=ctk.CTkFont(size=11),
text_color=("gray50", "gray55"), anchor="w").pack(anchor="w")
# ── Public API for Quick Flow ─────────────────────────────────────────────
def run_detect_headless(self, input_path: str, output_dir: str,
gap: float, min_dur: float, padding: float,
fps: float, autocut: bool,
progress_cb=None) -> list:
"""
Run detection synchronously (called from Quick Flow worker thread).
Returns list of singing segments.
"""
os.makedirs(output_dir, exist_ok=True)
stem = Path(input_path).stem
wav_path = os.path.join(output_dir, f"{stem}_audio.wav")
try:
extract_audio(input_path, wav_path, self.ffmpeg, progress_cb=progress_cb)
raw_segments = run_segmentation(wav_path, progress_cb=progress_cb)
if progress_cb:
progress_cb("Merging singing segments…")
singing = merge_singing_segments(raw_segments,
max_gap=gap, min_duration=min_dur, padding=padding)
if singing:
source_name = Path(input_path).name
export_edl(singing, os.path.join(output_dir, f"{stem}_singing.edl"),
fps=fps, title=f"{stem} - Singing", source_filename=source_name)
export_csv(singing, os.path.join(output_dir, f"{stem}_singing.csv"))
export_markers_csv(singing, os.path.join(output_dir, f"{stem}_markers.csv"), fps=fps)
export_json(singing, os.path.join(output_dir, f"{stem}_singing.json"))
if autocut:
export_ffmpeg_script(singing, input_path, output_dir)
if os.path.exists(wav_path):
os.remove(wav_path)
return singing
except Exception:
if os.path.exists(wav_path):
try:
os.remove(wav_path)
except Exception:
pass
raise
+64
View File
@@ -0,0 +1,64 @@
"""
Theme, DPI, and shared styling constants.
"""
import sys
# ── DPI awareness (call before any Tk/CTk window) ────────────────────────────
def setup_dpi():
if sys.platform == "win32":
try:
import ctypes
ctypes.windll.shcore.SetProcessDpiAwareness(2)
except Exception:
try:
import ctypes
ctypes.windll.user32.SetProcessDPIAware()
except Exception:
pass
def dpi_scale(widget) -> float:
try:
return max(1.0, widget.winfo_fpixels("1i") / 96.0)
except Exception:
return 1.0
# ── Color accents ─────────────────────────────────────────────────────────────
# Each feature area has its own accent color scheme: (light_mode, dark_mode)
ACCENT_BLUE = {
"fg": ("#3B82F6", "#2563EB"),
"hover": ("#2563EB", "#1D4ED8"),
}
ACCENT_PURPLE = {
"fg": ("#EDE9FE", "#3B1F6E"),
"text": ("#5B21B6", "#C4B5FD"),
"hover": ("#DDD6FE", "#4C2889"),
"preview": ("#7C3AED", "#A78BFA"),
}
ACCENT_GREEN = {
"fg": ("#10B981", "#059669"),
"hover": ("#059669", "#047857"),
}
ACCENT_AMBER = {
"fg": ("#F59E0B", "#D97706"),
"hover": ("#D97706", "#B45309"),
}
# Neutral button style
BTN_NEUTRAL = {
"fg_color": ("gray80", "gray25"),
"text_color": ("gray10", "gray90"),
"hover_color": ("gray70", "gray35"),
}
BTN_SIDE = {
"fg_color": ("gray82", "gray22"),
"text_color": ("gray10", "gray90"),
"hover_color": ("gray72", "gray32"),
}
+42
View File
@@ -0,0 +1,42 @@
"""
Timecode & duration formatting utilities.
"""
def fmt_dur(seconds: float) -> str:
"""Format a duration as H:MM:SS or M:SS for display."""
if seconds < 0:
return "…"
if seconds == 0:
return "—"
s = int(round(seconds))
h, rem = divmod(s, 3600)
m, sec = divmod(rem, 60)
return f"{h}:{m:02d}:{sec:02d}" if h else f"{m}:{sec:02d}"
def fmt_elapsed(s: float) -> str:
if s < 60:
return f"{int(s)}s"
m, sec = divmod(int(s), 60)
return f"{m}m {sec:02d}s"
def format_timecode(seconds: float) -> str:
"""Format seconds as HH:MM:SS."""
h = int(seconds // 3600)
m = int((seconds % 3600) // 60)
s = int(seconds % 60)
return f"{h:02d}:{m:02d}:{s:02d}"
def seconds_to_smpte(seconds: float, fps: float = 29.97) -> str:
"""Convert seconds to SMPTE timecode HH:MM:SS:FF."""
total_frames = int(seconds * fps)
ff = total_frames % int(round(fps))
total_seconds = total_frames // int(round(fps))
ss = total_seconds % 60
total_minutes = total_seconds // 60
mm = total_minutes % 60
hh = total_minutes // 60
return f"{hh:02d}:{mm:02d}:{ss:02d}:{ff:02d}"
+222
View File
@@ -0,0 +1,222 @@
//! Singing-segment detection.
//!
//! Rust owns the fast/IO parts (audio extraction via ffmpeg, progress events,
//! and all file exports). The Python sidecar owns *only* the ML step
//! (inaSpeechSegmenter + TensorFlow), which has no native equivalent.
//!
//! Protocol: the sidecar prints one JSON object per line to stdout:
//! {"type":"progress","message":"…"}
//! {"type":"result","segments":[{index,start,end,duration}, …],
//! "stats":{"music":123.4,"speech":456.7,…}}
//! {"type":"error","message":"…"}
use std::io::{BufRead, BufReader};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tauri::{AppHandle, Emitter};
use crate::tools::hide_window;
use crate::AppState;
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Segment {
pub index: u32,
pub start: f64,
pub end: f64,
pub duration: f64,
}
#[derive(Clone, Serialize)]
pub struct DetectResult {
pub segments: Vec<Segment>,
pub stats: Value,
}
#[derive(Clone, Serialize)]
struct DetectProgress {
message: String,
}
fn emit(app: &AppHandle, message: impl Into<String>) {
let _ = app.emit("detect-progress", DetectProgress { message: message.into() });
}
/// Locate the sidecar script: bundled resources first, dev tree as fallback.
fn sidecar_script() -> Option<PathBuf> {
if let Ok(p) = std::env::var("STREAM_STUDIO_SIDECAR") {
let pb = PathBuf::from(p);
if pb.exists() {
return Some(pb);
}
}
if let Ok(exe) = std::env::current_exe() {
if let Some(base) = exe.parent() {
for sub in ["sidecar", "resources/sidecar", "../sidecar"] {
let cand = base.join(sub).join("detect.py");
if cand.exists() {
return Some(cand);
}
}
}
}
// Dev fallback: <crate>/../sidecar/detect.py
let dev = Path::new(env!("CARGO_MANIFEST_DIR")).join("../sidecar/detect.py");
if dev.exists() {
return Some(dev);
}
None
}
/// Build the python invocation. Prefers the Windows `py -3.10` launcher
/// (TensorFlow does not yet support 3.12+), then a venv, then plain `python`.
fn python_command(script: &Path) -> Command {
if let Ok(custom) = std::env::var("STREAM_STUDIO_PYTHON") {
let mut c = Command::new(custom);
c.arg(script);
return c;
}
#[cfg(windows)]
{
// The py launcher lets us pin 3.10 regardless of the default install.
let mut c = Command::new("py");
c.args(["-3.10"]).arg(script);
return c;
}
#[allow(unreachable_code)]
{
let mut c = Command::new("python");
c.arg(script);
c
}
}
/// Extract 16 kHz mono WAV (the format inaSpeechSegmenter expects).
fn extract_audio(app: &AppHandle, ffmpeg: &str, video: &str, wav: &Path) -> Result<(), String> {
emit(app, "Extracting audio (16 kHz mono)…");
let mut cmd = Command::new(ffmpeg);
cmd.args([
"-y", "-i", video,
"-vn", "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1",
"-hide_banner", "-loglevel", "error",
&wav.to_string_lossy(),
]);
cmd.stdout(Stdio::null()).stderr(Stdio::piped());
hide_window(&mut cmd);
let out = cmd.output().map_err(|e| format!("ffmpeg audio extract failed: {e}"))?;
if !out.status.success() {
return Err(format!("Audio extraction failed:\n{}", String::from_utf8_lossy(&out.stderr)));
}
Ok(())
}
pub fn run(
app: &AppHandle,
state: &AppState,
ffmpeg: &str,
video: String,
gap: f64,
min_duration: f64,
padding: f64,
) -> Result<DetectResult, String> {
state.cancel.store(false, std::sync::atomic::Ordering::SeqCst);
let script = sidecar_script().ok_or(
"Detection sidecar not found. Expected sidecar/detect.py next to the app.",
)?;
// Temp WAV beside the source video.
let wav = Path::new(&video)
.with_extension("")
.to_string_lossy()
.to_string();
let wav = PathBuf::from(format!("{wav}.stream_studio.wav"));
extract_audio(app, ffmpeg, &video, &wav)?;
let mut cmd = python_command(&script);
cmd.args([
"--wav", &wav.to_string_lossy(),
"--gap", &gap.to_string(),
"--min-duration", &min_duration.to_string(),
"--padding", &padding.to_string(),
]);
cmd.stdout(Stdio::piped()).stderr(Stdio::piped());
cmd.env("PYTHONUNBUFFERED", "1").env("PYTHONIOENCODING", "utf-8");
hide_window(&mut cmd);
emit(app, "Loading detection model…");
let mut child = cmd.spawn().map_err(|e| {
format!("Could not start Python sidecar ({e}). Is Python 3.10 + inaSpeechSegmenter installed?")
})?;
let stdout = child.stdout.take().ok_or("No sidecar stdout")?;
let stderr = child.stderr.take();
*state.proc.lock().unwrap() = Some(child);
let mut result: Option<DetectResult> = None;
let mut err_msg: Option<String> = None;
for line in BufReader::new(stdout).lines() {
let line = match line {
Ok(l) => l,
Err(_) => break,
};
let line = line.trim();
if line.is_empty() {
continue;
}
let parsed: Value = match serde_json::from_str(line) {
Ok(v) => v,
Err(_) => continue, // ignore stray non-JSON output
};
match parsed.get("type").and_then(|t| t.as_str()) {
Some("progress") => {
if let Some(m) = parsed.get("message").and_then(|m| m.as_str()) {
emit(app, m);
}
}
Some("result") => {
let segments: Vec<Segment> = serde_json::from_value(
parsed.get("segments").cloned().unwrap_or(Value::Array(vec![])),
)
.unwrap_or_default();
let stats = parsed.get("stats").cloned().unwrap_or(Value::Null);
result = Some(DetectResult { segments, stats });
}
Some("error") => {
err_msg = parsed
.get("message")
.and_then(|m| m.as_str())
.map(|s| s.to_string());
}
_ => {}
}
}
let mut child = state.proc.lock().unwrap().take().ok_or("Sidecar vanished")?;
let status = child.wait().map_err(|e| format!("sidecar wait failed: {e}"))?;
let mut stderr_text = String::new();
if let Some(mut se) = stderr {
use std::io::Read;
let _ = se.read_to_string(&mut stderr_text);
}
let _ = std::fs::remove_file(&wav);
if state.cancel.load(std::sync::atomic::Ordering::SeqCst) {
return Err("__cancelled__".into());
}
if let Some(m) = err_msg {
return Err(m);
}
if let Some(r) = result {
return Ok(r);
}
if !status.success() {
let tail: String = stderr_text.chars().rev().take(800).collect::<String>().chars().rev().collect();
return Err(format!("Detection failed.\n{tail}"));
}
Err("Detection produced no result.".into())
}
+129
View File
@@ -0,0 +1,129 @@
// Detect view — find singing segments in one video, export editor markers.
import { h, clear, basename, fmtDur } from "../dom";
import { makeProgress, toast, renderSongs } from "../ui";
import {
pickVideo, detectSongs, cancel, exportSegments, onDetectProgress, reveal,
type Segment,
} from "../api";
export interface View { el: HTMLElement; onDrop: (paths: string[]) => void; setVideo: (p: string) => void; }
const FPS = ["29.97", "25", "23.976", "30", "60"];
function numField(label: string, value: number, step = "1"): [HTMLElement, HTMLInputElement] {
const input = h("input", { class: "input", type: "number", value: String(value), step, min: "0" }) as HTMLInputElement;
return [h("div", { class: "field" }, h("label", {}, label), input), input];
}
export function buildDetect(): View {
let running = false;
let segments: Segment[] = [];
let videoPath = "";
const videoInput = h("input", { class: "input grow", placeholder: "Select a video…", readonly: true }) as HTMLInputElement;
const [gapF, gapI] = numField("Merge gap (s)", 10);
const [minF, minI] = numField("Min duration (s)", 20);
const [padF, padI] = numField("Padding (s)", 2);
const fpsSel = h("select", { class: "input" }, ...FPS.map((f) => h("option", { value: f }, f))) as HTMLSelectElement;
const detectBtn = h("button", { class: "btn primary lg", disabled: true }, "🎤 Detect Songs") as HTMLButtonElement;
const stopBtn = h("button", { class: "btn danger", disabled: true }, "■ Stop") as HTMLButtonElement;
const exportBtn = h("button", { class: "btn", disabled: true }, "⬇ Export EDL / CSV") as HTMLButtonElement;
const revealBtn = h("button", { class: "btn ghost", style: "display:none" }, "Show in folder") as HTMLButtonElement;
const progress = makeProgress();
const statsEl = h("div", { class: "pills" });
const songsEl = h("div", { class: "songs" });
let lastExportDir = "";
function setVideo(p: string) {
videoPath = p;
videoInput.value = basename(p);
videoInput.title = p;
detectBtn.disabled = running || !p;
}
async function detect() {
if (!videoPath) return;
running = true;
detectBtn.disabled = true;
stopBtn.disabled = false;
exportBtn.disabled = true;
revealBtn.style.display = "none";
clear(songsEl); clear(statsEl);
progress.reset();
progress.indeterminate(true, "Starting detection…");
const un = await onDetectProgress((msg) => {
progress.indeterminate(true, msg);
progress.appendLog(msg);
});
try {
const res = await detectSongs(videoPath, +gapI.value, +minI.value, +padI.value);
segments = res.segments;
progress.indeterminate(false);
progress.set(1, `Found ${segments.length} singing segment(s).`);
renderSongs(songsEl, segments);
const totalSing = segments.reduce((a, s) => a + s.duration, 0);
statsEl.replaceChildren(
h("span", { class: "pill" }, `${segments.length} songs`),
h("span", { class: "pill" }, `singing ${fmtDur(totalSing)}`),
...Object.entries(res.stats || {}).map(([k, v]) =>
h("span", { class: "pill" }, `${k} ${fmtDur(v as number)}`)),
);
exportBtn.disabled = segments.length === 0;
if (!segments.length) toast("No singing detected — try a larger gap or smaller min duration.", "");
} catch (e) {
const msg = String(e);
progress.indeterminate(false);
if (msg.includes("__cancelled__")) progress.set(0, "Stopped by user.");
else { progress.set(0, "Detection failed."); toast(msg, "err"); }
} finally {
un();
running = false;
stopBtn.disabled = true;
detectBtn.disabled = !videoPath;
}
}
exportBtn.addEventListener("click", async () => {
if (!segments.length) return;
try {
const res = await exportSegments(segments, videoPath, null, +fpsSel.value);
lastExportDir = res.files[0] ?? "";
toast(`Exported ${res.files.length} files`, "ok");
if (lastExportDir) revealBtn.style.display = "";
} catch (e) { toast(String(e), "err"); }
});
revealBtn.addEventListener("click", () => lastExportDir && reveal(lastExportDir));
detectBtn.addEventListener("click", detect);
stopBtn.addEventListener("click", () => cancel());
const el = h("div", { class: "view" },
h("div", { class: "card" },
h("h2", {}, h("span", { class: "step" }, "1"), "Source Video"),
h("div", { class: "row" }, videoInput,
h("button", { class: "btn", onclick: async () => { const p = await pickVideo(); if (p) setVideo(p); } }, "Browse…")),
h("div", { class: "hint", style: "margin-top:8px" }, "Tip: drop a file here, or use a Concat output."),
),
h("div", { class: "card" },
h("h2", {}, h("span", { class: "step" }, "2"), "Detection Settings"),
h("div", { class: "row wrap" },
h("div", { class: "grow", style: "min-width:130px" }, gapF),
h("div", { class: "grow", style: "min-width:130px" }, minF),
h("div", { class: "grow", style: "min-width:130px" }, padF),
h("div", { class: "field", style: "min-width:120px" }, h("label", {}, "FPS (export)"), fpsSel)),
h("div", { class: "hint", style: "margin-top:8px" }, "Singing = \"music\". More chatting between songs → raise the gap. Short covers → lower the min duration."),
),
h("div", { class: "card" },
h("h2", {}, h("span", { class: "step" }, "3"), "Run"),
h("div", { class: "row", style: "margin-bottom:14px" }, detectBtn, stopBtn, exportBtn, revealBtn),
progress.root,
h("div", { style: "margin-top:12px" }, statsEl),
h("div", { style: "margin-top:10px" }, songsEl),
),
);
return { el, onDrop: (paths) => { if (paths[0]) setVideo(paths[0]); }, setVideo };
}
+114
View File
@@ -0,0 +1,114 @@
#!/usr/bin/env python3
"""
Stream Studio — singing-detection sidecar.
This is the ONLY Python in the project. It exists solely because the ML model
(inaSpeechSegmenter, which classifies singing voice as "music") is TensorFlow-
based and has no native Rust/JS equivalent. Everything else — UI, ffmpeg
concat, audio extraction, file exports — is handled by the Tauri/Rust host.
Contract (line-delimited JSON on stdout):
{"type": "progress", "message": "..."}
{"type": "result", "segments": [{"index","start","end","duration"}...],
"stats": {"music": 123.4, "speech": 456.7, ...}}
{"type": "error", "message": "..."}
The host extracts a 16 kHz mono WAV and passes it via --wav, so this script
never touches ffmpeg.
Run with Python 3.10 (TensorFlow does not support 3.12+):
py -3.10 detect.py --wav audio.wav --gap 10 --min-duration 20 --padding 2
"""
import argparse
import json
import sys
import time
def emit(obj: dict) -> None:
"""Write one JSON record per line and flush immediately."""
sys.stdout.write(json.dumps(obj, ensure_ascii=False) + "\n")
sys.stdout.flush()
def progress(msg: str) -> None:
emit({"type": "progress", "message": msg})
def merge_singing_segments(segments, max_gap, min_duration, padding):
"""Filter for 'music' segments and merge nearby ones into songs."""
music = sorted((s, e) for label, s, e in segments if label == "music")
if not music:
return []
merged = []
cur_start, cur_end = music[0]
for s, e in music[1:]:
if s - cur_end <= max_gap:
cur_end = max(cur_end, e)
else:
merged.append((cur_start, cur_end))
cur_start, cur_end = s, e
merged.append((cur_start, cur_end))
results = []
idx = 1
for s, e in merged:
if (e - s) >= min_duration:
ps = max(0.0, s - padding)
pe = e + padding
results.append({
"index": idx,
"start": round(ps, 3),
"end": round(pe, 3),
"duration": round(pe - ps, 3),
})
idx += 1
return results
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--wav", required=True)
ap.add_argument("--gap", type=float, default=10.0)
ap.add_argument("--min-duration", type=float, default=20.0)
ap.add_argument("--padding", type=float, default=2.0)
args = ap.parse_args()
try:
progress("Loading detection model (first run downloads weights)…")
from inaSpeechSegmenter import Segmenter
except ImportError:
emit({
"type": "error",
"message": (
"inaSpeechSegmenter is not installed for this Python.\n"
"Install it with:\n"
" py -3.10 -m pip install inaSpeechSegmenter tensorflow"
),
})
return 1
try:
t0 = time.time()
seg = Segmenter(vad_engine="smn", detect_gender=False)
progress("Analysing audio — this can take a while on long streams…")
raw = seg(args.wav)
progress(f"Segmentation done in {time.time() - t0:.0f}s ({len(raw)} raw segments).")
stats = {}
for label, s, e in raw:
stats[label] = round(stats.get(label, 0.0) + (e - s), 1)
songs = merge_singing_segments(raw, args.gap, args.min_duration, args.padding)
progress(f"Found {len(songs)} singing segment(s) after merge + filter.")
emit({"type": "result", "segments": songs, "stats": stats})
return 0
except Exception as exc: # noqa: BLE001 — surface any model/runtime error to the host
emit({"type": "error", "message": f"{type(exc).__name__}: {exc}"})
return 1
if __name__ == "__main__":
sys.exit(main())