Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 33c275a114 | |||
| b9573e8644 | |||
| 98a4d37162 |
185
README.md
185
README.md
@@ -1,185 +0,0 @@
|
||||
# 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`) |
|
||||
222
detect.rs
Normal file
222
detect.rs
Normal 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
detect.ts
Normal file
129
detect.ts
Normal 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 };
|
||||
}
|
||||
@@ -1,598 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Singing Segment Detector for Live Stream Recordings
|
||||
=====================================================
|
||||
Detects singing segments in long stream recordings and exports
|
||||
DaVinci Resolve-compatible EDL markers for fast video editing.
|
||||
|
||||
Pipeline:
|
||||
1. Extract audio from video (ffmpeg → 16kHz mono WAV)
|
||||
2. Run inaSpeechSegmenter to classify speech/music/noise
|
||||
3. Merge nearby "music" segments (singing) with configurable gap
|
||||
4. Export to EDL (Edit Decision List) for DaVinci Resolve import
|
||||
5. Optionally export CSV for review
|
||||
|
||||
Usage:
|
||||
python detect_singing.py "path/to/stream_recording.mp4"
|
||||
python detect_singing.py "path/to/stream_recording.mp4" --gap 15 --min-duration 30
|
||||
python detect_singing.py "path/to/stream_recording.mp4" --output-dir "D:/singing_edits"
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────
|
||||
# Step 1: Extract audio from video using ffmpeg
|
||||
# ──────────────────────────────────────────────
|
||||
def extract_audio(video_path: str, output_wav: str, ffmpeg_bin: str = "ffmpeg") -> str:
|
||||
"""Extract audio from video file as 16kHz mono WAV (required by inaSpeechSegmenter)."""
|
||||
print(f"\n[Step 1/4] Extracting audio from: {video_path}")
|
||||
print(f" Output WAV: {output_wav}")
|
||||
|
||||
cmd = [
|
||||
ffmpeg_bin,
|
||||
"-i", video_path,
|
||||
"-vn", # no video
|
||||
"-acodec", "pcm_s16le", # 16-bit PCM
|
||||
"-ar", "16000", # 16kHz sample rate
|
||||
"-ac", "1", # mono
|
||||
"-y", # overwrite
|
||||
output_wav,
|
||||
]
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=1800, # 30 min timeout for very long files
|
||||
)
|
||||
if result.returncode != 0:
|
||||
print(f" [ERROR] ffmpeg failed:\n{result.stderr[-500:]}")
|
||||
sys.exit(1)
|
||||
except FileNotFoundError:
|
||||
print(f" [ERROR] ffmpeg not found at '{ffmpeg_bin}'.")
|
||||
print(f" Make sure ffmpeg is installed and in your PATH.")
|
||||
sys.exit(1)
|
||||
|
||||
size_mb = os.path.getsize(output_wav) / (1024 * 1024)
|
||||
print(f" Audio extracted: {size_mb:.1f} MB")
|
||||
return output_wav
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────
|
||||
# Step 2: Run inaSpeechSegmenter
|
||||
# ──────────────────────────────────────────────
|
||||
def run_segmentation(wav_path: str) -> list:
|
||||
"""
|
||||
Run inaSpeechSegmenter on audio file.
|
||||
Returns list of (label, start_sec, end_sec) tuples.
|
||||
Labels: 'music', 'speech', 'male', 'female', 'noise', 'noEnergy'
|
||||
|
||||
NOTE: inaSpeechSegmenter classifies singing voice as 'music'.
|
||||
"""
|
||||
print(f"\n[Step 2/4] Running audio segmentation (this may take a while)...")
|
||||
|
||||
try:
|
||||
from inaSpeechSegmenter import Segmenter
|
||||
except ImportError:
|
||||
print(" [ERROR] inaSpeechSegmenter is not installed.")
|
||||
print(" Install it with: pip install inaSpeechSegmenter")
|
||||
sys.exit(1)
|
||||
|
||||
start_time = time.time()
|
||||
|
||||
# vad_engine='smn' → speech/music/noise detection
|
||||
# detect_gender=False → faster, we don't need gender info
|
||||
seg = Segmenter(vad_engine='smn', detect_gender=False)
|
||||
segments = seg(wav_path)
|
||||
|
||||
elapsed = time.time() - start_time
|
||||
print(f" Segmentation complete in {elapsed:.1f}s")
|
||||
print(f" Total segments found: {len(segments)}")
|
||||
|
||||
# Count segment types
|
||||
type_counts = {}
|
||||
type_durations = {}
|
||||
for label, start, end in segments:
|
||||
type_counts[label] = type_counts.get(label, 0) + 1
|
||||
type_durations[label] = type_durations.get(label, 0) + (end - start)
|
||||
|
||||
for label in sorted(type_counts.keys()):
|
||||
count = type_counts[label]
|
||||
dur = type_durations[label]
|
||||
print(f" {label:>10s}: {count:4d} segments, {format_timecode(dur)} total")
|
||||
|
||||
return segments
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────
|
||||
# Step 3: Filter & merge singing (music) segments
|
||||
# ──────────────────────────────────────────────
|
||||
def merge_singing_segments(
|
||||
segments: list,
|
||||
max_gap: float = 10.0,
|
||||
min_duration: float = 20.0,
|
||||
padding: float = 2.0,
|
||||
) -> list:
|
||||
"""
|
||||
Filter for 'music' segments (which includes singing) and merge
|
||||
nearby segments that are likely the same song.
|
||||
|
||||
Args:
|
||||
segments: Raw segmentation output [(label, start, end), ...]
|
||||
max_gap: Max gap (seconds) between music segments to merge them.
|
||||
Stream singers often have brief pauses, audience interaction
|
||||
between verses, etc. Default 10s works well.
|
||||
min_duration: Minimum duration (seconds) for a merged segment to be kept.
|
||||
Filters out short music stings, sound effects, etc.
|
||||
A typical song is 2-5 minutes, so 20s is a safe minimum.
|
||||
padding: Seconds to add before/after each segment for clean cuts.
|
||||
|
||||
Returns:
|
||||
List of dicts: [{"start": float, "end": float, "duration": float, "index": int}, ...]
|
||||
"""
|
||||
print(f"\n[Step 3/4] Merging singing segments (gap={max_gap}s, min={min_duration}s, pad={padding}s)")
|
||||
|
||||
# Extract only music segments
|
||||
music_segs = [(start, end) for label, start, end in segments if label == "music"]
|
||||
|
||||
if not music_segs:
|
||||
print(" No music segments found!")
|
||||
return []
|
||||
|
||||
# Sort by start time (should already be sorted, but just in case)
|
||||
music_segs.sort(key=lambda x: x[0])
|
||||
|
||||
# Merge segments with gaps smaller than max_gap
|
||||
merged = []
|
||||
current_start, current_end = music_segs[0]
|
||||
|
||||
for start, end in music_segs[1:]:
|
||||
if start - current_end <= max_gap:
|
||||
# Extend current segment
|
||||
current_end = max(current_end, end)
|
||||
else:
|
||||
# Save current segment and start new one
|
||||
merged.append((current_start, current_end))
|
||||
current_start, current_end = start, end
|
||||
|
||||
merged.append((current_start, current_end))
|
||||
|
||||
# Apply padding and minimum duration filter
|
||||
results = []
|
||||
idx = 1
|
||||
for start, end in merged:
|
||||
duration = end - start
|
||||
if duration >= min_duration:
|
||||
padded_start = max(0, start - padding)
|
||||
padded_end = end + padding
|
||||
results.append({
|
||||
"index": idx,
|
||||
"start": padded_start,
|
||||
"end": padded_end,
|
||||
"duration": padded_end - padded_start,
|
||||
"original_start": start,
|
||||
"original_end": end,
|
||||
})
|
||||
idx += 1
|
||||
|
||||
print(f" Found {len(results)} singing segments after merge+filter:")
|
||||
for seg in results:
|
||||
print(f" Song {seg['index']:2d}: "
|
||||
f"{format_timecode(seg['start'])} → {format_timecode(seg['end'])} "
|
||||
f"({seg['duration']:.0f}s)")
|
||||
|
||||
return results
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────
|
||||
# Step 4: Export formats
|
||||
# ──────────────────────────────────────────────
|
||||
|
||||
def export_edl(segments: list, output_path: str, fps: float = 29.97,
|
||||
title: str = "Singing Segments", source_filename: str = None):
|
||||
"""
|
||||
Export an EDL (Edit Decision List) file for DaVinci Resolve.
|
||||
|
||||
DaVinci Resolve import: File → Import → Timeline → select the .edl file.
|
||||
The EDL creates cut points at each singing segment's in/out points.
|
||||
|
||||
Args:
|
||||
source_filename: Original video filename (e.g. "stream_2026-04-01.mp4").
|
||||
Used as reel name + FROM CLIP NAME so DaVinci Resolve
|
||||
matches the EDL to the correct clip in the Media Pool.
|
||||
Without this, Resolve picks a random clip when multiple
|
||||
videos are loaded in the same project.
|
||||
"""
|
||||
print(f"\n[Step 4/4] Exporting EDL: {output_path}")
|
||||
|
||||
# DaVinci Resolve uses both the reel name column AND the "* FROM CLIP NAME:"
|
||||
# comment to identify which media clip an EDL edit refers to.
|
||||
#
|
||||
# Reel name: traditionally 8 chars max, but Resolve accepts longer names.
|
||||
# We use the filename without extension, truncated to a safe length.
|
||||
# FROM CLIP NAME: Resolve's preferred matching method. Must be the exact
|
||||
# filename (with extension) as it appears in the Media Pool.
|
||||
|
||||
if source_filename:
|
||||
clip_name = source_filename # full filename with extension
|
||||
reel_name = Path(source_filename).stem[:32] # stem, truncated for safety
|
||||
# Replace spaces with underscores in reel name (some EDL parsers choke on spaces)
|
||||
reel_name_safe = reel_name.replace(" ", "_")
|
||||
else:
|
||||
clip_name = None
|
||||
reel_name_safe = "001"
|
||||
|
||||
# Calculate record timecodes: sequential placement on the output timeline
|
||||
# Each segment is placed one after another, starting at 01:00:00:00
|
||||
rec_offset = 3600.0 # start at 01:00:00:00
|
||||
|
||||
with open(output_path, "w", encoding="utf-8") as f:
|
||||
f.write(f"TITLE: {title}\n")
|
||||
f.write("FCM: NON-DROP FRAME\n\n")
|
||||
|
||||
current_rec_pos = rec_offset
|
||||
for seg in segments:
|
||||
edit_num = f"{seg['index']:03d}"
|
||||
src_in = seconds_to_timecode(seg["start"], fps)
|
||||
src_out = seconds_to_timecode(seg["end"], fps)
|
||||
rec_in = seconds_to_timecode(current_rec_pos, fps)
|
||||
rec_out = seconds_to_timecode(current_rec_pos + seg["duration"], fps)
|
||||
|
||||
# EDL format: edit# reel_name track_type transition src_in src_out rec_in rec_out
|
||||
f.write(f"{edit_num} {reel_name_safe} V C {src_in} {src_out} {rec_in} {rec_out}\n")
|
||||
|
||||
# "* FROM CLIP NAME:" is the key line DaVinci uses for media matching.
|
||||
# It must exactly match the clip name shown in the Media Pool.
|
||||
if clip_name:
|
||||
f.write(f"* FROM CLIP NAME: {clip_name}\n")
|
||||
|
||||
f.write(f"* COMMENT: Song {seg['index']} - Duration {seg['duration']:.0f}s\n\n")
|
||||
|
||||
current_rec_pos += seg["duration"]
|
||||
|
||||
print(f" EDL file written with {len(segments)} edits")
|
||||
if clip_name:
|
||||
print(f" Source clip name: {clip_name}")
|
||||
|
||||
|
||||
def export_csv(segments: list, output_path: str):
|
||||
"""Export singing segments as CSV for review or further processing."""
|
||||
with open(output_path, "w", newline="", encoding="utf-8") as f:
|
||||
writer = csv.writer(f)
|
||||
writer.writerow(["index", "start_sec", "end_sec", "duration_sec",
|
||||
"start_timecode", "end_timecode"])
|
||||
for seg in segments:
|
||||
writer.writerow([
|
||||
seg["index"],
|
||||
f"{seg['start']:.2f}",
|
||||
f"{seg['end']:.2f}",
|
||||
f"{seg['duration']:.1f}",
|
||||
format_timecode(seg["start"]),
|
||||
format_timecode(seg["end"]),
|
||||
])
|
||||
print(f" CSV file written: {output_path}")
|
||||
|
||||
|
||||
def export_davinci_markers_csv(segments: list, output_path: str, fps: float = 29.97):
|
||||
"""
|
||||
Export a CSV file that can be used with DaVinci Resolve's
|
||||
'Import Markers from CSV' function (Resolve 18+).
|
||||
|
||||
Format: #, Color, Name, Start TC, End TC, Duration TC, Notes
|
||||
"""
|
||||
with open(output_path, "w", newline="", encoding="utf-8") as f:
|
||||
writer = csv.writer(f)
|
||||
# DaVinci Resolve Marker CSV header
|
||||
writer.writerow(["#", "Color", "Name", "Start TC", "End TC", "Duration TC", "Notes"])
|
||||
for seg in segments:
|
||||
start_tc = seconds_to_timecode(seg["start"], fps)
|
||||
end_tc = seconds_to_timecode(seg["end"], fps)
|
||||
dur_tc = seconds_to_timecode(seg["duration"], fps)
|
||||
writer.writerow([
|
||||
seg["index"],
|
||||
"Blue",
|
||||
f"Song {seg['index']}",
|
||||
start_tc,
|
||||
end_tc,
|
||||
dur_tc,
|
||||
f"Duration: {seg['duration']:.0f}s",
|
||||
])
|
||||
print(f" DaVinci markers CSV written: {output_path}")
|
||||
|
||||
|
||||
def export_json(segments: list, output_path: str):
|
||||
"""Export as JSON for programmatic use."""
|
||||
with open(output_path, "w", encoding="utf-8") as f:
|
||||
json.dump(segments, f, indent=2, ensure_ascii=False)
|
||||
print(f" JSON file written: {output_path}")
|
||||
|
||||
|
||||
def export_ffmpeg_concat(segments: list, video_path: str, output_path: str):
|
||||
"""
|
||||
Export a .bat/.sh script that uses ffmpeg to directly cut and
|
||||
concatenate all singing segments into a single video file.
|
||||
This is the fully automated option — no DaVinci needed.
|
||||
"""
|
||||
video_name = Path(video_path).stem
|
||||
ext = Path(video_path).suffix
|
||||
out_video = str(Path(output_path).parent / f"{video_name}_singing_only{ext}")
|
||||
|
||||
is_windows = os.name == "nt" or sys.platform == "win32"
|
||||
script_ext = ".bat" if is_windows else ".sh"
|
||||
script_path = str(Path(output_path).with_suffix(script_ext))
|
||||
|
||||
lines = []
|
||||
if is_windows:
|
||||
lines.append("@echo off")
|
||||
lines.append("REM Auto-generated ffmpeg script to extract singing segments")
|
||||
lines.append(f'REM Source: {video_path}')
|
||||
lines.append("")
|
||||
else:
|
||||
lines.append("#!/bin/bash")
|
||||
lines.append("# Auto-generated ffmpeg script to extract singing segments")
|
||||
lines.append(f'# Source: {video_path}')
|
||||
lines.append("")
|
||||
|
||||
# Create a concat file list
|
||||
concat_list_path = str(Path(output_path).parent / f"{video_name}_concat_list.txt")
|
||||
|
||||
# Generate individual segment extraction commands
|
||||
segment_files = []
|
||||
for seg in segments:
|
||||
seg_file = f"_seg_{seg['index']:03d}{ext}"
|
||||
seg_path = str(Path(output_path).parent / seg_file)
|
||||
segment_files.append(seg_file)
|
||||
|
||||
start = seg["start"]
|
||||
duration = seg["duration"]
|
||||
|
||||
cmd = (f'ffmpeg -y -ss {start:.2f} -i "{video_path}" '
|
||||
f'-t {duration:.2f} -c copy "{seg_path}"')
|
||||
lines.append(f"echo Extracting Song {seg['index']}...")
|
||||
lines.append(cmd)
|
||||
lines.append("")
|
||||
|
||||
# Generate concat list file
|
||||
lines.append(f"echo Creating concat list...")
|
||||
if is_windows:
|
||||
lines.append(f'(')
|
||||
for sf in segment_files:
|
||||
seg_path = str(Path(output_path).parent / sf)
|
||||
lines.append(f' echo file \'{seg_path}\'')
|
||||
lines.append(f') > "{concat_list_path}"')
|
||||
else:
|
||||
for i, sf in enumerate(segment_files):
|
||||
seg_path = str(Path(output_path).parent / sf)
|
||||
op = ">" if i == 0 else ">>"
|
||||
lines.append(f"echo \"file '{seg_path}'\" {op} \"{concat_list_path}\"")
|
||||
|
||||
lines.append("")
|
||||
lines.append(f"echo Concatenating all segments...")
|
||||
lines.append(f'ffmpeg -y -f concat -safe 0 -i "{concat_list_path}" -c copy "{out_video}"')
|
||||
lines.append("")
|
||||
|
||||
# Cleanup temp segment files
|
||||
lines.append("echo Cleaning up temp files...")
|
||||
for sf in segment_files:
|
||||
seg_path = str(Path(output_path).parent / sf)
|
||||
if is_windows:
|
||||
lines.append(f'del "{seg_path}"')
|
||||
else:
|
||||
lines.append(f'rm -f "{seg_path}"')
|
||||
if is_windows:
|
||||
lines.append(f'del "{concat_list_path}"')
|
||||
else:
|
||||
lines.append(f'rm -f "{concat_list_path}"')
|
||||
|
||||
lines.append("")
|
||||
lines.append(f'echo Done! Output: {out_video}')
|
||||
if is_windows:
|
||||
lines.append("pause")
|
||||
|
||||
with open(script_path, "w", encoding="utf-8") as f:
|
||||
f.write("\n".join(lines))
|
||||
|
||||
if not is_windows:
|
||||
os.chmod(script_path, 0o755)
|
||||
|
||||
print(f" FFmpeg concat script written: {script_path}")
|
||||
print(f" Run it to auto-generate: {out_video}")
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────
|
||||
# Export raw segmentation for debugging
|
||||
# ──────────────────────────────────────────────
|
||||
def export_raw_segments(segments: list, output_path: str):
|
||||
"""Export ALL raw segments (speech/music/noise) as CSV for debugging."""
|
||||
with open(output_path, "w", newline="", encoding="utf-8") as f:
|
||||
writer = csv.writer(f)
|
||||
writer.writerow(["label", "start_sec", "end_sec", "duration_sec",
|
||||
"start_timecode", "end_timecode"])
|
||||
for label, start, end in segments:
|
||||
writer.writerow([
|
||||
label,
|
||||
f"{start:.2f}",
|
||||
f"{end:.2f}",
|
||||
f"{end - start:.1f}",
|
||||
format_timecode(start),
|
||||
format_timecode(end),
|
||||
])
|
||||
print(f" Raw segments CSV written: {output_path}")
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────
|
||||
# Utility functions
|
||||
# ──────────────────────────────────────────────
|
||||
def format_timecode(seconds: float) -> str:
|
||||
"""Format seconds as HH:MM:SS."""
|
||||
h = int(seconds // 3600)
|
||||
m = int((seconds % 3600) // 60)
|
||||
s = int(seconds % 60)
|
||||
return f"{h:02d}:{m:02d}:{s:02d}"
|
||||
|
||||
|
||||
def seconds_to_timecode(seconds: float, fps: float = 29.97) -> str:
|
||||
"""Convert seconds to SMPTE timecode HH:MM:SS:FF."""
|
||||
total_frames = int(seconds * fps)
|
||||
ff = total_frames % int(round(fps))
|
||||
total_seconds = total_frames // int(round(fps))
|
||||
ss = total_seconds % 60
|
||||
total_minutes = total_seconds // 60
|
||||
mm = total_minutes % 60
|
||||
hh = total_minutes // 60
|
||||
return f"{hh:02d}:{mm:02d}:{ss:02d}:{ff:02d}"
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────
|
||||
# Main
|
||||
# ──────────────────────────────────────────────
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Detect singing segments in stream recordings and export DaVinci Resolve markers.",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog="""
|
||||
Examples:
|
||||
python detect_singing.py "D:\\streams\\2026-04-04.mp4"
|
||||
python detect_singing.py "D:\\streams\\2026-04-04.mp4" --gap 15 --min-duration 30
|
||||
python detect_singing.py "D:\\streams\\2026-04-04.mp4" --output-dir "D:\\singing_edits"
|
||||
python detect_singing.py "D:\\streams\\2026-04-04.mp4" --auto-cut
|
||||
|
||||
Tips:
|
||||
--gap 10 Merge music segments with <=10s gap (default). Increase if the singer
|
||||
often pauses >10s between verses in the same song.
|
||||
--min-duration 20 Ignore segments shorter than 20s (default). Decrease if
|
||||
the singer does very short songs or covers.
|
||||
--padding 2 Add 2s padding before/after each segment (default). Helps catch
|
||||
the very start/end of songs.
|
||||
--auto-cut Generate an ffmpeg script to automatically cut and concat all
|
||||
singing segments into a single video file.
|
||||
--fps 29.97 Frame rate for timecode calculation (default: 29.97 for NTSC).
|
||||
Use 25 for PAL or 23.976 for film.
|
||||
""",
|
||||
)
|
||||
|
||||
parser.add_argument("video", help="Path to the stream recording video file")
|
||||
parser.add_argument("--output-dir", "-o", default=None,
|
||||
help="Output directory (default: same directory as input video)")
|
||||
parser.add_argument("--gap", "-g", type=float, default=10.0,
|
||||
help="Max gap (seconds) to merge nearby singing segments (default: 10)")
|
||||
parser.add_argument("--min-duration", "-m", type=float, default=20.0,
|
||||
help="Minimum singing segment duration in seconds (default: 20)")
|
||||
parser.add_argument("--padding", "-p", type=float, default=2.0,
|
||||
help="Padding (seconds) before/after each segment (default: 2)")
|
||||
parser.add_argument("--fps", type=float, default=29.97,
|
||||
help="Video frame rate for timecode export (default: 29.97)")
|
||||
parser.add_argument("--keep-wav", action="store_true",
|
||||
help="Keep the extracted WAV file (default: delete after processing)")
|
||||
parser.add_argument("--auto-cut", action="store_true",
|
||||
help="Generate an ffmpeg script to auto-cut and concat singing segments")
|
||||
parser.add_argument("--raw-segments", action="store_true",
|
||||
help="Also export all raw segments (speech/music/noise) as CSV")
|
||||
parser.add_argument("--ffmpeg", default="ffmpeg",
|
||||
help="Path to ffmpeg binary (default: 'ffmpeg' from PATH)")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# Validate input
|
||||
video_path = os.path.abspath(args.video)
|
||||
if not os.path.isfile(video_path):
|
||||
print(f"[ERROR] Video file not found: {video_path}")
|
||||
sys.exit(1)
|
||||
|
||||
# Set output directory
|
||||
if args.output_dir:
|
||||
output_dir = os.path.abspath(args.output_dir)
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
else:
|
||||
output_dir = os.path.dirname(video_path)
|
||||
|
||||
video_stem = Path(video_path).stem
|
||||
|
||||
print("=" * 60)
|
||||
print(" Singing Segment Detector")
|
||||
print("=" * 60)
|
||||
print(f" Input: {video_path}")
|
||||
print(f" Output dir: {output_dir}")
|
||||
print(f" Settings: gap={args.gap}s, min={args.min_duration}s, pad={args.padding}s")
|
||||
print(f" FPS: {args.fps}")
|
||||
|
||||
# Step 1: Extract audio
|
||||
wav_path = os.path.join(output_dir, f"{video_stem}_audio.wav")
|
||||
extract_audio(video_path, wav_path, ffmpeg_bin=args.ffmpeg)
|
||||
|
||||
# Step 2: Run segmentation
|
||||
raw_segments = run_segmentation(wav_path)
|
||||
|
||||
# Step 3: Merge singing segments
|
||||
singing_segments = merge_singing_segments(
|
||||
raw_segments,
|
||||
max_gap=args.gap,
|
||||
min_duration=args.min_duration,
|
||||
padding=args.padding,
|
||||
)
|
||||
|
||||
if not singing_segments:
|
||||
print("\n[RESULT] No singing segments detected. Try lowering --min-duration or increasing --gap.")
|
||||
# Cleanup WAV
|
||||
if not args.keep_wav and os.path.exists(wav_path):
|
||||
os.remove(wav_path)
|
||||
sys.exit(0)
|
||||
|
||||
# Step 4: Export results
|
||||
edl_path = os.path.join(output_dir, f"{video_stem}_singing.edl")
|
||||
csv_path = os.path.join(output_dir, f"{video_stem}_singing.csv")
|
||||
markers_path = os.path.join(output_dir, f"{video_stem}_markers.csv")
|
||||
json_path = os.path.join(output_dir, f"{video_stem}_singing.json")
|
||||
|
||||
export_edl(singing_segments, edl_path, fps=args.fps,
|
||||
title=f"{video_stem} - Singing",
|
||||
source_filename=Path(video_path).name)
|
||||
export_csv(singing_segments, csv_path)
|
||||
export_davinci_markers_csv(singing_segments, markers_path, fps=args.fps)
|
||||
export_json(singing_segments, json_path)
|
||||
|
||||
if args.auto_cut:
|
||||
export_ffmpeg_concat(singing_segments, video_path, json_path)
|
||||
|
||||
if args.raw_segments:
|
||||
raw_csv_path = os.path.join(output_dir, f"{video_stem}_raw_segments.csv")
|
||||
export_raw_segments(raw_segments, raw_csv_path)
|
||||
|
||||
# Cleanup WAV unless --keep-wav
|
||||
if not args.keep_wav and os.path.exists(wav_path):
|
||||
os.remove(wav_path)
|
||||
print(f"\n Temp WAV file deleted.")
|
||||
|
||||
# Summary
|
||||
total_singing = sum(s["duration"] for s in singing_segments)
|
||||
print("\n" + "=" * 60)
|
||||
print(" DONE!")
|
||||
print("=" * 60)
|
||||
print(f" Songs detected: {len(singing_segments)}")
|
||||
print(f" Total singing time: {format_timecode(total_singing)}")
|
||||
print(f" EDL file: {edl_path}")
|
||||
print(f" DaVinci markers CSV: {markers_path}")
|
||||
print(f" Segments CSV: {csv_path}")
|
||||
print(f" Segments JSON: {json_path}")
|
||||
if args.auto_cut:
|
||||
script_ext = ".bat" if (os.name == "nt" or sys.platform == "win32") else ".sh"
|
||||
print(f" Auto-cut script: {json_path.replace('.json', script_ext)}")
|
||||
print()
|
||||
print(" To import into DaVinci Resolve:")
|
||||
print(" Option A: File → Import → Timeline → select the .edl file")
|
||||
print(" Option B: Timeline menu → Import Markers from CSV → select _markers.csv")
|
||||
print("=" * 60)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,56 +0,0 @@
|
||||
@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
|
||||
114
sidecar_detect.py
Normal file
114
sidecar_detect.py
Normal 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())
|
||||
Reference in New Issue
Block a user