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>
This commit is contained in:
185
README.md
Normal file
185
README.md
Normal 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
detect_singing.py
Normal file
598
detect_singing.py
Normal 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
setup_windows.bat
Normal file
56
setup_windows.bat
Normal 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
|
||||||
261
sing_detect.py
261
sing_detect.py
@@ -1,261 +0,0 @@
|
|||||||
import customtkinter as ctk
|
|
||||||
import tkinter as tk
|
|
||||||
from tkinter import filedialog, messagebox
|
|
||||||
import os
|
|
||||||
import csv
|
|
||||||
from datetime import timedelta
|
|
||||||
|
|
||||||
import math
|
|
||||||
import librosa
|
|
||||||
import soundfile as sf
|
|
||||||
|
|
||||||
from transformers import pipeline
|
|
||||||
from pyannote.core import Segment
|
|
||||||
|
|
||||||
# ------------------------
|
|
||||||
# Time formatting function
|
|
||||||
# ------------------------
|
|
||||||
def seconds_to_timecode(total_seconds):
|
|
||||||
"""
|
|
||||||
Convert float seconds -> "HH:MM:SS.mmm" for DaVinci Resolve CSV markers.
|
|
||||||
"""
|
|
||||||
td = timedelta(seconds=total_seconds)
|
|
||||||
hours, remainder = divmod(td.seconds, 3600)
|
|
||||||
minutes, secs = divmod(remainder, 60)
|
|
||||||
ms = int(td.microseconds / 1000)
|
|
||||||
return f"{hours:02d}:{minutes:02d}:{secs:02d}.{ms:03d}"
|
|
||||||
|
|
||||||
def chunk_audio(audio_path, chunk_length_s=10.0, stride_s=5.0, sr=16000):
|
|
||||||
"""
|
|
||||||
Loads the entire audio with librosa, then yields overlapping chunks
|
|
||||||
(chunk_data, chunk_start, chunk_end, sr).
|
|
||||||
|
|
||||||
- chunk_length_s = length (seconds) of each chunk
|
|
||||||
- stride_s = how much we "step back" within each chunk for overlap
|
|
||||||
e.g. chunk_length=10, stride=5 => chunk #0 covers
|
|
||||||
0-10s, chunk #1 covers 5-15s, chunk #2 covers 10-20s, etc.
|
|
||||||
- sr = sampling rate to load the audio
|
|
||||||
"""
|
|
||||||
audio, sr = librosa.load(audio_path, sr=sr)
|
|
||||||
total_len_s = len(audio) / sr
|
|
||||||
|
|
||||||
# Step in seconds, between the start points of consecutive chunks
|
|
||||||
# e.g. for chunk=10, stride=5 => step=10 - 5=5
|
|
||||||
# means next chunk starts 5s after the previous chunk start
|
|
||||||
step = chunk_length_s - stride_s
|
|
||||||
|
|
||||||
# Edge case: if stride_s >= chunk_length_s, you won't have overlap
|
|
||||||
if step <= 0:
|
|
||||||
step = chunk_length_s # no overlap
|
|
||||||
|
|
||||||
# Figure out how many chunks needed so we don’t exceed total length
|
|
||||||
# (a bit of math to ensure we handle trailing audio < chunk_length_s)
|
|
||||||
num_chunks = math.ceil((total_len_s - chunk_length_s) / step) + 1
|
|
||||||
if num_chunks < 1:
|
|
||||||
num_chunks = 1
|
|
||||||
|
|
||||||
for i in range(num_chunks):
|
|
||||||
chunk_start = i * step
|
|
||||||
chunk_end = chunk_start + chunk_length_s
|
|
||||||
|
|
||||||
# If the chunk goes beyond total_len_s, clamp it
|
|
||||||
if chunk_end > total_len_s:
|
|
||||||
chunk_end = total_len_s
|
|
||||||
|
|
||||||
# Convert times to sample indexes
|
|
||||||
start_sample = int(chunk_start * sr)
|
|
||||||
end_sample = int(chunk_end * sr)
|
|
||||||
|
|
||||||
# If our chunk_start >= total_len_s, we can stop
|
|
||||||
if chunk_start >= total_len_s:
|
|
||||||
break
|
|
||||||
|
|
||||||
# Slice the waveform
|
|
||||||
chunk_data = audio[start_sample:end_sample]
|
|
||||||
|
|
||||||
yield chunk_data, chunk_start, chunk_end, sr
|
|
||||||
|
|
||||||
class MusicMarkerApp(ctk.CTk):
|
|
||||||
def __init__(self):
|
|
||||||
super().__init__()
|
|
||||||
|
|
||||||
self.title("Music Marker Generator")
|
|
||||||
self.geometry("500x350")
|
|
||||||
|
|
||||||
# Variables to hold file paths & threshold
|
|
||||||
self.audio_file_path = tk.StringVar(value="")
|
|
||||||
self.output_csv_path = tk.StringVar(value="")
|
|
||||||
self.merge_threshold_var = tk.DoubleVar(value=2.0) # default: merge segments within 2 seconds
|
|
||||||
self.pipeline = None # We'll load it once on demand
|
|
||||||
|
|
||||||
self._create_widgets()
|
|
||||||
|
|
||||||
def _create_widgets(self):
|
|
||||||
# 1) Frame for selecting audio file
|
|
||||||
file_frame = ctk.CTkFrame(self)
|
|
||||||
file_frame.pack(pady=10, padx=10, fill="x")
|
|
||||||
|
|
||||||
file_label = ctk.CTkLabel(file_frame, text="Audio File:")
|
|
||||||
file_label.pack(side="left", padx=5)
|
|
||||||
|
|
||||||
file_entry = ctk.CTkEntry(file_frame, textvariable=self.audio_file_path, width=250)
|
|
||||||
file_entry.pack(side="left", padx=5)
|
|
||||||
|
|
||||||
file_button = ctk.CTkButton(file_frame, text="Browse", command=self._browse_audio_file)
|
|
||||||
file_button.pack(side="left", padx=5)
|
|
||||||
|
|
||||||
# 2) Frame for output CSV
|
|
||||||
output_frame = ctk.CTkFrame(self)
|
|
||||||
output_frame.pack(pady=10, padx=10, fill="x")
|
|
||||||
|
|
||||||
output_label = ctk.CTkLabel(output_frame, text="Output CSV:")
|
|
||||||
output_label.pack(side="left", padx=5)
|
|
||||||
|
|
||||||
output_entry = ctk.CTkEntry(output_frame, textvariable=self.output_csv_path, width=250)
|
|
||||||
output_entry.pack(side="left", padx=5)
|
|
||||||
|
|
||||||
output_button = ctk.CTkButton(output_frame, text="Browse", command=self._browse_output_csv)
|
|
||||||
output_button.pack(side="left", padx=5)
|
|
||||||
|
|
||||||
# 3) Merge Threshold
|
|
||||||
threshold_frame = ctk.CTkFrame(self)
|
|
||||||
threshold_frame.pack(pady=10, padx=10, fill="x")
|
|
||||||
|
|
||||||
threshold_label = ctk.CTkLabel(threshold_frame, text="Merge Gap (sec):")
|
|
||||||
threshold_label.pack(side="left", padx=5)
|
|
||||||
|
|
||||||
threshold_entry = ctk.CTkEntry(threshold_frame, textvariable=self.merge_threshold_var, width=50)
|
|
||||||
threshold_entry.pack(side="left", padx=5)
|
|
||||||
|
|
||||||
# 4) Run button
|
|
||||||
run_button = ctk.CTkButton(self, text="Run Music Detection", command=self._run_detection)
|
|
||||||
run_button.pack(pady=10)
|
|
||||||
|
|
||||||
# 5) Status label
|
|
||||||
self.status_label = ctk.CTkLabel(self, text="", wraplength=400, justify="left")
|
|
||||||
self.status_label.pack(pady=5)
|
|
||||||
|
|
||||||
def _browse_audio_file(self):
|
|
||||||
file_path = filedialog.askopenfilename(
|
|
||||||
title="Select Audio File",
|
|
||||||
filetypes=[("Audio Files", "*.wav *.mp3 *.flac *.m4a *.aac *.ogg *.wma *.aif *.aiff")]
|
|
||||||
)
|
|
||||||
if file_path:
|
|
||||||
self.audio_file_path.set(file_path)
|
|
||||||
|
|
||||||
def _browse_output_csv(self):
|
|
||||||
file_path = filedialog.asksaveasfilename(
|
|
||||||
title="Select Output CSV",
|
|
||||||
defaultextension=".csv",
|
|
||||||
filetypes=[("CSV Files", "*.csv")]
|
|
||||||
)
|
|
||||||
if file_path:
|
|
||||||
self.output_csv_path.set(file_path)
|
|
||||||
|
|
||||||
def _run_detection(self):
|
|
||||||
audio_path = self.audio_file_path.get().strip()
|
|
||||||
output_csv = self.output_csv_path.get().strip()
|
|
||||||
merge_threshold = self.merge_threshold_var.get()
|
|
||||||
|
|
||||||
if not audio_path or not os.path.isfile(audio_path):
|
|
||||||
messagebox.showerror("Error", "Please select a valid audio file.")
|
|
||||||
return
|
|
||||||
if not output_csv:
|
|
||||||
messagebox.showerror("Error", "Please specify an output CSV file.")
|
|
||||||
return
|
|
||||||
|
|
||||||
self._set_status("Loading model, please wait...")
|
|
||||||
|
|
||||||
# Load pipeline if not loaded yet
|
|
||||||
if self.pipeline is None:
|
|
||||||
try:
|
|
||||||
# Standard audio-classification pipeline
|
|
||||||
self.pipeline = pipeline(
|
|
||||||
"audio-classification",
|
|
||||||
model="MarekCech/GenreVim-Music-Detection-DistilHuBERT"
|
|
||||||
)
|
|
||||||
except Exception as e:
|
|
||||||
messagebox.showerror("Model Error", f"Could not load the model:\n{e}")
|
|
||||||
return
|
|
||||||
|
|
||||||
self._set_status("Chunking audio and running music detection...")
|
|
||||||
|
|
||||||
# We'll collect "music" segments from each chunk
|
|
||||||
music_segments = []
|
|
||||||
|
|
||||||
# Manually chunk the audio & classify each chunk
|
|
||||||
try:
|
|
||||||
for chunk_data, chunk_start, chunk_end, sr in chunk_audio(
|
|
||||||
audio_path,
|
|
||||||
chunk_length_s=10.0,
|
|
||||||
stride_s=5.0,
|
|
||||||
sr=16000
|
|
||||||
):
|
|
||||||
# The pipeline expects a waveform plus sampling_rate, not a path
|
|
||||||
# So pass the chunk_data + sr
|
|
||||||
results = self.pipeline(chunk_data, sampling_rate=sr)
|
|
||||||
if not results:
|
|
||||||
continue
|
|
||||||
|
|
||||||
# best guess from the pipeline for this chunk
|
|
||||||
best_guess = max(results, key=lambda x: x["score"])
|
|
||||||
if best_guess["label"].lower() == "music" and best_guess["score"] > 0.5:
|
|
||||||
music_segments.append(Segment(chunk_start, chunk_end))
|
|
||||||
|
|
||||||
except Exception as e:
|
|
||||||
messagebox.showerror("Detection Error", f"Error running music detection:\n{e}")
|
|
||||||
return
|
|
||||||
|
|
||||||
# Sort segments by start time
|
|
||||||
music_segments.sort(key=lambda seg: seg.start)
|
|
||||||
|
|
||||||
# Merge close segments based on threshold
|
|
||||||
merged_segments = []
|
|
||||||
if not music_segments:
|
|
||||||
self._set_status("No music segments found. No CSV generated.")
|
|
||||||
return
|
|
||||||
else:
|
|
||||||
current_start = music_segments[0].start
|
|
||||||
current_end = music_segments[0].end
|
|
||||||
|
|
||||||
for seg in music_segments[1:]:
|
|
||||||
# If the new segment starts within X seconds of the old segment, merge them
|
|
||||||
if seg.start <= current_end + merge_threshold:
|
|
||||||
current_end = max(current_end, seg.end)
|
|
||||||
else:
|
|
||||||
merged_segments.append(Segment(current_start, current_end))
|
|
||||||
current_start = seg.start
|
|
||||||
current_end = seg.end
|
|
||||||
|
|
||||||
# finalize last
|
|
||||||
merged_segments.append(Segment(current_start, current_end))
|
|
||||||
|
|
||||||
# Write CSV for DaVinci Resolve
|
|
||||||
try:
|
|
||||||
with open(output_csv, "w", newline="", encoding="utf-8") as csvfile:
|
|
||||||
writer = csv.writer(csvfile)
|
|
||||||
# DaVinci Resolve CSV marker columns
|
|
||||||
writer.writerow(["Name", "Start", "End", "Color", "Marker Type"])
|
|
||||||
|
|
||||||
for i, seg in enumerate(merged_segments, start=1):
|
|
||||||
start_tc = seconds_to_timecode(seg.start)
|
|
||||||
end_tc = seconds_to_timecode(seg.end)
|
|
||||||
name = f"Music Segment #{i}"
|
|
||||||
color = "Green"
|
|
||||||
marker_type = "Comment"
|
|
||||||
writer.writerow([name, start_tc, end_tc, color, marker_type])
|
|
||||||
|
|
||||||
self._set_status(
|
|
||||||
f"Done! Found {len(merged_segments)} music segments.\n"
|
|
||||||
f"CSV saved to: {output_csv}"
|
|
||||||
)
|
|
||||||
except Exception as e:
|
|
||||||
messagebox.showerror("File Error", f"Could not write CSV:\n{e}")
|
|
||||||
|
|
||||||
def _set_status(self, msg):
|
|
||||||
self.status_label.configure(text=msg)
|
|
||||||
self.update_idletasks()
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
app = MusicMarkerApp()
|
|
||||||
app.mainloop()
|
|
||||||
Reference in New Issue
Block a user