diff --git a/src/gst_webrtc/gpu_sink/core.py b/src/gst_webrtc/gpu_sink/core.py index d18d27d..61d391c 100644 --- a/src/gst_webrtc/gpu_sink/core.py +++ b/src/gst_webrtc/gpu_sink/core.py @@ -41,14 +41,12 @@ def __init__( output_format: str = "RGBA", queue_size: int = 4, drop_when_full: bool = True, - copy_frame: bool = True, ) -> None: if output_format not in {"RGBA", "RGB", "BGR", "GRAY8"}: raise ValueError("output_format must be one of RGBA|RGB|BGR|GRAY8") self.name = name or f"appsink-{uuid.uuid4().hex[:8]}" self.output_format = output_format self.drop_when_full = drop_when_full - self.copy_frame = copy_frame self._q: "queue.Queue[Frame]" = queue.Queue(maxsize=max(1, queue_size)) self._sink: Optional[Gst.Element] = None @@ -203,11 +201,7 @@ def _on_new_sample(self, sink: Gst.Element): # -> Gst.FlowReturn return Gst.FlowReturn.OK try: arr = self._map_to_numpy(map_info, width, height, str(fmt)) - # Always copy to ensure safety across threads and buffer reuse. - if not self.copy_frame: - # Minimal impl note: zero-copy is unsafe without keeping the buffer mapped. - # Fallback to copy for correctness. - pass + # Always copy: zero-copy is unsafe without keeping the buffer mapped. arr = arr.copy() finally: buffer.unmap(map_info) diff --git a/src/gst_webrtc/sender/core.py b/src/gst_webrtc/sender/core.py index 710edb2..9fe207c 100644 --- a/src/gst_webrtc/sender/core.py +++ b/src/gst_webrtc/sender/core.py @@ -324,11 +324,10 @@ def _on_ice_state(self, _webrtc, _pspec) -> None: elif nick == "failed": self._restart_ice() else: - # If disconnected persists >3s, restart ICE + # If disconnected persists >1s, restart ICE if self._ice_disconnected_at is not None: import time as _t - # Faster reaction: 1.0s if _t.time() - self._ice_disconnected_at > 1.0: self._restart_ice() diff --git a/tests/e2e/.gitignore b/tests/e2e/.gitignore index 82efd22..0108268 100644 --- a/tests/e2e/.gitignore +++ b/tests/e2e/.gitignore @@ -2,3 +2,4 @@ logs/ report.json artifacts/ gpu_sink_report.json +frames_msid/ diff --git a/tests/e2e/README.md b/tests/e2e/README.md index c628b3c..f4002ea 100644 --- a/tests/e2e/README.md +++ b/tests/e2e/README.md @@ -6,18 +6,18 @@ Quick start: ``` pixi install -./e2e/smoke.sh +./tests/e2e/smoke.sh ``` What it does: - Starts `signal-server` via `go run . serve --addr :18080`. -- Runs `receiver.py` headlessly and `sender.py` via `pixi run`. +- Runs `receiver.py` and `sender_sw.py` (software x264enc) headlessly via `pixi run`. - Watches logs for successful WebRTC link/answer. -- Writes a JSON report to `e2e/report.json` and exits non-zero on failure. +- Writes a JSON report to `tests/e2e/report.json` and exits non-zero on failure. Artifacts: -- Logs: `e2e/logs/{signal,receiver,sender}.log` -- Report: `e2e/report.json` +- Logs: `tests/e2e/logs/{signal,receiver,sender}.log` +- Report: `tests/e2e/report.json` Environment: - `TIMEOUT_SEC` (optional): overall wait timeout (default 25). @@ -32,20 +32,20 @@ Run: ``` pixi install -./e2e/gpu_sink.sh +./tests/e2e/gpu_sink.sh ``` What it does: - Starts the Go signaling server on `:18080`. - Runs `gpu_sink_save.py` as the receiver to pull NumPy frames and save PNGs. -- Runs `sender.py` to produce an H264 RTP stream. +- Runs `sender_sw.py` to produce an H264 RTP stream. - Waits until the requested number of frames are saved, then verifies images using PIL. -- Writes a JSON report to `e2e/gpu_sink_report.json`. +- Writes a JSON report to `tests/e2e/gpu_sink_report.json`. Artifacts: -- Saved frames: `e2e/artifacts/gpu_frames/*.png` -- Logs: `e2e/logs/{signal_gpu,gpu_sink_receiver,gpu_sink_sender}.log` -- Report: `e2e/gpu_sink_report.json` +- Saved frames: `tests/e2e/artifacts/gpu_frames/*.png` +- Logs: `tests/e2e/logs/{signal_gpu,gpu_sink_receiver,gpu_sink_sender}.log` +- Report: `tests/e2e/gpu_sink_report.json` Environment: - `FRAMES` (optional): number of frames to save (default 5) diff --git a/tests/e2e/appsink_msid_e2e.py b/tests/e2e/appsink_msid_e2e.py index f0d9eb3..173b553 100644 --- a/tests/e2e/appsink_msid_e2e.py +++ b/tests/e2e/appsink_msid_e2e.py @@ -17,7 +17,6 @@ import argparse import asyncio import pathlib -import sys from typing import Dict, List import numpy as np @@ -37,11 +36,6 @@ gi.require_version("GstApp", "1.0") from gi.repository import Gst, GstApp # type: ignore -# Ensure project root on sys.path when running from tests/e2e/ -ROOT = pathlib.Path(__file__).resolve().parents[2] -if str(ROOT) not in sys.path: - sys.path.insert(0, str(ROOT)) - from gst_webrtc import init_gst from gst_webrtc.receiver import WebRTCReceiver from gst_webrtc.sender import WebRTCSender diff --git a/tests/e2e/gpu_sink.sh b/tests/e2e/gpu_sink.sh index abb5ef4..991acff 100644 --- a/tests/e2e/gpu_sink.sh +++ b/tests/e2e/gpu_sink.sh @@ -5,7 +5,7 @@ set -Eeuo pipefail IFS=$'\n\t' ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")"/../.. && pwd)" -E2E_DIR="$ROOT_DIR/e2e" +E2E_DIR="$ROOT_DIR/tests/e2e" LOG_DIR="$E2E_DIR/logs" ART_DIR="$E2E_DIR/artifacts" OUT_DIR="$ART_DIR/gpu_frames" diff --git a/tests/e2e/gpu_sink_save.py b/tests/e2e/gpu_sink_save.py index 6b517fd..f470f65 100644 --- a/tests/e2e/gpu_sink_save.py +++ b/tests/e2e/gpu_sink_save.py @@ -6,7 +6,7 @@ pixi run python tests/e2e/gpu_sink_save.py --frames 5 --out ./frames_gpu Prereqs: - - Start signaling server (see project docs) and a sender (`tests/e2e/sender.py`). + - Start signaling server (see project docs) and a sender (`tests/e2e/sender_sw.py`). - Pillow is recommended: `pixi run python -m pip install pillow` if missing. """ @@ -28,7 +28,8 @@ gi.require_version("Gst", "1.0") gi.require_version("GstWebRTC", "1.0") gi.require_version("GstSdp", "1.0") -from gi.repository import Gst # type: ignore +gi.require_version("GstApp", "1.0") +from gi.repository import Gst, GstApp # type: ignore from gst_webrtc import init_gst from gst_webrtc.gpu_sink import GpuFrameSink @@ -64,10 +65,25 @@ async def main(nframes: int, out_dir: Path, out_fmt: str, timeout: float) -> Non # Start receiver loop recv_task = asyncio.create_task(receiver.run()) - # Wait for appsink to be added by receiver, then bind + # Wait for the receiver to add its appsink, then bind. The receiver renames + # the appsink to the incoming stream's msid, so discover it by element type + # rather than by GpuFrameSink's original name. + def _find_appsink_name(): + it = receiver.pipe.iterate_recurse() + while True: + res, elem = it.next() + if res == Gst.IteratorResult.OK: + if isinstance(elem, GstApp.AppSink): + return elem.get_name() + elif res == Gst.IteratorResult.RESYNC: + it.resync() + else: + return None + for _ in range(200): # ~2s max - el = receiver.pipe.get_by_name(sink.name) - if el is not None: + found = _find_appsink_name() + if found is not None: + sink.name = found sink.bind(receiver.pipe) print(f"[gpu-sink-test] bound appsink: {sink.name}") break @@ -81,7 +97,6 @@ async def main(nframes: int, out_dir: Path, out_fmt: str, timeout: float) -> Non continue arr: np.ndarray = frame.array - h, w = arr.shape[:2] info = { "shape": arr.shape, "dtype": str(arr.dtype), diff --git a/tests/e2e/inference_buffer_v2_ipc.py b/tests/e2e/inference_buffer_v2_ipc.py index bb835f6..aecf572 100644 --- a/tests/e2e/inference_buffer_v2_ipc.py +++ b/tests/e2e/inference_buffer_v2_ipc.py @@ -73,7 +73,12 @@ def main() -> int: # Step 1: Child writes v1; parent verifies visibility (only for CUDA) v1 = 1.2345 parent_conn.send(("write1", v1)) - tag, mean1 = parent_conn.recv() + msg = parent_conn.recv() + if msg[0] == "child_error": + print(f"FAIL: child raised: {msg[1]}") + proc.terminate(); proc.join(5) + return 6 + tag, mean1 = msg assert tag == "done1" if is_cuda: torch.cuda.synchronize() @@ -92,7 +97,12 @@ def main() -> int: if is_cuda: torch.cuda.synchronize() parent_conn.send(("parent_written2", v2)) - tag, mean2, ok2 = parent_conn.recv() + msg = parent_conn.recv() + if msg[0] == "child_error": + print(f"FAIL: child raised: {msg[1]}") + proc.terminate(); proc.join(5) + return 7 + tag, mean2, ok2 = msg assert tag == "seen2" if is_cuda: print(f"parent->child CUDA propagation: child_mean={mean2:.6f} child_allclose={ok2}") diff --git a/tests/e2e/ipc_two_procs/run.sh b/tests/e2e/ipc_two_procs/run.sh index aacdb9a..37b0185 100644 --- a/tests/e2e/ipc_two_procs/run.sh +++ b/tests/e2e/ipc_two_procs/run.sh @@ -11,7 +11,7 @@ cleanup() { [[ -p "$workdir/w_out" ]] && rm -f "$workdir/w_out" [[ -p "$workdir/r_in" ]] && rm -f "$workdir/r_in" [[ -p "$workdir/r_out" ]] && rm -f "$workdir/r_out" - [[ -f "$workdir" ]] && rm -rf "$workdir" || true + [[ -d "$workdir" ]] && rm -rf "$workdir" || true } trap cleanup EXIT diff --git a/tests/e2e/receiver.py b/tests/e2e/receiver.py index d3e9ee1..6f892c9 100644 --- a/tests/e2e/receiver.py +++ b/tests/e2e/receiver.py @@ -1,37 +1,26 @@ import asyncio import os -import gi - -gi.require_version("Gst", "1.0") -gi.require_version("GstWebRTC", "1.0") -gi.require_version("GstSdp", "1.0") - -from gi.repository import Gst, GstSdp, GstWebRTC # noqa: F401 - from gst_webrtc import init_gst from gst_webrtc.receiver import WebRTCReceiver -from gst_webrtc.ws_signal.signal_client import SignalClient # noqa: F401 ROOM = os.environ.get("ROOM", "demo") SIGNAL_URL = os.environ.get("SIGNAL_URL", "ws://127.0.0.1:18080/ws") STUN = os.environ.get("STUN", "stun://stun.example.com") TURN = os.environ.get( "TURN", "turn://USERNAME:PASSWORD@TURN_HOST:3478?transport=udp" -) # not used +) init_gst() -wh = "width=640,height=480" -fr = "framerate=30/1" queue = "queue max-size-buffers=1 max-size-time=0 max-size-bytes=0 leaky=downstream" def h264_decode_bin_sink() -> str: desc = f""" capsfilter caps="application/x-rtp" ! rtph264depay ! h264parse ! {queue} ! -nvh264dec ! {queue} ! -videoconvert ! autovideosink +avdec_h264 ! {queue} ! +videoconvert ! fakesink sync=false """ return desc diff --git a/tests/e2e/receiver_msid_save.py b/tests/e2e/receiver_msid_save.py index 48c64f7..f824d7b 100644 --- a/tests/e2e/receiver_msid_save.py +++ b/tests/e2e/receiver_msid_save.py @@ -37,9 +37,9 @@ def _rtp_h264_to_appsink_desc(appsink_name: str = "recv_app", to_format: str = "RGBA") -> str: - # Keep CPU decode for stability across environments + # CPU decode for stability across environments (no GPU context needed) q = "queue max-size-buffers=1 max-size-time=0 max-size-bytes=0 leaky=downstream" - dec = f"nvh264dec ! {q}" + dec = f"avdec_h264 ! {q}" return ( f"capsfilter caps=\"application/x-rtp\" ! rtph264depay ! h264parse ! " f"{dec} ! videoconvert ! video/x-raw,format={to_format} ! {q} ! " @@ -58,7 +58,6 @@ def _frame_from_sample(sample: Gst.Sample, expect_fmt: str = "RGBA") -> np.ndarr s = caps.get_structure(0) if caps else None width = int(s.get_value("width")) if s and s.has_field("width") else 0 height = int(s.get_value("height")) if s and s.has_field("height") else 0 - fmt = str(s.get_value("format")) if s and s.has_field("format") else expect_fmt ok, map_info = buf.map(Gst.MapFlags.READ) if not ok: @@ -72,14 +71,9 @@ def _frame_from_sample(sample: Gst.Sample, expect_fmt: str = "RGBA") -> np.ndarr bpr = nbytes // height row_pixels = bpr // ch if ch > 0 else 0 arr = np.frombuffer(mv, dtype=np.uint8) - if ch == 4: - arr = arr.reshape(height, row_pixels, 4) - if width > 0 and row_pixels != width: - arr = arr[:, :width, :] - else: - arr = arr.reshape(height, row_pixels, 3) - if width > 0 and row_pixels != width: - arr = arr[:, :width, :] + arr = arr.reshape(height, row_pixels, ch) + if width > 0 and row_pixels != width: + arr = arr[:, :width, :] return arr.copy() finally: buf.unmap(map_info) diff --git a/tests/e2e/sender.py b/tests/e2e/sender.py index 47b466b..d052332 100644 --- a/tests/e2e/sender.py +++ b/tests/e2e/sender.py @@ -6,49 +6,24 @@ import gi gi.require_version("Gst", "1.0") -gi.require_version("GstWebRTC", "1.0") -gi.require_version("GstSdp", "1.0") -from gi.repository import Gst, GstSdp, GstWebRTC # noqa: F401 +from gi.repository import Gst from gst_webrtc import init_gst from gst_webrtc.sender import WebRTCSender -from gst_webrtc.ws_signal.signal_client import SignalClient # noqa: F401 ROOM = os.environ.get("ROOM", "demo") SIGNAL_URL = os.environ.get("SIGNAL_URL", "ws://127.0.0.1:18080/ws") STUN = os.environ.get("STUN", "stun://stun.example.com") TURN = os.environ.get( "TURN", "turn://USERNAME:PASSWORD@TURN_HOST:3478?transport=udp" -) # not used +) init_gst() -wh = "width=640,height=480" -fr = "framerate=30/1" queue = "queue max-size-buffers=1 max-size-time=0 max-size-bytes=0 leaky=downstream" -def create_test_video_source(name: str) -> str: - """Build a low-latency H264 RTP source with an identifiable name. - - NOTE: We insert an `identity name=` so the multi-source case is easy to - debug on the sender side. This does not affect WebRTC semantics directly, - but helps correlate logs and pads per source. - """ - src = f"""\ -videotestsrc is-live=true pattern=ball ! video/x-raw,{wh},{fr} ! \ -identity name={name} silent=true ! \ -{queue} ! \ -nvh264enc ! \ -h264parse config-interval=-1 ! video/x-h264,alignment=au ! \ -{queue} ! \ -rtph264pay aggregate-mode=zero-latency pt=96 ! \ -capsfilter caps=\"application/x-rtp,media=video,encoding-name=H264,payload=96,clock-rate=90000\" -""" - return src - - # RealSense D405 color stream parameters (pushed into GStreamer via appsrc). RS_W, RS_H, RS_FPS = ( int(os.environ.get("RS_W", 640)), diff --git a/tests/e2e/sender_sw.py b/tests/e2e/sender_sw.py index bf27be1..6012f79 100644 --- a/tests/e2e/sender_sw.py +++ b/tests/e2e/sender_sw.py @@ -4,24 +4,10 @@ Use x264enc (CPU) to avoid GPU dependency during CI/E2E runs. Run: - ROOM=demo SIGNAL_URL=ws://127.0.0.1:18080/ws pixi run python e2e/sender_sw.py + ROOM=demo SIGNAL_URL=ws://127.0.0.1:18080/ws pixi run python tests/e2e/sender_sw.py """ import asyncio -import pathlib -import sys - -import gi - -gi.require_version("Gst", "1.0") -gi.require_version("GstWebRTC", "1.0") -gi.require_version("GstSdp", "1.0") -from gi.repository import Gst # type: ignore # noqa: F401 - -# Ensure project root on sys.path when running from e2e/ -ROOT = pathlib.Path(__file__).resolve().parents[1] -if str(ROOT) not in sys.path: - sys.path.insert(0, str(ROOT)) from gst_webrtc import init_gst from gst_webrtc.sender import WebRTCSender diff --git a/tests/e2e/smoke.sh b/tests/e2e/smoke.sh index 35977b3..df4737a 100644 --- a/tests/e2e/smoke.sh +++ b/tests/e2e/smoke.sh @@ -1,7 +1,7 @@ #!/usr/bin/env bash # Minimal end-to-end smoke test orchestrator # - Starts Go signal server on :18080 -# - Runs Python receiver.py (headless) and sender.py via pixi +# - Runs Python receiver.py and sender_sw.py (software x264enc) via pixi # - Watches logs for success patterns # - Writes e2e/report.json and exits non-zero on failure @@ -9,7 +9,7 @@ set -Eeuo pipefail IFS=$'\n\t' ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")"/../.. && pwd)" -E2E_DIR="$ROOT_DIR/e2e" +E2E_DIR="$ROOT_DIR/tests/e2e" LOG_DIR="$E2E_DIR/logs" REPORT_JSON="$E2E_DIR/report.json" @@ -72,7 +72,7 @@ echo "[e2e] signal-server is up (pid=$server_pid)" echo "[e2e] starting receiver.py (headless)" ( cd "$ROOT_DIR" - ROOM="$ROOM" SIGNAL_URL="$SIGNAL_URL" E2E_HEADLESS=1 PYTHONUNBUFFERED=1 \ + ROOM="$ROOM" SIGNAL_URL="$SIGNAL_URL" PYTHONUNBUFFERED=1 \ exec pixi run python -u tests/e2e/receiver.py \ >"$LOG_DIR/receiver.log" 2>&1 ) & @@ -81,12 +81,12 @@ receiver_pid=$! # Give receiver a moment to join room sleep 1 -echo "[e2e] starting sender.py" +echo "[e2e] starting sender_sw.py" ( cd "$ROOT_DIR" - # Use headless-friendly encoder in CI to avoid GPU dependency - ROOM="$ROOM" SIGNAL_URL="$SIGNAL_URL" E2E_HEADLESS=1 PYTHONUNBUFFERED=1 \ - exec pixi run python -u tests/e2e/sender.py \ + # Software x264enc sender: no GPU or camera dependency, runs anywhere + ROOM="$ROOM" SIGNAL_URL="$SIGNAL_URL" PYTHONUNBUFFERED=1 \ + exec pixi run python -u tests/e2e/sender_sw.py \ >"$LOG_DIR/sender.log" 2>&1 ) & sender_pid=$!