Claude came up with this:
# avi_recorder.py - MJPEG + PCM audio AVI recorder for the OpenMV N6
#
# Records camera frames as Motion-JPEG video and the on-board MEMS
# microphone as signed-16-bit PCM audio, interleaved into a single
# standard RIFF/AVI file that plays in VLC, ffmpeg, QuickTime, etc.
#
# Tested against the OpenMV N6 board interface:
# https://docs.openmv.io/v5.0.0/openmvcam/quickref/openmv-n6.html
#
# --------------------------------------------------------------------------
# Quick start
# --------------------------------------------------------------------------
# import csi, time
# from avi_recorder import AviWriter
#
# cam = csi.CSI()
# cam.reset()
# cam.pixformat(csi.RGB565) # writer JPEG-compresses each frame
# cam.framesize(csi.QVGA) # 320x240
#
# avi = AviWriter("recording.avi", 320, 240, quality=90,
# audio_frequency=16000, audio_channels=1)
#
# clock = time.clock()
# for _ in range(300): # ~record until you decide to stop
# clock.tick()
# avi.add_frame(cam.snapshot())
#
# avi.close() # finalizes header + index, stops audio
#
# Or as a context manager:
# with AviWriter("clip.avi", 320, 240) as avi:
# for _ in range(300):
# avi.add_frame(cam.snapshot())
#
# --------------------------------------------------------------------------
# Notes
# --------------------------------------------------------------------------
# * Video FPS is *measured* from wall-clock time between the first and last
# frame, then written back into the header on close(). You do not need to
# know the frame rate in advance.
# * Audio is captured asynchronously via audio.start_streaming(); buffers are
# queued and flushed (interleaved) just before each video frame is written,
# so A/V stays roughly in sync. Any tail audio is flushed on close().
# * add_frame() accepts either an OpenMV image.Image (RGB565/GRAYSCALE - it is
# JPEG compressed for you) or a bytes/bytearray of an already-encoded JPEG
# (e.g. when the sensor is configured for csi.JPEG pixformat: pass
# img.bytearray()).
# * Only 16-bit PCM is emitted (that is what the audio module produces).
import struct
import time
try:
import audio
except ImportError: # allows importing the module on a board without audio
audio = None
class AviWriter:
# AVIMAINHEADER.dwFlags
_AVIF_HASINDEX = 0x00000010
_AVIF_ISINTERLEAVED = 0x00000100
# AVIINDEXENTRY.dwFlags
_AVIIF_KEYFRAME = 0x00000010
def __init__(self, path, width, height, quality=90,
record_audio=True, audio_channels=1,
audio_frequency=16000, audio_gain_db=24):
self.path = path
self.width = int(width)
self.height = int(height)
self.quality = int(quality)
self.record_audio = bool(record_audio) and (audio is not None)
self.audio_channels = int(audio_channels)
self.audio_frequency = int(audio_frequency)
self.audio_gain_db = int(audio_gain_db)
self._bits = 16
self._block_align = self.audio_channels * (self._bits // 8)
self._avg_bps = self.audio_frequency * self._block_align
# running state
self.frames = 0
self._audio_bytes = 0
self._index = [] # list of (fourcc, flags, movi_offset, size)
self._movi_off = 0 # bytes written into the movi payload so far
self._audio_q = [] # queued PCM buffers from the streaming cb
self._max_jpeg = 0
self._max_audio = 0
self._t0 = None
self._t1 = None
self._closed = False
# Open file and reserve the (fixed-size) preamble. It is rewritten
# with the final rates/counts in close().
self._f = open(path, "wb")
self._preamble_len = self._write_preamble(0.0, 0, 0)
self._movi_data_start = self._f.tell()
if self.record_audio:
audio.init(channels=self.audio_channels,
frequency=self.audio_frequency,
gain_db=self.audio_gain_db)
audio.start_streaming(self._audio_cb)
# ------------------------------------------------------------------ #
# Public API
# ------------------------------------------------------------------ #
def add_frame(self, img, quality=None):
"""Append one video frame. `img` is an OpenMV image (compressed here)
or bytes/bytearray of an already-encoded JPEG."""
if self._closed:
raise OSError("AviWriter is closed")
now = time.ticks_us()
if self._t0 is None:
self._t0 = now
self._t1 = now
if isinstance(img, (bytes, bytearray, memoryview)):
data = img
else:
q = self.quality if quality is None else int(quality)
# compress() JPEG-encodes the frame in place and returns it.
data = img.compress(quality=q).bytearray()
# Interleave: flush audio captured up to this point, then the frame.
self._flush_audio()
self._write_chunk(b"00dc", data, keyframe=True)
self.frames += 1
n = len(data)
if n > self._max_jpeg:
self._max_jpeg = n
def close(self):
"""Stop audio, flush, write the index, patch the header, close file."""
if self._closed:
return
self._closed = True
if self.record_audio:
try:
audio.stop_streaming()
except Exception:
pass
self._flush_audio() # drain any tail buffers
self._write_index()
self._finalize()
self._f.close()
# context manager sugar
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
return False
# ------------------------------------------------------------------ #
# Audio capture
# ------------------------------------------------------------------ #
def _audio_cb(self, pcmbuf):
# The driver reuses pcmbuf, so copy the bytes out immediately.
self._audio_q.append(bytes(pcmbuf))
def _flush_audio(self):
if not self._audio_q:
return
# Detach the queue so a concurrent callback appends to a fresh list.
q = self._audio_q
self._audio_q = []
for buf in q:
n = len(buf)
if not n:
continue
self._write_chunk(b"01wb", buf, keyframe=False)
self._audio_bytes += n
if n > self._max_audio:
self._max_audio = n
# ------------------------------------------------------------------ #
# RIFF chunk writing
# ------------------------------------------------------------------ #
def _write_chunk(self, fourcc, data, keyframe):
n = len(data)
# idx1 offset is relative to the 'movi' FOURCC (which sits 4 bytes
# before the first payload chunk), hence +4.
off = self._movi_off + 4
f = self._f
f.write(fourcc)
f.write(struct.pack("<I", n))
f.write(data)
pad = n & 1 # chunks are word-aligned
if pad:
f.write(b"\x00")
self._index.append(
(fourcc, self._AVIIF_KEYFRAME if keyframe else 0, off, n))
self._movi_off += 8 + n + pad
def _write_index(self):
f = self._f
f.write(b"idx1")
f.write(struct.pack("<I", 16 * len(self._index)))
for fourcc, flags, off, size in self._index:
f.write(fourcc)
f.write(struct.pack("<III", flags, off, size))
# ------------------------------------------------------------------ #
# Header (preamble) - fixed size, written twice (placeholder + final)
# ------------------------------------------------------------------ #
def _finalize(self):
file_size = self._f.tell()
movi_list_size = 4 + self._movi_off # 'movi' fourcc + payload
riff_size = file_size - 8
if self.frames >= 2 and self._t0 is not None:
secs = time.ticks_diff(self._t1, self._t0) / 1000000.0
fps = (self.frames - 1) / secs if secs > 0 else 30.0
else:
fps = 30.0
self._f.seek(0)
self._write_preamble(fps, movi_list_size, riff_size)
self._f.seek(file_size)
def _write_preamble(self, fps, movi_list_size, riff_size):
hdrl = self._build_hdrl(fps)
buf = bytearray()
# RIFF header
buf += b"RIFF" + struct.pack("<I", riff_size) + b"AVI "
# hdrl LIST
buf += b"LIST" + struct.pack("<I", len(hdrl) + 4) + b"hdrl" + hdrl
# movi LIST header (payload follows this preamble)
buf += b"LIST" + struct.pack("<I", movi_list_size) + b"movi"
self._f.write(buf)
return len(buf)
def _build_hdrl(self, fps):
streams = 2 if self.record_audio else 1
micro = int(round(1000000.0 / fps)) if fps > 0 else 33333
rate = int(round(fps * 1000)) if fps > 0 else 30000
scale = 1000
max_bytes_per_sec = (self._avg_bps +
int(self._max_jpeg * (fps if fps > 0 else 30)))
suggested = max(self._max_jpeg, self._max_audio, 1)
# --- avih (AVIMAINHEADER, 56 bytes body) ---
avih_body = struct.pack(
"<14I",
micro, # dwMicroSecPerFrame
max_bytes_per_sec, # dwMaxBytesPerSec
0, # dwPaddingGranularity
self._AVIF_HASINDEX | self._AVIF_ISINTERLEAVED, # dwFlags
self.frames, # dwTotalFrames
0, # dwInitialFrames
streams, # dwStreams
suggested, # dwSuggestedBufferSize
self.width, # dwWidth
self.height, # dwHeight
0, 0, 0, 0) # dwReserved[4]
avih = b"avih" + struct.pack("<I", len(avih_body)) + avih_body
# --- video stream list (vids / MJPG) ---
vid_strh_body = struct.pack(
"<4s4sI2H8I4h",
b"vids", b"MJPG",
0, # dwFlags
0, 0, # wPriority, wLanguage
0, # dwInitialFrames
scale, # dwScale
rate, # dwRate (rate/scale = fps)
0, # dwStart
self.frames, # dwLength (frames)
max(self._max_jpeg, 1), # dwSuggestedBufferSize
0, # dwQuality
0, # dwSampleSize (0 = variable)
0, 0, self.width, self.height) # rcFrame
vid_strh = b"strh" + struct.pack("<I", len(vid_strh_body)) + vid_strh_body
vid_strf_body = struct.pack(
"<IiiHH4sIiiII",
40, # biSize
self.width, # biWidth
self.height, # biHeight
1, # biPlanes
24, # biBitCount
b"MJPG", # biCompression
self.width * self.height * 3, # biSizeImage
0, 0, # biX/YPelsPerMeter
0, 0) # biClrUsed, biClrImportant
vid_strf = b"strf" + struct.pack("<I", len(vid_strf_body)) + vid_strf_body
vid_strl_body = b"strl" + vid_strh + vid_strf
vid_strl = b"LIST" + struct.pack("<I", len(vid_strl_body)) + vid_strl_body
hdrl = avih + vid_strl
# --- audio stream list (auds / PCM) ---
if self.record_audio:
aud_blocks = self._audio_bytes // self._block_align
aud_strh_body = struct.pack(
"<4s4sI2H8I4h",
b"auds", b"\x00\x00\x00\x00",
0, # dwFlags
0, 0, # wPriority, wLanguage
0, # dwInitialFrames
self._block_align, # dwScale
self._avg_bps, # dwRate (rate/scale = samples/sec)
0, # dwStart
aud_blocks, # dwLength (in blocks/samples)
max(self._max_audio, 1), # dwSuggestedBufferSize
0, # dwQuality
self._block_align, # dwSampleSize
0, 0, 0, 0) # rcFrame (unused for audio)
aud_strh = b"strh" + struct.pack("<I", len(aud_strh_body)) + aud_strh_body
aud_strf_body = struct.pack(
"<HHIIHH",
1, # wFormatTag = WAVE_FORMAT_PCM
self.audio_channels, # nChannels
self.audio_frequency, # nSamplesPerSec
self._avg_bps, # nAvgBytesPerSec
self._block_align, # nBlockAlign
self._bits) # wBitsPerSample
aud_strf = b"strf" + struct.pack("<I", len(aud_strf_body)) + aud_strf_body
aud_strl_body = b"strl" + aud_strh + aud_strf
aud_strl = b"LIST" + struct.pack("<I", len(aud_strl_body)) + aud_strl_body
hdrl += aud_strl
return hdrl
Which you then use like:
import csi, time
from avi_recorder import AviWriter
cam = csi.CSI()
cam.reset()
cam.pixformat(csi.RGB565) # writer JPEG-compresses each frame
cam.framesize(csi.QVGA) # 320x240
avi = AviWriter("recording.avi", 320, 240, quality=90,
audio_frequency=16000, audio_channels=1)
record_secs = 10
target_fps = 30
frame_ms = 1000 // target_fps
start = time.ticks_ms()
while time.ticks_diff(time.ticks_ms(), start) < record_secs * 1000:
t = time.ticks_ms()
avi.add_frame(cam.snapshot())
dt = time.ticks_diff(time.ticks_ms(), t)
if dt < frame_ms:
time.sleep_ms(frame_ms - dt)
avi.close() # finalizes header + index, stops audio
A few design points worth flagging:
FPS is measured, not assumed. add_frame() timestamps with time.ticks_us(), and close() seeks back and rewrites the fixed-size preamble with the real frame rate, total frame count, and audio length — so you don’t have to declare a frame rate up front and it stays honest even if capture rate drifts.
Audio comes in through audio.start_streaming(self._audio_cb); the callback copies each buffer (bytes(pcmbuf), since the driver reuses it) into a queue, and the queue is flushed as 01wb chunks immediately before each 00dc video chunk. That keeps the interleave roughly time-ordered, and any tail buffers are drained after stop_streaming() in close(). Video length is derived from wall-clock time while audio length comes from the actual sample count, so the two durations line up.
add_frame() takes either an image.Image (RGB565/GRAYSCALE — it calls img.compress(quality=...).bytearray() for you) or raw JPEG bytes.
The index (idx1) is written with offsets relative to the movi FOURCC and AVIIF_KEYFRAME set on every JPEG frame, which is what makes it seekable in VLC/QuickTime rather than just linearly playable.
One thing to sanity-check on real hardware: I stubbed the audio module for the host test, so confirm your firmware build’s exact audio API (audio.init(channels=, frequency=, gain_db=) / start_streaming / stop_streaming) matches — the docs snippet I pulled shows exactly that signature, but if stop_streaming is named differently in v5.0.0 the close() call is the only spot to adjust.
Sources: OpenMV N6 board interface
recording.zip (20.9 MB)
