Audio Visualizer
Live microphone spectrum on the LEDs: five render styles (bars, mirror, segments, dots, wave) and colour themes.
How to run
This app is a single Python file that talks directly to the BUSY Bar HTTP API: no libraries, no setup. Connect your bar via USB and run:
python app.py
Over USB the bar is always at 10.0.4.20. On Wi-Fi, pass the bar's IP
with --host <address>.
No bar at hand? Run the
BUSY Bar Emulator
and point the app at it: python app.py --host 127.0.0.1:8080.
Source code
app.py
#!/usr/bin/env python3
"""Audio visualizer: live spectrum bars from your microphone.
Live spectrum from the mic in several render styles (bars, mirror, segments,
dots, wave), each recolourable with a theme, plus floating peak-hold caps.
python3 app.py # BUSY Bar over USB (always 10.0.4.20)
python3 app.py --host 127.0.0.1:8080 # emulator or a Wi-Fi bar
python3 app.py --style segments --theme fire
python3 app.py --demo # no mic/ffmpeg: cycles every style + theme
Options: --style {bars,mirror,segments,dots,wave}
--theme {classic,fire,ocean,aurora,rainbow} --fps N
Each frame is rendered to a single 72x16 image and pushed as one image element.
Individual rectangles cost ~3.6 ms each on the device, so a busy style (segments
can reach 100+ rects) either crawls or trips the element cap; one full-frame
image is a flat ~50 ms regardless, so every style runs smoothly.
Live capture needs macOS with `ffmpeg` installed (`brew install ffmpeg`) and
microphone access; it reads the built-in mic via avfoundation. The --demo mode
needs neither and is handy for a quick look.
"""
import argparse
import json
import math
import shutil
import struct
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import zlib
APP = "audio-visualizer"
SAMPLE_RATE = 22050
CHUNK_SAMPLES = 2048
NUM_BANDS = 24
FREQ_MIN = 60.0
FREQ_MAX = 8000.0
# 24 target frequencies, log-spaced from 60 Hz to 8000 Hz
_log_min = math.log(FREQ_MIN)
_log_max = math.log(FREQ_MAX)
BAND_FREQS = [
math.exp(_log_min + (_log_max - _log_min) * i / (NUM_BANDS - 1))
for i in range(NUM_BANDS)
]
W, H = 72, 16
DISPLAY_H = H
# Colour themes as vertical gradient stops: (position 0..1, (r, g, b) in 0..1).
# Position 0 is the bottom row of the display, 1 is the top. A bar samples the
# palette from its base up to its current peak, so quiet bands stay in the cool
# low colours and loud bands climb into the hot top colours.
THEMES = {
"classic": [(0.0, (0.00, 0.78, 0.00)), (0.5, (1.00, 0.78, 0.00)), (1.0, (1.00, 0.12, 0.00))],
"fire": [(0.0, (0.45, 0.00, 0.00)), (0.35, (1.00, 0.25, 0.00)), (0.7, (1.00, 0.65, 0.00)), (1.0, (1.00, 1.00, 0.75))],
"ocean": [(0.0, (0.00, 0.10, 0.55)), (0.45, (0.00, 0.50, 0.95)), (0.8, (0.00, 0.90, 0.95)), (1.0, (0.80, 1.00, 1.00))],
"aurora": [(0.0, (0.00, 0.35, 0.20)), (0.4, (0.00, 0.85, 0.50)), (0.7, (0.20, 0.95, 0.75)), (1.0, (0.65, 0.30, 0.95))],
}
THEME_NAMES = list(THEMES.keys()) + ["rainbow"]
# Render styles: the shape/layout of the visualiser (independent of --theme).
STYLE_NAMES = ["bars", "mirror", "segments", "dots", "wave"]
# Bright cap that floats on top of each bar at its recent peak.
PEAK_COLOR = (255, 255, 255)
# Peak caps fall this many pixels per second, then are scaled to the frame rate.
PEAK_FALL_PER_SEC = 9.0
# ---------------------------------------------------------------------------
# HTTP helpers — push one full-frame image per frame
# ---------------------------------------------------------------------------
def _base(host):
host = host.replace("http://", "").replace("https://", "").rstrip("/")
return "http://" + host
def _png(pixels):
"""72x16 flat list of (r, g, b) -> minimal RGBA PNG bytes (stdlib only)."""
raw = bytearray()
for y in range(H):
raw.append(0)
base = y * W
for x in range(W):
r, g, b = pixels[base + x]
raw += bytes((r, g, b, 255))
def _chunk(tag, data):
c = tag + data
return struct.pack(">I", len(data)) + c + struct.pack(">I", zlib.crc32(c) & 0xffffffff)
return (b"\x89PNG\r\n\x1a\n"
+ _chunk(b"IHDR", struct.pack(">IIBBBBB", W, H, 8, 6, 0, 0, 0))
+ _chunk(b"IDAT", zlib.compress(bytes(raw), 6))
+ _chunk(b"IEND", b""))
# The device briefly locks an asset while a draw reads it; re-uploading the same
# name too soon returns HTTP 508, so rotate through a few filenames.
_RING = 4
_frame_no = 0
def _post(host, path, data, content_type):
req = urllib.request.Request(_base(host) + path, data=data, method="POST",
headers={"Content-Type": content_type})
with urllib.request.urlopen(req, timeout=5) as r:
return r.getcode()
def show(host, pixels):
"""Upload the frame PNG and draw it as one image. Returns the draw status."""
global _frame_no
fn = "frame%d.png" % (_frame_no % _RING)
_frame_no += 1
try:
_post(host, "/api/assets/upload?application_name=%s&file=%s" % (APP, fn),
_png(pixels), "application/octet-stream")
body = {"application_name": APP,
"elements": [{"id": "frame", "type": "image", "path": fn, "x": 0, "y": 0}]}
return _post(host, "/api/display/draw", json.dumps(body).encode(), "application/json")
except urllib.error.HTTPError as e:
return e.code
except urllib.error.URLError as e:
raise RuntimeError(f"push failed: {e}") from e
def _clear(host):
qs = urllib.parse.urlencode({"application_name": APP})
req = urllib.request.Request(_base(host) + "/api/display/draw?" + qs, method="DELETE")
try:
with urllib.request.urlopen(req, timeout=5) as r:
return r.getcode()
except urllib.error.HTTPError as e:
return e.code
except urllib.error.URLError as e:
raise RuntimeError(f"clear failed: {e}") from e
# ---------------------------------------------------------------------------
# Spectrum: Goertzel algorithm (pure Python, no numpy)
# ---------------------------------------------------------------------------
def goertzel_magnitude(samples, freq, sample_rate):
"""Compute the magnitude of a single frequency bin using the Goertzel algorithm."""
n = len(samples)
k = freq * n / sample_rate
omega = 2.0 * math.pi * k / n
cos_w = math.cos(omega)
coeff = 2.0 * cos_w
q1 = 0.0
q2 = 0.0
for s in samples:
q0 = coeff * q1 - q2 + s
q2 = q1
q1 = q0
real = q1 - q2 * math.cos(omega)
imag = q2 * math.sin(omega)
return math.sqrt(real * real + imag * imag)
def compute_band_magnitudes(samples):
"""Return list of 24 raw magnitudes, one per band."""
return [goertzel_magnitude(samples, f, SAMPLE_RATE) for f in BAND_FREQS]
# ---------------------------------------------------------------------------
# Colour / palette
# ---------------------------------------------------------------------------
def _hsv(h, s, v):
"""HSV (all 0..1) -> (r, g, b) in 0..1."""
i = int(h * 6.0) % 6
f = h * 6.0 - int(h * 6.0)
p, q, t = v * (1 - s), v * (1 - f * s), v * (1 - (1 - f) * s)
return [(v, t, p), (q, v, p), (p, v, t), (p, q, v), (t, p, v), (v, p, q)][i]
def _sample(stops, t):
"""Linearly interpolate a list of (pos, (r, g, b)) gradient stops at t in 0..1."""
t = max(0.0, min(1.0, t))
prev = stops[0]
for stop in stops:
if t <= stop[0]:
(p0, c0), (p1, c1) = prev, stop
span = (p1 - p0) or 1.0
k = (t - p0) / span
return tuple(c0[j] + (c1[j] - c0[j]) * k for j in range(3))
prev = stop
return stops[-1][1]
def _theme_rgb(band_i, t, theme, num_bands):
"""Single (r, g, b) in 0..1 for a band at absolute height fraction t in 0..1."""
t = max(0.0, min(1.0, t))
if theme == "rainbow":
return _hsv(band_i / max(1, num_bands), 1.0, 0.35 + 0.65 * t)
return _sample(THEMES.get(theme, THEMES["classic"]), t)
def _c(band_i, t, theme, n):
"""Theme colour as an (r, g, b) 0..255 tuple."""
r, g, b = _theme_rgb(band_i, t, theme, n)
return (max(0, min(255, int(r * 255 + 0.5))),
max(0, min(255, int(g * 255 + 0.5))),
max(0, min(255, int(b * 255 + 0.5))))
# ---------------------------------------------------------------------------
# Pixel buffer + per-style rasterisers (write pixels, not rect elements)
# ---------------------------------------------------------------------------
def _blank():
return [(0, 0, 0)] * (W * H)
def _px(buf, x, y, rgb):
if 0 <= x < W and 0 <= y < H:
buf[y * W + x] = rgb
def _col2(buf, x, y, rgb):
_px(buf, x, y, rgb)
_px(buf, x + 1, y, rgb)
def _raster_bars(buf, heights, peaks, theme):
"""Vertical gradient bars anchored at the bottom, floating peak-hold caps."""
n = len(heights)
for i, h in enumerate(heights):
x = i * 3
for y in range(H - h, H):
_col2(buf, x, y, _c(i, (H - 1 - y) / (H - 1), theme, n))
ph = int(peaks[i])
if ph > h and ph > 0:
_col2(buf, x, H - ph, PEAK_COLOR)
def _raster_mirror(buf, heights, peaks, theme):
"""Bars grow symmetrically from the horizontal centre, up and down."""
n = len(heights)
mid = H // 2
for i, h in enumerate(heights):
x = i * 3
half = h // 2
for y in range(mid - half, mid):
t = (mid - y) / max(1, half) * (h / H)
_col2(buf, x, y, _c(i, t, theme, n))
for y in range(mid, mid + half):
t = (y - mid + 1) / max(1, half) * (h / H)
_col2(buf, x, y, _c(i, t, theme, n))
phalf = int(peaks[i]) // 2
if phalf > half and phalf > 0:
_col2(buf, x, mid - phalf, PEAK_COLOR)
_col2(buf, x, mid + phalf - 1, PEAK_COLOR)
# LED-block geometry for the segmented style: 2 px lit block, 1 px dark gap.
SEG_BLOCK, SEG_GAP = 2, 1
SEG_PITCH = SEG_BLOCK + SEG_GAP
SEG_SLOTS = 6
def _seg_block(k):
"""(y, height) of block slot k (0 = bottom), clamped into the display."""
y = H - SEG_BLOCK - k * SEG_PITCH
bh = SEG_BLOCK
if y < 0:
bh += y
y = 0
return y, bh
def _raster_segments(buf, heights, peaks, theme):
"""Discrete stacked LED blocks per band (classic hardware VU meter)."""
n = len(heights)
for i, h in enumerate(heights):
x = i * 3
lit = int(round(h / H * SEG_SLOTS))
for k in range(lit):
y, bh = _seg_block(k)
if bh <= 0:
continue
t = 1.0 - (y + bh / 2.0) / H
col = _c(i, t, theme, n)
for yy in range(y, y + bh):
_col2(buf, x, yy, col)
peak_slot = int(round(peaks[i] / H * SEG_SLOTS))
if peak_slot > lit and peak_slot > 0:
y, bh = _seg_block(peak_slot - 1)
for yy in range(y, y + bh):
_col2(buf, x, yy, PEAK_COLOR)
def _raster_dots(buf, heights, peaks, theme):
"""Only a bouncing dot at each band top, with a falling peak-hold dot."""
n = len(heights)
for i, h in enumerate(heights):
x = i * 3
if h > 0:
dh = 2
y0 = min(H - dh, H - h)
col = _c(i, h / H, theme, n)
for yy in range(y0, y0 + dh):
_col2(buf, x, yy, col)
ph = int(peaks[i])
if ph > h and ph > 0:
_col2(buf, x, H - ph, PEAK_COLOR)
def _raster_wave(buf, heights, peaks, theme):
"""A continuous contour line connecting the band tops (oscilloscope-style)."""
n = len(heights)
tops = [H - max(1, h) for h in heights]
for i, h in enumerate(heights):
x = i * 3
col = _c(i, max(1, h) / H, theme, n)
_col2(buf, x, tops[i], col)
if i < n - 1:
lo, hi = min(tops[i], tops[i + 1]), max(tops[i], tops[i + 1])
for yy in range(lo, hi + 1):
_px(buf, x + 2, yy, col)
_RASTERISERS = {
"bars": _raster_bars,
"mirror": _raster_mirror,
"segments": _raster_segments,
"dots": _raster_dots,
"wave": _raster_wave,
}
def build_pixels(heights, peaks, theme, style="bars"):
"""Rasterise the chosen style into a flat 72x16 pixel buffer."""
buf = _blank()
_RASTERISERS.get(style, _raster_bars)(buf, heights, peaks, theme)
return buf
def magnitudes_to_heights(magnitudes, running_max):
"""Convert raw magnitudes to pixel heights 0..16, updating running_max in place."""
heights = []
for mag in magnitudes:
running_max[0] = max(running_max[0] * 0.995, 1.0)
if mag > running_max[0]:
running_max[0] = mag
h = int(16.0 * math.log1p(mag) / math.log1p(running_max[0]))
heights.append(max(0, min(16, h)))
return heights
def smooth_heights(new_heights, old_heights, decay=0.75):
"""Attack/decay smoothing: bars fall at decay rate per frame."""
return [max(n, int(o * decay)) for n, o in zip(new_heights, old_heights)]
# ---------------------------------------------------------------------------
# argparse
# ---------------------------------------------------------------------------
def parse_args():
p = argparse.ArgumentParser(description="Audio visualizer for BUSY Bar")
p.add_argument("--host", default="10.0.4.20")
p.add_argument("--fps", type=int, default=15, help="draw updates per second (default: 15)")
p.add_argument("--style", default="bars", choices=STYLE_NAMES,
help="render style / shape (default: bars)")
p.add_argument("--theme", default="classic", choices=THEME_NAMES,
help="colour theme (default: classic)")
p.add_argument("--device", default="auto", help="ffmpeg avfoundation audio input, e.g. :1 (default: auto)")
p.add_argument("--demo", action="store_true",
help="run a synthetic-audio demo (no mic/ffmpeg) that cycles the styles and themes")
p.add_argument("--test", action="store_true")
return p.parse_args()
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def _test_mode(host, theme, style):
"""Draw one synthetic frame and exit."""
heights = [3, 5, 8, 12, 16, 14, 10, 7, 9, 13, 15, 11, 8, 6, 4, 7, 10, 14, 12, 9, 6, 4, 3, 2]
peaks = [min(H, h + 3) for h in heights]
status = show(host, build_pixels(heights, peaks, theme, style))
print(f"test: drew 1 frame ({style}/{theme}) status {status}")
sys.exit(0)
# One (style, theme) pair per style so a viewer sees every look once.
DEMO_COMBOS = [
("bars", "fire"),
("mirror", "aurora"),
("segments", "classic"),
("dots", "ocean"),
("wave", "rainbow"),
]
DEMO_SECONDS_PER_COMBO = 1.6
# ~100 BPM groove; the whole cycle equals len(DEMO_COMBOS) * DEMO_SECONDS_PER_COMBO.
_DEMO_BEAT_HZ = 100.0 / 60.0
def _synth_heights(t):
"""Music-like spectrum for band 0..23 at time t seconds (no audio needed)."""
phase = (t * _DEMO_BEAT_HZ) % 1.0
kick = math.exp(-6.0 * phase)
off = (phase + 0.5) % 1.0
snare = math.exp(-9.0 * off)
swell = 0.85 + 0.15 * math.sin(t * 0.5)
mel_center = 6.0 + 8.0 * (0.5 + 0.5 * math.sin(t * 0.8))
heights = []
for i in range(NUM_BANDS):
if i <= 5:
wob = 0.5 + 0.5 * math.sin(t * 2.0 + i * 0.7)
val = (0.55 + 0.45 * wob) * (0.30 + 0.80 * kick)
elif i <= 14:
mel = math.exp(-((i - mel_center) ** 2) / 6.0)
groove = 0.5 + 0.5 * math.sin(t * 3.3 + i * 0.5)
val = 0.20 + 0.75 * mel * (0.4 + 0.6 * groove) + 0.20 * snare
else:
shimmer = 0.5 + 0.5 * math.sin(t * 11.0 + i * 1.7)
hat = math.exp(-12.0 * off)
val = (0.10 + 0.35 * shimmer) * (0.5 + 0.9 * hat)
val *= swell
heights.append(max(0, min(H, int(round(val * H)))))
return heights
def _demo_mode(host, fps):
"""Drive the real render pipeline with a synthetic spectrum, cycling looks."""
interval = 1.0 / max(1, fps)
peak_fall = max(0.3, PEAK_FALL_PER_SEC / max(1, fps))
smooth = [0] * NUM_BANDS
peaks = [0.0] * NUM_BANDS
t0 = time.monotonic()
print(f"audio-visualizer demo -> {_base(host)} (Ctrl-C to stop)")
try:
while True:
t = time.monotonic() - t0
style, theme = DEMO_COMBOS[int(t / DEMO_SECONDS_PER_COMBO) % len(DEMO_COMBOS)]
new_h = _synth_heights(t)
smooth = smooth_heights(new_h, smooth)
for i in range(NUM_BANDS):
peaks[i] = max(float(smooth[i]), peaks[i] - peak_fall)
try:
show(host, build_pixels(smooth, peaks, theme, style))
except RuntimeError:
pass
time.sleep(interval)
except KeyboardInterrupt:
pass
finally:
try:
_clear(host)
except Exception:
pass
print("stopped.")
def pick_device():
"""List avfoundation audio devices and pick the built-in microphone:
prefer a name containing MacBook, else the first non-iPhone device."""
r = subprocess.run(["ffmpeg", "-hide_banner", "-f", "avfoundation",
"-list_devices", "true", "-i", ""],
capture_output=True, text=True, timeout=10)
devices = []
in_audio = False
for line in r.stderr.splitlines():
if "audio devices" in line.lower():
in_audio = True
continue
if in_audio:
m = __import__("re").search(r"\[(\d+)\] (.+)$", line)
if m:
devices.append((m.group(1), m.group(2).strip()))
elif "]" not in line:
break
if not devices:
return ":0"
for idx, name in devices:
if "macbook" in name.lower():
print(f"microfoon: [{idx}] {name}")
return f":{idx}"
for idx, name in devices:
if "iphone" not in name.lower():
print(f"microfoon: [{idx}] {name}")
return f":{idx}"
return f":{devices[0][0]}"
def main():
args = parse_args()
if args.test:
_test_mode(args.host, args.theme, args.style)
return
if args.demo:
_demo_mode(args.host, args.fps)
return
if shutil.which("ffmpeg") is None:
print("ffmpeg not found. Install it with: brew install ffmpeg")
sys.exit(1)
frame_interval = 1.0 / max(1, args.fps)
peak_fall = max(0.3, PEAK_FALL_PER_SEC / max(1, args.fps))
running_max = [1.0]
smooth = [0] * NUM_BANDS
peaks = [0.0] * NUM_BANDS
last_draw = 0.0
last_chunk = None
ffmpeg_proc = None
_draw_error_printed = False
if args.device == "auto":
try:
args.device = pick_device()
except Exception as e:
print(f"device detection failed ({e}), using :0")
args.device = ":0"
cmd = [
"ffmpeg",
"-hide_banner", "-loglevel", "error",
"-f", "avfoundation",
"-i", args.device,
"-ac", "1",
"-ar", str(SAMPLE_RATE),
"-f", "s16le",
"-",
]
print(f"audio-visualizer -> {_base(args.host)} (Ctrl-C to stop)")
try:
ffmpeg_proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
while True:
chunk_bytes = ffmpeg_proc.stdout.read(CHUNK_SAMPLES * 2)
if chunk_bytes and len(chunk_bytes) == CHUNK_SAMPLES * 2:
last_chunk = chunk_bytes
_hb_count = globals().setdefault("_hb", [0])
_hb_count[0] += 1
if _hb_count[0] % 22 == 0:
samples = struct.unpack(f"<{CHUNK_SAMPLES}h", chunk_bytes)
rms = (sum(v * v for v in samples) / CHUNK_SAMPLES) ** 0.5
note = "" if rms >= 20 else " (stilte: check microfoontoestemming)"
print(f"audio niveau: {rms:7.1f}{note}")
elif ffmpeg_proc.poll() is not None:
err = ffmpeg_proc.stderr.read().decode("utf-8", "ignore").strip()
print("ffmpeg stopped unexpectedly:")
for line in err.splitlines()[-5:]:
print(f" {line}")
sys.exit(1)
now = time.monotonic()
if now - last_draw >= frame_interval:
last_draw = now
if last_chunk is not None:
samples = list(struct.unpack(f"<{CHUNK_SAMPLES}h", last_chunk))
mags = compute_band_magnitudes(samples)
new_h = magnitudes_to_heights(mags, running_max)
smooth = smooth_heights(new_h, smooth)
for i in range(NUM_BANDS):
peaks[i] = max(float(smooth[i]), peaks[i] - peak_fall)
try:
status = show(args.host, build_pixels(smooth, peaks, args.theme, args.style))
if status == 409:
if not _draw_error_printed:
print("display busy (409), retrying")
_draw_error_printed = True
else:
_draw_error_printed = False
except RuntimeError as e:
if not _draw_error_printed:
print(f"draw error: {e}")
_draw_error_printed = True
except KeyboardInterrupt:
pass
finally:
if ffmpeg_proc is not None:
ffmpeg_proc.terminate()
try:
ffmpeg_proc.wait(timeout=3)
except subprocess.TimeoutExpired:
ffmpeg_proc.kill()
try:
_clear(args.host)
except Exception:
pass
print("stopped.")
if __name__ == "__main__":
main()