pymembus

Python bindings for libmembus, a small shared-memory IPC library.

pymembus is useful when multiple local processes need to exchange data with low overhead:

• raw shared memory blocks with memmap
• broadcast message queues with memmsg
• command channels with memcmd
• fixed-schema shared key/value state with memkv
• video ring buffers with memvid
• audio ring buffers with memaud
• variable-length packetized record rings with mempkt
• simple readiness polling with select()

All shared-memory names in the examples use POSIX-style names such as "/myshare". On Linux, stale objects can remain after a crash; each type has a remove(name) helper for cleanup.

Contents

• Install
• Build From Source
• Run Tests
• Quick Start
• API Guide
• Diagnostics
• Command Line Helpers
• Troubleshooting
• Development Notes

Install

From PyPI:

python3 -m pip install pymembus

If you build from source, install system build dependencies first. On Debian or Ubuntu:

sudo apt-get update
sudo apt-get install -y build-essential git cmake libboost-all-dev
sudo apt-get install -y python3 python3-pip

Optional tools used by this repository's CMake documentation targets:

sudo apt-get install -y doxygen graphviz go-md2man

Build From Source

Install this checkout into your active Python environment:

python3 -m pip install .

Source wheels built from this repository install both the Python extension and the bundled libmembus shared library into the pymembus package directory. The extension uses a package-relative runtime path, so no LD_LIBRARY_PATH setup is needed after installation.

Or build with CMake directly:

cmake -S . -B ./bld -DCMAKE_BUILD_TYPE=Release
cmake --build ./bld -j

The source build fetches pinned third-party dependencies when needed:

• libmembus v2.1.0
• pybind11 v2.13.6

libmembus 2.1.0 requires C++20 and Boost stacktrace support. The Python build requires CMake 3.30 or newer.

To uninstall a pip install:

python3 -m pip uninstall -y pymembus

To build distribution artifacts:

python3 setup.py sdist
python3 setup.py bdist_wheel

Run Tests

The test suite uses pytest. If you built with CMake:

cmake -S . -B ./bld -DCMAKE_BUILD_TYPE=Release
cmake --build ./bld --target pymembus-test

You can also run pytest from the repository root:

python3 -m pytest -v

Plain pytest imports either an installed pymembus package or the module built under bld/lib. If neither exists yet, run the CMake build above or install the checkout with python3 -m pip install . first.

Wheel builds made through scikit-build do not run the CMake test target during install; run tests explicitly with one of the commands above.

The pytest configuration in pyproject.toml limits collection to src/pytest/py and adds bld/lib to PYTHONPATH, so it will not try to run vendored tests under bld/_deps.

The suite covers maps, messages, commands, key/value stores, video/audio formats, NumPy buffer sharing, diagnostics, and select(). NumPy-specific tests are skipped when NumPy is not installed.

Quick Start

import pymembus

if hasattr(pymembus, "pymembus"):
    pymembus = pymembus.pymembus

name = "/quickstart"
pymembus.memmsg.remove(name)

tx = pymembus.memmsg()
rx = pymembus.memmsg()

assert tx.open(name, 1024, True, True)      # writer, create
assert rx.open(name, 1024, False, False)    # reader, attach

assert tx.write("hello")
message, overrun = rx.read_with_overrun(0)

assert message == "hello"
assert not overrun

rx.close()
tx.close()
pymembus.memmsg.remove(name)

API Guide

The snippets below assume the normalized import pattern from the quick start:

import pymembus

if hasattr(pymembus, "pymembus"):
    pymembus = pymembus.pymembus

Raw Shared Memory: memmap

memmap gives direct access to a named shared-memory block.

import pymembus

name = "/my_map"
pymembus.memmap.remove(name)

writer = pymembus.memmap()
reader = pymembus.memmap()

assert writer.open(name, 1024, True, True)
assert writer.write("hello") == 5

assert reader.open(name, 0, False, False, True)  # read-only attach
assert reader.read(5) == "hello"

view = memoryview(writer)
assert view.shape == (1024,)
assert not view.readonly

readonly_view = memoryview(reader)
assert readonly_view.readonly

del readonly_view
del view
reader.close()
writer.close()
pymembus.memmap.remove(name)

Parameters for memmap.open(name, size, create=False, new=False, read_only=False):

• create: create the object if it does not exist
• new: remove any existing object first
• read_only: attach without write permissions

Broadcast Messages: memmsg

memmsg is a single-writer, multi-reader message queue. Every reader receives every message independently.

name = "/my_messages"
pymembus.memmsg.remove(name)

tx = pymembus.memmsg()
rx1 = pymembus.memmsg()
rx2 = pymembus.memmsg()

