Adding data (aka json) to the py_lib (#1318)

Co-authored-by: claire <claire@rclaire.ai>
Co-authored-by: Luke Curley <kixelated@gmail.com>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: 4 people

justfile

@@ -68,7 +68,11 @@ ci:
 	# Run the Python checks.
 	uv run ruff check py/
 	uv run ruff format --check py/
-	uv run --package moq-lite pyright
+	# Sync moq-lite's dev group (pytest) first, then override moq-ffi with a source build;
+	# --no-sync on the pyright invocation keeps uv from reinstalling the registry moq-ffi.
+	uv sync --package moq-lite
+	uv run maturin develop -m rs/moq-ffi/Cargo.toml --uv
+	uv run --package moq-lite --no-sync pyright
 
 	# Run the tofu checks.
 	(cd cdn && just check)
@@ -120,8 +124,9 @@ test *args:
 
 	# Run the Python tests.
 	if command -v uv &> /dev/null; then
+		uv sync --package moq-lite
 		uv run maturin develop -m rs/moq-ffi/Cargo.toml --uv
-		uv run --package moq-lite pytest py/moq-lite/tests/
+		uv run --package moq-lite --no-sync pytest py/moq-lite/tests/
 	fi
 
 # Automatically fix some issues.

py/moq-lite/moq_lite/__init__.py

@@ -2,8 +2,8 @@
 
 from .client import Client
 from .origin import Announced, AnnouncedBroadcast, Announcement, OriginConsumer, OriginProducer
-from .publish import BroadcastProducer, MediaProducer
-from .subscribe import BroadcastConsumer, CatalogConsumer, Container, MediaConsumer
+from .publish import BroadcastProducer, GroupProducer, MediaProducer, TrackProducer
+from .subscribe import BroadcastConsumer, CatalogConsumer, Container, GroupConsumer, MediaConsumer, TrackConsumer
 from .types import Audio, Catalog, Dimensions, Frame, Video
 
 __all__ = [
@@ -19,9 +19,13 @@
     "Client",
     "Dimensions",
     "Frame",
+    "GroupConsumer",
+    "GroupProducer",
     "MediaConsumer",
     "MediaProducer",
     "OriginConsumer",
     "OriginProducer",
+    "TrackConsumer",
+    "TrackProducer",
     "Video",
 ]

py/moq-lite/moq_lite/publish.py

@@ -2,7 +2,12 @@
 
 from __future__ import annotations
 
-from moq_ffi import MoqBroadcastProducer, MoqMediaProducer
+from typing import TYPE_CHECKING
+
+from moq_ffi import MoqBroadcastProducer, MoqGroupProducer, MoqMediaProducer, MoqTrackProducer
+
+if TYPE_CHECKING:
+    from .subscribe import BroadcastConsumer, GroupConsumer, TrackConsumer
 
 
 class MediaProducer:
@@ -18,6 +23,57 @@ def finish(self) -> None:
         self._inner.finish()
 
 
+class GroupProducer:
+    """Writes frames into a single group on a track."""
+
+    def __init__(self, inner: MoqGroupProducer) -> None:
+        self._inner = inner
+
+    @property
+    def sequence(self) -> int:
+        """The sequence number of this group within the track."""
+        return self._inner.sequence()
+
+    def consume(self) -> GroupConsumer:
+        """Create a consumer that reads frames from this group."""
+        from .subscribe import GroupConsumer
+
+        return GroupConsumer(self._inner.consume())
+
+    def write_frame(self, payload: bytes) -> None:
+        self._inner.write_frame(payload)
+
+    def finish(self) -> None:
+        self._inner.finish()
+
+
+class TrackProducer:
+    """Track producer — write arbitrary byte payloads with no codec required.
+
+    Same pattern as moq-boy's status/command tracks.
+    """
+
+    def __init__(self, inner: MoqTrackProducer) -> None:
+        self._inner = inner
+
+    def append_group(self) -> GroupProducer:
+        """Start a new group; write frames into it, then finish()."""
+        return GroupProducer(self._inner.append_group())
+
+    def write_frame(self, payload: bytes) -> None:
+        """Convenience: write a single-frame group in one call."""
+        self._inner.write_frame(payload)
+
+    def consume(self) -> TrackConsumer:
+        """Create a consumer that reads directly from this producer's track."""
+        from .subscribe import TrackConsumer
+
+        return TrackConsumer(self._inner.consume())
+
+    def finish(self) -> None:
+        self._inner.finish()
+
+
 class BroadcastProducer:
     """Wraps MoqBroadcastProducer with a cleaner interface."""
 
@@ -27,5 +83,15 @@ def __init__(self) -> None:
     def publish_media(self, format: str, init: bytes) -> MediaProducer:
         return MediaProducer(self._inner.publish_media(format, init))
 
+    def publish_track(self, name: str) -> TrackProducer:
+        """Create a track — send any bytes, no codec validation."""
+        return TrackProducer(self._inner.publish_track(name))
+
+    def consume(self) -> BroadcastConsumer:
+        """Create a consumer that reads from this broadcast's tracks."""
+        from .subscribe import BroadcastConsumer
+
+        return BroadcastConsumer(self._inner.consume())
+
     def finish(self) -> None:
         self._inner.finish()

py/moq-lite/moq_lite/subscribe.py

@@ -2,7 +2,14 @@
 
 from __future__ import annotations
 
-from moq_ffi import Container, MoqBroadcastConsumer, MoqCatalogConsumer, MoqMediaConsumer
+from moq_ffi import (
+    Container,
+    MoqBroadcastConsumer,
+    MoqCatalogConsumer,
+    MoqGroupConsumer,
+    MoqMediaConsumer,
+    MoqTrackConsumer,
+)
 
 from .types import Catalog, Frame
 
@@ -26,6 +33,84 @@ def cancel(self) -> None:
         self._inner.cancel()
 
 
+class GroupConsumer:
+    """Async iterator of byte payloads within a single group."""
+
+    def __init__(self, inner: MoqGroupConsumer) -> None:
+        self._inner = inner
+
+    @property
+    def sequence(self) -> int:
+        """The sequence number of this group within the track."""
+        return self._inner.sequence()
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self) -> bytes:
+        frame = await self._inner.read_frame()
+        if frame is None:
+            raise StopAsyncIteration
+        return frame
+
+    def cancel(self) -> None:
+        self._inner.cancel()
+
+
+class TrackConsumer:
+    """Async iterator of groups from a track.
+
+    Each group is itself an async iterator of byte payloads. Same pattern as
+    moq-boy's status/command tracks (one frame per group), but multi-frame
+    groups are also supported.
+    """
+
+    def __init__(self, inner: MoqTrackConsumer) -> None:
+        self._inner = inner
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self) -> GroupConsumer:
+        group = await self.recv_group()
+        if group is None:
+            raise StopAsyncIteration
+        return group
+
+    async def recv_group(self) -> GroupConsumer | None:
+        """Return the next group in arrival order. Returns `None` when the track ends.
+
+        Groups are returned as they arrive on the wire, which may be out of sequence
+        order. Use this for live consumption where latency matters more than order.
+        """
+        group = await self._inner.recv_group()
+        if group is None:
+            return None
+        return GroupConsumer(group)
+
+    async def next_group(self) -> GroupConsumer | None:
+        """Return the next group in sequence order, skipping forward if behind.
+
+        Returns `None` when the track ends. Use this when order matters more than
+        latency; `recv_group` is preferred for live consumption.
+        """
+        group = await self._inner.next_group()
+        if group is None:
+            return None
+        return GroupConsumer(group)
+
+    async def read_frame(self) -> bytes | None:
+        """Read the first frame of the next group.
+
+        Convenience for tracks using one-frame-per-group (like moq-boy's
+        status/command tracks). Returns `None` when the track ends.
+        """
+        return await self._inner.read_frame()
+
+    def cancel(self) -> None:
+        self._inner.cancel()
+
+
 class CatalogConsumer:
     """Wraps MoqCatalogConsumer as an async iterator of Catalog."""
 
@@ -54,6 +139,10 @@ def __init__(self, inner: MoqBroadcastConsumer) -> None:
     def subscribe_catalog(self) -> CatalogConsumer:
         return CatalogConsumer(self._inner.subscribe_catalog())
 
+    def subscribe_track(self, name: str) -> TrackConsumer:
+        """Subscribe to a track — receive arbitrary byte payloads."""
+        return TrackConsumer(self._inner.subscribe_track(name))
+
     def subscribe_media(self, name: str, container: Container, max_latency_ms: int) -> MediaConsumer:
         return MediaConsumer(self._inner.subscribe_media(name, container, max_latency_ms))
 

py/moq-lite/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "moq-lite"
-version = "0.0.2"
+version = "0.0.3"
 description = "Ergonomic Python wrapper for MoQ (Media over QUIC)"
 readme = "README.md"
 license = "MIT OR Apache-2.0"