import subprocess import numpy as np from core.paths import _bin def x_to_t(x: float, width: float, view_start: float, view_dur: float) -> float: """Map a pixel x in [0,width] to a time in [view_start, view_start+view_dur].""" if width <= 0 or view_dur <= 0: return view_start return view_start + (x / width) * view_dur def t_to_x(t: float, width: float, view_start: float, view_dur: float) -> int: """Map a time to a pixel x in [0,width] (rounded).""" if width <= 0 or view_dur <= 0: return 0 return int(round((t - view_start) / view_dur * width)) def peaks(samples, buckets: int = 128) -> list[float]: """Reduce a 1-D sample array to *buckets* normalized peak magnitudes (0..1).""" if samples is None or len(samples) == 0: return [0.0] * buckets a = np.abs(np.asarray(samples, dtype="float32")) idx = np.linspace(0, len(a), buckets + 1).astype(int) out = [float(a[idx[i]:idx[i + 1]].max()) if idx[i + 1] > idx[i] else 0.0 for i in range(buckets)] m = max(out) or 1.0 return [v / m for v in out] def load_region_samples(path: str, start: float, duration: float, sr: int = 8000) -> np.ndarray: """Decode a [start, start+duration] mono slice via ffmpeg for waveform preview (low sample rate). Returns a float32 numpy array, empty on failure.""" cmd = [ _bin("ffmpeg"), "-ss", str(start), "-i", path, "-t", str(duration), "-vn", "-ac", "1", "-ar", str(sr), "-f", "f32le", "-loglevel", "error", "pipe:1", ] try: proc = subprocess.run(cmd, capture_output=True, timeout=60) except Exception: return np.zeros(0, dtype="float32") if proc.returncode != 0 or not proc.stdout: return np.zeros(0, dtype="float32") buf = proc.stdout buf = buf[: len(buf) - (len(buf) % 4)] return np.frombuffer(buf, dtype="float32")