assert tx.open(name, 4096, True, True)
assert rx1.open(name, 4096, False, False)
assert rx2.open(name, 4096, False, False)

assert tx.write("frame-ready")
assert rx1.read_with_overrun(0) == ("frame-ready", False)
assert rx2.read_with_overrun(0) == ("frame-ready", False)

tx.close()
rx1.close()
rx2.close()

Use poll() for non-blocking readiness checks:

if rx1.poll():
    msg = rx1.read(0)

When a reader falls behind far enough that the writer overwrites unread messages, read_with_overrun() returns ("", True).

For binary payloads, use read_bytes() or read_bytes_with_overrun():

assert tx.write_bytes(b"\xff\x00\x80")
payload, overrun = rx1.read_bytes_with_overrun(wait=0)
assert payload == b"\xff\x00\x80"

Both methods accept the same wait argument as the text reads. An empty read returns b""; the overrun variant returns (b"", True) on overrun. write() already accepts both str and bytes; write_bytes() explicitly accepts bytes. Empty messages remain invalid. Existing read() and read_with_overrun() return UTF-8 text. A decoding error consumes the message for that reader, so choose a binary read before reading arbitrary bytes.

The same binary methods are available on memcmd. Raw memmap objects also provide read_bytes(sz=-1); their existing write() accepts bytes.

Command Channels: memcmd

memcmd is a multi-writer command channel. It is useful for control paths, for example sending commands from a UI process to a capture process.

name = "/camera_commands"
pymembus.memcmd.remove(name)

receiver = pymembus.memcmd()
sender = pymembus.memcmd()

assert receiver.open(name, 1024, True, True)  # bReader=True, bCreate=True
assert sender.open(name, 1024)

assert sender.write("pan-left")
cmd, overrun = receiver.read_with_overrun(0)
assert cmd == "pan-left"
assert not overrun

sender.close()
receiver.close()

Shared State: memkv

memkv is a fixed-schema key/value store. The owner creates the store and sets slot names; any process can then read or write values.

name = "/camera_state"
pymembus.memkv.remove(name)

owner = pymembus.memkv()
assert owner.create(name, 3, 15, 63, True)
assert owner.setName(0, "mode")
assert owner.setName(1, "count")
assert owner.setName(2, "status")

peer = pymembus.memkv()
assert peer.open(name)

assert peer.setValue("mode", "auto")
value, stale = peer.getValue("mode")
assert value == "auto"
assert not stale

epoch = peer.getEpoch()
assert owner.setValue("status", "ready")
changed, epoch = peer.getChanged(epoch)
assert changed == {"status": "ready"}

peer.close()
owner.close()

getValue() returns (value, stale). stale is true if the lock-free read did not settle before its retry limit.

maxNameLen and maxValueLen must produce an aligned slot layout. In practice, choose values where maxNameLen + maxValueLen + 2 is divisible by 8, such as 15, 63 or 8, 14.

Video Ring Buffers: memvid

memvid stores packed video frames in a shared ring buffer. Use video_format values instead of old numeric bits-per-pixel values.

name = "/video"
pymembus.memvid.remove(name)

video = pymembus.memvid()
assert video.open(
    name,
    True,
    640,
    480,
    pymembus.video_format.rgb24,
    30,
    4,
)

slot = video.getPtr(0)
assert video.setVpts(slot, 123456)
assert video.setApts(slot, 123000)
assert video.next(1) == 1

assert video.getSeq() == 1
assert video.getFrameSeq(slot) == 1
assert video.getFormatName() == "RGB24"

frame = memoryview(video[slot])
assert frame.shape == (480, 640, 3)

del frame
video.close()
pymembus.memvid.remove(name)

Supported video formats:

• gray8
• rgb24
• bgr24
• rgba32
• bgra32
• yuyv422
• uyvy422
• userType (opaque / custom fixed-size format; geometry supplied by the caller via scanwidth, identity carried in fourcc / guid)

memvid.open() accepts optional trailing parameters after bufs: scanwidth, align, frameextra, fourcc, guid (16 bytes), and meta. See Stream Identity And Per-Frame Metadata.

NumPy can view frame buffers without copying:

import numpy as np

frame = np.array(video[slot], copy=False)
frame[10, 10] = [255, 0, 0]

del frame
video.close()
pymembus.memvid.remove(name)

Audio Ring Buffers: memaud

memaud stores PCM audio buffers in a shared ring buffer. Use audio_format values instead of old numeric bits-per-sample values.

name = "/audio"
pymembus.memaud.remove(name)

