g2g-python 0.6.0

# Minimal stdlib-only hosted element for the M198 step-2 zero-copy test.
# Stands in for a gst-python-ml `backend/g2g` element shell: it receives the
# frame as a writable buffer-protocol object and proves an in-place write
# reaches the Rust Frame's memory, with no numpy / cv2 dependency.


class EchoTransform:
    """Bumps the first byte in place and echoes geometry/format as a blob."""

    def g2g_process(self, buf, width, height, fmt, meta):
        mv = memoryview(buf)
        assert not mv.readonly, "frame buffer must be writable"
        assert mv.nbytes == width * height * 4, "expected RGBA geometry"
        # In-place mutation: this write lands directly in the Rust frame buffer.
        mv[0] = (mv[0] + 1) % 256
        # Attach an analytics result through the sink (the AnalyticsBackend
        # mirror): label id 7, a box, and a confidence.
        meta.add_object(7, 1.0, 2.0, 3.0, 4.0, 0.9)
        # Attach opaque tagged side-data (the FrameIO.append_blob mirror), e.g.
        # an embedding's serialized bytes.
        meta.add_blob("embedding", bytes([1, 2, 3, 4]))

    def g2g_process_batch(self, buffers, width, height, fmt, meta):
        # Stand-in for batched inference: sum the batch's first bytes into the
        # anchor (buffers[0]), and attach one detection whose label is the
        # batch size, proving N inputs reached one Python call.
        total = 0
        for b in buffers:
            total = (total + memoryview(b)[0]) % 256
        memoryview(buffers[0])[0] = total
        meta.add_object(len(buffers), 0.0, 0.0, 1.0, 1.0, 1.0)


class AudioTranscriber:
    """A payload element: reads an audio buffer, emits a much shorter text one.

    The transcription shape. Nothing about the output fits in the input buffer,
    so it goes back through meta.emit instead of an in-place write.
    """

    def g2g_process_payload(self, buffers, caps, meta):
        samples = memoryview(buffers[0])
        assert not samples.readonly, "payload buffer must be writable"
        meta.add_blob("caps", caps.encode("utf-8"))
        meta.emit(("heard %d bytes" % samples.nbytes).encode("utf-8"))


class SpeechSynthesizer:
    """The other direction: a short text payload in, a long audio one out.

    Synthesized speech runs for as long as its own samples, so it states a
    duration instead of inheriting the text buffer's.
    """

    SAMPLE_RATE = 16000
    SAMPLES_PER_CHARACTER = 100
    BYTES_PER_SAMPLE = 2

    def g2g_process_payload(self, buffers, caps, meta):
        text = bytes(memoryview(buffers[0]))
        samples = len(text) * self.SAMPLES_PER_CHARACTER
        meta.emit(
            bytes(samples * self.BYTES_PER_SAMPLE),
            duration_ns=samples * 1_000_000_000 // self.SAMPLE_RATE,
        )


class ChunkedSynthesizer:
    """Emits several buffers from one input, the streaming-TTS shape.

    Speech generated a chunk at a time, and the separation family's stems, both
    hand back more than one buffer per input buffer.
    """

    CHUNKS = 3

    def g2g_process_payload(self, buffers, caps, meta):
        text = bytes(memoryview(buffers[0]))
        for chunk in range(self.CHUNKS):
            meta.emit(text + str(chunk).encode("utf-8"))


class StreamingSynthesizer:
    """Chunked speech whose chunks play one after another, not all at once.

    Each chunk states the presentation time it starts at, counted from the
    element properties the test sets (chunks, chunk_duration, first_pts).
    """

    def g2g_process_payload(self, buffers, caps, meta):
        text = bytes(memoryview(buffers[0]))
        for chunk in range(self.chunks):
            meta.emit(
                text + str(chunk).encode("utf-8"),
                duration_ns=self.chunk_duration,
                pts_ns=self.first_pts + chunk * self.chunk_duration,
            )


class UnstampedSynthesizer:
    """Emits one buffer with no presentation time, for a sink to present on
    arrival."""

    def g2g_process_payload(self, buffers, caps, meta):
        import g2g

        meta.emit(bytes(memoryview(buffers[0])), pts_ns=g2g.PTS_NONE)


class ThreadedTransform:
    """Stages a detection from a *worker thread*, not the calling thread.

    The native MetaSink is created on the host's element thread; an `unsendable`
    sink trips pyo3's thread-affinity check the moment another thread touches it,
    so this cross-thread `add_object` would raise. A `Sync` (Mutex-backed) sink
    accepts it. Models an ML element that parallelizes post-processing.
    """

    def g2g_process(self, buf, width, height, fmt, meta):
        import threading

        err = []

        def work():
            try:
                meta.add_object(11, 0.0, 0.0, 1.0, 1.0, 0.5)
            except BaseException as e:  # noqa: BLE001 - surface any failure
                err.append(repr(e))

        t = threading.Thread(target=work)
        t.start()
        t.join()  # releases the GIL so the worker thread can run
        if err:
            raise RuntimeError("cross-thread add_object failed: " + err[0])


class RetainingTransform:
    """Misbehaves: stashes the frame buffer view on self, so it outlives the
    g2g_process call. The host must reject this (the retained pointer would
    dangle once the frame is freed downstream) rather than risk a use-after-free.
    """

    def g2g_process(self, buf, width, height, fmt, meta):
        # Retain a writable view past return: this is the contract violation the
        # host's export-counter guard catches.
        self.saved = memoryview(buf)


class CounterSource:
    """A source: writes its frame index into byte 0, ends after three frames."""

    def __init__(self):
        self.n = 0

    def g2g_produce(self, buf, width, height, fmt, meta):
        if self.n >= 3:
            return False  # end of stream
        memoryview(buf)[0] = self.n
        self.n += 1
        return True


class PropEcho:
    """Echoes forwarded element properties back as metadata, proving the host
    set them on the instance: a gst-style `model-name` reaches `self.model_name`,
    and an int property keeps its int type (used directly as a detection label,
    which the native sink requires to be an integer)."""

    def g2g_process(self, buf, width, height, fmt, meta):
        model = getattr(self, "model_name", "<unset>")
        device = getattr(self, "device", "<unset>")
        # Forwarded as a Python int; add_object's label requires an integer, so
        # passing it straight through fails loudly if it arrived as a string.
        batch = getattr(self, "batch_size", 0)
        meta.add_blob("model_name", model.encode("utf-8"))
        meta.add_blob("device", device.encode("utf-8"))
        meta.add_blob("language", getattr(self, "language", "<unset>").encode("utf-8"))
        meta.add_object(batch, 0.0, 0.0, 1.0, 1.0, 1.0)

    def g2g_process_batch(self, buffers, width, height, fmt, meta):
        self.g2g_process(buffers[0], width, height, fmt, meta)


class DeclaredProps:
    """States the properties it has, the way a gst-python-ml element does, so the
    host can refuse a pipeline naming one it has not."""

    def g2g_properties(self):
        return ["model_name", "device"]

    def g2g_process(self, buf, width, height, fmt, meta):
        meta.add_blob("model_name", getattr(self, "model_name", "<unset>").encode("utf-8"))