engineering/zebra-report
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

add zebrad: pulseaudio→websocket introspector for chat.html

Browse source 
single C file (~620 LOC), no third-party deps beyond libpulse.
closes a receive loop for web/chat.html: a browser tab cannot read
another tab's PulseAudio state, so each peer runs zebrad locally
to forward decoded frames to ws://127.0.0.1:7777.

components inline:
  * SHA-1 (RFC 3174) + base64 encode — for the WebSocket handshake;
    verified against the canonical RFC 6455 example
    (key dGhlIHNhbXBsZSBub25jZQ== → accept s3pPLMBiTxaQ9kYGzzhZRbK+xOo=)
  * CRC-32 (IEEE 802.3) — verifies DATA + HELLO frame payloads
  * UART symbol decoder with adaptive threshold:
    running peak with 2-second half-life decay, threshold = 50% of peak,
    falling-edge start detect, sample at 1.5..8.5 × sps from edge
  * frame assembler: magic-sync ZB, type dispatch OFFER/READY/DATA/HELLO,
    overflow + corrupt-length guards, resync on bad checksum
  * embedded WebSocket server: TCP listener on 127.0.0.1:7777,
    handshake responder, binary frames only (opcode 0x82), no masking
    (server→client), any inbound data closes the client (browser
    auto-reconnects); max 16 concurrent clients
  * pulseaudio: pa_mainloop integration, record stream on @DEFAULT_MONITOR@
    by default, PA_SAMPLE_FLOAT32LE mono @ 8 kHz, 10 ms fragsize, peak
    energy extracted from |abs(sample)|

adaptive baud: starts at ZEBRA_BAUD_HANDSHAKE (50). On valid READY frame
locks in negotiated baud from payload. Matches the chat.html send path.

binds 127.0.0.1 only — never reachable from the LAN, even though the
chat.html page itself may be served over HTTPS from a public domain.

smoke tested: PA connect + stream ready + WS handshake (RFC example
verified) + clean shutdown. -Wall -Wextra clean.

Makefile: zebrad now part of `make all`; clean target updated.
README: new section documenting it.
This commit is contained in:
Russell Ballestrini 2026-05-27 15:57:22 -04:00
parent 5d76d713a6
commit c8a755513f
3 changed files with 719 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

9
Makefile
View file

@ -4,9 +4,9 @@ LDFLAGS = $(shell pkg-config --libs libpulse) -lrt -lpthread
PULSE = src/pulse.c PULSE = src/pulse.c
.PHONY: all clean serve blog test test-all .PHONY: all clean serve blog test test-all zebrad
all: tx rx chat bt carrier all: tx rx chat bt carrier zebrad
test: test/unit test: test/unit
@./test/unit @./test/unit
@ -43,6 +43,9 @@ bt: src/bt.c $(PULSE)
carrier: src/carrier.c carrier: src/carrier.c
$(CC) $(CFLAGS) -o $@ $^ $(LDFLAGS) $(CC) $(CFLAGS) -o $@ $^ $(LDFLAGS)
zebrad: src/zebrad.c include/zebra.h
$(CC) $(CFLAGS) -o $@ src/zebrad.c $(LDFLAGS) -lm
blog: blog:
python3 blog/build.py python3 blog/build.py
@ -50,5 +53,5 @@ serve: blog
cd web && python3 -m http.server 8765 cd web && python3 -m http.server 8765
clean: clean:
rm -f tx rx chat bt carrier test/unit test/integration test/functional rm -f tx rx chat bt carrier zebrad test/unit test/integration test/functional
rm -rf web/blog/001-volume-modem web/blog/002-sse-chatroom web/blog/index.html rm -rf web/blog/001-volume-modem web/blog/002-sse-chatroom web/blog/index.html

25
README.md
View file

