ReviewOS
Reload

also looking at this

home-lang/home

feat(media): add bounded PCM capture, owned snapshots and A/V streaming

#917
Merged chris wants to merge feat/pcm-av-streaming into main
11 files +683 -0

Review threads live on the whole diff, not on one commit, so none are shown here - a thread's line means something in the branch's final form, and painting it into an intermediate step would put it on code it is not about.

.github/workflows/streaming-av.ymladded+29-0
@@ -0,0 +1,29 @@
1name: Streaming A/V modules
2on:
3 push:
4 branches: [feat/pcm-av-streaming]
5 pull_request:
6 paths: ['packages/video/src/containers/av_stream.zig', 'packages/video/src/core/pcm_timeline.zig', 'packages/video/src/core/recording_policy.zig', 'packages/audio/src/io/native_capture*', '.github/workflows/streaming-av.yml']
7permissions:
8 contents: read
9jobs:
10 modules:
11 strategy:
12 fail-fast: false
13 matrix:
14 os: [ubuntu-24.04, ubuntu-24.04-arm]
15 zig: ['0.15.2', 'master']
16 runs-on: ${{ matrix.os }}
17 steps:
18 - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262
19 - uses: mlugg/setup-zig@d1434d08867e3ee9daa34448df10607b98908d29
20 with:
21 version: ${{ matrix.zig }}
22 - run: sudo apt-get update && sudo apt-get install -y libasound2-dev
23 - run: |
24 zig test packages/video/src/core/pcm_timeline.zig
25 zig test packages/video/src/core/recording_policy.zig
26 zig test packages/video/src/containers/av_stream.zig
27 cc -std=c11 -Wall -Wextra -Werror packages/audio/src/io/native_capture.c packages/audio/src/io/native_capture_test.c -lasound -o /tmp/home-capture-test
28 /tmp/home-capture-test
29 zig test packages/audio/src/io/native_capture.zig packages/audio/src/io/native_capture.c -lc -lasound
packages/audio/src/io/io.zigmodified+3-0
@@ -37,3 +37,6 @@ test "io module" {
3737 _ = wasapi;
3838 _ = recorder;
3939}
40
41// Opt-in Linux S16LE capture; compile native_capture.c and link libasound.
42pub const native_capture = ;
packages/audio/src/io/native_capture.cadded+77-0
@@ -0,0 +1,77 @@
1/* Header-checked Linux ALSA input. No subprocess, FFmpeg or unbounded queue. */
2#define _POSIX_C_SOURCE 200809L
3#include "native_capture.h"
4#include <alsa/asoundlib.h>
5#include <stdlib.h>
6#include <errno.h>
7#include <time.h>
8#include <limits.h>
9struct home_audio_capture {
10 snd_pcm_t *pcm;
11 snd_pcm_status_t *status;
12 home_capture_config config;
13 snd_pcm_uframes_t buffer_frames;
14 uint64_t anchor,samples;
15 int started,anchored;
16};
17static uint64_t now_ns(clockid_t clock) {
18 struct timespec ts;if(clock_gettime(clock,&ts)!=0)return 0;
19 return (uint64_t)ts.tv_sec*1000000000u+(uint64_t)ts.tv_nsec;
20}
21void home_audio_close(home_audio_capture *s) {
22 if(!s)return;
23 if(s->pcm){snd_pcm_drop(s->pcm);snd_pcm_close(s->pcm);}
24 if(s->status)snd_pcm_status_free(s->status);
25 free(s);
26}
27int home_audio_open(const home_capture_config *cfg,const char *device,home_audio_capture **out) {
28 if(!out)return -1;
29 *out=NULL;
30 if(!cfg || !device || !device[0] || cfg->sample_rate<8000 || cfg->sample_rate>192000 || cfg->channels<1 || cfg->channels>2 || cfg->block_frames<1 || cfg->block_frames>4096)return -1;
31 home_audio_capture *s=calloc(1,sizeof(*s));if(!s)return -2;s->config=*cfg;
32 snd_pcm_hw_params_t *hw=NULL;snd_pcm_sw_params_t *sw=NULL;
33 if(snd_pcm_open(&s->pcm,device,SND_PCM_STREAM_CAPTURE,SND_PCM_NONBLOCK)<0)goto fail;
34 if(snd_pcm_hw_params_malloc(&hw)<0 || snd_pcm_sw_params_malloc(&sw)<0 || snd_pcm_status_malloc(&s->status)<0)goto fail;
35 snd_pcm_uframes_t period=cfg->block_frames,buffer=cfg->sample_rate/5;
36 if(snd_pcm_hw_params_any(s->pcm,hw)<0 || snd_pcm_hw_params_set_access(s->pcm,hw,SND_PCM_ACCESS_RW_INTERLEAVED)<0 || snd_pcm_hw_params_set_format(s->pcm,hw,SND_PCM_FORMAT_S16_LE)<0 || snd_pcm_hw_params_set_channels(s->pcm,hw,cfg->channels)<0 || snd_pcm_hw_params_set_rate(s->pcm,hw,cfg->sample_rate,0)<0 || snd_pcm_hw_params_set_period_size_near(s->pcm,hw,&period,NULL)<0 || snd_pcm_hw_params_set_buffer_size_near(s->pcm,hw,&buffer)<0 || snd_pcm_hw_params(s->pcm,hw)<0)goto fail;
37 s->buffer_frames=buffer;
38 if(snd_pcm_sw_params_current(s->pcm,sw)<0 || snd_pcm_sw_params_set_tstamp_mode(s->pcm,sw,SND_PCM_TSTAMP_ENABLE)<0 || snd_pcm_sw_params_set_tstamp_type(s->pcm,sw,SND_PCM_TSTAMP_TYPE_MONOTONIC)<0 || snd_pcm_sw_params(s->pcm,sw)<0 || snd_pcm_prepare(s->pcm)<0)goto fail;
39 snd_pcm_hw_params_free(hw);snd_pcm_sw_params_free(sw);*out=s;return 0;
40fail:
41 if(hw)snd_pcm_hw_params_free(hw);
42 if(sw)snd_pcm_sw_params_free(sw);
43 home_audio_close(s);return -3;
44}
45int home_audio_read(home_audio_capture *s,uint8_t *data,uint32_t capacity,uint32_t timeout_ms,uint32_t *frames,uint64_t *timestamp,int64_t *error_ns) {
46 if(!s || !data || !frames || !timestamp || !error_ns || capacity<s->config.block_frames || timeout_ms>1000)return -1;
47 *frames=0;*timestamp=0;*error_ns=0;
48 if(!s->started){if(snd_pcm_start(s->pcm)<0)return -3;s->started=1;}
49 const uint64_t deadline=now_ns(CLOCK_MONOTONIC)+(uint64_t)timeout_ms*1000000;
50 snd_pcm_sframes_t n;
51 for(;;) {
52 n=snd_pcm_readi(s->pcm,data,s->config.block_frames);
53 if(n>0)break;
54 if(n<0 && n!=-EAGAIN && n!=-EINTR)return -3;
55 uint64_t now=now_ns(CLOCK_MONOTONIC);if(now>=deadline)return 1;
56 int wait=(int)((deadline-now+999999)/1000000);
57 int rc=snd_pcm_wait(s->pcm,wait);if(rc<0 && rc!=-EINTR)return -3;
58 if(rc==0)return 1;
59 }
60 if(snd_pcm_status(s->pcm,s->status)<0 || snd_pcm_status_get_state(s->status)!=SND_PCM_STATE_RUNNING)return -3;
61 snd_htimestamp_t stamp;snd_pcm_status_get_htstamp(s->status,&stamp);
62 const uint64_t mono=now_ns(CLOCK_MONOTONIC),boot=now_ns(CLOCK_BOOTTIME);
63 const snd_pcm_uframes_t available=snd_pcm_status_get_avail(s->status);
64 if(!mono || boot<mono || available>s->buffer_frames)return -3;
65 const uint64_t status_ns=(uint64_t)stamp.tv_sec*1000000000u+(uint64_t)stamp.tv_nsec;
66 const uint64_t backlog=((uint64_t)available+(uint64_t)n)*1000000000u/s->config.sample_rate;
67 if(status_ns<backlog)return -3;
68 const uint64_t measured=status_ns-backlog+(boot-mono);
69 if(!s->anchored){s->anchor=measured;s->anchored=1;}
70 __uint128_t value=(__uint128_t)s->anchor+(__uint128_t)s->samples*1000000000u/s->config.sample_rate;
71 if(value>UINT64_MAX || s->samples>UINT64_MAX-(uint64_t)n)return -3;
72 const uint64_t nominal=(uint64_t)value;
73 uint64_t delta=measured>nominal ? measured-nominal : nominal-measured;
74 if(delta>100000000u)return -3;
75 *error_ns=measured>=nominal ? (int64_t)delta : -(int64_t)delta;
76 *timestamp=nominal;*frames=(uint32_t)n;s->samples+=(uint64_t)n;return 0;
77}
packages/audio/src/io/native_capture.hadded+13-0
@@ -0,0 +1,13 @@
1/* Home ALSA capture bridge, MIT. Opaque system ABI; S16LE interleaved PCM. */
2#ifndef HOME_NATIVE_CAPTURE_H
3#define HOME_NATIVE_CAPTURE_H
4#include <stdint.h>
5typedef struct home_audio_capture home_audio_capture;
6typedef struct { uint32_t sample_rate,channels,block_frames; } home_capture_config;
7int home_audio_open(const home_capture_config *,const char *device,home_audio_capture **);
8/* 0 success, 1 timeout, negative error. timestamp is first sample in CLOCK_BOOTTIME
9 * ns. Anchored sample counting avoids scheduler jitter; measured error is returned.
10 * XRUNs/disconnects or clock error >100 ms fail explicitly; no silent recovery. */
11int home_audio_read(home_audio_capture *,uint8_t *,uint32_t capacity_frames,uint32_t timeout_ms,uint32_t *frames,uint64_t *timestamp_ns,int64_t *clock_error_ns);
12void home_audio_close(home_audio_capture *);
13#endif
packages/audio/src/io/native_capture.zigadded+42-0
@@ -0,0 +1,42 @@
1//! Home opaque, header-checked ALSA capture. Build C boundary and link libasound.
2//! Calls serialize; close only after read returns. S16LE mono/stereo, BOOTTIME ns.
3const std = ;
4const Handle = opaque {};
5pub const Config = extern struct { sample_rate: u32, channels: u32, block_frames: u32 };
6extern fn home_audio_open(*const Config, [*:0]const u8, *?*Handle) c_int;
7extern fn home_audio_read(*Handle, [*]u8, u32, u32, *u32, *u64, *i64) c_int;
8extern fn home_audio_close(*Handle) void;
9pub const Block = struct { frames: u32, timestamp_ns: u64, clock_error_ns: i64 };
10pub const Capture = struct {
11 handle: *Handle,
12 config: Config,
13 failed: bool = false,
14 pub fn init(cfg: Config, device: [:0]const u8) !Capture {
15 var out: ?*Handle = null;
16 try check(home_audio_open(&cfg, device.ptr, &out));
17 return .{ .handle = out.?, .config = cfg };
18 }
19 pub fn deinit(s: *Capture) void {
20 home_audio_close(s.handle);
21 }
22 pub fn read(s: *Capture, data: []u8, timeout_ms: u32) !?Block {
23 if (s.failed) return error.CaptureFailed;
24 if (data.len < @as(u64, s.config.block_frames) * s.config.channels * 2 or timeout_ms > 1000) return error.InvalidConfiguration;
25 var block: Block = undefined;
26 const rc = home_audio_read(s.handle, data.ptr, @intCast(data.len / (s.config.channels * 2)), timeout_ms, &block.frames, &block.timestamp_ns, &block.clock_error_ns);
27 if (rc == 1) return null;
28 check(rc) catch |err| {
29 s.failed = true;
30 return err;
31 };
32 return block;
33 }
34};
35fn check(rc: c_int) !void {
36 return switch (rc) {
37 0 => {},
38 -1 => error.InvalidConfiguration,
39 -2 => error.OutOfMemory,
40 else => error.CaptureFailed,
41 };
42}
packages/audio/src/io/native_capture_test.cadded+16-0
@@ -0,0 +1,16 @@
1#include "native_capture.h"
2#include <assert.h>
3#include <stdint.h>
4#include <stddef.h>
5int main(void) {
6 home_capture_config cfg={48000,2,480};
7 home_audio_capture *capture=NULL;
8 assert(home_audio_open(&cfg,"null",&capture)==0);
9 uint8_t bytes[1920];uint32_t frames=0;uint64_t timestamp=0;int64_t error=0;
10 assert(home_audio_read(capture,bytes,480,100,&frames,&timestamp,&error)==0);
11 assert(frames>0 && frames<=480 && timestamp>0 && error==0);
12 home_audio_close(capture);
13 capture=NULL;cfg.channels=3;
14 assert(home_audio_open(&cfg,"null",&capture)==-1 && !capture);
15 return 0;
16}
packages/video/STREAMING_AV.mdadded+46-0
@@ -0,0 +1,46 @@
1# Bounded recording support
2
3`src/containers/av_stream.zig` streams Matroska video plus lossless S16LE PCM
4through a caller-owned synchronous sink. Supports MJPEG, AVC and HEVC; AVC/HEVC
5samples must be four-byte NAL-length prefixed and accompanied by avcC/hvcC
6CodecPrivate. Each packet is one cluster, using microsecond timecodes and one
7common epoch. Per-track timestamps must increase; cross-track ordering may differ.
8The writer keeps a fixed header buffer and no frame collection. A sink failure
9poisons further writes. Final duration, seek cues and audio compression are absent.
10
11`src/core/pcm_timeline.zig` owns a fixed PCM arena and descriptor queue. It rejects
12capacity exhaustion without mutation, checks time arithmetic, and clips at sample
13boundaries for initial trimming and segment splits. Borrowed output chunks expire
14at the next mutation; consume them synchronously. It does not resample or correct
15drift. Preserve the capture clock and gaps; never infer capture time from callbacks.
16
17`src/core/recording_policy.zig` checks soft elapsed-time and byte thresholds. The
18caller generates paths and requests a keyframe; it must not cut predicted frames.
19It suppresses duplicate requests while a split is pending and ignores empty output.
20
21`packages/audio/src/io/native_capture.{c,h,zig}` provides explicit Linux ALSA
22S16LE capture using checked system headers and an opaque handle. Compile the C
23source with the system compiler and link libasound. Rate 8–192kHz, mono/stereo,
24block size 1–4096; nonblocking reads have a bounded timeout. It anchors the sample
25clock from ALSA MONOTONIC timestamps, converts to Linux BOOTTIME, then advances
26by sample count. Reported clock error is measured against ALSA status; >100ms
27error, XRUN or device removal fails instead of silently dropping/recovering audio.
28This is a new opt-in input backend; the older recorder abstraction is unchanged.
29The existing simplified AAC implementations have not been adopted or represented
30as a standards-verified capture encoder.
31
32Call read and close serially. No queues, threads or implicit microphone access are
33created by the Home adapter. Consumers must bound their own input/output queues.
34Physical microphone latency, camera offset, suspend, drift and long recordings
35still require hardware validation. No FFmpeg or Libav dependency is used.
36
37Targeted Zig tests (0.15.2 and current Home toolchain):
38
39```sh
40zig test packages/video/src/core/pcm_timeline.zig
41zig test packages/video/src/core/recording_policy.zig
42zig test packages/video/src/containers/av_stream.zig
43```
44
45The Linux native smoke explicitly opens ALSA's `null` test device, never the
46user's default microphone. See `.github/workflows/streaming-av.yml`.
packages/video/src/containers/av_stream.zigadded+154-0
@@ -0,0 +1,154 @@
1//! Home streaming Matroska video + lossless S16LE PCM. One cluster per packet;
2//! microsecond timecodes, independent per-track ordering, no file-sized buffer.
3const std = ;
4pub const Sink = struct { context: *anyopaque, write: *const fn (*anyopaque, []const u8) anyerror!void };
5pub const Config = struct { width: u32, height: u32, fps_num: u32, fps_den: u32, codec: u32, sample_rate: u32, channels: u32, codec_private: []const u8 = &.{} };
6const Buffer = struct {
7 data: [16384]u8 = undefined,
8 len: usize = 0,
9 fn bytes(s: *Buffer, data: []const u8) !void {
10 if (data.len > s.data.len - s.len) return error.HeaderTooLarge;
11 @memcpy(s.data[s.len..][0..data.len], data);
12 s.len += data.len;
13 }
14 fn slice(s: *const Buffer) []const u8 {
15 return s.data[0..s.len];
16 }
17 fn number(s: *Buffer, value: u64) !void {
18 var b: [8]u8 = undefined;
19 std.mem.writeInt(u64, &b, value, .big);
20 var n: usize = 0;
21 while (n < 7 and b[n] == 0) : (n += 1) {}
22 try s.bytes(b[n..]);
23 }
24 fn vint(s: *Buffer, value: u64) !void {
25 var n: usize = 1;
26 while (n < 8 and value >= (@as(u64, 1) << @intCast(n * 7)) - 1) : (n += 1) {}
27 if (value >= (@as(u64, 1) << 56) - 1) return error.PacketTooLarge;
28 var b: [8]u8 = undefined;
29 std.mem.writeInt(u64, &b, value | (@as(u64, 1) << @intCast(n * 7)), .big);
30 try s.bytes(b[8 - n ..]);
31 }
32 fn element(s: *Buffer, id: u32, value: []const u8) !void {
33 try s.number(id);
34 try s.vint(value.len);
35 try s.bytes(value);
36 }
37 fn uint(s: *Buffer, id: u32, value: u64) !void {
38 var b: Buffer = .{};
39 try b.number(value);
40 try s.element(id, b.slice());
41 }
42};
43pub const Writer = struct {
44 sink: Sink,
45 config: Config,
46 last_video: ?u64 = null,
47 last_audio: ?u64 = null,
48 failed: bool = false,
49 pub fn init(sink: Sink, cfg: Config) !Writer {
50 if (cfg.width == 0 or cfg.height == 0 or cfg.width > 16384 or cfg.height > 16384 or cfg.fps_num == 0 or cfg.fps_den == 0 or cfg.fps_num > 1000000 or cfg.fps_den > 1000000 or cfg.codec < 1 or cfg.codec > 3 or cfg.sample_rate < 8000 or cfg.sample_rate > 192000 or cfg.channels < 1 or cfg.channels > 2 or cfg.codec_private.len > 8192) return error.InvalidConfiguration;
51 if (cfg.codec != 3 and (cfg.codec_private.len < (if (cfg.codec == 1) @as(usize, 7) else 23) or cfg.codec_private[0] != 1)) return error.InvalidConfiguration;
52 var ebml: Buffer = .{};
53 try ebml.uint(0x4286, 1);
54 try ebml.uint(0x42f7, 1);
55 try ebml.uint(0x42f2, 4);
56 try ebml.uint(0x42f3, 8);
57 try ebml.element(0x4282, "matroska");
58 try ebml.uint(0x4287, 4);
59 try ebml.uint(0x4285, 2);
60 var header: Buffer = .{};
61 try header.element(0x1a45dfa3, ebml.slice());
62 try header.bytes(&.{ 0x18, 0x53, 0x80, 0x67, 0x01, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff });
63 var info: Buffer = .{};
64 try info.uint(0x2ad7b1, 1000);
65 try info.element(0x4d80, "Home");
66 try info.element(0x5741, "Home synchronized A/V");
67 try header.element(0x1549a966, info.slice());
68 var tracks: Buffer = .{};
69 var video_info: Buffer = .{};
70 try video_info.uint(0xb0, cfg.width);
71 try video_info.uint(0xba, cfg.height);
72 var track: Buffer = .{};
73 try track.uint(0xd7, 1);
74 try track.uint(0x73c5, 1);
75 try track.uint(0x83, 1);
76 try track.uint(0x9c, 0);
77 try track.element(0x86, if (cfg.codec == 3) "V_MJPEG" else if (cfg.codec == 1) "V_MPEG4/ISO/AVC" else "V_MPEGH/ISO/HEVC");
78 try track.uint(0x23e383, @as(u64, 1_000_000_000) * cfg.fps_den / cfg.fps_num);
79 if (cfg.codec != 3) try track.element(0x63a2, cfg.codec_private);
80 try track.element(0xe0, video_info.slice());
81 try tracks.element(0xae, track.slice());
82 var audio_info: Buffer = .{};
83 var rate: [8]u8 = undefined;
84 std.mem.writeInt(u64, &rate, @bitCast(@as(f64, @floatFromInt(cfg.sample_rate))), .big);
85 try audio_info.element(0xb5, &rate);
86 try audio_info.uint(0x9f, cfg.channels);
87 try audio_info.uint(0x6264, 16);
88 track = .{};
89 try track.uint(0xd7, 2);
90 try track.uint(0x73c5, 2);
91 try track.uint(0x83, 2);
92 try track.uint(0x9c, 0);
93 try track.element(0x86, "A_PCM/INT/LIT");
94 try track.element(0xe1, audio_info.slice());
95 try tracks.element(0xae, track.slice());
96 try header.element(0x1654ae6b, tracks.slice());
97 try sink.write(sink.context, header.slice());
98 // Codec-private bytes were consumed synchronously, not retained.
99 var owned = cfg;
100 owned.codec_private = &.{};
101 return .{ .sink = sink, .config = owned };
102 }
103 fn block(s: *Writer, track: u8, data: []const u8, timestamp_ns: u64, key: bool) !void {
104 if (s.failed) return error.WriterFailed;
105 if (data.len == 0 or data.len > std.math.maxInt(u32)) return error.InvalidPacket;
106 const last = if (track == 1) s.last_video else s.last_audio;
107 if (last) |t| if (timestamp_ns <= t) return error.InvalidTimestamp;
108 var time: Buffer = .{};
109 try time.uint(0xe7, timestamp_ns / 1000);
110 var size: Buffer = .{};
111 try size.vint(data.len + 4);
112 var header: Buffer = .{};
113 try header.number(0x1f43b675);
114 try header.vint(time.len + 1 + size.len + 4 + data.len);
115 try header.bytes(time.slice());
116 try header.bytes(&.{0xa3});
117 try header.bytes(size.slice());
118 try header.bytes(&.{ 0x80 | track, 0, 0, if (key) @as(u8, 0x80) else 0 });
119 errdefer s.failed = true;
120 try s.sink.write(s.sink.context, header.slice());
121 try s.sink.write(s.sink.context, data);
122 if (track == 1) s.last_video = timestamp_ns else s.last_audio = timestamp_ns;
123 }
124 pub fn video(s: *Writer, data: []const u8, timestamp_ns: u64, key: bool) !void {
125 if (s.config.codec == 3 and (data.len < 4 or !std.mem.startsWith(u8, data, &.{ 255, 216 }) or !std.mem.endsWith(u8, data, &.{ 255, 217 }))) return error.InvalidPacket;
126 try s.block(1, data, timestamp_ns, key);
127 }
128 pub fn audio(s: *Writer, data: []const u8, timestamp_ns: u64) !void {
129 if (data.len == 0 or data.len % (s.config.channels * 2) != 0) return error.InvalidPcm;
130 try s.block(2, data, timestamp_ns, true);
131 }
132};
133test "Matroska has real PCM track, allows interleaved track times, and poisons on I/O failure" {
134 const Out = struct {
135 data: std.ArrayList(u8) = .empty,
136 fail: bool = false,
137 fn write(ctx: *anyopaque, data: []const u8) anyerror!void {
138 const s: *@This() = @ptrCast(@alignCast(ctx));
139 if (s.fail) return error.Full;
140 try s.data.appendSlice(std.testing.allocator, data);
141 }
142 };
143 var out: Out = .{};
144 defer out.data.deinit(std.testing.allocator);
145 var writer = try Writer.init(.{ .context = &out, .write = Out.write }, .{ .width = 32, .height = 32, .fps_num = 30, .fps_den = 1, .codec = 3, .sample_rate = 48000, .channels = 2 });
146 try std.testing.expect(std.mem.indexOf(u8, out.data.items, "A_PCM/INT/LIT") != null);
147 try writer.video(&.{ 255, 216, 255, 217 }, 1000000, true);
148 try writer.audio(&.{ 1, 0, 2, 0 }, 0);
149 try std.testing.expectError(error.InvalidPcm, writer.audio(&.{1}, 100));
150 try std.testing.expectError(error.InvalidTimestamp, writer.audio(&.{ 1, 0, 2, 0 }, 0));
151 out.fail = true;
152 try std.testing.expectError(error.Full, writer.audio(&.{ 1, 0, 2, 0 }, 2000000));
153 try std.testing.expectError(error.WriterFailed, writer.video(&.{ 255, 216, 255, 217 }, 3000000, true));
154}
packages/video/src/core/pcm_timeline.zigadded+128-0
@@ -0,0 +1,128 @@
1//! Bounded PCM ownership and sample-accurate clipping on a common nanosecond clock.
2//! Single serialized owner; returned data expires on the next push/mutation.
3const std = ;
4pub fn at(origin: u64, frames: u64, rate: u32) !u64 {
5 if (rate < 8000 or rate > 192000) return error.InvalidRate;
6 const value = @as(u128, origin) + @as(u128, frames) * 1_000_000_000 / rate;
7 if (value > std.math.maxInt(u64)) return error.TimestampOverflow;
8 return @intCast(value);
9}
10pub fn before(origin: u64, cut: u64, rate: u32, limit: usize) usize {
11 if (cut <= origin) return 0;
12 return @intCast(@min(limit, (@as(u128, cut - origin) * rate + 999_999_999) / 1_000_000_000));
13}
14const Entry = struct { offset: usize, frames: usize, consumed: usize = 0, timestamp: u64 };
15pub const Chunk = struct { data: []const u8, timestamp: u64, frames: usize };
16pub const Queue = struct {
17 allocator: std.mem.Allocator,
18 data: []u8,
19 entries: []Entry,
20 rate: u32,
21 channels: u32,
22 capacity: usize,
23 head: usize = 0,
24 count: usize = 0,
25 used: usize = 0,
26 write_pos: usize = 0,
27 last_end: ?u64 = null,
28 trimmed: u64 = 0,
29 pub fn init(a: std.mem.Allocator, rate: u32, channels: u32, frames: usize, blocks: usize) !Queue {
30 if (rate < 8000 or rate > 192000 or channels < 1 or channels > 2 or frames == 0 or frames > rate * 10 or blocks == 0 or blocks > 4096) return error.InvalidConfiguration;
31 const data = try a.alloc(u8, frames * channels * 2);
32 errdefer a.free(data);
33 const entries = try a.alloc(Entry, blocks);
34 return .{ .allocator = a, .data = data, .entries = entries, .rate = rate, .channels = channels, .capacity = frames };
35 }
36 pub fn deinit(s: *Queue) void {
37 s.allocator.free(s.data);
38 s.allocator.free(s.entries);
39 }
40 pub fn push(s: *Queue, data: []const u8, timestamp: u64) !void {
41 const stride = s.channels * 2;
42 if (data.len == 0 or data.len % stride != 0) return error.InvalidPcm;
43 const frames = data.len / stride;
44 if (frames > s.capacity - s.used or s.count == s.entries.len) return error.BufferFull;
45 var ts = timestamp;
46 if (s.last_end) |end| {
47 if (ts < end) {
48 if (end - ts > (1_000_000_000 + s.rate - 1) / s.rate) return error.InvalidTimestamp;
49 ts = end;
50 }
51 }
52 const end = try at(ts, frames, s.rate);
53 const first = @min(frames, s.capacity - s.write_pos);
54 @memcpy(s.data[s.write_pos * stride ..][0 .. first * stride], data[0 .. first * stride]);
55 @memcpy(s.data[0 .. (frames - first) * stride], data[first * stride ..]);
56 s.entries[(s.head + s.count) % s.entries.len] = .{ .offset = s.write_pos, .frames = frames, .timestamp = ts };
57 s.write_pos = (s.write_pos + frames) % s.capacity;
58 s.count += 1;
59 s.used += frames;
60 s.last_end = end;
61 }
62 fn consume(s: *Queue, e: *Entry, n: usize) void {
63 e.consumed += n;
64 s.used -= n;
65 if (e.consumed == e.frames) {
66 s.head = (s.head + 1) % s.entries.len;
67 s.count -= 1;
68 }
69 }
70 /// Emit samples with time >= minimum and < cut. Wrap is returned as separate
71 /// contiguous chunks; never duplicate a sample at a segment boundary.
72 pub fn take(s: *Queue, minimum: u64, cut: u64) !?Chunk {
73 while (s.count != 0) {
74 const e = &s.entries[s.head];
75 const end = before(e.timestamp, cut, s.rate, e.frames);
76 const begin = @max(e.consumed, @min(end, before(e.timestamp, minimum, s.rate, e.frames)));
77 if (begin > e.consumed) {
78 const n = begin - e.consumed;
79 s.trimmed += n;
80 s.consume(e, n);
81 if (s.count == 0 or begin == e.frames) continue;
82 }
83 if (end <= e.consumed) return null;
84 const offset = (e.offset + e.consumed) % s.capacity;
85 const n = @min(end - e.consumed, s.capacity - offset);
86 const result = Chunk{ .data = s.data[offset * s.channels * 2 ..][0 .. n * s.channels * 2], .timestamp = try at(e.timestamp, e.consumed, s.rate), .frames = n };
87 s.consume(e, n);
88 return result;
89 }
90 return null;
91 }
92};
93test "PCM splits across samples and arena wrap without loss or duplicate frames" {
94 var q = try Queue.init(std.testing.allocator, 48000, 1, 8, 4);
95 defer q.deinit();
96 try q.push(&.{ 1, 0, 2, 0, 3, 0, 4, 0, 5, 0, 6, 0 }, 1000000);
97 const first = (try q.take(1000000, 1062500)).?;
98 try std.testing.expectEqualSlices(u8, &.{ 1, 0, 2, 0, 3, 0 }, first.data);
99 try std.testing.expect((try q.take(1000000, 1062500)) == null);
100 const next = (try q.take(1062500, 1125000)).?;
101 try std.testing.expectEqualSlices(u8, &.{ 4, 0, 5, 0, 6, 0 }, next.data);
102 try q.push(&.{ 7, 0, 8, 0, 9, 0, 10, 0 }, 1125000);
103 const wrap = (try q.take(1125000, 2000000)).?;
104 try std.testing.expectEqualSlices(u8, &.{ 7, 0, 8, 0 }, wrap.data);
105 try std.testing.expectEqualSlices(u8, &.{ 9, 0, 10, 0 }, (try q.take(1125000, 2000000)).?.data);
106 try std.testing.expectEqual(@as(usize, 0), q.used);
107}
108test "PCM rejects full queues, invalid inputs and overflow before modifying history" {
109 var q = try Queue.init(std.testing.allocator, 48000, 1, 4, 2);
110 defer q.deinit();
111 try q.push(&.{ 1, 0, 2, 0, 3, 0, 4, 0 }, 1000000);
112 try std.testing.expectError(error.BufferFull, q.push(&.{ 5, 0 }, 2000000));
113 _ = try q.take(1041667, 2000000);
114 try std.testing.expectEqual(@as(u64, 3), q.trimmed);
115 try std.testing.expectError(error.InvalidPcm, q.push(&.{1}, 2000000));
116 try std.testing.expectError(error.InvalidTimestamp, q.push(&.{ 1, 0 }, 1));
117 try std.testing.expectError(error.TimestampOverflow, at(std.math.maxInt(u64), 1, 48000));
118 try std.testing.expectEqual(@as(u64, 1_000_000_000), try at(0, 44100, 44100));
119}
120test "PCM partial allocation failure cleanup" {
121 const Check = struct {
122 fn run(a: std.mem.Allocator) !void {
123 var q = try Queue.init(a, 48000, 2, 1024, 32);
124 defer q.deinit();
125 }
126 };
127 try std.testing.checkAllAllocationFailures(std.testing.allocator, Check.run, .{});
128}
packages/video/src/core/recording_policy.zigadded+18-0
@@ -0,0 +1,18 @@
1//! Soft rotation thresholds: caller stages a keyframe-aligned split when due.
2const std = ;
3pub const Policy = struct {
4 duration_ns: u64 = 0,
5 max_bytes: u64 = 0,
6 pub fn due(s: Policy, frames: u64, elapsed_ns: u64, bytes: u64, pending: bool) bool {
7 return frames > 0 and !pending and ((s.duration_ns != 0 and elapsed_ns >= s.duration_ns) or (s.max_bytes != 0 and bytes >= s.max_bytes));
8 }
9};
10test "thresholds include equality, ignore empty and pending segments, and can disable" {
11 const p = Policy{ .duration_ns = 100, .max_bytes = 200 };
12 try std.testing.expect(!p.due(0, 100, 200, false));
13 try std.testing.expect(!p.due(1, 100, 200, true));
14 try std.testing.expect(!p.due(1, 99, 199, false));
15 try std.testing.expect(p.due(1, 100, 0, false));
16 try std.testing.expect(p.due(1, 0, 200, false));
17 try std.testing.expect(!(Policy{}).due(1, 100, 200, false));
18}
packages/video/src/video.zigmodified+5-0
@@ -1293,3 +1293,8 @@ test "SampleFormat" {
12931293 try std.testing.expectEqual(@as(u8, 2), SampleFormat.s16le.bytesPerSample());
12941294 try std.testing.expectEqual(@as(u8, 4), SampleFormat.f32le.bytesPerSample());
12951295}
1296
1297// Bounded capture building blocks; independently testable on Zig 0.15.2+.
1298pub const av_stream = ;
1299pub const pcm_timeline = ;
1300pub const recording_policy = ;