53 lines
2.2 KiB
Python
53 lines
2.2 KiB
Python
import time
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import cv2
|
|
import numpy as np
|
|
import rclpy
|
|
from rclpy.qos import qos_profile_sensor_data
|
|
from sensor_msgs.msg import CompressedImage
|
|
from camera_test_pkg import gstreamer_camera as camera
|
|
|
|
|
|
class CameraTest(unittest.TestCase):
|
|
def test_pipeline_formats_and_validation(self):
|
|
camera.Gst.init(None)
|
|
for fmt in ['auto', 'UYVY', 'YUY2', 'MJPEG']:
|
|
pipeline = camera.Gst.parse_launch(camera.capture_pipeline('/dev/video0', 640, 480, 30, fmt))
|
|
pipeline.set_state(camera.Gst.State.NULL)
|
|
for args in [('/tmp/other', 640, 480, 30, 'auto'),
|
|
('/dev/video0', 0, 480, 30, 'auto'),
|
|
('/dev/video0', 640, 480, 30, 'bad')]:
|
|
with self.assertRaises(ValueError):
|
|
camera.capture_pipeline(*args)
|
|
|
|
def test_synthetic_frame_to_ros_jpeg(self):
|
|
# Odd width exercises RGB row padding without accessing physical hardware.
|
|
pipeline = ('videotestsrc is-live=true ! video/x-raw,format=RGB,width=641,height=480,framerate=30/1 ! '
|
|
'appsink name=camera_sink max-buffers=1 drop=true sync=false')
|
|
rclpy.init()
|
|
node = None
|
|
try:
|
|
with patch.object(camera, 'capture_pipeline', return_value=pipeline):
|
|
node = camera.GstreamerCameraTest()
|
|
frames = []
|
|
subscription = node.create_subscription(CompressedImage, '/camera_test/image_jpeg', frames.append,
|
|
qos_profile_sensor_data)
|
|
end = time.monotonic() + 8
|
|
while not frames and time.monotonic() < end:
|
|
rclpy.spin_once(node, timeout_sec=0.1)
|
|
self.assertTrue(frames, 'No ROS JPEG received')
|
|
decoded = cv2.imdecode(np.frombuffer(frames[0].data, dtype=np.uint8), cv2.IMREAD_COLOR)
|
|
self.assertEqual(decoded.shape[1], 640)
|
|
self.assertEqual(frames[0].format, 'jpeg')
|
|
node.destroy_subscription(subscription)
|
|
finally:
|
|
if node:
|
|
node.destroy_node()
|
|
rclpy.shutdown()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|