@ -41,7 +41,7 @@ early with a clear message.
## Targets ## Targets
### `make all` — five binaries ### `make all` — six binaries
| Binary | Source | Role | | Binary | Source | Role |
|-----------|----------------|------| |-----------|----------------|------|
@ -50,6 +50,7 @@ early with a clear message.
| `chat` | `src/chat.c` | bidirectional tx+rx, line-based chat UI | | `chat` | `src/chat.c` | bidirectional tx+rx, line-based chat UI |
| `bt` | `src/bt.c` | "battle toads" dual-channel stereo UART (2× throughput) | | `bt` | `src/bt.c` | "battle toads" dual-channel stereo UART (2× throughput) |
| `carrier` | `src/carrier.c`| publishes a silent PA sink so volume reads have something to read | | `carrier` | `src/carrier.c`| publishes a silent PA sink so volume reads have something to read |
| `zebrad` | `src/zebrad.c` | PA→WS introspector for `web/chat.html` |
All link against `libpulse`, `librt`, and `libpthread`. Headers come from All link against `libpulse`, `librt`, and `libpthread`. Headers come from
`include/zebra.h` (protocol constants) and `include/modem.h` (inline encode/ `include/zebra.h` (protocol constants) and `include/modem.h` (inline encode/
@ -107,6 +108,28 @@ the blog and serves `web/` on port 8765 for local preview.
The chat UI (`web/chat.html`) is plain static HTML and works directly under The chat UI (`web/chat.html`) is plain static HTML and works directly under
`make serve` — open `http://127.0.0.1:8765/chat.html`. `make serve` — open `http://127.0.0.1:8765/chat.html`.
### `make zebrad` — PulseAudio → WebSocket introspector
`web/chat.html` modulates audio output via Web Audio `GainNode` (mic stays off).
A browser tab cannot read another tab's PA state, so to close the receive
loop each peer runs `zebrad`. Decoded frames are forwarded over a local
WebSocket; `chat.html` auto-connects to `ws://127.0.0.1:7777`.
```bash
make zebrad
./zebrad --verbose # default: @DEFAULT_MONITOR@, port 7777
./zebrad --source <name> --port 7777
```
Single C file, no third-party deps beyond `libpulse`. Embedded WebSocket
server: SHA-1 + base64 inline, server→client binary frames only (RFC 6455
opcode 0x82); any inbound data closes the socket (browser auto-reconnects).
Bound to `127.0.0.1` only — never accessible from the LAN.
Adaptive baud: starts at `ZEBRA_BAUD_HANDSHAKE` (50), watches for a `READY`
frame, locks in the negotiated rate from its payload. Peak tracker has
2-second half-life decay so threshold adapts to room volume.
### `make clean` ### `make clean`
Removes: Removes:

689
src/zebrad.c Normal file
View file

@ -0,0 +1,689 @@
/*
* zebrad — zebra-report introspector daemon.
*
* Listens to a PulseAudio monitor source for volume-modulated chat frames
* emitted by web/chat.html (Web Audio GainNode swings MARK/SPACE), decodes
* per the protocol in include/zebra.h plus the DATA/HELLO extensions, and
* broadcasts decoded frames to local WebSocket clients on 127.0.0.1:7777.
*
* Each chat.html tab connects via ws://127.0.0.1:7777 and receives every
* decoded frame as a raw binary WebSocket message.
*
* Single file. No third-party deps beyond libpulse (already linked by the
* rest of the codebase). Minimal embedded WebSocket server: SHA-1 + base64
* implemented inline; only opcode 0x82 (binary, FIN) ever transmitted; any
* client frame closes the socket (browser auto-reconnects).
*
* usage:
* ./zebrad # @DEFAULT_MONITOR@, port 7777
* ./zebrad --verbose
* ./zebrad --source alsa_output.X.monitor --port 7777
*/
#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#include <signal.h>
#include <getopt.h>
#include <ctype.h>
#include <math.h>
#include <sys/socket.h>
#include <sys/uio.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <pulse/pulseaudio.h>
#include <pulse/error.h>
#include "zebra.h"
/* --- protocol extensions not in zebra.h --- */
#define T_OFFER ZEBRA_HS_TYPE_OFFER
#define T_READY ZEBRA_HS_TYPE_READY
#define T_DATA 0x03
#define T_HELLO 0x04
/* --- defaults --- */
#define DEFAULT_PORT 7777
#define DEFAULT_RATE 8000
#define DEFAULT_SOURCE "@DEFAULT_MONITOR@"
#define MAX_CLIENTS 16
#define WS_RECV_BUF 4096
#define FRAME_BUF_MAX 1024
#define ENERGY_NO_CARRIER 0.02
#define THRESHOLD_RATIO 0.50
#define PEAK_DECAY_HALF_S 2.0 /* peak-tracker half-life, seconds */
static int verbose = 0;
static const char *opt_source = DEFAULT_SOURCE;
static int opt_port = DEFAULT_PORT;
static int opt_rate = DEFAULT_RATE;
static int opt_baud = ZEBRA_BAUD_HANDSHAKE;
/* ============================================================ */
/* SHA-1 — RFC 3174 */
/* ============================================================ */
typedef struct { uint32_t h[5]; uint64_t bits; uint8_t buf[64]; int blen; } sha1_t;
static uint32_t rol(uint32_t x, int n) { return (x << n) | (x >> (32 - n)); }
static void sha1_block(sha1_t *s, const uint8_t *b) {
uint32_t w[80];
for (int i = 0; i < 16; i++)
w[i] = ((uint32_t)b[i*4] << 24) | ((uint32_t)b[i*4+1] << 16) |
((uint32_t)b[i*4+2] << 8) | (uint32_t)b[i*4+3];
for (int i = 16; i < 80; i++)
w[i] = rol(w[i-3] ^ w[i-8] ^ w[i-14] ^ w[i-16], 1);
uint32_t a=s->h[0], bb=s->h[1], c=s->h[2], d=s->h[3], e=s->h[4];
for (int i = 0; i < 80; i++) {
uint32_t f, k;
if (i < 20) { f = (bb & c) | ((~bb) & d); k = 0x5A827999; }
else if (i < 40) { f = bb ^ c ^ d; k = 0x6ED9EBA1; }
else if (i < 60) { f = (bb & c) | (bb & d) | (c & d); k = 0x8F1BBCDC; }
else { f = bb ^ c ^ d; k = 0xCA62C1D6; }
uint32_t t = rol(a, 5) + f + e + k + w[i];
e = d; d = c; c = rol(bb, 30); bb = a; a = t;
}
s->h[0]+=a; s->h[1]+=bb; s->h[2]+=c; s->h[3]+=d; s->h[4]+=e;
}
static void sha1_init(sha1_t *s) {
s->h[0]=0x67452301; s->h[1]=0xEFCDAB89; s->h[2]=0x98BADCFE;
s->h[3]=0x10325476; s->h[4]=0xC3D2E1F0;
s->bits = 0; s->blen = 0;
}
static void sha1_update(sha1_t *s, const uint8_t *data, size_t len) {
s->bits += (uint64_t)len * 8;
while (len) {
int take = 64 - s->blen;
if ((size_t)take > len) take = len;
memcpy(s->buf + s->blen, data, take);
s->blen += take; data += take; len -= take;
if (s->blen == 64) { sha1_block(s, s->buf); s->blen = 0; }
}
}
static void sha1_final(sha1_t *s, uint8_t out[20]) {
s->buf[s->blen++] = 0x80;
if (s->blen > 56) {
memset(s->buf + s->blen, 0, 64 - s->blen);
sha1_block(s, s->buf); s->blen = 0;
}
memset(s->buf + s->blen, 0, 56 - s->blen);
for (int i = 0; i < 8; i++) s->buf[56+i] = (s->bits >> (56 - i*8)) & 0xFF;
sha1_block(s, s->buf);
for (int i = 0; i < 5; i++) {
out[i*4+0] = (s->h[i] >> 24) & 0xFF; out[i*4+1] = (s->h[i] >> 16) & 0xFF;
out[i*4+2] = (s->h[i] >> 8) & 0xFF; out[i*4+3] = s->h[i] & 0xFF;
}
}
/* ============================================================ */
/* base64 encode */
/* ============================================================ */
static const char B64[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
static int base64_encode(const uint8_t *in, size_t inlen, char *out, size_t outlen) {
size_t need = ((inlen + 2) / 3) * 4 + 1;
if (outlen < need) return -1;
size_t i = 0, o = 0;
while (i + 3 <= inlen) {
out[o++] = B64[(in[i] >> 2) & 0x3F];
out[o++] = B64[((in[i] & 0x03) << 4) | (in[i+1] >> 4)];
out[o++] = B64[((in[i+1] & 0x0F) << 2) | (in[i+2] >> 6)];
out[o++] = B64[in[i+2] & 0x3F];
i += 3;
}
if (i < inlen) {
out[o++] = B64[(in[i] >> 2) & 0x3F];
if (i + 1 == inlen) {
out[o++] = B64[(in[i] & 0x03) << 4];
out[o++] = '='; out[o++] = '=';
} else {
out[o++] = B64[((in[i] & 0x03) << 4) | (in[i+1] >> 4)];
out[o++] = B64[(in[i+1] & 0x0F) << 2]; out[o++] = '=';
}
}
out[o] = '\0';
return (int)o;
}
/* ============================================================ */
/* CRC-32 (IEEE 802.3) */
/* ============================================================ */
static uint32_t crc32_tbl[256];
static void crc32_init(void) {
for (int i = 0; i < 256; i++) {
uint32_t c = i;
for (int j = 0; j < 8; j++) c = (c & 1) ? (0xEDB88320 ^ (c >> 1)) : (c >> 1);
crc32_tbl[i] = c;
}
}
static uint32_t crc32_calc(const uint8_t *data, size_t n) {
uint32_t c = 0xFFFFFFFF;
for (size_t i = 0; i < n; i++) c = crc32_tbl[(c ^ data[i]) & 0xFF] ^ (c >> 8);
return c ^ 0xFFFFFFFF;
}
static uint8_t xor_checksum(const uint8_t *data, size_t n) {
uint8_t x = 0; for (size_t i = 0; i < n; i++) x ^= data[i]; return x;
}
/* ============================================================ */
/* UART decoder */
/* ============================================================ */
typedef enum { U_HUNT, U_SAMPLE } uart_state_t;
typedef struct {
int sample_rate;
int baud;
double sps; /* samples per symbol */
uart_state_t state;
uint8_t byte;
int bit_idx;
int cur_sample;
int prev_above;
double peak;
double peak_decay; /* multiplier per sample */
} uart_t;
static void uart_set_baud(uart_t *u, int baud) {
u->baud = baud;
u->sps = (double)u->sample_rate / (double)baud;
}
static void uart_init(uart_t *u, int rate, int baud) {
u->sample_rate = rate;
uart_set_baud(u, baud);
u->state = U_HUNT;
u->byte = 0; u->bit_idx = 0; u->cur_sample = 0;
u->prev_above = 1;
u->peak = 0;
/* exponential decay: half-life PEAK_DECAY_HALF_S */
u->peak_decay = pow(0.5, 1.0 / (PEAK_DECAY_HALF_S * (double)rate));
}
/* Returns 1 if a byte decoded (in *out), 0 otherwise. */
static int uart_push(uart_t *u, double energy, uint8_t *out) {
u->peak = u->peak * u->peak_decay;
if (energy > u->peak) u->peak = energy;
if (u->peak < ENERGY_NO_CARRIER) {
u->state = U_HUNT; u->prev_above = 1;
return 0;
}
double thr = u->peak * THRESHOLD_RATIO;
int above = (energy >= thr);
if (u->state == U_HUNT) {
if (u->prev_above && !above) {
u->state = U_SAMPLE;
u->cur_sample = 1;
u->byte = 0; u->bit_idx = 0;
}
} else {
u->cur_sample++;
double center = (1.5 + (double)u->bit_idx) * u->sps;
if ((double)u->cur_sample >= center) {
uint8_t bit = above ? 1 : 0;
u->byte |= (bit << u->bit_idx);
u->bit_idx++;
if (u->bit_idx == 8) {
*out = u->byte;
u->state = U_HUNT; u->prev_above = 1;
return 1;
}
}
}
u->prev_above = above;
return 0;
}
/* ============================================================ */
/* frame assembler */
/* ============================================================ */
typedef struct {
uint8_t buf[FRAME_BUF_MAX];
size_t len;
} frame_t;
static void frame_init(frame_t *f) { f->len = 0; }
static void frame_drop_until_magic(frame_t *f) {
size_t i = 0;
while (i + 1 < f->len &&
!(f->buf[i] == ZEBRA_HS_MAGIC_0 && f->buf[i+1] == ZEBRA_HS_MAGIC_1)) {
i++;
}
if (i == 0) return;
memmove(f->buf, f->buf + i, f->len - i);
f->len -= i;
}
static void frame_consume(frame_t *f, size_t n) {
memmove(f->buf, f->buf + n, f->len - n);
f->len -= n;
}
/* Returns: frame length if complete (copied into *out, max FRAME_BUF_MAX),
* 0 if needs more bytes, -1 on corrupt (rewound) input. */
static int frame_push(frame_t *f, uint8_t byte, uint8_t *out) {
if (f->len >= FRAME_BUF_MAX) {
/* overflow guard: drop everything */
f->len = 0;
}
f->buf[f->len++] = byte;
frame_drop_until_magic(f);
if (f->len < 3) return 0;
uint8_t t = f->buf[2];
if (t == T_OFFER || t == T_READY) {
if (f->len < ZEBRA_HS_FRAME_LEN) return 0;
if (xor_checksum(f->buf, 5) != f->buf[5]) {
frame_consume(f, 2); return -1;
}
memcpy(out, f->buf, ZEBRA_HS_FRAME_LEN);
int n = ZEBRA_HS_FRAME_LEN;
frame_consume(f, n);
return n;
}
if (t == T_DATA) {
if (f->len < 9) return 0;
uint16_t ln = f->buf[7] | (f->buf[8] << 8);
if (ln > 512) { frame_consume(f, 2); return -1; }
size_t need = 9 + ln + 4;
if (f->len < need) return 0;
uint32_t got = (uint32_t)f->buf[9+ln]
| ((uint32_t)f->buf[9+ln+1] << 8)
| ((uint32_t)f->buf[9+ln+2] << 16)
| ((uint32_t)f->buf[9+ln+3] << 24);
uint32_t calc = crc32_calc(f->buf, 9 + ln);
if (got != calc) { frame_consume(f, 2); return -1; }
memcpy(out, f->buf, need);
frame_consume(f, need);
return (int)need;
}
if (t == T_HELLO) {
if (f->len < 10) return 0;
uint8_t hlen = f->buf[9];
if (hlen > 64) { frame_consume(f, 2); return -1; }
size_t need = 10 + hlen + 4;
if (f->len < need) return 0;
uint32_t got = (uint32_t)f->buf[10+hlen]
| ((uint32_t)f->buf[10+hlen+1] << 8)
| ((uint32_t)f->buf[10+hlen+2] << 16)
| ((uint32_t)f->buf[10+hlen+3] << 24);
uint32_t calc = crc32_calc(f->buf, 10 + hlen);
if (got != calc) { frame_consume(f, 2); return -1; }
memcpy(out, f->buf, need);
frame_consume(f, need);
return (int)need;
}
/* unknown type */
frame_consume(f, 2);
return -1;
}
/* ============================================================ */
/* WebSocket: handshake + binary frame send */
/* ============================================================ */
typedef enum { C_HANDSHAKE, C_OPEN } client_state_t;
typedef struct {
int fd;
client_state_t state;
char rbuf[WS_RECV_BUF];
size_t rlen;
pa_io_event *io;
} client_t;
static client_t clients[MAX_CLIENTS];
static int listener_fd = -1;
static pa_io_event *listener_io = NULL;
static pa_mainloop *mloop = NULL;
static pa_mainloop_api *mapi = NULL;
static pa_context *pa_ctx = NULL;
static pa_stream *pa_str = NULL;
static uart_t uart;
static frame_t fasm;
static void client_close(client_t *c) {
if (c->fd < 0) return;
if (c->io) mapi->io_free(c->io);
close(c->fd);
c->fd = -1; c->io = NULL; c->state = C_HANDSHAKE; c->rlen = 0;
if (verbose) fprintf(stderr, "zebrad: client closed\n");
}
static int set_nonblock(int fd) {
int fl = fcntl(fd, F_GETFL, 0);
return fcntl(fd, F_SETFL, fl | O_NONBLOCK);
}
/* find Sec-WebSocket-Key header value */
static const char *find_header(const char *buf, const char *name, size_t *vlen) {
size_t nlen = strlen(name);
const char *p = buf;
while ((p = strstr(p, "\r\n")) != NULL) {
p += 2;
if (strncasecmp(p, name, nlen) == 0 && p[nlen] == ':') {
p += nlen + 1;
while (*p == ' ' || *p == '\t') p++;
const char *end = strstr(p, "\r\n");
if (!end) return NULL;
*vlen = end - p;
return p;
}
}
return NULL;
}
#define WS_MAGIC "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
static int ws_handshake_respond(client_t *c, const char *key, size_t klen) {
sha1_t s;
sha1_init(&s);
sha1_update(&s, (const uint8_t *)key, klen);
sha1_update(&s, (const uint8_t *)WS_MAGIC, strlen(WS_MAGIC));
uint8_t digest[20];
sha1_final(&s, digest);
char accept[64];
if (base64_encode(digest, 20, accept, sizeof(accept)) < 0) return -1;
char resp[512];
int n = snprintf(resp, sizeof(resp),
"HTTP/1.1 101 Switching Protocols\r\n"
"Upgrade: websocket\r\n"
"Connection: Upgrade\r\n"
"Sec-WebSocket-Accept: %s\r\n"
"\r\n", accept);
if (n < 0 || n >= (int)sizeof(resp)) return -1;
ssize_t w = send(c->fd, resp, n, MSG_NOSIGNAL);
return (w == n) ? 0 : -1;
}
/* send a single binary WS frame (FIN=1, opcode=0x2, MASK=0) */
static int ws_send_binary(int fd, const uint8_t *payload, size_t len) {
uint8_t hdr[10];
size_t hlen = 0;
hdr[hlen++] = 0x82; /* FIN + binary */
if (len < 126) {
hdr[hlen++] = (uint8_t)len;
} else if (len < 65536) {
hdr[hlen++] = 126;
hdr[hlen++] = (len >> 8) & 0xFF;
hdr[hlen++] = len & 0xFF;
} else {
hdr[hlen++] = 127;
for (int i = 0; i < 8; i++) hdr[hlen++] = (len >> ((7-i)*8)) & 0xFF;
}
struct iovec iov[2] = { { hdr, hlen }, { (void*)payload, len } };
ssize_t want = (ssize_t)(hlen + len);
ssize_t got = writev(fd, iov, 2);
return (got == want) ? 0 : -1;
}
static void ws_broadcast(const uint8_t *frame, size_t len) {
for (int i = 0; i < MAX_CLIENTS; i++) {
if (clients[i].fd >= 0 && clients[i].state == C_OPEN) {
if (ws_send_binary(clients[i].fd, frame, len) < 0) {
client_close(&clients[i]);
}
}
}
}
/* ============================================================ */
/* mainloop io: accept + per-client read (handshake then quiet) */
/* ============================================================ */
static void on_client_read(pa_mainloop_api *a, pa_io_event *e, int fd,
pa_io_event_flags_t flags, void *u) {
(void)a; (void)e;
client_t *c = (client_t *)u;
if (flags & (PA_IO_EVENT_HANGUP | PA_IO_EVENT_ERROR)) {
client_close(c); return;
}
if (c->state == C_HANDSHAKE) {
ssize_t r = recv(fd, c->rbuf + c->rlen, sizeof(c->rbuf) - c->rlen - 1, 0);
if (r <= 0) { client_close(c); return; }
c->rlen += r;
c->rbuf[c->rlen] = '\0';
if (strstr(c->rbuf, "\r\n\r\n")) {
size_t klen = 0;
const char *key = find_header(c->rbuf, "Sec-WebSocket-Key", &klen);
if (!key || klen < 16 || klen > 32) { client_close(c); return; }
if (ws_handshake_respond(c, key, klen) < 0) { client_close(c); return; }
c->state = C_OPEN;
if (verbose) fprintf(stderr, "zebrad: client upgraded\n");
}
} else {
/* C_OPEN: any inbound data → drop client (we are TX-only). */
uint8_t junk[256];
ssize_t r = recv(fd, junk, sizeof(junk), 0);
if (r <= 0) { client_close(c); return; }
/* ignore content; could parse close frame here */
}
}
static void on_accept(pa_mainloop_api *a, pa_io_event *e, int fd,
pa_io_event_flags_t flags, void *u) {
(void)e; (void)u; (void)flags;
int cfd = accept(fd, NULL, NULL);
if (cfd < 0) return;
set_nonblock(cfd);
int one = 1; setsockopt(cfd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));
/* find a free slot */
for (int i = 0; i < MAX_CLIENTS; i++) {
if (clients[i].fd < 0) {
clients[i].fd = cfd; clients[i].state = C_HANDSHAKE; clients[i].rlen = 0;
clients[i].io = a->io_new(a, cfd, PA_IO_EVENT_INPUT, on_client_read, &clients[i]);
return;
}
}
close(cfd);
fprintf(stderr, "zebrad: too many clients, rejected\n");
}
static int ws_listener_start(int port) {
listener_fd = socket(AF_INET, SOCK_STREAM, 0);
if (listener_fd < 0) return -1;
int one = 1;
setsockopt(listener_fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
struct sockaddr_in addr;
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); /* 127.0.0.1 only */
addr.sin_port = htons(port);
if (bind(listener_fd, (struct sockaddr *)&addr, sizeof(addr)) < 0) return -1;
if (listen(listener_fd, 8) < 0) return -1;
set_nonblock(listener_fd);
listener_io = mapi->io_new(mapi, listener_fd, PA_IO_EVENT_INPUT, on_accept, NULL);
return 0;
}
/* ============================================================ */
/* decoded frame handler */
/* ============================================================ */
static void on_decoded_frame(const uint8_t *frame, size_t len) {
if (len < 3) return;
uint8_t t = frame[2];
if (verbose)
fprintf(stderr, "zebrad: frame type=0x%02x len=%zu\n", t, len);
/* adaptive baud: lock in negotiated rate on READY (commit) */
if (t == T_READY && len >= 6) {
unsigned nb = frame[3] | ((unsigned)frame[4] << 8);
if (nb >= ZEBRA_BAUD_MIN && nb <= (unsigned)ZEBRA_BAUD_MAX) {
uart_set_baud(&uart, (int)nb);
if (verbose) fprintf(stderr, "zebrad: baud → %u\n", nb);
}
}
ws_broadcast(frame, len);
}
/* ============================================================ */
/* PulseAudio */
/* ============================================================ */
static void pa_stream_read_cb(pa_stream *s, size_t nbytes, void *u) {
(void)u; (void)nbytes;
const void *data = NULL;
size_t bytes = 0;
if (pa_stream_peek(s, &data, &bytes) < 0) return;
if (bytes == 0) return;
if (!data) { pa_stream_drop(s); return; } /* hole, no audio data */
/* samples are float32 little-endian, mono */
size_t n_samples = bytes / sizeof(float);
const float *fp = (const float *)data;
uint8_t outbuf[FRAME_BUF_MAX];
for (size_t i = 0; i < n_samples; i++) {
double e = fp[i];
if (e < 0) e = -e;
uint8_t byte;
if (uart_push(&uart, e, &byte)) {
int r = frame_push(&fasm, byte, outbuf);
if (r > 0) on_decoded_frame(outbuf, (size_t)r);
}
}
pa_stream_drop(s);
}
static void pa_stream_state_cb(pa_stream *s, void *u) {
(void)u;
switch (pa_stream_get_state(s)) {
case PA_STREAM_READY:
if (verbose) fprintf(stderr, "zebrad: pa stream ready\n");
break;
case PA_STREAM_FAILED:
fprintf(stderr, "zebrad: pa stream failed: %s\n",
pa_strerror(pa_context_errno(pa_ctx)));
mapi->quit(mapi, 1);
break;
default: break;
}
}
static void pa_context_state_cb(pa_context *c, void *u) {
(void)u;
switch (pa_context_get_state(c)) {
case PA_CONTEXT_READY: {
pa_sample_spec ss;
ss.format = PA_SAMPLE_FLOAT32LE;
ss.rate = opt_rate;
ss.channels = 1;
pa_str = pa_stream_new(c, "zebrad-rx", &ss, NULL);
if (!pa_str) {
fprintf(stderr, "zebrad: pa_stream_new failed\n");
mapi->quit(mapi, 1); return;
}
pa_stream_set_state_callback(pa_str, pa_stream_state_cb, NULL);
pa_stream_set_read_callback(pa_str, pa_stream_read_cb, NULL);
pa_buffer_attr ba;
ba.maxlength = (uint32_t)-1;
ba.fragsize = pa_usec_to_bytes(10000, &ss); /* 10ms fragments */
ba.minreq = (uint32_t)-1;
ba.tlength = (uint32_t)-1;
ba.prebuf = (uint32_t)-1;
pa_stream_flags_t fl = PA_STREAM_ADJUST_LATENCY;
if (pa_stream_connect_record(pa_str, opt_source, &ba, fl) < 0) {
fprintf(stderr, "zebrad: connect_record failed: %s\n",
pa_strerror(pa_context_errno(c)));
mapi->quit(mapi, 1);
}
break;
}
case PA_CONTEXT_FAILED:
fprintf(stderr, "zebrad: pa context failed: %s\n",
pa_strerror(pa_context_errno(c)));
mapi->quit(mapi, 1);
break;
default: break;
}
}
/* ============================================================ */
/* main */
/* ============================================================ */
static void on_sigint(int sig) { (void)sig; if (mapi) mapi->quit(mapi, 0); }
static void usage(const char *prog) {
fprintf(stderr,
"usage: %s [--source SRC] [--port N] [--rate HZ] [--baud N] [--verbose]\n"
"\n"
" --source SRC PulseAudio source name (default: @DEFAULT_MONITOR@)\n"
" --port N WebSocket listen port (default: %d, bound to 127.0.0.1)\n"
" --rate HZ audio sample rate (default: %d)\n"
" --baud N initial baud rate (default: %d, auto-adjusts on READY)\n"
" --verbose log decoded frames + state transitions to stderr\n",
prog, DEFAULT_PORT, DEFAULT_RATE, ZEBRA_BAUD_HANDSHAKE);
}
int main(int argc, char **argv) {
static struct option longopts[] = {
{ "source", required_argument, 0, 's' },
{ "port", required_argument, 0, 'p' },
{ "rate", required_argument, 0, 'r' },
{ "baud", required_argument, 0, 'b' },
{ "verbose", no_argument, 0, 'v' },
{ "help", no_argument, 0, 'h' },
{ 0, 0, 0, 0 }
};
int ch;
while ((ch = getopt_long(argc, argv, "s:p:r:b:vh", longopts, NULL)) != -1) {
switch (ch) {
case 's': opt_source = optarg; break;
case 'p': opt_port = atoi(optarg); break;
case 'r': opt_rate = atoi(optarg); break;
case 'b': opt_baud = atoi(optarg); break;
case 'v': verbose = 1; break;
case 'h': usage(argv[0]); return 0;
default: usage(argv[0]); return 2;
}
}
crc32_init();
for (int i = 0; i < MAX_CLIENTS; i++) clients[i].fd = -1;
frame_init(&fasm);
uart_init(&uart, opt_rate, opt_baud);
signal(SIGINT, on_sigint);
signal(SIGTERM, on_sigint);
signal(SIGPIPE, SIG_IGN);
mloop = pa_mainloop_new();
if (!mloop) { fprintf(stderr, "zebrad: pa_mainloop_new failed\n"); return 1; }
mapi = pa_mainloop_get_api(mloop);
if (ws_listener_start(opt_port) < 0) {
fprintf(stderr, "zebrad: cannot bind 127.0.0.1:%d: %s\n",
opt_port, strerror(errno));
return 1;
}
fprintf(stderr, "zebrad: ws://127.0.0.1:%d source=%s rate=%dHz baud=%d\n",
opt_port, opt_source, opt_rate, opt_baud);
pa_ctx = pa_context_new(mapi, "zebrad");
pa_context_set_state_callback(pa_ctx, pa_context_state_cb, NULL);
if (pa_context_connect(pa_ctx, NULL, 0, NULL) < 0) {
fprintf(stderr, "zebrad: pa_context_connect: %s\n",
pa_strerror(pa_context_errno(pa_ctx)));
return 1;
}
int retval = 0;
pa_mainloop_run(mloop, &retval);
if (pa_str) { pa_stream_disconnect(pa_str); pa_stream_unref(pa_str); }
if (pa_ctx) { pa_context_disconnect(pa_ctx); pa_context_unref(pa_ctx); }
if (listener_io) mapi->io_free(listener_io);
if (listener_fd >= 0) close(listener_fd);
for (int i = 0; i < MAX_CLIENTS; i++) if (clients[i].fd >= 0) client_close(&clients[i]);
pa_mainloop_free(mloop);
return retval;
}