diff --git a/Makefile b/Makefile index 7f6f2a1..dff0609 100644 --- a/Makefile +++ b/Makefile @@ -4,9 +4,9 @@ LDFLAGS = $(shell pkg-config --libs libpulse) -lrt -lpthread 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/unit @@ -43,6 +43,9 @@ bt: src/bt.c $(PULSE) carrier: src/carrier.c $(CC) $(CFLAGS) -o $@ $^ $(LDFLAGS) +zebrad: src/zebrad.c include/zebra.h + $(CC) $(CFLAGS) -o $@ src/zebrad.c $(LDFLAGS) -lm + blog: python3 blog/build.py @@ -50,5 +53,5 @@ serve: blog cd web && python3 -m http.server 8765 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 diff --git a/README.md b/README.md index fb885ec..90d2c86 100644 --- a/README.md +++ b/README.md @@ -41,7 +41,7 @@ early with a clear message. ## Targets -### `make all` — five binaries +### `make all` — six binaries | Binary | Source | Role | |-----------|----------------|------| @@ -50,6 +50,7 @@ early with a clear message. | `chat` | `src/chat.c` | bidirectional tx+rx, line-based chat UI | | `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 | +| `zebrad` | `src/zebrad.c` | PA→WS introspector for `web/chat.html` | All link against `libpulse`, `librt`, and `libpthread`. Headers come from `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 `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 --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` Removes: diff --git a/src/zebrad.c b/src/zebrad.c new file mode 100644 index 0000000..3b8a3f2 --- /dev/null +++ b/src/zebrad.c @@ -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 +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +#include +#include + +#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; +}