cristhian
6f76074709
validate / validate (push) Successful in 59m15s
A source that documents itself through a specific downstream node depends on that node to be understood, and that node already depends on this one for frames. The README pointed at one by name and the CI workflow still carried a comment about the dataflow that wired it, which no longer exists. The output is documented by its type instead: wire `image` into anything that takes std/media/v1/Image. PREVIEW covers seeing the frames without naming anyone. Generic words like "a viewer" or "a consumer" stay, since they describe a role rather than a repository.
149 lines
4.4 KiB
Python
149 lines
4.4 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_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()),
|
|
])
|