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
.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+41-0
@@ -0,0 +1,41 @@
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 Handle = opaque {};
4pub const Config = extern struct { sample_rate: u32, channels: u32, block_frames: u32 };
5extern fn home_audio_open(*const Config, [*:0]const u8, *?*Handle) c_int;
6extern fn home_audio_read(*Handle, [*]u8, u32, u32, *u32, *u64, *i64) c_int;
7extern fn home_audio_close(*Handle) void;
8pub const Block = struct { frames: u32, timestamp_ns: u64, clock_error_ns: i64 };
9pub const Capture = struct {
10 handle: *Handle,
11 config: Config,
12 failed: bool = false,
13 pub fn init(cfg: Config, device: [:0]const u8) !Capture {
14 var out: ?*Handle = null;
15 try check(home_audio_open(&cfg, device.ptr, &out));
16 return .{ .handle = out.?, .config = cfg };
17 }
18 pub fn deinit(s: *Capture) void {
19 home_audio_close(s.handle);
20 }
21 pub fn read(s: *Capture, data: []u8, timeout_ms: u32) !?Block {
22 if (s.failed) return error.CaptureFailed;
23 if (data.len < @as(u64, s.config.block_frames) * s.config.channels * 2 or timeout_ms > 1000) return error.InvalidConfiguration;
24 var block: Block = undefined;
25 const rc = home_audio_read(s.handle, data.ptr, @intCast(s.config.block_frames), timeout_ms, &block.frames, &block.timestamp_ns, &block.clock_error_ns);
26 if (rc == 1) return null;
27 check(rc) catch |err| {
28 s.failed = true;
29 return err;
30 };
31 return block;
32 }
33};
34fn check(rc: c_int) !void {
35 return switch (rc) {
36 0 => {},
37 -1 => error.InvalidConfiguration,
38 -2 => error.OutOfMemory,
39 else => error.CaptureFailed,
40 };
41}
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+65-0
@@ -0,0 +1,65 @@
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`.
47
48`Queue.pushRolling` adds bounded circular PCM history. It validates the entire
49input before evicting whole oldest blocks to meet both sample and descriptor
50limits. `retention_floor` reports the end of the discarded prefix. Invalid or
51oversized blocks leave history untouched. `window(minimum, cut, cursor)` reads
52non-destructively with sample-exact clipping, including arena wrap. Each replay
53uses a fresh cursor; chunks and cursors expire on mutation. `Chunk.end_ns` uses
54the original block clock to avoid accumulating fractional sample rounding.
55Consumers can align an encoded keyframe ring to the PCM retention floor and
56replay separate snapshots without consuming live audio. If eviction passes a
57live consumer's watermark, report loss rather than silently producing a gap.
58No encoder, thread or file I/O is performed by these history methods.
59
60`Queue.cloneWindow(allocator, minimum, cut)` freezes just the selected sample
61window into independent storage. It preserves capture timestamps and sample phase,
62handles a wrapped source block even when all 4096 descriptors are occupied, and
63frees partial allocations on failure. The source is unchanged. Serialize cloning
64with source mutation; after return the copy may move to an independent worker and
65outlive the source. No audio encoder or worker runtime is imported by this module.
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+262-0
@@ -0,0 +1,262 @@
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, end_ns: u64 };
16pub const Cursor = struct { block: usize = 0, frame: usize = 0 };
17pub const Queue = struct {
18 allocator: std.mem.Allocator,
19 data: []u8,
20 entries: []Entry,
21 rate: u32,
22 channels: u32,
23 capacity: usize,
24 head: usize = 0,
25 count: usize = 0,
26 used: usize = 0,
27 write_pos: usize = 0,
28 last_end: ?u64 = null,
29 trimmed: u64 = 0,
30 retention_floor: ?u64 = null,
31 pub fn init(a: std.mem.Allocator, rate: u32, channels: u32, frames: usize, blocks: usize) !Queue {
32 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;
33 const data = try a.alloc(u8, frames * channels * 2);
34 errdefer a.free(data);
35 const entries = try a.alloc(Entry, blocks);
36 return .{ .allocator = a, .data = data, .entries = entries, .rate = rate, .channels = channels, .capacity = frames };
37 }
38 pub fn deinit(s: *Queue) void {
39 s.allocator.free(s.data);
40 s.allocator.free(s.entries);
41 }
42 pub fn push(s: *Queue, data: []const u8, timestamp: u64) !void {
43 const stride = s.channels * 2;
44 if (data.len == 0 or data.len % stride != 0) return error.InvalidPcm;
45 const frames = data.len / stride;
46 if (frames > s.capacity - s.used or s.count == s.entries.len) return error.BufferFull;
47 const ts = timestamp;
48 if (s.last_end) |end| if (ts < end) return error.InvalidTimestamp;
49 const end = try at(ts, frames, s.rate);
50 const first = @min(frames, s.capacity - s.write_pos);
51 @memcpy(s.data[s.write_pos * stride ..][0 .. first * stride], data[0 .. first * stride]);
52 @memcpy(s.data[0 .. (frames - first) * stride], data[first * stride ..]);
53 s.entries[(s.head + s.count) % s.entries.len] = .{ .offset = s.write_pos, .frames = frames, .timestamp = ts };
54 s.write_pos = (s.write_pos + frames) % s.capacity;
55 s.count += 1;
56 s.used += frames;
57 s.last_end = end;
58 }
59 fn consume(s: *Queue, e: *Entry, n: usize) void {
60 e.consumed += n;
61 s.used -= n;
62 if (e.consumed == e.frames) {
63 s.head = (s.head + 1) % s.entries.len;
64 s.count -= 1;
65 }
66 }
67 /// Circular ownership: reject invalid input before eviction, then discard
68 /// whole oldest blocks until the new block fits. No allocation after init.
69 pub fn pushRolling(s: *Queue, data: []const u8, timestamp: u64) !void {
70 const stride = s.channels * 2;
71 if (data.len == 0 or data.len % stride != 0) return error.InvalidPcm;
72 const frames = data.len / stride;
73 if (frames > s.capacity) return error.BufferFull;
74 if (s.last_end) |end| if (timestamp < end) return error.InvalidTimestamp;
75 _ = try at(timestamp, frames, s.rate);
76 while (frames > s.capacity - s.used or s.count == s.entries.len) {
77 const e = &s.entries[s.head];
78 const n = e.frames - e.consumed;
79 s.retention_floor = try at(e.timestamp, e.frames, s.rate);
80 s.trimmed += n;
81 s.consume(e, n);
82 }
83 try s.push(data, timestamp);
84 }
85 /// Non-consuming sample window. Cursor and returned slices expire on any
86 /// mutation. Separate cursors can replay the same history independently.
87 pub fn window(s: *const Queue, minimum: u64, cut: u64, cursor: *Cursor) !?Chunk {
88 while (cursor.block < s.count) {
89 const e = s.entries[(s.head + cursor.block) % s.entries.len];
90 const begin = @max(@max(e.consumed, cursor.frame), before(e.timestamp, minimum, s.rate, e.frames));
91 const end = before(e.timestamp, cut, s.rate, e.frames);
92 if (begin >= end) {
93 if (end < e.frames) return null;
94 cursor.block += 1;
95 cursor.frame = 0;
96 continue;
97 }
98 const offset = (e.offset + begin) % s.capacity;
99 const n = @min(end - begin, s.capacity - offset);
100 cursor.frame = begin + n;
101 return .{ .data = s.data[offset * s.channels * 2 ..][0 .. n * s.channels * 2], .timestamp = try at(e.timestamp, begin, s.rate), .frames = n, .end_ns = try at(e.timestamp, begin + n, s.rate) };
102 }
103 return null;
104 }
105 /// Freeze a selected sample window into independent, minimally sized storage.
106 /// Caller serializes source mutation; destination may then move to a worker.
107 pub fn cloneWindow(s: *const Queue, a: std.mem.Allocator, minimum: u64, cut: u64) !Queue {
108 var cursor: Cursor = .{};
109 var frames: usize = 0;
110 var blocks: usize = 0;
111 while (try s.window(minimum, cut, &cursor)) |chunk| {
112 frames += chunk.frames;
113 blocks += 1;
114 }
115 // One wrapped source block can become two contiguous descriptors.
116 // A full source therefore needs at most entries.len + 1 descriptors.
117 const data = try a.alloc(u8, @max(frames, 1) * s.channels * 2);
118 errdefer a.free(data);
119 const entries = try a.alloc(Entry, @max(blocks, 1));
120 errdefer a.free(entries);
121 var result = Queue{ .allocator = a, .data = data, .entries = entries, .rate = s.rate, .channels = s.channels, .capacity = @max(frames, 1) };
122 cursor = .{};
123 while (try s.window(minimum, cut, &cursor)) |chunk| try result.push(chunk.data, chunk.timestamp);
124 return result;
125 }
126 /// Emit samples with time >= minimum and < cut. Wrap is returned as separate
127 /// contiguous chunks; never duplicate a sample at a segment boundary.
128 pub fn take(s: *Queue, minimum: u64, cut: u64) !?Chunk {
129 while (s.count != 0) {
130 const e = &s.entries[s.head];
131 const end = before(e.timestamp, cut, s.rate, e.frames);
132 const begin = @max(e.consumed, @min(end, before(e.timestamp, minimum, s.rate, e.frames)));
133 if (begin > e.consumed) {
134 const n = begin - e.consumed;
135 s.trimmed += n;
136 s.consume(e, n);
137 if (s.count == 0 or begin == e.frames) continue;
138 }
139 if (end <= e.consumed) return null;
140 const offset = (e.offset + e.consumed) % s.capacity;
141 const n = @min(end - e.consumed, s.capacity - offset);
142 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, .end_ns = try at(e.timestamp, e.consumed + n, s.rate) };
143 s.consume(e, n);
144 return result;
145 }
146 return null;
147 }
148};
149test "PCM splits across samples and arena wrap without loss or duplicate frames" {
150 var q = try Queue.init(std.testing.allocator, 48000, 1, 8, 4);
151 defer q.deinit();
152 try q.push(&.{ 1, 0, 2, 0, 3, 0, 4, 0, 5, 0, 6, 0 }, 1000000);
153 const first = (try q.take(1000000, 1062500)).?;
154 try std.testing.expectEqualSlices(u8, &.{ 1, 0, 2, 0, 3, 0 }, first.data);
155 try std.testing.expect((try q.take(1000000, 1062500)) == null);
156 const next = (try q.take(1062500, 1125000)).?;
157 try std.testing.expectEqualSlices(u8, &.{ 4, 0, 5, 0, 6, 0 }, next.data);
158 try q.push(&.{ 7, 0, 8, 0, 9, 0, 10, 0 }, 1125000);
159 const wrap = (try q.take(1125000, 2000000)).?;
160 try std.testing.expectEqualSlices(u8, &.{ 7, 0, 8, 0 }, wrap.data);
161 try std.testing.expectEqualSlices(u8, &.{ 9, 0, 10, 0 }, (try q.take(1125000, 2000000)).?.data);
162 try std.testing.expectEqual(@as(usize, 0), q.used);
163}
164test "PCM rejects full queues, invalid inputs and overflow before modifying history" {
165 var q = try Queue.init(std.testing.allocator, 48000, 1, 4, 2);
166 defer q.deinit();
167 try q.push(&.{ 1, 0, 2, 0, 3, 0, 4, 0 }, 1000000);
168 try std.testing.expectError(error.BufferFull, q.push(&.{ 5, 0 }, 2000000));
169 _ = try q.take(1041667, 2000000);
170 try std.testing.expectEqual(@as(u64, 3), q.trimmed);
171 try std.testing.expectError(error.InvalidPcm, q.push(&.{1}, 2000000));
172 try std.testing.expectError(error.InvalidTimestamp, q.push(&.{ 1, 0 }, 1));
173 try std.testing.expectError(error.TimestampOverflow, at(std.math.maxInt(u64), 1, 48000));
174 const previous = q.last_end.?;
175 try std.testing.expectError(error.InvalidTimestamp, q.push(&.{ 1, 0 }, previous - 1));
176 try std.testing.expectEqual(previous, q.last_end.?);
177 try std.testing.expectEqual(@as(u64, 1_000_000_000), try at(0, 44100, 44100));
178}
179test "PCM partial allocation failure cleanup" {
180 const Check = struct {
181 fn run(a: std.mem.Allocator) !void {
182 var q = try Queue.init(a, 48000, 2, 1024, 32);
183 defer q.deinit();
184 }
185 };
186 try std.testing.checkAllAllocationFailures(std.testing.allocator, Check.run, .{});
187}
188
189test "rolling PCM snapshots are repeatable across wrap and exact sample boundaries" {
190 var q = try Queue.init(std.testing.allocator, 44100, 1, 8, 3);
191 defer q.deinit();
192 try q.pushRolling(&.{ 1, 0, 2, 0, 3, 0, 4, 0, 5, 0 }, 1000000);
193 try q.pushRolling(&.{ 6, 0, 7, 0, 8, 0, 9, 0 }, try at(1000000, 5, 44100));
194 try std.testing.expectEqual(@as(u64, 5), q.trimmed);
195 try std.testing.expectEqual(@as(usize, 4), q.used);
196 const start = try at(try at(1000000, 5, 44100), 1, 44100);
197 for (0..2) |_| {
198 var cursor: Cursor = .{};
199 const a = (try q.window(start, std.math.maxInt(u64), &cursor)).?;
200 try std.testing.expectEqualSlices(u8, &.{ 7, 0, 8, 0 }, a.data);
201 const b = (try q.window(start, std.math.maxInt(u64), &cursor)).?;
202 try std.testing.expectEqualSlices(u8, &.{ 9, 0 }, b.data);
203 try std.testing.expect((try q.window(start, std.math.maxInt(u64), &cursor)) == null);
204 try std.testing.expectEqual(@as(usize, 4), q.used);
205 }
206 const floor = q.retention_floor;
207 try std.testing.expectError(error.InvalidTimestamp, q.pushRolling(&.{ 1, 0 }, 0));
208 var oversized: [18]u8 = undefined;
209 @memset(&oversized, 0);
210 try std.testing.expectError(error.BufferFull, q.pushRolling(&oversized, q.last_end.?));
211 try std.testing.expectEqual(floor, q.retention_floor);
212 try std.testing.expectEqual(@as(usize, 4), q.used);
213}
214test "rolling PCM metadata limit and window clipping preserve gaps" {
215 var q = try Queue.init(std.testing.allocator, 8000, 2, 16, 2);
216 defer q.deinit();
217 const block = [_]u8{ 1, 0, 2, 0, 3, 0, 4, 0 };
218 try q.pushRolling(&block, 0);
219 try q.pushRolling(&block, 1000000);
220 try q.pushRolling(&block, 2000000);
221 try std.testing.expectEqual(@as(u64, 2), q.trimmed);
222 var cursor: Cursor = .{};
223 const chunk = (try q.window(1125000, 2000000, &cursor)).?;
224 try std.testing.expectEqual(@as(usize, 1), chunk.frames);
225 try std.testing.expectEqual(@as(u64, 1250000), chunk.end_ns);
226 try std.testing.expect((try q.window(1125000, 2000000, &cursor)) == null);
227}
228
229test "frozen PCM windows own wrapped clipped samples and clean up allocation failures" {
230 const Check = struct {
231 fn run(a: std.mem.Allocator) !void {
232 var q = try Queue.init(a, 44100, 1, 6, 3);
233 defer q.deinit();
234 try q.pushRolling(&.{ 1, 0, 2, 0, 3, 0, 4, 0 }, 1000);
235 try q.pushRolling(&.{ 5, 0, 6, 0, 7, 0, 8, 0 }, try at(1000, 4, 44100));
236 const begin = try at(try at(1000, 4, 44100), 1, 44100);
237 var frozen = try q.cloneWindow(a, begin, std.math.maxInt(u64));
238 defer frozen.deinit();
239 try q.pushRolling(&.{ 9, 0, 10, 0, 11, 0, 12, 0 }, q.last_end.?);
240 var cursor: Cursor = .{};
241 var frames: usize = 0;
242 var values: [6]u8 = undefined;
243 while (try frozen.window(0, std.math.maxInt(u64), &cursor)) |chunk| {
244 @memcpy(values[frames * 2 ..][0..chunk.data.len], chunk.data);
245 frames += chunk.frames;
246 }
247 try std.testing.expectEqualSlices(u8, &.{ 6, 0, 7, 0, 8, 0 }, &values);
248 }
249 };
250 try std.testing.checkAllAllocationFailures(std.testing.allocator, Check.run, .{});
251}
252
253test "freeze supports a full descriptor queue with a wrapped final block" {
254 var q = try Queue.init(std.testing.allocator, 48000, 1, 8193, 4096);
255 defer q.deinit();
256 for (0..4097) |i| try q.pushRolling(&.{ 1, 0, 2, 0 }, try at(0, i * 2, 48000));
257 var frozen = try q.cloneWindow(std.testing.allocator, 0, std.math.maxInt(u64));
258 defer frozen.deinit();
259 try std.testing.expectEqual(@as(usize, 4096), q.count);
260 try std.testing.expectEqual(@as(usize, 4097), frozen.count);
261 try std.testing.expectEqual(q.used, frozen.used);
262}
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 = ;