audio = pymembus.memaud()
assert audio.open(
    name,
    True,
    2,
    pymembus.audio_format.s16le,
    48000,
    50,
    4,
)

slot = audio.getPtr(0)
assert audio.setPts(slot, 123456)
assert audio.next(1) == 1

assert audio.getChannels() == 2
assert audio.getSampleRate() == 48000
assert audio.getFormatName() == "S16LE"

buf = memoryview(audio[slot])
assert buf.shape == (960, 2)

del buf
audio.close()
pymembus.memaud.remove(name)

Supported audio formats:

• u8
• s16le
• s24le
• s32le
• f32le
• f64le
• userType (opaque / custom fixed-size format; payload size supplied by the caller)

s24le is exposed as raw bytes because Python and NumPy do not have a native 24-bit integer scalar type. memaud.open() accepts the same optional trailing identity parameters as memvid.open() (align, frameextra, fourcc, guid, meta) plus payloadSize for audio_format.userType — see Stream Identity And Per-Frame Metadata.

Stream Identity And Per-Frame Metadata

memvid, memaud, and mempkt share a common set of identity and metadata features backed by the libmembus 2.1.0 header layout.

Stream identity is set at create time and read back on either side:

guid = bytes(range(16))  # any 16-byte identity, or omit for none

video = pymembus.memvid()
video.open(
    "/id_demo", True, 640, 480, pymembus.video_format.rgb24, 30, 4,
    fourcc=0x34363248,          # 'H264' little-endian fourcc, for example
    guid=guid,                  # 16-byte GUID (raises ValueError if not 16 bytes)
    meta=b"stream-header",      # opaque main user buffer copied in at create
)

assert video.getFourcc() == 0x34363248
assert video.getGuid() == guid            # returns None when no GUID was set
assert video.getMeta() == b"stream-header"  # returns None when no meta was set
assert video.getMetaSize() == len(b"stream-header")
assert video.getVersion() >= 2
assert video.getAlign() >= 8

Each frame or sample buffer can also carry a small per-frame user blob when the ring is created with frameextra > 0:

video.open("/user_demo", True, 640, 480, pymembus.video_format.rgb24, 30, 4,
           frameextra=64)

assert video.setUserData(0, b"caption")
assert video.getUserData(0) == b"caption"   # None when frameextra == 0
assert video.getUserLen(0) == len(b"caption")

getFrameExtra() reports the per-frame capacity. setUserData() copies up to getFrameExtra() bytes. The same fourcc / guid / meta / frameextra / setUserData / getUserData API is available on memaud; mempkt carries the equivalent per-record metadata through the meta argument of write() and the returned info dict.

For opaque fixed-size audio payloads, pass audio_format.userType and provide payloadSize:

audio.open("/opaque_audio", True, 1, pymembus.audio_format.userType,
           48000, 50, 4, payloadSize=4096, fourcc=0x55534552)

assert audio.getBytesPerSample() == 0
assert audio.getBufSize() == 4096
assert memoryview(audio.getBuf(0)).shape == (4096,)

Packetized Records: mempkt

mempkt is a single-writer, multi-reader ring for variable-length, opaque records. Unlike memvid/memaud, which store fixed-size frames, mempkt is meant for compressed or packetized streams such as MJPEG, H.264 access units, RTSP payloads, or muxed A/V. A fixed descriptor ring provides O(1) addressing and overrun detection while the variable payloads live in a separate packed byte arena.

name = "/packets"
pymembus.mempkt.remove(name)

writer = pymembus.mempkt()
reader = pymembus.mempkt()

# open(name, create, bufs, arenasz, maxrec, align=0, fourcc=0, meta=b"")
assert writer.open(name, True, 8, 1 << 20, 65536, 0, 0, b"stream-header")
assert reader.open_existing(name)

# write(payload, kind=pkt_kind.data, track=0, pts=0, meta=b"")
idx = writer.write(b"\x00\x01\x02frame-bytes",
                   pymembus.pkt_kind.video, track=0, pts=123456,
                   meta=b"per-record-metadata")

# getRecord() returns None on a torn/lapped/invalid read, otherwise
# (payload: bytes, meta: bytes, info: dict).
record = reader.getRecord(idx - 1)
assert record is not None
payload, meta, info = record

assert payload == b"\x00\x01\x02frame-bytes"
assert meta == b"per-record-metadata"
assert info["kind"] == pymembus.pkt_kind.video
assert info["pts"] == 123456

reader.close()
writer.close()
pymembus.mempkt.remove(name)

open() parameters:

