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