# --- 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 torch
from dotenv import find_dotenv, load_dotenv
from pathlib import Path
from transformers import pipeline
# Knowledge/.env sets HF_TOKEN - authenticated HF Hub requests get higher rate limits
load_dotenv(find_dotenv(usecwd=True))
device = "cuda:0" if torch.cuda.is_available() else "cpu"
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)
# --- 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 time
import librosa
import sounddevice as sd
from IPython.display import HTML, display
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
# The pipelines emit config warnings on every call; at one window a second that
# buries the output. Errors still get through.
from transformers.utils import logging as hf_logging
hf_logging.set_verbosity_error()
CHUNK_SECONDS = 3.0 # how much trailing audio each classification sees
HOP_SECONDS = 1.0 # how often that window is re-classified
SILENCE_DBFS = -45.0 # windows quieter than this are not sent to the models
SHOW_ROWS = 12 # most recent rows kept on screen; the full log feeds the charts
CLAP_SR, AST_SR = 48_000, 16_000 # each model's native feature-extractor rate
MY_SOUNDS = ["speech", "typing on a keyboard", "a fan or air conditioner",
"music", "silence", "a door closing"]
# Substring of the *PortAudio* device name, which is not the ALSA card id: this mic
# is card id "U2K" but shows up as "UGREEN camera 2K: USB Audio (hw:0,0)". Run
# `python -c "import sounddevice; print(sounddevice.query_devices())"` to list them.
MIC_HINT = "UGREEN" # None -> just take the first input device
def pick_microphone(hint=MIC_HINT):
"Return (device_index, native_sample_rate) for a capture device, preferring `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()]
idx, info = (match or inputs)[0]
return idx, int(info["default_samplerate"])
# rich draws the panel; 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 x.size else 0.0
return 20 * np.log10(r) if r > 0 else float("-inf")
def _view(elapsed, seconds, dbfs, rows, note="", fill=None):
"One frame of the live view: header stats over one table row per classified window."
head = Table.grid(padding=(0, 2))
head.add_column(style="dim", justify="right")
head.add_column()
meter = "#" * int(max(0.0, min(1.0, (dbfs + 60) / 60)) * 24)
head.add_row("elapsed", f"{elapsed:5.1f}s / {seconds:.0f}s")
head.add_row("level", f"{dbfs:6.1f} dBFS [dim]{meter}[/]")
if fill is not None:
done = int(max(0.0, min(1.0, fill)) * 24)
head.add_row("next hop", f"[dim]{'=' * done}{'.' * (24 - done)}[/] {fill:4.0%}")
if note:
head.add_row("status", f"[dim]{note}[/]")
head.add_row("totals", f"{len(rows)} window(s), "
f"{sum(r['ast_ms'] + r['clap_ms'] for r in rows) / 1000:.1f}s classifying")
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("window", justify="right", width=13, no_wrap=True)
table.add_column("dBFS", justify="right", width=6, no_wrap=True)
table.add_column("ms", justify="right", width=9, no_wrap=True)
table.add_column("AST (527 fixed classes)", overflow="ellipsis", ratio=1)
table.add_column("CLAP (your labels)", overflow="ellipsis", ratio=1)
shown = rows[-SHOW_ROWS:]
if len(rows) > len(shown):
table.add_row("...", "", "", "", f"[dim]{len(rows) - len(shown)} earlier[/]",
"[dim]all of them are still in the charts[/]")
for r in shown:
table.add_row(
str(r["window"]), f"{r['from']:.1f}-{r['at']:.1f}s", f"{r['dbfs']:.0f}",
f"{r['ast_ms'] + r['clap_ms']:.0f}",
f"{r['ast_label']} [dim]{r['ast_score']:.2f}[/]",
f"[green]{r['clap_label']}[/] [dim]{r['clap_score']:.2f}[/]")
if not rows:
table.add_row("-", "", "", "", "[dim italic](nothing classified yet)[/]", "")
return Panel(Group(head, Text(""), table),
title="[cyan]live audio classification[/]", border_style="cyan",
padding=(0, 1))
def live_classify(seconds=30, chunk_seconds=CHUNK_SECONDS, hop_seconds=HOP_SECONDS,
labels=MY_SOUNDS):
"""Classify a sliding window of live microphone audio with AST and CLAP at once.
A `chunk_seconds` window slides over the stream and is re-classified every
`hop_seconds`, so each moment of audio is scored about three times with
different context. Unlike streaming ASR there is nothing to de-duplicate: the
overlap simply smooths the score trajectory, which is what the chart plots.
Both models stay resident so they see the SAME window - that comparison is the
whole point of the section, and it is what a sequential load/free would destroy.
In fp16 the pair costs 0.51 GB (measured) and about 104 ms per window, so a 1 s
hop runs at a real-time factor near 0.10.
Returns (log, labels); log feeds the charts.
"""
import queue as _queue
mic, native_sr = pick_microphone()
q, overflows = _queue.Queue(), 0
def on_audio(indata, frames, time_info, status):
"PortAudio callback thread - keep it cheap, just hand the samples over."
nonlocal overflows
if status.input_overflow:
overflows += 1
q.put(indata[:, 0].copy()) # copy: PortAudio reuses this buffer
prompts = [f"the sound of {s}" for s in labels]
view = display(_html(_view(0.0, seconds, float("-inf"), [], note="starting")),
display_id=True)
buf = np.zeros(0, dtype="float32")
window_n, hop_n, fresh = int(chunk_seconds * native_sr), int(hop_seconds * native_sr), 0
log = []
with sd.InputStream(device=mic, samplerate=native_sr, channels=1, dtype="float32",
blocksize=int(native_sr * 0.1), callback=on_audio):
_note(f"[bold]mic [{mic}][/] @ {native_sr} Hz -> "
f"{chunk_seconds:.0f}s window every {hop_seconds:.1f}s, "
f"listening {seconds:.0f}s - make some noise",
title="[cyan]capture[/]", style="cyan")
t0 = time.perf_counter()
try:
while time.perf_counter() - t0 < seconds:
chunks = []
try:
chunks.append(q.get(timeout=0.2))
except _queue.Empty:
pass
while True: # drain whatever piled up while the last window ran
try:
chunks.append(q.get_nowait())
except _queue.Empty:
break
if chunks:
fresh += sum(c.size for c in chunks)
buf = np.concatenate([buf, *chunks])[-window_n:] # slide, do not grow
elapsed = time.perf_counter() - t0
if fresh < hop_n or buf.size < int(0.8 * native_sr):
view.update(_html(_view(elapsed, seconds, _dbfs(buf), log,
fill=fresh / hop_n, note="collecting")))
continue
fresh = 0
# Resample the window, not the live stream. CLAP's pipeline rejects a
# dict and assumes the array is ALREADY at 48 kHz, so the 48 kHz copy
# is the source of truth and AST's 16 kHz copy is derived from it.
a48 = (librosa.resample(buf, orig_sr=native_sr, target_sr=CLAP_SR)
if native_sr != CLAP_SR else buf)
dbfs = _dbfs(a48)
if dbfs < SILENCE_DBFS:
view.update(_html(_view(
elapsed, seconds, dbfs, log,
note=f"below {SILENCE_DBFS:.0f} dBFS - window skipped")))
continue
a16 = librosa.resample(a48, orig_sr=CLAP_SR, target_sr=AST_SR)
t1 = time.perf_counter()
ast_out = tagger({"array": a16, "sampling_rate": AST_SR})
ast_ms = 1000 * (time.perf_counter() - t1)
t1 = time.perf_counter()
clap_out = zsc(a48, candidate_labels=prompts)
clap_ms = 1000 * (time.perf_counter() - t1)
scores = {r["label"]: float(r["score"]) for r in clap_out}
top = max(scores, key=scores.get)
log.append({
"window": len(log) + 1, "at": elapsed,
"from": max(0.0, elapsed - buf.size / native_sr), "dbfs": dbfs,
"ast_ms": ast_ms, "clap_ms": clap_ms,
"ast_label": ast_out[0]["label"], "ast_score": float(ast_out[0]["score"]),
"clap_label": top.replace("the sound of ", ""),
"clap_score": scores[top], "clap_scores": scores,
})
view.update(_html(_view(elapsed, seconds, dbfs, log)))
except KeyboardInterrupt:
view.update(_html(_view(time.perf_counter() - t0, seconds, float("-inf"),
log, note="stopped")))
n = len(log)
total_ms = sum(r["ast_ms"] + r["clap_ms"] for r in log)
rtf = (total_ms / 1000) / (n * hop_seconds) if n else float("nan")
summary = Table.grid(padding=(0, 2))
summary.add_column(style="dim", justify="right")
summary.add_column()
summary.add_row("windows", f"{n} [dim]({chunk_seconds:.0f}s window, "
f"{hop_seconds:.1f}s hop)[/]")
summary.add_row("mean AST", f"{sum(r['ast_ms'] for r in log) / max(n, 1):.0f} ms")
summary.add_row("mean CLAP", f"{sum(r['clap_ms'] for r in log) / max(n, 1):.0f} ms")
summary.add_row("real-time factor", f"{rtf:.3f}"
+ ("" if rtf < 1 else " [red](slower than audio)[/]"))
if overflows:
summary.add_row("overflows", f"[red]{overflows}[/] (not keeping up)")
display(_html(Panel(summary, title="[dim]done[/]", border_style="dim", padding=(0, 1))))
return log, labels
def classification_charts(log, labels, hop_seconds=HOP_SECONDS):
"""Two ECharts views of the run (pyecharts is the repo standard for all charts).
The score trajectory is the interesting one: CLAP re-scores every label on every
window, so you watch the model change its mind as the room changes. The latency
chart underneath is the cost of having done so.
"""
from pyecharts import options as opts
from pyecharts.charts import Bar, Line, Page
if not log:
_note("no windows classified - nothing rose above the silence gate",
title="[yellow]empty[/]", style="yellow")
return None
x = [f"{r['at']:.1f}" for r in log]
trend = Line(init_opts=opts.InitOpts(width="760px", height="400px")).add_xaxis(x)
for s in labels:
trend.add_yaxis(s, [round(r["clap_scores"].get(f"the sound of {s}", 0.0), 4)
for r in log],
is_smooth=True, label_opts=opts.LabelOpts(is_show=False))
trend.set_global_opts(
title_opts=opts.TitleOpts(title="CLAP score per label over time",
subtitle="zero-shot: these labels are yours, not the model's"),
xaxis_opts=opts.AxisOpts(name="seconds"),
yaxis_opts=opts.AxisOpts(name="score", max_=1),
tooltip_opts=opts.TooltipOpts(trigger="axis"),
legend_opts=opts.LegendOpts(pos_top="10%", type_="scroll"),
)
budget = 1000 * hop_seconds
lat = (
Bar(init_opts=opts.InitOpts(width="760px", height="360px"))
.add_xaxis(x)
.add_yaxis("AST ms", [round(r["ast_ms"], 1) for r in log], stack="t")
.add_yaxis("CLAP ms", [round(r["clap_ms"], 1) for r in log], stack="t")
.set_global_opts(
title_opts=opts.TitleOpts(
title="Classification latency per window",
subtitle=f"stacked; under {budget:.0f} ms keeps up with a {hop_seconds:.1f}s hop"),
xaxis_opts=opts.AxisOpts(name="seconds"),
yaxis_opts=opts.AxisOpts(name="milliseconds"),
tooltip_opts=opts.TooltipOpts(trigger="axis"),
legend_opts=opts.LegendOpts(pos_top="10%"),
)
.set_series_opts(
label_opts=opts.LabelOpts(is_show=False),
markline_opts=opts.MarkLineOpts(
data=[opts.MarkLineItem(y=budget, name="real-time budget")]),
)
)
return Page().add(trend, lat).render_notebook()
# fp16 halves the pair to 0.51 GB with identical predictions (measured: AST "Static"
# 0.681 either way), so both stay resident and score the same window.
tagger = pipeline("audio-classification", model="MIT/ast-finetuned-audioset-10-10-0.4593",
device=device, top_k=5,
dtype=torch.float16 if device != "cpu" else torch.float32)
zsc = pipeline("zero-shot-audio-classification", model="laion/clap-htsat-unfused",
device=device, dtype=torch.float16 if device != "cpu" else torch.float32)
rvram("AST + CLAP live")
LOG, LABELS = live_classify(seconds=30)
del tagger, zsc
free_memory()
rvram("after live mic")
classification_charts(LOG, LABELS)