录像 API

Recording utilities - pure-python MPEG-TS muxing for EncodedFrame streams.

TsWriter writes a single .ts event clip; HlsWriter cuts keyframe-aligned segments plus an m3u8 live playlist; PrerollBuffer keeps a ring of encoded frames so an app can dump “seconds before the event” when something happens.

No ffmpeg is required: Annex-B payloads from the EncodedPublisher are packetised into PES/TS directly, without re-encoding.

with HlsWriter("/var/tmp/hls", segment_seconds=6.0) as hls:
    for frame in client.subscribe_encoded("main"):
        hls.write(frame)
# serve /var/tmp/hls/index.m3u8 over HTTP for hls.js

class neoruntime_ipc_sdk.recording.TsWriter(path, codec='h264')

Bases: object

Single-file MPEG-TS recorder for EncodedFrame Annex-B payloads.

w = TsWriter("event.ts", codec="h264")
for frame in client.subscribe_encoded("main"):
    w.write(frame)
w.close()

__init__(path, codec='h264')

write(frame)

Packetise one encoded frame into the TS stream.

close()

class neoruntime_ipc_sdk.recording.HlsWriter(out_dir, segment_seconds=6.0, window=5, codec='h264')

Bases: object

Keyframe-aligned HLS segmenter with a live-window m3u8 playlist.

Every segment is self-contained (PAT+PMT first) so hls.js can join the stream at any segment boundary. The playlist is rewritten atomically (tmp + os.replace) after every segment close.

hls = HlsWriter("/srv/hls", segment_seconds=6.0, window=5)
for frame in client.subscribe_encoded("main"):
    hls.write(frame)      # blocks of frames -> seg000001.ts ...
hls.close()               # appends #EXT-X-ENDLIST

__init__(out_dir, segment_seconds=6.0, window=5, codec='h264')

write(frame)

Append a frame; cuts a segment on keyframes once it is long enough.

close()

Finalise the last segment and end the playlist.

class neoruntime_ipc_sdk.recording.PrerollBuffer(seconds=10.0)

Bases: object

Ring buffer of encoded frames for “seconds before the event” clips.

preroll = PrerollBuffer(seconds=10.0)
for frame in client.subscribe_encoded("main"):
    preroll.push(frame)
    if event_detected:
        writer = preroll.dump("event.ts")
        # keep writing live frames to `writer`, then writer.close()

__init__(seconds=10.0)

push(frame)

Add a frame and evict frames older than the window.

property frames: list[EncodedFrame]

dump(path, on_frame=None)

Flush buffered preroll into a new .ts file and return the writer.

on_frame(writer) is called right after the preroll flush - the typical use is starting a live-tail thread that keeps calling writer.write() until the event window ends.

TsWriter

class neoruntime_ipc_sdk.TsWriter(path, codec='h264')

Single-file MPEG-TS recorder for EncodedFrame Annex-B payloads.

w = TsWriter("event.ts", codec="h264")
for frame in client.subscribe_encoded("main"):
    w.write(frame)
w.close()

__init__(path, codec='h264')

write(frame)

Packetise one encoded frame into the TS stream.

close()

HlsWriter

class neoruntime_ipc_sdk.HlsWriter(out_dir, segment_seconds=6.0, window=5, codec='h264')

Keyframe-aligned HLS segmenter with a live-window m3u8 playlist.

Every segment is self-contained (PAT+PMT first) so hls.js can join the stream at any segment boundary. The playlist is rewritten atomically (tmp + os.replace) after every segment close.

hls = HlsWriter("/srv/hls", segment_seconds=6.0, window=5)
for frame in client.subscribe_encoded("main"):
    hls.write(frame)      # blocks of frames -> seg000001.ts ...
hls.close()               # appends #EXT-X-ENDLIST

__init__(out_dir, segment_seconds=6.0, window=5, codec='h264')

write(frame)

Append a frame; cuts a segment on keyframes once it is long enough.

close()

Finalise the last segment and end the playlist.

PrerollBuffer

class neoruntime_ipc_sdk.PrerollBuffer(seconds=10.0)

Ring buffer of encoded frames for “seconds before the event” clips.

preroll = PrerollBuffer(seconds=10.0)
for frame in client.subscribe_encoded("main"):
    preroll.push(frame)
    if event_detected:
        writer = preroll.dump("event.ts")
        # keep writing live frames to `writer`, then writer.close()

__init__(seconds=10.0)

push(frame)

Add a frame and evict frames older than the window.

property frames: list[EncodedFrame]

dump(path, on_frame=None)

Flush buffered preroll into a new .ts file and return the writer.

on_frame(writer) is called right after the preroll flush - the typical use is starting a live-tail thread that keeps calling writer.write() until the event window ends.

使用示例

录制 MP4（TsWriter）

from neoruntime_ipc_sdk import EncodedMediaClient, TsWriter

media = EncodedMediaClient()
writer = TsWriter("/data/aipc/recordings/clip.mp4", codec="h264")

try:
    # 订阅编码流，按 PTS 写入
    for frame in media.subscribe("main"):
        writer.write(frame)
        if enough_recorded(frame):
            break
finally:
    writer.close()  # 必须 close 以落盘 moov/索引

HLS 直播切片

from neoruntime_ipc_sdk import EncodedMediaClient, HlsWriter

media = EncodedMediaClient()

# 每 6s 一个分片，保留最近 5 个分片的播放列表
hls = HlsWriter("/data/aipc/recordings/live", segment_seconds=6.0, window=5)

try:
    for frame in media.subscribe("main"):
        hls.write(frame)
finally:
    hls.close()

预滚录像（事件前缓存）

from neoruntime_ipc_sdk import EncodedMediaClient, PrerollBuffer

media = EncodedMediaClient()

# 环形缓存最近 10s 编码帧
preroll = PrerollBuffer(seconds=10.0)

for frame in media.subscribe("main"):
    preroll.push(frame)
    if event_detected():
        # dump：预滚内容 + 后续帧写成 ts 文件
        writer = preroll.dump("/data/aipc/recordings/alert.ts")
        # ... 继续向 writer.write(frame) 追加事件后内容
        writer.close()
        break

自定义帧回调

# dump 的 on_frame 在写出每一帧前被调用，可用来过滤/修改
def mark(frame):
    print(f"writing pts={frame.pts_ns}")
    return frame

writer = preroll.dump("/data/aipc/recordings/alert.ts", on_frame=mark)