from display_grid import CameraGrid from drm_outputs import DrmOutput from gst_sink import appsrc_cmd, va_grid_cmd def _out(): return DrmOutput( name="HDMI-A-7", sysfs="card0-HDMI-A-7", node="/dev/dri/card0", driver="xe", connector_id=538, status="connected", ) def test_appsrc_queue_buffers_default_two(): cmd = appsrc_cmd(_out(), 1920, 1080, 30) assert "max-size-buffers=2" in cmd def test_appsrc_queue_buffers_one(): cmd = appsrc_cmd(_out(), 1920, 1080, 30, queue_buffers=1) assert "max-size-buffers=1" in cmd assert "max-size-buffers=2" not in cmd def test_appsrc_i420_blocksize(): cmd = appsrc_cmd(_out(), 1920, 1080, 30, pixel_format="i420") joined = " ".join(cmd) assert "format=i420" in joined assert "blocksize=3110400" in joined # 1920*1080*3//2 assert "format=bgra" not in joined def test_va_grid_cmd_pins_arc_compositor_and_twenty_tiles(): g = CameraGrid(width=1920, height=1080, cols=5, rows=4) tile_fds = list(range(4, 24)) cmd = va_grid_cmd( _out(), g, ingest_w=640, ingest_h=360, fps=15, tile_fds=tile_fds, chrome_fd=3, queue_buffers=1) joined = " ".join(cmd) assert "varenderD129compositor" in joined assert "varenderD129postproc" in joined assert "kmssink" in joined assert "connector-id=538" in joined assert "driver-name=xe" in joined assert joined.count("fdsrc") == 21 # chrome + 20 tiles x, y, w, h = g.inner_rect(0) assert f"xpos={x}" in joined assert f"ypos={y}" in joined assert "format=i420" in joined assert "blocksize=345600" in joined # 640*360*1.5 def test_va_grid_cmd_h264_uses_arc_decoder(): g = CameraGrid(width=1920, height=1080, cols=5, rows=4) paths = [f"/tmp/cam-{i:02d}.h264" for i in range(1, 21)] cmd = va_grid_cmd( _out(), g, ingest_w=640, ingest_h=360, fps=15, tile_h264_paths=paths, chrome_fd=3) joined = " ".join(cmd) assert "varenderD129h264dec" in joined assert joined.count("h264parse") == 20 assert "fdsrc" in joined # chrome only assert joined.count("filesrc") == 20