g2g-python 0.6.0

//! M198 step 2: the live per-frame path, embedded CPython (`python` feature).
//!
//! Drives `PyTransform` against a stdlib-only hosted element fixture and proves
//! the zero-copy buffer-protocol contract: Python writes into the frame's own
//! System memory in place, and that write is observable on the frame that flows
//! downstream. Needs libpython at build + run time, so the whole file compiles
//! away without the feature.
#![cfg(feature = "python")]

use g2g_core::memory::SystemSlice;
use g2g_core::{
    AsyncElement, Caps, Dim, Frame, FrameTiming, G2gError, MemoryDomain, OutputSink,
    PipelinePacket, PushOutcome, Rate, RawVideoFormat,
};
use g2g_python::PyTransform;

#[derive(Default)]
struct CollectSink {
    packets: Vec<PipelinePacket>,
}

impl OutputSink for CollectSink {
    fn poll_push(
        &mut self,
        _cx: &mut core::task::Context<'_>,
        packet_slot: &mut Option<PipelinePacket>,
    ) -> core::task::Poll<Result<PushOutcome, G2gError>> {
        let packet = packet_slot.take().expect("poll_push without a packet");

        self.packets.push(packet);
        core::task::Poll::Ready(Ok(PushOutcome::Accepted))
    }
}

fn frame_2x1_rgba(first: u8) -> Frame {
    // 2x1 RGBA = 8 bytes; only the first byte's value matters to the assertion.
    let mut bytes = vec![0u8; 8];
    bytes[0] = first;
    Frame {
        domain: MemoryDomain::System(SystemSlice::from_boxed(bytes.into_boxed_slice())),
        timing: FrameTiming {
            pts_ns: 0,
            dts_ns: 0,
            duration_ns: 0,
            capture_ns: 0,
            arrival_ns: 0,
            keyframe: false,
        },
        sequence: 0,
        meta: Default::default(),
    }
}

#[test]
fn python_writes_into_frame_memory_in_place() {
    // Put the fixture element on the interpreter's import path. Set before the
    // first GIL acquisition so it is on sys.path at interpreter init.
    std::env::set_var(
        "PYTHONPATH",
        concat!(env!("CARGO_MANIFEST_DIR"), "/tests/fixtures"),
    );

    let mut el = PyTransform::new("echo_element", "EchoTransform").with_draw_label(true);
    let caps = Caps::RawVideo {
        format: RawVideoFormat::Rgba8,
        width: Dim::Fixed(2),
        height: Dim::Fixed(1),
        framerate: Rate::Fixed(30),
        interlace: g2g_core::Interlace::Any,
    };
    // Instantiates the Python class under the GIL.
    el.configure_pipeline(&caps).unwrap();

    let mut sink = CollectSink::default();
    let rt = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    rt.block_on(el.process(PipelinePacket::DataFrame(frame_2x1_rgba(10)), &mut sink))
        .unwrap();

    // Exactly one frame flowed, and the first byte was incremented (10 -> 11)
    // by Python writing into the Rust buffer through the buffer protocol.
    assert_eq!(sink.packets.len(), 1);
    let PipelinePacket::DataFrame(frame) = &sink.packets[0] else {
        panic!("expected a DataFrame downstream");
    };
    let Some(slice) = frame.domain.as_system_slice() else {
        panic!("expected System memory");
    };
    assert_eq!(slice[0], 11, "in-place write did not reach Rust");
}

#[test]
fn worker_is_reused_across_frames() {
    std::env::set_var(
        "PYTHONPATH",
        concat!(env!("CARGO_MANIFEST_DIR"), "/tests/fixtures"),
    );

    let mut el = PyTransform::new("echo_element", "EchoTransform");
    let caps = Caps::RawVideo {
        format: RawVideoFormat::Rgba8,
        width: Dim::Fixed(2),
        height: Dim::Fixed(1),
        framerate: Rate::Fixed(30),
        interlace: g2g_core::Interlace::Any,
    };
    el.configure_pipeline(&caps).unwrap();

    let mut sink = CollectSink::default();
    let rt = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    // Three frames through the one persistent worker thread (the reply channel
    // is capacity-1 and reused, so this also exercises that it cycles cleanly).
    for first in [10u8, 20, 30] {
        rt.block_on(el.process(PipelinePacket::DataFrame(frame_2x1_rgba(first)), &mut sink))
            .unwrap();
    }

    let got: Vec<u8> = sink
        .packets
        .iter()
        .map(|p| match p {
            PipelinePacket::DataFrame(f) => {
                f.domain.as_system_slice().expect("expected System memory")[0]
            }
            _ => panic!("expected DataFrames"),
        })
        .collect();
    assert_eq!(
        got,
        vec![11, 21, 31],
        "each frame should be incremented in place"
    );
    assert_eq!(el.emitted_count(), 3);
}