cristhian
a6fa3ce32b
dora's typed image is an Arrow struct of width, height, encoding and data -- it carries inside the payload what this node has been reading out of message metadata. The decoder now takes either shape: a struct describes itself and its metadata is ignored, a flat UInt8 buffer is described by width/height/ encoding beside it, and everything after unpacking is common. The declared contract stays std/core/v1/UInt8, deliberately. Type compatibility is directional -- Image satisfies a Bytes input, but a std/core/v1/UInt8 output wired into an std/media/v1/Image input is a validate error -- and dora-zed declares image_bgr as std/core/v1/UInt8. Flipping this manifest now would break the camera. Accepting both shapes first is what lets the producers move before the contract does. output_framing is not part of this. It selects the wire format, raw or arrow-ipc, and is not required for images: a struct payload was verified to arrive identically under both framings, and validate does not ask for arrow-ipc when an output is annotated std/media/v1/Image. It earns its keep when the wire must be self-describing, such as crossing languages or recording.
339 lines
10 KiB
Python
339 lines
10 KiB
Python
"""Tests that need neither a dataflow nor a display."""
|
|
|
|
import io
|
|
|
|
import cv2
|
|
import numpy as np
|
|
import pyarrow as pa
|
|
import pytest
|
|
from PIL import Image
|
|
|
|
from dora_image_view import bbox, decode, depth, intrinsics
|
|
from dora_image_view.view import Bboxes, View
|
|
|
|
WIDTH, HEIGHT = 12, 8
|
|
|
|
|
|
def _meta(encoding: str, width: int | None = None, height: int | None = None) -> dict:
|
|
"""Build message metadata as a node sending a flat buffer attaches it."""
|
|
return {"encoding": encoding, "width": width, "height": height}
|
|
|
|
|
|
class FakeStorage:
|
|
"""Stands in for the Arrow array dora hands to the node."""
|
|
|
|
def __init__(self, data: np.ndarray) -> None:
|
|
self._data = data
|
|
|
|
def to_numpy(self) -> np.ndarray:
|
|
return self._data
|
|
|
|
|
|
@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_bgr8_round_trips_unchanged(frame):
|
|
decoded = decode.to_bgr(FakeStorage(frame.ravel()), _meta("bgr8", WIDTH, HEIGHT))
|
|
|
|
np.testing.assert_array_equal(decoded, frame)
|
|
|
|
|
|
def test_rgb8_round_trips_unchanged(frame):
|
|
rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
|
|
|
|
decoded = decode.to_bgr(FakeStorage(rgb.ravel()), _meta("rgb8", WIDTH, HEIGHT))
|
|
|
|
np.testing.assert_array_equal(decoded, frame)
|
|
|
|
|
|
@pytest.mark.parametrize(("suffix", "encoding"), [
|
|
(".png", "png"),
|
|
(".jpg", "jpeg"),
|
|
(".bmp", "bmp"),
|
|
(".webp", "webp"),
|
|
])
|
|
def test_self_describing_encodings_need_no_dimensions(frame, suffix, encoding):
|
|
ok, buffer = cv2.imencode(suffix, frame)
|
|
assert ok
|
|
|
|
decoded = decode.to_bgr(FakeStorage(buffer.ravel()), _meta(encoding))
|
|
|
|
assert decoded.shape == (HEIGHT, WIDTH, 3)
|
|
|
|
|
|
def test_yuv420_is_decoded(frame):
|
|
planes = cv2.cvtColor(frame, cv2.COLOR_BGR2YUV_I420)
|
|
|
|
decoded = decode.to_bgr(FakeStorage(planes.ravel()), _meta("yuv420", WIDTH, HEIGHT))
|
|
|
|
assert decoded.shape == (HEIGHT, WIDTH, 3)
|
|
|
|
|
|
def test_avif_is_decoded(frame):
|
|
buffer = io.BytesIO()
|
|
Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)).save(buffer, format="AVIF")
|
|
payload = np.frombuffer(buffer.getvalue(), dtype=np.uint8)
|
|
|
|
decoded = decode.to_bgr(FakeStorage(payload), _meta("avif"))
|
|
|
|
assert decoded.shape == (HEIGHT, WIDTH, 3)
|
|
|
|
|
|
def test_decoded_frame_is_writable():
|
|
storage = FakeStorage(np.arange(2 * 3 * 3, dtype=np.uint8))
|
|
|
|
decoded = decode.to_bgr(storage, _meta("bgr8", 3, 2))
|
|
|
|
# The frame is annotated in place, so it must not be a read-only view.
|
|
decoded[0, 0] = 0
|
|
|
|
|
|
def test_raw_encodings_need_dimensions():
|
|
storage = FakeStorage(np.zeros(18, dtype=np.uint8))
|
|
with pytest.raises(RuntimeError, match="width"):
|
|
decode.to_bgr(storage, _meta("bgr8"))
|
|
|
|
|
|
def test_short_payload_is_rejected():
|
|
storage = FakeStorage(np.zeros(4, dtype=np.uint8))
|
|
with pytest.raises(RuntimeError, match="expected at least"):
|
|
decode.to_bgr(storage, _meta("bgr8", 3, 2))
|
|
|
|
|
|
def test_unsupported_encoding_is_rejected():
|
|
storage = FakeStorage(np.zeros(4, dtype=np.uint8))
|
|
with pytest.raises(RuntimeError, match="unsupported image encoding"):
|
|
decode.to_bgr(storage, _meta("tiff"))
|
|
|
|
|
|
def test_xywh_is_converted_to_corners():
|
|
value = pa.array(
|
|
[
|
|
{
|
|
"bbox": np.array([10.0, 20.0, 4.0, 6.0]),
|
|
"conf": np.array([0.5]),
|
|
"labels": np.array(["cat"]),
|
|
},
|
|
],
|
|
)
|
|
|
|
boxes = bbox.parse(value, "xywh")
|
|
|
|
np.testing.assert_allclose(boxes.corners, [[8.0, 17.0, 12.0, 23.0]])
|
|
assert list(boxes.labels) == ["cat"]
|
|
|
|
|
|
def test_headless_show_draws_without_a_window():
|
|
view = View(headless=True, text="hello")
|
|
view.bboxes = Bboxes(
|
|
corners=np.array([[1.0, 1.0, 5.0, 5.0]]),
|
|
conf=np.array([0.9]),
|
|
labels=np.array(["cat"]),
|
|
)
|
|
frame = np.zeros((32, 32, 3), dtype=np.uint8)
|
|
view.panels["image"] = frame
|
|
|
|
assert view.show() is False
|
|
assert frame.any(), "the overlays should have marked the frame"
|
|
|
|
|
|
def test_show_with_no_panels_is_a_no_op():
|
|
assert View(headless=True).show() is False
|
|
|
|
|
|
def test_panels_are_tiled_left_to_right_at_a_common_height():
|
|
view = View(headless=True)
|
|
view.panels["image"] = np.zeros((20, 10, 3), dtype=np.uint8)
|
|
# Half the height and half the width: scaling to the common height should
|
|
# double both, so it contributes 20 columns, not 10.
|
|
view.panels["depth"] = np.zeros((10, 10, 3), dtype=np.uint8)
|
|
|
|
composite = view.compose()
|
|
|
|
assert composite.shape[0] == 20
|
|
assert composite.shape[1] == 10 + 20
|
|
|
|
|
|
def test_depth_decodes_millimetres_and_reshapes():
|
|
values = np.array([100.0, 200.0, 300.0, 400.0], dtype=np.float32)
|
|
|
|
decoded = depth.to_millimetres(FakeStorage(values), {"shape": [2, 2]})
|
|
|
|
np.testing.assert_allclose(decoded, [[100.0, 200.0], [300.0, 400.0]])
|
|
|
|
|
|
def test_depth_reinterprets_a_float32_payload_sent_as_bytes():
|
|
values = np.array([1500.0, 2500.0], dtype=np.float32)
|
|
|
|
decoded = depth.to_millimetres(
|
|
FakeStorage(values.view(np.uint8)),
|
|
{"shape": [1, 2], "dtype": "float32"},
|
|
)
|
|
|
|
np.testing.assert_allclose(decoded, [[1500.0, 2500.0]])
|
|
|
|
|
|
def test_depth_converts_metres_to_millimetres():
|
|
decoded = depth.to_millimetres(
|
|
FakeStorage(np.array([1.5, 2.0], dtype=np.float32)),
|
|
{"shape": [1, 2], "units": "m"},
|
|
)
|
|
|
|
np.testing.assert_allclose(decoded, [[1500.0, 2000.0]])
|
|
|
|
|
|
def test_depth_normalises_non_finite_values_to_missing():
|
|
values = np.array([np.inf, np.nan, 500.0, 0.0], dtype=np.float32)
|
|
|
|
decoded = depth.to_millimetres(FakeStorage(values), {"shape": [2, 2]})
|
|
|
|
np.testing.assert_allclose(decoded, [[0.0, 0.0], [500.0, 0.0]])
|
|
|
|
|
|
def test_depth_without_a_shape_is_an_error():
|
|
with pytest.raises(RuntimeError, match="shape"):
|
|
depth.to_millimetres(FakeStorage(np.zeros(4, dtype=np.float32)), {})
|
|
|
|
|
|
def test_colorize_leaves_missing_depth_black_and_colours_the_rest():
|
|
depth_mm = np.array([[0.0, 500.0], [1000.0, 1500.0]], dtype=np.float32)
|
|
|
|
frame = depth.colorize(depth_mm)
|
|
|
|
assert frame.shape == (2, 2, 3)
|
|
np.testing.assert_array_equal(frame[0, 0], [0, 0, 0])
|
|
assert frame[1, 1].any(), "valid depth should be coloured"
|
|
|
|
|
|
def test_depth_range_is_pinned_by_the_environment(monkeypatch):
|
|
monkeypatch.setenv("VIEW_DEPTH_MIN_MM", "200")
|
|
monkeypatch.setenv("VIEW_DEPTH_MAX_MM", "600")
|
|
|
|
assert depth.resolve_range(np.array([[1000.0]], dtype=np.float32)) == (200.0, 600.0)
|
|
|
|
|
|
def test_one_pinned_end_leaves_the_other_auto_ranged(monkeypatch):
|
|
monkeypatch.setenv("VIEW_DEPTH_MIN_MM", "0")
|
|
|
|
low, high = depth.resolve_range(np.array([[100.0, 900.0]], dtype=np.float32))
|
|
|
|
assert low == 0.0
|
|
assert high > 0.0
|
|
|
|
|
|
def test_depth_range_of_an_empty_frame_does_not_divide_by_zero():
|
|
low, high = depth.resolve_range(np.zeros((2, 2), dtype=np.float32))
|
|
|
|
assert high > low
|
|
|
|
|
|
def test_camera_info_reads_the_compact_four_value_form():
|
|
values = np.array([600.0, 601.0, 320.0, 240.0], dtype=np.float32)
|
|
|
|
model = intrinsics.parse(FakeStorage(values), {"distortion": [0.1, 0.2]})
|
|
|
|
assert (model.fx, model.fy, model.cx, model.cy) == (600.0, 601.0, 320.0, 240.0)
|
|
assert model.distortion == (0.1, 0.2)
|
|
|
|
|
|
def test_camera_info_reads_a_ros_k_matrix():
|
|
# Row-major K: [fx, 0, cx, 0, fy, cy, 0, 0, 1].
|
|
k = np.array(
|
|
[600.0, 0.0, 320.0, 0.0, 601.0, 240.0, 0.0, 0.0, 1.0],
|
|
dtype=np.float32,
|
|
)
|
|
|
|
model = intrinsics.parse(FakeStorage(k), {})
|
|
|
|
assert (model.fx, model.fy, model.cx, model.cy) == (600.0, 601.0, 320.0, 240.0)
|
|
|
|
|
|
def test_camera_info_reinterprets_a_payload_sent_as_bytes():
|
|
values = np.array([600.0, 601.0, 320.0, 240.0], dtype=np.float32)
|
|
|
|
model = intrinsics.parse(FakeStorage(values.view(np.uint8)), {})
|
|
|
|
assert model.fx == 600.0
|
|
assert model.cy == 240.0
|
|
|
|
|
|
def test_camera_info_that_is_too_short_is_an_error():
|
|
with pytest.raises(RuntimeError, match="camera_info"):
|
|
intrinsics.parse(FakeStorage(np.zeros(3, dtype=np.float32)), {})
|
|
|
|
|
|
def test_intrinsics_are_read_from_frame_metadata():
|
|
model = intrinsics.from_metadata(
|
|
{"fx": 600.0, "fy": 601.0, "cx": 320.0, "cy": 240.0},
|
|
)
|
|
|
|
assert model is not None
|
|
assert (model.fx, model.cy) == (600.0, 240.0)
|
|
|
|
|
|
def test_frame_metadata_without_intrinsics_yields_none():
|
|
assert intrinsics.from_metadata({"width": 640, "height": 480}) is None
|
|
|
|
|
|
def test_the_camera_model_is_drawn_on_the_composite():
|
|
view = View(headless=True)
|
|
view.panels["image"] = np.zeros((40, 80, 3), dtype=np.uint8)
|
|
|
|
without = view.compose().copy()
|
|
view.intrinsics = intrinsics.Intrinsics(600.0, 601.0, 320.0, 240.0)
|
|
with_model = view.compose()
|
|
|
|
assert not np.array_equal(without, with_model), "the model should be drawn"
|
|
|
|
|
|
def _image_struct(frame: np.ndarray, encoding: str = "bgr8"):
|
|
"""Build a payload shaped like std/media/v1/Image."""
|
|
return pa.array(
|
|
[{
|
|
"width": frame.shape[1],
|
|
"height": frame.shape[0],
|
|
"encoding": encoding,
|
|
"data": frame.tobytes(),
|
|
}],
|
|
type=pa.struct([
|
|
("width", pa.uint32()),
|
|
("height", pa.uint32()),
|
|
("encoding", pa.utf8()),
|
|
("data", pa.large_binary()),
|
|
]),
|
|
)
|
|
|
|
|
|
def test_image_struct_is_decoded(frame):
|
|
# No metadata at all: the struct describes itself.
|
|
decoded = decode.to_bgr(_image_struct(frame), {})
|
|
|
|
np.testing.assert_array_equal(decoded, frame)
|
|
|
|
|
|
def test_image_struct_wins_over_message_metadata(frame):
|
|
# Metadata disagreeing with the struct must not be consulted.
|
|
decoded = decode.to_bgr(_image_struct(frame), _meta("rgb8", 1, 1))
|
|
|
|
np.testing.assert_array_equal(decoded, frame)
|
|
|
|
|
|
def test_image_struct_carrying_an_encoded_frame_is_decoded(frame):
|
|
ok, encoded = cv2.imencode(".png", frame)
|
|
assert ok
|
|
payload = np.frombuffer(encoded.tobytes(), dtype=np.uint8).reshape(1, -1)
|
|
|
|
decoded = decode.to_bgr(_image_struct(payload, "png"), {})
|
|
|
|
np.testing.assert_array_equal(decoded, frame)
|
|
|
|
|
|
def test_a_decoded_image_struct_frame_is_writable(frame):
|
|
decoded = decode.to_bgr(_image_struct(frame), {})
|
|
decoded[0, 0] = (1, 2, 3)
|
|
|
|
assert tuple(decoded[0, 0]) == (1, 2, 3)
|