diff --git a/core/exporters.py b/core/exporters.py deleted file mode 100644 index 6e22306..0000000 --- a/core/exporters.py +++ /dev/null @@ -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), - ]) diff --git a/core/fftools.py b/core/fftools.py deleted file mode 100644 index d3a7ac1..0000000 --- a/core/fftools.py +++ /dev/null @@ -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 diff --git a/core/segmenter.py b/core/segmenter.py deleted file mode 100644 index b30bba8..0000000 --- a/core/segmenter.py +++ /dev/null @@ -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, - } diff --git a/detect.rs b/detect.rs new file mode 100644 index 0000000..342ece4 --- /dev/null +++ b/detect.rs @@ -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, + pub stats: Value, +} + +#[derive(Clone, Serialize)] +struct DetectProgress { + message: String, +} + +fn emit(app: &AppHandle, message: impl Into) { + let _ = app.emit("detect-progress", DetectProgress { message: message.into() }); +} + +/// Locate the sidecar script: bundled resources first, dev tree as fallback. +fn sidecar_script() -> Option { + 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: /../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 { + 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 = None; + let mut err_msg: Option = 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 = 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::().chars().rev().collect(); + return Err(format!("Detection failed.\n{tail}")); + } + Err("Detection produced no result.".into()) +} diff --git a/detect.ts b/detect.ts new file mode 100644 index 0000000..9d8be28 --- /dev/null +++ b/detect.ts @@ -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 }; +} diff --git a/sidecar_detect.py b/sidecar_detect.py new file mode 100644 index 0000000..c4f93c9 --- /dev/null +++ b/sidecar_detect.py @@ -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()) diff --git a/ui/tab_detect.py b/ui/tab_detect.py deleted file mode 100644 index 5b1d9bd..0000000 --- a/ui/tab_detect.py +++ /dev/null @@ -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 diff --git a/ui/theme.py b/ui/theme.py deleted file mode 100644 index f317f0b..0000000 --- a/ui/theme.py +++ /dev/null @@ -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"), -} diff --git a/utils/formats.py b/utils/formats.py deleted file mode 100644 index a5edc59..0000000 --- a/utils/formats.py +++ /dev/null @@ -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}"