• bufs: number of descriptor slots in the ring
• arenasz: payload arena size in bytes; size it with headroom (several times maxrec) so slow readers are not livelocked
• maxrec: largest single record (payload plus per-record metadata) accepted
• align: record alignment (power of two, >= 8); 0 selects the default
• fourcc: optional 32-bit stream identity (0 = none)
• meta: optional main user buffer copied in at create, read back with getMeta()

write() accepts either bytes or str payloads and returns the new descriptor write-pointer slot index, or -1 on failure (not open, read-only, or payload + meta exceeds maxrec). Record kinds are pkt_kind.data, pkt_kind.video, and pkt_kind.audio.

Because the arena is overwritten in place, readers must copy a record out and then re-check the arena write cursor to confirm the bytes were not lapped mid-copy. getRecord() does this for you and returns None when a record was torn, lapped, or the slot index is out of bounds. The returned info dict carries seq, wcursor, pts, kind, track, len, and userlen.

Use waitForFrame(wait_ms, lastSeq) to block until getSeq() advances past lastSeq, and getPtr(offset) to compute a slot index relative to the current write pointer (for example getPtr(-1) for the most recently written slot).

Buffer Lifetime And Read-Only Views

memmap, memvid, and memaud expose Python's buffer protocol. A writer that created a share exports writable buffers. A reader opened read-only, or with open_existing() for video/audio, exports read-only buffers.

Keep one lifetime rule in mind for exported buffers: release memoryview or NumPy arrays before calling close() on the owning object. pymembus intentionally raises RuntimeError if a video or audio mapping is closed while exported frame buffers still exist. For memmap, close() marks the object closed and defers the actual unmap until exported views are gone, so existing views do not point at unmapped shared memory.

Waiting On Multiple Sources: select()

select(wait_ms, conditions) polls a list of Python callables and returns the zero-based index of the first ready condition, or -1 on timeout.

idx = pymembus.select(100, [
    lambda: video.getSeq() > last_video_seq,
    lambda: commands.poll(),
])

if idx == 0:
    read_video_frame()
elif idx == 1:
    handle_command()

The conditions should be cheap, non-consuming readiness checks.

Diagnostics

Most API calls return False, -1, or an empty string on failure. Check last_error() or last_error_message() immediately after a failed call:

missing = pymembus.memmap()

if not missing.open("/does-not-exist", 0, False):
    assert pymembus.last_error() == pymembus.errc.open_failed
    print(pymembus.last_error_message())

Common error codes include:

• open_failed
• create_failed
• map_failed
• size_mismatch
• invalid_layout
• not_open
• access_denied
• message_too_large
• lock_timeout
• timeout
• overrun

Command Line Helpers

After installation, the package may install a pymembus helper command on Linux:

pymembus help
pymembus files
pymembus info version
sudo pymembus uninstall

pymembus info <variable> accepts values such as name, description, url, version, build, company, author, lib, include, bin, and share.

Troubleshooting

ModuleNotFoundError: No module named 'pymembus'

Build the extension first:

cmake -S . -B ./bld -DCMAKE_BUILD_TYPE=Release
cmake --build ./bld -j

Then run tests from the repository root:

python3 -m pytest -v

The repository's pytest config adds bld/lib to PYTHONPATH.

ImportError: liblibmembus.so: cannot open shared object file

This means an older or incomplete install has the Python extension but not its native libmembus dependency. Reinstall from the current checkout:

python3 -m pip install --force-reinstall .

Current source installs include liblibmembus.so in the pymembus package directory and set the extension runpath to load it from there.

Pytest tries to run pybind11 tests

This happens when pytest recursively scans build artifacts. The repository config sets:

norecursedirs = ["bld", "_skbuild", "dist", "build", ".git", "*.egg-info"]

If you use a custom pytest command, prefer:

python3 -m pytest -v src/pytest/py

A share already exists or has an invalid layout

Remove stale shared-memory objects before creating a fresh one:

pymembus.memmsg.remove("/my_messages")
pymembus.memvid.remove("/video")
pymembus.memaud.remove("/audio")

The 2.1.0 libmembus wire formats validate shared-memory headers. Old shares created by earlier library versions may be rejected with invalid_layout.

Doxygen or Graphviz warnings during build

The CMake build may emit documentation warnings from Doxygen or Graphviz. They do not affect the Python extension or pytest suite.

Development Notes

• Project metadata lives in PROJECT.txt.
• The Python extension source is in src/py/cpp/main.cpp.
• Tests live in src/pytest/py/test.py.
• The fallback libmembus dependency is configured in src/libmembus.cmake.
• See UPDATE.md for the libmembus 2.1.0 migration report.

References

• Python
• CMake
• pip
• git
• Boost
• pybind11
• libmembus