Recording 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.

Usage Examples

Record MP4 (TsWriter)

from neoruntime_ipc_sdk import EncodedMediaClient, TsWriter

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

try:
    # Subscribe to the encoded stream and write frames by PTS
    for frame in media.subscribe("main"):
        writer.write(frame)
        if enough_recorded(frame):
            break
finally:
    writer.close()  # must close to finalize the moov/index

HLS live segmentation

from neoruntime_ipc_sdk import EncodedMediaClient, HlsWriter

media = EncodedMediaClient()

# One 6s segment, playlist keeps the latest 5 segments
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()

Preroll recording (pre-event cache)

from neoruntime_ipc_sdk import EncodedMediaClient, PrerollBuffer

media = EncodedMediaClient()

# Ring buffer holding the last 10s of encoded frames
preroll = PrerollBuffer(seconds=10.0)

for frame in media.subscribe("main"):
    preroll.push(frame)
    if event_detected():
        # dump: preroll contents + subsequent frames -> ts file
        writer = preroll.dump("/data/aipc/recordings/alert.ts")
        # ... keep appending post-event frames via writer.write(frame)
        writer.close()
        break

Custom frame callback

# dump's on_frame runs before each frame is written — use it to
# filter or modify frames
def mark(frame):
    print(f"writing pts={frame.pts_ns}")
    return frame

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