cristhian
8a7bda5867
Both inputs now declare dora's Arrow struct {width, height, encoding, data}
rather than a flat buffer plus loose metadata. The payload describes itself, so
nothing about its shape has to be agreed out of band, and the frame travels as
one Arrow value instead of a buffer whose meaning lives in a sidecar dict.
Depth uses the same type. dora's registry has no depth type, and a depth map is
an image whose pixels are distances -- the reading ROS takes as well -- so it
arrives with a single-channel encoding: 32FC1/float32 or 16UC1/mono16/uint16.
`units` stays in message metadata because the struct has no field for it.
decode.IMAGE_TYPE is the single definition of the struct; the example source
and the tests build payloads from it rather than respelling the schema.
The older shapes are still decoded -- a flat UInt8 buffer with
width/height/encoding for image, a flat Float32 buffer with shape/units for
depth -- so a producer's payload can move before its manifest does. That is
only half a migration path: type compatibility is directional, so a producer
still declaring std/core/v1/UInt8 fails validate against an
std/media/v1/Image input regardless of what it sends, and has to update its
manifest too.
Verified: 41 tests, validate clean with 4 contracts injected, the dataflow runs
to a clean exit, and a composite rendered through the struct path is identical
to the one rendered before the migration.
375 lines
11 KiB
Python
375 lines
11 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=decode.IMAGE_TYPE,
|
|
)
|
|
|
|
|
|
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)
|
|
|
|
|
|
def test_depth_is_decoded_from_an_image_struct():
|
|
values = np.array([[100.0, 200.0], [300.0, 400.0]], dtype=np.float32)
|
|
|
|
# No shape metadata: the struct carries its own dimensions.
|
|
decoded = depth.to_millimetres(_image_struct(values, "32FC1"), {})
|
|
|
|
np.testing.assert_allclose(decoded, values)
|
|
|
|
|
|
def test_depth_image_struct_reads_a_uint16_encoding():
|
|
values = np.array([[1500, 2500]], dtype=np.uint16)
|
|
|
|
decoded = depth.to_millimetres(_image_struct(values, "16UC1"), {})
|
|
|
|
np.testing.assert_allclose(decoded, [[1500.0, 2500.0]])
|
|
|
|
|
|
def test_depth_image_struct_still_honours_the_units_metadata():
|
|
values = np.array([[1.5, 2.0]], dtype=np.float32)
|
|
|
|
decoded = depth.to_millimetres(_image_struct(values, "32FC1"), {"units": "m"})
|
|
|
|
np.testing.assert_allclose(decoded, [[1500.0, 2000.0]])
|
|
|
|
|
|
def test_depth_image_struct_with_an_unsupported_encoding_is_rejected():
|
|
values = np.zeros((1, 2), dtype=np.float32)
|
|
|
|
with pytest.raises(RuntimeError, match="unsupported depth encoding"):
|
|
depth.to_millimetres(_image_struct(values, "rgb8"), {})
|
|
|
|
|
|
def test_a_decoded_depth_struct_is_writable():
|
|
values = np.array([[100.0, 0.0]], dtype=np.float32)
|
|
|
|
decoded = depth.to_millimetres(_image_struct(values, "32FC1"), {})
|
|
decoded[0, 0] = 42.0
|
|
|
|
assert decoded[0, 0] == 42.0
|