# --- standalone setup ----------------------------------------------------------
# Lifted from the Setup section and the helper cells above so this demo runs on its
# own in a fresh kernel - no earlier cell has to have been executed first.
import ctypes
import ctypes.util
import gc
import numpy as np
import time
import torch
from dotenv import find_dotenv, load_dotenv
from pathlib import Path
# Knowledge/.env sets HF_TOKEN - authenticated HF Hub requests get higher rate limits
load_dotenv(find_dotenv(usecwd=True))
def vram(tag=""):
"Report current GPU memory (allocated / reserved). No-op on CPU."
if torch.cuda.is_available():
alloc = torch.cuda.memory_allocated() / 1e9
reserved = torch.cuda.memory_reserved() / 1e9
print(f"VRAM {tag:16s} {alloc:5.2f} GB allocated / {reserved:5.2f} GB reserved")
def free_memory():
"GC then release cached CPU/GPU memory. Call right after `del model`.\n\n `del` drops the Python reference; this reclaims the RAM and hands the\n freed VRAM back to the CUDA allocator so usage stays flat across cells.\n "
gc.collect()
if torch.cuda.is_available():
torch.cuda.empty_cache()
torch.cuda.ipc_collect()
# glibc keeps freed CPU allocations in its arenas instead of returning them
# to the OS, so RSS compounds across model sections (cpu-offloaded weights
# live in system RAM). malloc_trim(0) hands the freed arenas back. See
# dl-visualization-and-memory.instructions.md - not optional on a 12 GB box.
try:
ctypes.CDLL(ctypes.util.find_library("c") or "libc.so.6").malloc_trim(0)
except Exception:
pass
# All downloads (samples, HF cache) go to DL_tasks/datasets/ (gitignored)
DATA_DIR = Path("../../datasets")
DATA_DIR.mkdir(exist_ok=True)
SR = 16000
FRAME = SR // 100 # 10 ms
# --- the demo ------------------------------------------------------------------
# sounddevice needs the PortAudio runtime (libportaudio2) and ALSA nodes under
# /dev/snd. On the knowledge-lab LXC those are passed in from the Proxmox host by
# `av_devices` in infra/proxmox/variables.tf - the camera's video node and its
# USB-Audio card are separate passthroughs.
import io
import sounddevice as sd
from IPython.display import HTML, display
from IPython.utils.capture import capture_output
from rich import box
from rich.console import Console, Group
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
# Silero is a STREAMING model, and this demo uses it as one. The other live demos in
# this repo fake streaming by re-decoding a sliding window, because Whisper, AST and
# CLAP are offline models with no notion of partial input. Silero is different: it
# carries recurrent state and consumes one small frame at a time, so here the model
# really is fed 32 ms at a time and answers before you have stopped talking.
VAD_FRAME = 512 # Silero accepts EXACTLY this many samples at 16 kHz. 480 and
# 1024 both raise inside the TorchScript interpreter - it is a
# hard contract, not a hint.
LISTEN_SECONDS = 30 # interrupt the kernel to stop early
UI_EVERY = 0.25 # seconds between view refreshes; the model runs every frame
# Endpointing by hysteresis: one threshold flickers on and off around the boundary,
# so opening and closing use different ones, and a segment has to persist to count.
START_PROB = 0.50 # frame probability that opens a segment
END_PROB = 0.35 # ... and the lower bar it must fall under to close it
MIN_SILENCE_MS = 300 # quiet this long ends a segment (shorter = choppier)
MIN_SPEECH_MS = 120 # segments briefer than this are discarded as blips
SHOW_ROWS = 12
MIC_HINT = "UGREEN" # substring of the PortAudio device name; None -> first input
def pick_microphone(hint=MIC_HINT):
"Return the index of a capture device, preferring one whose name matches `hint`."
inputs = [(i, d) for i, d in enumerate(sd.query_devices()) if d["max_input_channels"] > 0]
if not inputs:
raise RuntimeError(
"no audio capture device: PortAudio sees no ALSA card. Check that /dev/snd "
"exists and holds controlC*/pcmC*D*c nodes - on the LXC that means adding "
"them to av_devices and running `just tf-apply`."
)
match = [(i, d) for i, d in inputs if hint and hint.lower() in d["name"].lower()]
return (match or inputs)[0][0]
# rich draws the panels; the in-place update is an IPython display handle. rich's own
# Live goes through ipywidgets in Jupyter and appends a fresh view per refresh
# instead of replacing one, so render with rich and update with the handle.
# force_jupyter=False is load-bearing: left on, Console.print() calls display()
# itself and every frame leaks an extra output cell.
_console = Console(record=True, file=io.StringIO(), width=100,
force_terminal=True, force_jupyter=False)
_FRAGMENT = '<pre style="font-family:ui-monospace,monospace;line-height:1.3;margin:0">{code}</pre>'
def _html(renderable):
"Render a rich object to an HTML fragment suitable for display handle updates."
_console.print(renderable)
out = _console.export_html(inline_styles=True, code_format=_FRAGMENT)
_console.file = io.StringIO() # reset between frames, else the HTML grows forever
return HTML(out)
def _note(text, title=None, style="dim"):
"A one-off rich line, so nothing in this cell falls back to a bare print."
display(_html(Panel(Text.from_markup(text), title=title, border_style=style,
padding=(0, 1))))
def _grid(**rows):
"Two-column rich grid - the shape used for every stats block in this cell."
g = Table.grid(padding=(0, 2))
g.add_column(style="dim", justify="right")
g.add_column()
for k, v in rows.items():
g.add_row(k.replace("_", " "), str(v))
return g
def rvram(tag=""):
"vram(), rendered through rich instead of print. No-op on CPU."
if not torch.cuda.is_available():
return
display(_html(Panel(
_grid(**{"allocated": f"{torch.cuda.memory_allocated() / 1e9:.2f} GB",
"reserved": f"{torch.cuda.memory_reserved() / 1e9:.2f} GB"}),
title=f"[dim]VRAM {tag}[/]", border_style="dim", padding=(0, 1))))
def _dbfs(x):
"RMS level of a float32 block, in dBFS. -inf for digital silence."
r = float(np.sqrt((x ** 2).mean())) if len(x) else 0.0
return 20 * np.log10(r) if r > 0 else float("-inf")
def _view(elapsed, seconds, prob, dbfs, segments, live_start=None, note=""):
"One frame of the live view: header stats over one table row per finished segment."
speaking = live_start is not None
head = Table.grid(padding=(0, 2))
head.add_column(style="dim", justify="right")
head.add_column()
bar = int(max(0.0, min(1.0, prob)) * 24)
colour = "green" if speaking else "yellow"
head.add_row("elapsed", f"{elapsed:5.1f}s / {seconds:.0f}s")
head.add_row("speech prob", f"[{colour}]{'#' * bar}[/][dim]{'.' * (24 - bar)}[/] {prob:.2f}")
head.add_row("level", f"{dbfs:6.1f} dBFS")
head.add_row("state", f"[green]SPEECH since {live_start:.1f}s[/]" if speaking
else "[dim]silence[/]")
if note:
head.add_row("status", f"[dim]{note}[/]")
voiced = sum(s["end"] - s["start"] for s in segments)
head.add_row("totals", f"{len(segments)} segment(s), {voiced:.1f}s voiced"
+ (f", {voiced / elapsed:.0%} of elapsed" if elapsed > 0 else ""))
table = Table(box=box.SIMPLE_HEAD, pad_edge=False, expand=True,
header_style="dim", border_style="dim")
table.add_column("#", justify="right", style="dim", width=4, no_wrap=True)
table.add_column("start", justify="right", width=8, no_wrap=True)
table.add_column("end", justify="right", width=8, no_wrap=True)
table.add_column("dur", justify="right", width=7, no_wrap=True)
table.add_column("mean p", justify="right", width=8, no_wrap=True)
table.add_column("peak p", justify="right", width=8, no_wrap=True)
shown = segments[-SHOW_ROWS:]
if len(segments) > len(shown):
table.add_row("...", "", "", "", "", f"[dim]{len(segments) - len(shown)} earlier[/]")
for s in shown:
table.add_row(str(s["n"]), f"{s['start']:.2f}s", f"{s['end']:.2f}s",
f"{s['end'] - s['start']:.2f}s", f"{s['mean_p']:.2f}",
f"{s['peak_p']:.2f}")
if speaking:
table.add_row(f"[green]{len(segments) + 1}[/]", f"[green]{live_start:.2f}s[/]",
"[green]...[/]", f"[green]{elapsed - live_start:.2f}s[/]", "", "")
if not segments and not speaking:
table.add_row("-", "", "", "", "", "[dim italic](no speech yet)[/]")
return Panel(Group(head, Text(""), table),
title="[cyan]live voice activity detection[/]", border_style="cyan",
padding=(0, 1))
def live_vad(seconds=LISTEN_SECONDS):
"""Stream the microphone through Silero one 32 ms frame at a time.
The capture stream is opened at 16 kHz with blocksize=VAD_FRAME, so every
PortAudio callback delivers exactly one frame the model will accept - no
resampling, no re-blocking, no frame-alignment arithmetic. (Verified that the
device really honours 16 kHz rather than silently returning 44.1 kHz data with
the wrong label: 80000 samples took 5.4 s of wall clock, not 1.8 s.)
Silero costs 0.33 ms per 32 ms frame here - a real-time factor of 0.010 - so
unlike the ASR and classification demos there is no latency budget to manage.
Returns (segments, trace); both feed vad_charts().
"""
import queue as _queue
mic = pick_microphone()
q, overflows = _queue.Queue(), 0
def on_audio(indata, frames, time_info, status):
"PortAudio callback thread - keep it cheap, just hand the frame over."
nonlocal overflows
if status.input_overflow:
overflows += 1
q.put(indata[:, 0].copy()) # copy: PortAudio reuses this buffer
model.reset_states() # recurrent state must not carry over from an earlier run
view = display(_html(_view(0.0, seconds, 0.0, float("-inf"), [], note="starting")),
display_id=True)
segments, trace = [], []
live_start, quiet_ms, run_probs = None, 0.0, []
frame_ms = 1000 * VAD_FRAME / SR
last_ui = -UI_EVERY
with sd.InputStream(device=mic, samplerate=SR, channels=1, dtype="float32",
blocksize=VAD_FRAME, callback=on_audio):
_note(f"[bold]mic [{mic}][/] @ {SR} Hz -> {VAD_FRAME}-sample frames "
f"({frame_ms:.0f} ms), listening {seconds:.0f}s - talk, then pause",
title="[cyan]capture[/]", style="cyan")
t0 = time.perf_counter()
try:
while time.perf_counter() - t0 < seconds:
try:
frame = q.get(timeout=0.2)
except _queue.Empty:
continue
if frame.size != VAD_FRAME: # never seen in practice; the contract is hard
continue
with torch.no_grad():
prob = float(model(torch.from_numpy(frame), SR).item())
elapsed = time.perf_counter() - t0
level = _dbfs(frame)
trace.append({"at": elapsed, "prob": prob, "dbfs": level})
# Hysteresis state machine: open on START_PROB, close only after
# MIN_SILENCE_MS below END_PROB, discard anything under MIN_SPEECH_MS.
if live_start is None:
if prob >= START_PROB:
live_start, quiet_ms, run_probs = elapsed, 0.0, [prob]
else:
run_probs.append(prob)
quiet_ms = quiet_ms + frame_ms if prob < END_PROB else 0.0
if quiet_ms >= MIN_SILENCE_MS:
end = elapsed - quiet_ms / 1000
if (end - live_start) * 1000 >= MIN_SPEECH_MS:
segments.append({
"n": len(segments) + 1, "start": live_start, "end": end,
"mean_p": float(np.mean(run_probs)),
"peak_p": float(np.max(run_probs)),
})
live_start, quiet_ms, run_probs = None, 0.0, []
if elapsed - last_ui >= UI_EVERY: # model every frame, redraw far less
last_ui = elapsed
view.update(_html(_view(elapsed, seconds, prob, level, segments,
live_start)))
except KeyboardInterrupt:
view.update(_html(_view(time.perf_counter() - t0, seconds, 0.0,
float("-inf"), segments, live_start, note="stopped")))
if live_start is not None: # speech still open when the clock ran out
segments.append({"n": len(segments) + 1, "start": live_start,
"end": time.perf_counter() - t0,
"mean_p": float(np.mean(run_probs)),
"peak_p": float(np.max(run_probs))})
voiced = sum(s["end"] - s["start"] for s in segments)
total = trace[-1]["at"] if trace else 0.0
display(_html(Panel(_grid(**{
"frames": f"{len(trace)} x {frame_ms:.0f} ms",
"segments": str(len(segments)),
"voiced": f"{voiced:.1f}s of {total:.1f}s ({voiced / total:.0%})" if total else "-",
"overflows": f"[red]{overflows}[/]" if overflows else "0",
}), title="[dim]done[/]", border_style="dim", padding=(0, 1))))
return segments, trace
def vad_charts(segments, trace):
"""Two ECharts views of the run (pyecharts is the repo standard for all charts).
The probability trace with both thresholds drawn on it is the useful picture: you
can see directly why a boundary landed where it did, and why one threshold would
have chattered where two do not.
"""
from pyecharts import options as opts
from pyecharts.charts import Bar, Line, Page
if not trace:
_note("no frames captured - nothing to chart", title="[yellow]empty[/]",
style="yellow")
return None
x = [round(t["at"], 2) for t in trace]
mask = [0] * len(trace)
for s in segments:
for i, t in enumerate(trace):
if s["start"] <= t["at"] <= s["end"]:
mask[i] = 1
prob = (
Line(init_opts=opts.InitOpts(width="820px", height="400px"))
.add_xaxis(x)
.add_yaxis("speech probability", [round(t["prob"], 4) for t in trace],
is_symbol_show=False, is_smooth=False,
areastyle_opts=opts.AreaStyleOpts(opacity=0.15),
label_opts=opts.LabelOpts(is_show=False))
.add_yaxis("endpointed speech", mask, is_step=True, is_symbol_show=False,
label_opts=opts.LabelOpts(is_show=False))
.set_global_opts(
title_opts=opts.TitleOpts(
title="Silero speech probability, frame by frame",
subtitle=f"{1000 * VAD_FRAME / SR:.0f} ms frames - open at "
f"{START_PROB}, close under {END_PROB} for {MIN_SILENCE_MS} ms"),
xaxis_opts=opts.AxisOpts(name="seconds", type_="value"),
yaxis_opts=opts.AxisOpts(name="probability", max_=1),
tooltip_opts=opts.TooltipOpts(trigger="axis"),
legend_opts=opts.LegendOpts(pos_top="10%"),
)
.set_series_opts(markline_opts=opts.MarkLineOpts(data=[
opts.MarkLineItem(y=START_PROB, name="open"),
opts.MarkLineItem(y=END_PROB, name="close"),
]))
)
if segments:
durations = Bar(init_opts=opts.InitOpts(width="820px", height="320px"))
durations.add_xaxis([str(s["n"]) for s in segments])
durations.add_yaxis("duration (s)",
[round(s["end"] - s["start"], 2) for s in segments])
durations.add_yaxis("mean probability",
[round(s["mean_p"], 3) for s in segments])
durations.set_global_opts(
title_opts=opts.TitleOpts(title="Endpointed segments",
subtitle="what the state machine actually committed"),
xaxis_opts=opts.AxisOpts(name="segment"),
tooltip_opts=opts.TooltipOpts(trigger="axis"),
legend_opts=opts.LegendOpts(pos_top="10%"),
)
durations.set_series_opts(label_opts=opts.LabelOpts(is_show=False))
return Page().add(prob, durations).render_notebook()
return prob.render_notebook()
# torch.hub prints its cache path on load; swallow it so the cell's output stays rich.
torch.hub.set_dir(str(DATA_DIR / "torch_hub"))
with capture_output():
model, _utils = torch.hub.load("snakers4/silero-vad", "silero_vad", trust_repo=True)
_note("Silero VAD loaded [dim](~1.8M params, CPU is plenty - 0.33 ms per 32 ms frame)[/]",
title="[dim]model[/]")
SEGMENTS, TRACE = live_vad(seconds=30)
del model
free_memory()
rvram("after live vad")
vad_charts(SEGMENTS, TRACE)