feat: support multiple camera previews and concurrent agent code navigation

2026-10-20 10:10:48 +09:00
parent c17763f59d
commit af54e875df
13 changed files with 525 additions and 160 deletions
@@ -0,0 +1,67 @@
+import ast
+from pathlib import Path
+import re
+import threading
+from types import SimpleNamespace
+from typing import Optional
+import unittest
+
+
+class CameraTests(unittest.TestCase):
+    def setUp(self):
+        self.now = 10.0
+        source = ast.parse(Path(__file__).with_name('bridge.py').read_text())
+        node = next(item for item in source.body if isinstance(item, ast.ClassDef) and item.name == 'SlamNode')
+        names = ('_discover_cameras', '_on_camera_image', 'camera_sources', 'camera_frame_jpeg')
+        methods = [item for item in node.body if isinstance(item, ast.FunctionDef) and item.name in names]
+        scope = {'re': re, 'Optional': Optional, 'CompressedImage': object,
+                 'time': SimpleNamespace(time=lambda: self.now), '_CAMERA_FRAME_MAX_AGE_SECONDS': 3,
+                 'qos_profile_sensor_data': object()}
+        exec(compile(ast.Module(body=methods, type_ignores=[]), 'camera-bridge', 'exec'), scope)
+        self.node = type('CameraNode', (), {name: scope[name] for name in names})()
+        self.node._lock = threading.Lock()
+        self.node._camera_frames = {}
+        self.node._camera_subscriptions = {}
+        self.node._camera_image = None
+        self.node._last_camera_image_wall_time = None
+        self.node._camera_image_count = 0
+        self.node.camera_image_topic = '/camera_test/image_jpeg'
+        self.topics = []
+        self.callbacks = {}
+        self.node.get_topic_names_and_types = lambda: self.topics
+        def subscribe(kind, topic, callback, qos):
+            self.callbacks[topic] = callback
+            return topic
+        self.node.create_subscription = subscribe
+        self.node.destroy_subscription = lambda sub: self.callbacks.pop(sub)
+
+    def test_independent_frames_and_disconnect(self):
+        for key in ['video0', 'video2']:
+            self.topics.append((f'/camera_test/cameras/{key}/image_jpeg', ['sensor_msgs/msg/CompressedImage']))
+        self.topics += [('/other/image_jpeg', ['sensor_msgs/msg/CompressedImage']),
+                        ('/camera_test/cameras/video9/image_jpeg', ['sensor_msgs/msg/Image'])]
+        self.node._discover_cameras()
+        self.assertEqual(len(self.callbacks), 2)
+        for topic, callback in self.callbacks.items():
+            callback(SimpleNamespace(data=topic.encode(), format='jpeg'))
+        self.assertEqual([item['id'] for item in self.node.camera_sources()], ['video0', 'video2'])
+        self.assertIn(b'video2', self.node.camera_frame_jpeg('video2'))
+        self.assertIsNone(self.node.camera_frame_jpeg('../video0'))
+        self.now += 4
+        self.assertTrue(all(not item['available'] for item in self.node.camera_sources()))
+        self.assertIsNone(self.node.camera_frame_jpeg('video0'))
+        self.topics = self.topics[:1]
+        self.node._discover_cameras()
+        self.assertNotIn('video2', self.node._camera_frames)
+
+    def test_single_camera_compatibility_and_non_jpeg(self):
+        self.node._on_camera_image(SimpleNamespace(data=b'jpeg', format='jpeg'))
+        self.assertEqual(self.node.camera_frame_jpeg(), b'jpeg')
+        self.assertEqual(self.node.camera_frame_jpeg('legacy'), b'jpeg')
+        self.node._on_camera_image(SimpleNamespace(data=b'png', format='png'))
+        self.assertIsNone(self.node.camera_frame_jpeg('legacy'))
+        self.assertFalse(self.node.camera_sources()[0]['available'])
+
+
+if __name__ == '__main__':
+    unittest.main()