38 lines
1.4 KiB
Python
38 lines
1.4 KiB
Python
import subprocess
|
|
|
|
import numpy as np
|
|
|
|
from core.paths import _bin
|
|
|
|
|
|
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")
|