cristhian
7628a8515b
validate / validate (push) Successful in 59m14s
CAPTURE_PATH defaulted to index 0, which is a coin flip on a machine with one camera. A UVC camera exposes a second device node for its metadata stream -- /dev/video1 beside /dev/video0 -- and that node opens happily and never yields an image, so the lowest device number is wrong about as often as it is right. With neither CAMERA_ID nor CAPTURE_PATH set, the node now opens each /dev/video* in numeric order and keeps the first that actually delivers a frame. Reading one is the test, because opening is not evidence. Discovery finding nothing falls back to index 0 rather than raising, which keeps the failure in one place: the first read, where CI publishes a placeholder and a normal run raises with a message naming the device. CAMERA_ID is still worth preferring on a robot. Neither an index nor discovery order is stable across reboots or across which port a camera is in; a by-id name is.
194 lines
6.0 KiB
Python
194 lines
6.0 KiB
Python
"""Tests that need neither a camera nor a dataflow."""
|
|
|
|
import os
|
|
|
|
import cv2
|
|
import numpy as np
|
|
import pyarrow as pa
|
|
import pytest
|
|
|
|
from dora_webcam import camera, encode, image
|
|
|
|
WIDTH, HEIGHT = 12, 8
|
|
|
|
|
|
@pytest.fixture
|
|
def frame() -> np.ndarray:
|
|
rng = np.random.default_rng(seed=0)
|
|
return rng.integers(0, 255, (HEIGHT, WIDTH, 3), dtype=np.uint8)
|
|
|
|
|
|
def test_discovery_skips_a_device_that_opens_but_never_yields_a_frame(monkeypatch):
|
|
# A UVC camera exposes a metadata node beside the capture node; it opens
|
|
# and reads nothing, so the lowest device number is the wrong one to take.
|
|
monkeypatch.setattr(camera, "_devices", lambda: ["/dev/video0", "/dev/video1"])
|
|
monkeypatch.setattr(camera, "_yields_frame", lambda path: path == "/dev/video1")
|
|
|
|
assert camera.discover() == "/dev/video1"
|
|
|
|
|
|
def test_discovery_returns_none_when_nothing_captures(monkeypatch):
|
|
monkeypatch.setattr(camera, "_devices", lambda: ["/dev/video0"])
|
|
monkeypatch.setattr(camera, "_yields_frame", lambda path: False)
|
|
|
|
assert camera.discover() is None
|
|
|
|
|
|
def test_devices_are_ordered_numerically(monkeypatch):
|
|
monkeypatch.setattr(
|
|
camera.glob, "glob",
|
|
lambda pattern: ["/dev/video10", "/dev/video2", "/dev/video1"],
|
|
)
|
|
|
|
assert camera._devices() == ["/dev/video1", "/dev/video2", "/dev/video10"]
|
|
|
|
|
|
def test_an_unset_device_is_discovered(monkeypatch):
|
|
monkeypatch.setattr(camera, "discover", lambda: "/dev/video3")
|
|
|
|
assert camera.resolve(None, None) == "/dev/video3"
|
|
|
|
|
|
def test_an_unset_device_falls_back_when_nothing_is_found(monkeypatch):
|
|
# Falling back rather than raising keeps the failure in one place: the
|
|
# first read, which under CI publishes a placeholder instead of raising.
|
|
monkeypatch.setattr(camera, "discover", lambda: None)
|
|
|
|
assert camera.resolve(None, None) == camera.FALLBACK
|
|
|
|
|
|
def test_an_explicit_device_is_not_overridden_by_discovery(monkeypatch):
|
|
monkeypatch.setattr(camera, "discover", lambda: "/dev/video9")
|
|
|
|
assert camera.resolve(None, "/dev/video4") == "/dev/video4"
|
|
|
|
|
|
def test_a_numeric_capture_path_is_read_as_an_index():
|
|
assert camera.resolve(None, "1") == 1
|
|
|
|
|
|
def test_a_device_path_is_passed_through():
|
|
assert camera.resolve(None, "/dev/video2") == "/dev/video2"
|
|
|
|
|
|
def test_an_unmatched_camera_id_is_an_error():
|
|
with pytest.raises(RuntimeError, match="no camera matching id"):
|
|
camera.resolve("nothing-matches-this", "0")
|
|
|
|
|
|
def test_camera_id_resolves_through_by_id(tmp_path, monkeypatch):
|
|
# A by-id directory of one symlink, standing in for /dev/v4l/by-id.
|
|
by_id = tmp_path / "by-id"
|
|
by_id.mkdir()
|
|
(by_id / "usb-Acme_Webcam_1234-video-index0").symlink_to("/dev/video3")
|
|
monkeypatch.setattr(camera, "BY_ID", str(by_id) + os.sep)
|
|
|
|
assert camera.resolve("acme_webcam", "0") == 3
|
|
|
|
|
|
def test_camera_id_wins_over_capture_path(tmp_path, monkeypatch):
|
|
by_id = tmp_path / "by-id"
|
|
by_id.mkdir()
|
|
(by_id / "usb-Acme_Webcam-video-index0").symlink_to("/dev/video5")
|
|
monkeypatch.setattr(camera, "BY_ID", str(by_id) + os.sep)
|
|
|
|
assert camera.resolve("acme", "/dev/video9") == 5
|
|
|
|
|
|
def test_bgr8_is_passed_through_unchanged(frame):
|
|
np.testing.assert_array_equal(encode.encode(frame, "bgr8", 95), frame)
|
|
|
|
|
|
def test_rgb8_swaps_the_channels(frame):
|
|
encoded = encode.encode(frame, "rgb8", 95)
|
|
|
|
np.testing.assert_array_equal(cv2.cvtColor(encoded, cv2.COLOR_RGB2BGR), frame)
|
|
|
|
|
|
@pytest.mark.parametrize("encoding", ["jpeg", "png", "bmp", "webp"])
|
|
def test_compressed_encodings_round_trip(frame, encoding):
|
|
encoded = encode.encode(frame, encoding, 100)
|
|
|
|
decoded = cv2.imdecode(np.frombuffer(encoded.tobytes(), np.uint8), cv2.IMREAD_COLOR)
|
|
|
|
assert decoded.shape == frame.shape
|
|
|
|
|
|
def test_yuv420_is_one_and_a_half_bytes_per_pixel(frame):
|
|
encoded = encode.encode(frame, "yuv420", 95)
|
|
|
|
assert encoded.size == WIDTH * HEIGHT * 3 // 2
|
|
|
|
|
|
def test_an_unsupported_encoding_is_rejected(frame):
|
|
with pytest.raises(RuntimeError, match="unsupported encoding"):
|
|
encode.encode(frame, "tiff", 95)
|
|
|
|
|
|
def test_packing_produces_the_image_struct(frame):
|
|
packed = image.pack(frame, "bgr8", WIDTH, HEIGHT)
|
|
|
|
assert packed.type == image.IMAGE_TYPE
|
|
row = packed[0].as_py()
|
|
assert (row["width"], row["height"], row["encoding"]) == (WIDTH, HEIGHT, "bgr8")
|
|
np.testing.assert_array_equal(
|
|
np.frombuffer(row["data"], np.uint8).reshape(HEIGHT, WIDTH, 3),
|
|
frame,
|
|
)
|
|
|
|
|
|
def test_packing_records_image_size_not_payload_size(frame):
|
|
# A compressed payload is a 1-D byte run; the receiver still needs the
|
|
# image's dimensions, so they are recorded rather than inferred.
|
|
encoded = encode.encode(frame, "jpeg", 95)
|
|
|
|
row = image.pack(encoded, "jpeg", WIDTH, HEIGHT)[0].as_py()
|
|
|
|
assert (row["width"], row["height"]) == (WIDTH, HEIGHT)
|
|
assert len(row["data"]) != WIDTH * HEIGHT * 3
|
|
|
|
|
|
def test_the_placeholder_frame_is_labelled():
|
|
frame = camera.placeholder("no frame from 0")
|
|
|
|
assert frame.shape == (*camera.PLACEHOLDER, 3)
|
|
assert frame.any(), "the message should have been drawn"
|
|
|
|
|
|
class FakeCapture:
|
|
"""Stands in for cv2.VideoCapture."""
|
|
|
|
def __init__(self, ok: bool) -> None:
|
|
self._ok = ok
|
|
|
|
def read(self):
|
|
return self._ok, np.zeros((4, 4, 3), dtype=np.uint8)
|
|
|
|
|
|
def test_a_failed_read_raises_outside_ci():
|
|
with pytest.raises(RuntimeError, match="could not read a frame"):
|
|
camera.read(FakeCapture(ok=False), 0, ci=False)
|
|
|
|
|
|
def test_a_failed_read_yields_a_placeholder_under_ci():
|
|
frame = camera.read(FakeCapture(ok=False), 0, ci=True)
|
|
|
|
assert frame.shape == (*camera.PLACEHOLDER, 3)
|
|
|
|
|
|
def test_a_good_read_is_returned_as_is():
|
|
frame = camera.read(FakeCapture(ok=True), 0, ci=False)
|
|
|
|
assert frame.shape == (4, 4, 3)
|
|
|
|
|
|
def test_the_struct_matches_the_published_type():
|
|
# A consumer respells this schema in its own repository; they share no
|
|
# package, so the field names and types are the entire contract.
|
|
assert image.IMAGE_TYPE == pa.struct([
|
|
("width", pa.uint32()),
|
|
("height", pa.uint32()),
|
|
("encoding", pa.utf8()),
|
|
("data", pa.large_binary()),
|
|
])
|