From 75cceaeb2ad451d46711679c085ff0580d7a2ef7 Mon Sep 17 00:00:00 2001 From: Andy Liu Date: Sun, 2 Aug 2026 18:05:06 +0800 Subject: [PATCH 1/3] Bound camera RPC waits and feed shutdown --- anki_vector/camera.py | 197 ++++++++++++++++++++++++++++++------------ tests/__init__.py | 1 + tests/test_camera.py | 160 ++++++++++++++++++++++++++++++++++ 3 files changed, 304 insertions(+), 54 deletions(-) create mode 100644 tests/__init__.py create mode 100644 tests/test_camera.py diff --git a/anki_vector/camera.py b/anki_vector/camera.py index f0845c6..332095f 100644 --- a/anki_vector/camera.py +++ b/anki_vector/camera.py @@ -29,15 +29,13 @@ "CameraComponent", "CameraConfig", "CameraImage"] import asyncio -from concurrent.futures import CancelledError +from concurrent import futures import io -import time import sys +import threading +import time -from . import annotate, connection, util -from .events import Events -from .exceptions import VectorCameraFeedException, VectorCameraImageCaptureException -from .messaging import protocol +import grpc try: import numpy as np @@ -49,6 +47,16 @@ except ImportError: sys.exit("Cannot import from PIL: Do `pip3 install --user Pillow` to install") +from . import annotate, connection, util +from .events import Events +from .exceptions import (VectorCameraFeedException, + VectorCameraImageCaptureException, + VectorTimeoutException) +from .messaging import protocol + + +_CAMERA_RPC_TIMEOUT = 10.0 + def _convert_to_pillow_image(image_data: bytes) -> Image.Image: """Convert raw image bytes to a Pillow Image.""" @@ -278,6 +286,7 @@ def __init__(self, robot): self._latest_image: CameraImage = None self._latest_image_id: int = None self._camera_feed_task: asyncio.Task = None + self._camera_feed_stream = None self._enabled = False self._config = None # type CameraConfig self._gain = 0.0 @@ -398,7 +407,15 @@ def image_annotator(self) -> annotate.ImageAnnotator: """ return self._image_annotator - def init_camera_feed(self) -> None: + async def _start_camera_feed(self) -> None: + """Create the camera feed task on the connection event loop.""" + if not self._camera_feed_task or self._camera_feed_task.done(): + self._enabled = True + self._camera_feed_task = asyncio.ensure_future( + self._request_and_handle_images()) + + def init_camera_feed( + self, timeout: float = _CAMERA_RPC_TIMEOUT) -> None: """Begin camera feed task. .. testcode:: @@ -409,47 +426,82 @@ def init_camera_feed(self) -> None: robot.camera.init_camera_feed() image = robot.camera.latest_image image.raw_image.show() + + :param timeout: Number of seconds to wait for the feed task to be + scheduled on the connection thread. """ - if not self._camera_feed_task or self._camera_feed_task.done(): - self._enabled = True - self._camera_feed_task = self.conn.loop.create_task(self._request_and_handle_images()) - - def close_camera_feed(self) -> None: - """Cancel camera feed task.""" - if self._camera_feed_task: - self._enabled = False - self._camera_feed_task.cancel() - future = self.conn.run_coroutine(self._camera_feed_task) + if timeout <= 0: + raise ValueError('timeout must be greater than zero') + if threading.current_thread() is self.conn.thread: + asyncio.ensure_future(self._start_camera_feed()) + return + + future = self.conn.run_coroutine(self._start_camera_feed()) + try: + future.result(timeout=timeout) + except futures.TimeoutError as exc: + future.cancel() + raise VectorTimeoutException(None) from exc + + async def _close_camera_feed(self, timeout: float) -> None: + """Cancel and await the camera feed task on its owning event loop.""" + deadline = self.conn.loop.time() + timeout + self._enabled = False + task = self._camera_feed_task + self._camera_feed_task = None + stream = self._camera_feed_stream + if stream is not None: + stream.cancel() + if task and not task.done(): try: - future.result() - except CancelledError: - self.logger.debug('Camera feed task was cancelled. This is expected during disconnection.') - # wait for streaming to end, up to 10 seconds - iterations = 0 - max_iterations = 100 - while self.image_streaming_enabled(): - time.sleep(0.1) - iterations += 1 - if iterations > max_iterations: - # leave loop, even if streaming is still enabled - # because other SDK functions will still work and - # the RPC should have had enough time to finish - # which means we _should_ be in a good state. - self.logger.info('Camera Feed closed, but streaming on' - ' robot remained enabled. This is unexpected.') - break - self._camera_feed_task = None - - async def _image_streaming_enabled(self) -> bool: + await asyncio.wait_for(task, timeout=timeout) + except asyncio.TimeoutError as exc: + task.cancel() + raise VectorTimeoutException(None) from exc + + while True: + remaining = deadline - self.conn.loop.time() + if remaining <= 0: + raise VectorTimeoutException(None) + enabled = await self._image_streaming_enabled( + min(1.0, remaining)) + if not enabled: + return + await asyncio.sleep(min(0.1, remaining)) + + def close_camera_feed( + self, timeout: float = _CAMERA_RPC_TIMEOUT) -> None: + """Cancel the camera feed task within a bounded time. + + :param timeout: Number of seconds to wait for the stream to close. + """ + if timeout <= 0: + raise ValueError('timeout must be greater than zero') + if not self._camera_feed_task: + return + if threading.current_thread() is self.conn.thread: + asyncio.ensure_future(self._close_camera_feed(timeout)) + return + + future = self.conn.run_coroutine(self._close_camera_feed(timeout)) + try: + future.result(timeout=timeout + 1.0) + except futures.TimeoutError as exc: + future.cancel() + raise VectorTimeoutException(None) from exc + + async def _image_streaming_enabled(self, timeout: float) -> bool: """request streaming enabled status from the robot""" request = protocol.IsImageStreamingEnabledRequest() - response = await self.conn.grpc_interface.IsImageStreamingEnabled(request) + response = await self.conn.grpc_interface.IsImageStreamingEnabled( + request, timeout=timeout) enabled = False if response: enabled = response.is_image_streaming_enabled return enabled - def image_streaming_enabled(self) -> bool: + def image_streaming_enabled( + self, timeout: float = _CAMERA_RPC_TIMEOUT) -> bool: """True if image streaming is enabled on the robot .. testcode:: @@ -461,9 +513,18 @@ def image_streaming_enabled(self) -> bool: print("Robot is streaming video") else: print("Robot is not streaming video") + + :param timeout: Number of seconds to wait for the status RPC. """ - future = self.conn.run_coroutine(self._image_streaming_enabled()) - return future.result() + if timeout <= 0: + raise ValueError('timeout must be greater than zero') + future = self.conn.run_coroutine( + self._image_streaming_enabled(timeout)) + try: + return future.result(timeout=timeout + 1.0) + except futures.TimeoutError as exc: + future.cancel() + raise VectorTimeoutException(None) from exc def _unpack_image(self, msg: protocol.CameraFeedResponse) -> None: """Processes raw data from the robot into a more useful image structure.""" @@ -484,27 +545,45 @@ def _unpack_image(self, msg: protocol.CameraFeedResponse) -> None: async def _request_and_handle_images(self) -> None: """Queries and listens for camera feed events from the robot. Received events are parsed by a helper function.""" + stream = None try: req = protocol.CameraFeedRequest() - async for evt in self.grpc_interface.CameraFeed(req): + stream = self.grpc_interface.CameraFeed(req) + self._camera_feed_stream = stream + async for evt in stream: # If the camera feed is disabled after stream is setup, exit the stream # (the camera feed on the robot is disabled internally on stream exit) if not self._enabled: self.logger.warning('Camera feed has been disabled. Enable the feed to start/continue receiving camera feed data') return self._unpack_image(evt) - except CancelledError: - self.logger.debug('Camera feed task was cancelled. This is expected during disconnection.') + except asyncio.CancelledError: + self.logger.debug( + 'Camera feed task was cancelled. This is expected during ' + 'disconnection.') + raise + except grpc.RpcError as exc: + if exc.code() != grpc.StatusCode.CANCELLED: + raise + finally: + self._camera_feed_stream = None + if stream is not None: + stream.cancel() @connection.on_connection_thread() - async def capture_single_image(self, enable_high_resolution: bool = False) -> CameraImage: + async def capture_single_image( + self, + enable_high_resolution: bool = False, + timeout: float = _CAMERA_RPC_TIMEOUT) -> CameraImage: """Request to capture a single image from the robot's camera. This call requests the robot to capture an image and returns the - received image, formatted as a Pillow image. This differs from `latest_image`, - which maintains the last image received from the camera feed (if enabled). + received image, formatted as a Pillow image. This differs from + `latest_image`, which maintains the last image received from the camera + feed (if enabled). - Note that when the camera feed is enabled this call returns the `latest_image`. + Note that when the camera feed is enabled this call returns the + `latest_image`. .. testcode:: @@ -514,16 +593,26 @@ async def capture_single_image(self, enable_high_resolution: bool = False) -> Ca image = robot.camera.capture_single_image() image.raw_image.show() - :param enable_high_resolution: Enable/disable request for high resolution images. The default resolution - is 640x360, while the high resolution is 1280x720. + :param enable_high_resolution: Enable/disable request for high + resolution images. The default resolution is 640x360, while the + high resolution is 1280x720. + :param timeout: Number of seconds to wait for the gRPC response. """ + if timeout <= 0: + raise ValueError('timeout must be greater than zero') if self._enabled: - self.logger.warning('Camera feed is enabled. Receiving image from the feed at default resolution.') + self.logger.warning( + 'Camera feed is enabled. Receiving image from the feed at ' + 'default resolution.') return self._latest_image if enable_high_resolution: - self.logger.warning('Capturing a high resolution (1280*720) image. Image events for this frame need to be scaled.') - req = protocol.CaptureSingleImageRequest(enable_high_resolution=enable_high_resolution) - res = await self.grpc_interface.CaptureSingleImage(req) + self.logger.warning( + 'Capturing a high resolution (1280*720) image. Image events ' + 'for this frame need to be scaled.') + req = protocol.CaptureSingleImageRequest( + enable_high_resolution=enable_high_resolution) + res = await self.grpc_interface.CaptureSingleImage( + req, timeout=timeout) if res and res.data: image = _convert_to_pillow_image(res.data) return CameraImage(image, self._image_annotator, res.image_id) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..51afbd1 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Tests for the Vector SDK.""" diff --git a/tests/test_camera.py b/tests/test_camera.py new file mode 100644 index 0000000..338ef27 --- /dev/null +++ b/tests/test_camera.py @@ -0,0 +1,160 @@ +"""Tests for bounded camera gRPC calls and feed cancellation.""" + +import asyncio +from concurrent import futures +import io +from types import SimpleNamespace +import unittest +from unittest import mock + +from PIL import Image + +from anki_vector import camera + + +class _CameraGrpcInterface: + + def __init__(self, image_data): + self.image_data = image_data + self.capture_timeout = None + self.status_timeout = None + + async def CaptureSingleImage(self, _request, timeout=None): + self.capture_timeout = timeout + return SimpleNamespace(data=self.image_data, image_id=11) + + async def IsImageStreamingEnabled(self, _request, timeout=None): + self.status_timeout = timeout + return SimpleNamespace(is_image_streaming_enabled=True) + + +class _HangingCameraStream: + + def __init__(self): + self.cancelled = False + + def __aiter__(self): + return self + + async def __anext__(self): + await asyncio.Future() + + def cancel(self): + self.cancelled = True + + +class _StreamGrpcInterface: + + def __init__(self, stream): + self.stream = stream + + def CameraFeed(self, _request): + return self.stream + + +class _TimedOutFuture: + + def __init__(self): + self.result_timeout = None + self.cancelled = False + + def result(self, timeout=None): + self.result_timeout = timeout + raise futures.TimeoutError() + + def cancel(self): + self.cancelled = True + + +def _jpeg_data(): + output = io.BytesIO() + Image.new('RGB', (8, 6), color=(20, 40, 60)).save( + output, format='JPEG') + return output.getvalue() + + +class CameraTests(unittest.TestCase): + + def setUp(self): + self.loop = asyncio.new_event_loop() + self.addCleanup(self.loop.close) + + def _make_component(self, grpc_interface): + connection = SimpleNamespace(grpc_interface=grpc_interface) + robot = SimpleNamespace(conn=connection) + component = object.__new__(camera.CameraComponent) + component._robot = robot + component._enabled = False + component._image_annotator = mock.Mock() + component._camera_feed_task = None + component._camera_feed_stream = None + component.logger = mock.Mock() + return component + + def _capture(self, component, **kwargs): + capture = camera.CameraComponent.capture_single_image.__wrapped__ + return self.loop.run_until_complete(capture(component, **kwargs)) + + def test_capture_passes_timeout_to_grpc(self): + grpc_interface = _CameraGrpcInterface(_jpeg_data()) + component = self._make_component(grpc_interface) + + image = self._capture(component, timeout=2.5) + + self.assertEqual(grpc_interface.capture_timeout, 2.5) + self.assertEqual(image.image_id, 11) + self.assertEqual(image.raw_image.size, (8, 6)) + + def test_capture_rejects_non_positive_timeout(self): + component = self._make_component(None) + + with self.assertRaises(ValueError): + self._capture(component, timeout=0) + + def test_streaming_status_passes_timeout_to_grpc(self): + grpc_interface = _CameraGrpcInterface(_jpeg_data()) + component = self._make_component(grpc_interface) + + enabled = self.loop.run_until_complete( + component._image_streaming_enabled(1.5)) + + self.assertTrue(enabled) + self.assertEqual(grpc_interface.status_timeout, 1.5) + + def test_cancelled_camera_task_cancels_grpc_iterator(self): + stream = _HangingCameraStream() + component = self._make_component(_StreamGrpcInterface(stream)) + component._enabled = True + task = self.loop.create_task( + component._request_and_handle_images()) + self.loop.run_until_complete(asyncio.sleep(0)) + + task.cancel() + with self.assertRaises(asyncio.CancelledError): + self.loop.run_until_complete(task) + + self.assertTrue(stream.cancelled) + + def test_close_camera_feed_has_outer_timeout(self): + timed_out_future = _TimedOutFuture() + + def run_coroutine(coroutine): + coroutine.close() + return timed_out_future + + connection = SimpleNamespace( + run_coroutine=run_coroutine, + thread=object()) + component = self._make_component(None) + component._robot.conn = connection + component._camera_feed_task = object() + + with self.assertRaises(camera.VectorTimeoutException): + component.close_camera_feed(timeout=0.25) + + self.assertEqual(timed_out_future.result_timeout, 1.25) + self.assertTrue(timed_out_future.cancelled) + + +if __name__ == '__main__': + unittest.main() From 80c1a7b86f10f799587775ce4953ba9cb722a222 Mon Sep 17 00:00:00 2001 From: Andy Liu Date: Sun, 2 Aug 2026 18:17:43 +0800 Subject: [PATCH 2/3] Minimize camera timeout fix --- anki_vector/camera.py | 130 ++++++++++++++---------------------------- tests/__init__.py | 1 - tests/test_camera.py | 126 ++++++++-------------------------------- 3 files changed, 66 insertions(+), 191 deletions(-) delete mode 100644 tests/__init__.py diff --git a/anki_vector/camera.py b/anki_vector/camera.py index 332095f..b64406a 100644 --- a/anki_vector/camera.py +++ b/anki_vector/camera.py @@ -32,7 +32,6 @@ from concurrent import futures import io import sys -import threading import time import grpc @@ -407,15 +406,7 @@ def image_annotator(self) -> annotate.ImageAnnotator: """ return self._image_annotator - async def _start_camera_feed(self) -> None: - """Create the camera feed task on the connection event loop.""" - if not self._camera_feed_task or self._camera_feed_task.done(): - self._enabled = True - self._camera_feed_task = asyncio.ensure_future( - self._request_and_handle_images()) - - def init_camera_feed( - self, timeout: float = _CAMERA_RPC_TIMEOUT) -> None: + def init_camera_feed(self) -> None: """Begin camera feed task. .. testcode:: @@ -426,69 +417,43 @@ def init_camera_feed( robot.camera.init_camera_feed() image = robot.camera.latest_image image.raw_image.show() - - :param timeout: Number of seconds to wait for the feed task to be - scheduled on the connection thread. """ - if timeout <= 0: - raise ValueError('timeout must be greater than zero') - if threading.current_thread() is self.conn.thread: - asyncio.ensure_future(self._start_camera_feed()) - return - - future = self.conn.run_coroutine(self._start_camera_feed()) - try: - future.result(timeout=timeout) - except futures.TimeoutError as exc: - future.cancel() - raise VectorTimeoutException(None) from exc - - async def _close_camera_feed(self, timeout: float) -> None: - """Cancel and await the camera feed task on its owning event loop.""" - deadline = self.conn.loop.time() + timeout - self._enabled = False - task = self._camera_feed_task - self._camera_feed_task = None - stream = self._camera_feed_stream - if stream is not None: - stream.cancel() - if task and not task.done(): - try: - await asyncio.wait_for(task, timeout=timeout) - except asyncio.TimeoutError as exc: - task.cancel() - raise VectorTimeoutException(None) from exc + if not self._camera_feed_task or self._camera_feed_task.done(): + self._enabled = True + self._camera_feed_task = self.conn.loop.create_task(self._request_and_handle_images()) - while True: - remaining = deadline - self.conn.loop.time() - if remaining <= 0: - raise VectorTimeoutException(None) - enabled = await self._image_streaming_enabled( - min(1.0, remaining)) - if not enabled: - return - await asyncio.sleep(min(0.1, remaining)) - - def close_camera_feed( - self, timeout: float = _CAMERA_RPC_TIMEOUT) -> None: + def close_camera_feed(self, timeout: float = _CAMERA_RPC_TIMEOUT) -> None: """Cancel the camera feed task within a bounded time. :param timeout: Number of seconds to wait for the stream to close. """ if timeout <= 0: raise ValueError('timeout must be greater than zero') - if not self._camera_feed_task: - return - if threading.current_thread() is self.conn.thread: - asyncio.ensure_future(self._close_camera_feed(timeout)) - return - - future = self.conn.run_coroutine(self._close_camera_feed(timeout)) - try: - future.result(timeout=timeout + 1.0) - except futures.TimeoutError as exc: - future.cancel() - raise VectorTimeoutException(None) from exc + if self._camera_feed_task: + deadline = time.monotonic() + timeout + self._enabled = False + if self._camera_feed_stream is not None: + self._camera_feed_stream.cancel() + else: + self._camera_feed_task.cancel() + future = self.conn.run_coroutine(self._camera_feed_task) + try: + future.result(timeout=timeout) + except futures.CancelledError: + self.logger.debug( + 'Camera feed task was cancelled. This is expected during ' + 'disconnection.') + except futures.TimeoutError as exc: + future.cancel() + raise VectorTimeoutException(None) from exc + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise VectorTimeoutException(None) + if not self.image_streaming_enabled(timeout=min(1.0, remaining)): + break + time.sleep(min(0.1, remaining)) + self._camera_feed_task = None async def _image_streaming_enabled(self, timeout: float) -> bool: """request streaming enabled status from the robot""" @@ -500,8 +465,7 @@ async def _image_streaming_enabled(self, timeout: float) -> bool: enabled = response.is_image_streaming_enabled return enabled - def image_streaming_enabled( - self, timeout: float = _CAMERA_RPC_TIMEOUT) -> bool: + def image_streaming_enabled(self, timeout: float = _CAMERA_RPC_TIMEOUT) -> bool: """True if image streaming is enabled on the robot .. testcode:: @@ -518,10 +482,9 @@ def image_streaming_enabled( """ if timeout <= 0: raise ValueError('timeout must be greater than zero') - future = self.conn.run_coroutine( - self._image_streaming_enabled(timeout)) + future = self.conn.run_coroutine(self._image_streaming_enabled(timeout)) try: - return future.result(timeout=timeout + 1.0) + return future.result(timeout=timeout) except futures.TimeoutError as exc: future.cancel() raise VectorTimeoutException(None) from exc @@ -578,12 +541,10 @@ async def capture_single_image( """Request to capture a single image from the robot's camera. This call requests the robot to capture an image and returns the - received image, formatted as a Pillow image. This differs from - `latest_image`, which maintains the last image received from the camera - feed (if enabled). + received image, formatted as a Pillow image. This differs from `latest_image`, + which maintains the last image received from the camera feed (if enabled). - Note that when the camera feed is enabled this call returns the - `latest_image`. + Note that when the camera feed is enabled this call returns the `latest_image`. .. testcode:: @@ -593,26 +554,19 @@ async def capture_single_image( image = robot.camera.capture_single_image() image.raw_image.show() - :param enable_high_resolution: Enable/disable request for high - resolution images. The default resolution is 640x360, while the - high resolution is 1280x720. + :param enable_high_resolution: Enable/disable request for high resolution images. The default resolution + is 640x360, while the high resolution is 1280x720. :param timeout: Number of seconds to wait for the gRPC response. """ if timeout <= 0: raise ValueError('timeout must be greater than zero') if self._enabled: - self.logger.warning( - 'Camera feed is enabled. Receiving image from the feed at ' - 'default resolution.') + self.logger.warning('Camera feed is enabled. Receiving image from the feed at default resolution.') return self._latest_image if enable_high_resolution: - self.logger.warning( - 'Capturing a high resolution (1280*720) image. Image events ' - 'for this frame need to be scaled.') - req = protocol.CaptureSingleImageRequest( - enable_high_resolution=enable_high_resolution) - res = await self.grpc_interface.CaptureSingleImage( - req, timeout=timeout) + self.logger.warning('Capturing a high resolution (1280*720) image. Image events for this frame need to be scaled.') + req = protocol.CaptureSingleImageRequest(enable_high_resolution=enable_high_resolution) + res = await self.grpc_interface.CaptureSingleImage(req, timeout=timeout) if res and res.data: image = _convert_to_pillow_image(res.data) return CameraImage(image, self._image_annotator, res.image_id) diff --git a/tests/__init__.py b/tests/__init__.py deleted file mode 100644 index 51afbd1..0000000 --- a/tests/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Tests for the Vector SDK.""" diff --git a/tests/test_camera.py b/tests/test_camera.py index 338ef27..c2591b3 100644 --- a/tests/test_camera.py +++ b/tests/test_camera.py @@ -1,34 +1,32 @@ """Tests for bounded camera gRPC calls and feed cancellation.""" import asyncio -from concurrent import futures -import io from types import SimpleNamespace import unittest -from unittest import mock - -from PIL import Image from anki_vector import camera class _CameraGrpcInterface: - def __init__(self, image_data): - self.image_data = image_data + def __init__(self, stream=None): + self.stream = stream self.capture_timeout = None self.status_timeout = None async def CaptureSingleImage(self, _request, timeout=None): self.capture_timeout = timeout - return SimpleNamespace(data=self.image_data, image_id=11) + return SimpleNamespace(data=b'') async def IsImageStreamingEnabled(self, _request, timeout=None): self.status_timeout = timeout - return SimpleNamespace(is_image_streaming_enabled=True) + return SimpleNamespace(is_image_streaming_enabled=False) + + def CameraFeed(self, _request): + return self.stream -class _HangingCameraStream: +class _CameraStream: def __init__(self): self.cancelled = False @@ -43,36 +41,6 @@ def cancel(self): self.cancelled = True -class _StreamGrpcInterface: - - def __init__(self, stream): - self.stream = stream - - def CameraFeed(self, _request): - return self.stream - - -class _TimedOutFuture: - - def __init__(self): - self.result_timeout = None - self.cancelled = False - - def result(self, timeout=None): - self.result_timeout = timeout - raise futures.TimeoutError() - - def cancel(self): - self.cancelled = True - - -def _jpeg_data(): - output = io.BytesIO() - Image.new('RGB', (8, 6), color=(20, 40, 60)).save( - output, format='JPEG') - return output.getvalue() - - class CameraTests(unittest.TestCase): def setUp(self): @@ -80,81 +48,35 @@ def setUp(self): self.addCleanup(self.loop.close) def _make_component(self, grpc_interface): - connection = SimpleNamespace(grpc_interface=grpc_interface) - robot = SimpleNamespace(conn=connection) component = object.__new__(camera.CameraComponent) - component._robot = robot + component._robot = SimpleNamespace( + conn=SimpleNamespace(grpc_interface=grpc_interface)) component._enabled = False - component._image_annotator = mock.Mock() - component._camera_feed_task = None component._camera_feed_stream = None - component.logger = mock.Mock() + component.logger = SimpleNamespace(debug=lambda *_: None, error=lambda *_: None) return component - def _capture(self, component, **kwargs): - capture = camera.CameraComponent.capture_single_image.__wrapped__ - return self.loop.run_until_complete(capture(component, **kwargs)) - - def test_capture_passes_timeout_to_grpc(self): - grpc_interface = _CameraGrpcInterface(_jpeg_data()) + def test_camera_rpcs_receive_deadlines(self): + grpc_interface = _CameraGrpcInterface() component = self._make_component(grpc_interface) - - image = self._capture(component, timeout=2.5) - - self.assertEqual(grpc_interface.capture_timeout, 2.5) - self.assertEqual(image.image_id, 11) - self.assertEqual(image.raw_image.size, (8, 6)) - - def test_capture_rejects_non_positive_timeout(self): - component = self._make_component(None) - - with self.assertRaises(ValueError): - self._capture(component, timeout=0) - - def test_streaming_status_passes_timeout_to_grpc(self): - grpc_interface = _CameraGrpcInterface(_jpeg_data()) - component = self._make_component(grpc_interface) - - enabled = self.loop.run_until_complete( - component._image_streaming_enabled(1.5)) - - self.assertTrue(enabled) - self.assertEqual(grpc_interface.status_timeout, 1.5) - - def test_cancelled_camera_task_cancels_grpc_iterator(self): - stream = _HangingCameraStream() - component = self._make_component(_StreamGrpcInterface(stream)) + capture = camera.CameraComponent.capture_single_image.__wrapped__ + self.loop.run_until_complete(capture(component, timeout=2.5)) + self.loop.run_until_complete(component._image_streaming_enabled(1.5)) + self.assertEqual( + (grpc_interface.capture_timeout, grpc_interface.status_timeout), + (2.5, 1.5)) + + def test_camera_stream_is_cancelled_with_task(self): + stream = _CameraStream() + component = self._make_component(_CameraGrpcInterface(stream)) component._enabled = True - task = self.loop.create_task( - component._request_and_handle_images()) + task = self.loop.create_task(component._request_and_handle_images()) self.loop.run_until_complete(asyncio.sleep(0)) - task.cancel() with self.assertRaises(asyncio.CancelledError): self.loop.run_until_complete(task) - self.assertTrue(stream.cancelled) - def test_close_camera_feed_has_outer_timeout(self): - timed_out_future = _TimedOutFuture() - - def run_coroutine(coroutine): - coroutine.close() - return timed_out_future - - connection = SimpleNamespace( - run_coroutine=run_coroutine, - thread=object()) - component = self._make_component(None) - component._robot.conn = connection - component._camera_feed_task = object() - - with self.assertRaises(camera.VectorTimeoutException): - component.close_camera_feed(timeout=0.25) - - self.assertEqual(timed_out_future.result_timeout, 1.25) - self.assertTrue(timed_out_future.cancelled) - if __name__ == '__main__': unittest.main() From 6cd0ad256ef20208cb56ea7e75f92b0276e208f2 Mon Sep 17 00:00:00 2001 From: Andy Liu Date: Sun, 2 Aug 2026 19:13:51 +0800 Subject: [PATCH 3/3] Remove camera regression tests --- tests/test_camera.py | 82 -------------------------------------------- 1 file changed, 82 deletions(-) delete mode 100644 tests/test_camera.py diff --git a/tests/test_camera.py b/tests/test_camera.py deleted file mode 100644 index c2591b3..0000000 --- a/tests/test_camera.py +++ /dev/null @@ -1,82 +0,0 @@ -"""Tests for bounded camera gRPC calls and feed cancellation.""" - -import asyncio -from types import SimpleNamespace -import unittest - -from anki_vector import camera - - -class _CameraGrpcInterface: - - def __init__(self, stream=None): - self.stream = stream - self.capture_timeout = None - self.status_timeout = None - - async def CaptureSingleImage(self, _request, timeout=None): - self.capture_timeout = timeout - return SimpleNamespace(data=b'') - - async def IsImageStreamingEnabled(self, _request, timeout=None): - self.status_timeout = timeout - return SimpleNamespace(is_image_streaming_enabled=False) - - def CameraFeed(self, _request): - return self.stream - - -class _CameraStream: - - def __init__(self): - self.cancelled = False - - def __aiter__(self): - return self - - async def __anext__(self): - await asyncio.Future() - - def cancel(self): - self.cancelled = True - - -class CameraTests(unittest.TestCase): - - def setUp(self): - self.loop = asyncio.new_event_loop() - self.addCleanup(self.loop.close) - - def _make_component(self, grpc_interface): - component = object.__new__(camera.CameraComponent) - component._robot = SimpleNamespace( - conn=SimpleNamespace(grpc_interface=grpc_interface)) - component._enabled = False - component._camera_feed_stream = None - component.logger = SimpleNamespace(debug=lambda *_: None, error=lambda *_: None) - return component - - def test_camera_rpcs_receive_deadlines(self): - grpc_interface = _CameraGrpcInterface() - component = self._make_component(grpc_interface) - capture = camera.CameraComponent.capture_single_image.__wrapped__ - self.loop.run_until_complete(capture(component, timeout=2.5)) - self.loop.run_until_complete(component._image_streaming_enabled(1.5)) - self.assertEqual( - (grpc_interface.capture_timeout, grpc_interface.status_timeout), - (2.5, 1.5)) - - def test_camera_stream_is_cancelled_with_task(self): - stream = _CameraStream() - component = self._make_component(_CameraGrpcInterface(stream)) - component._enabled = True - task = self.loop.create_task(component._request_and_handle_images()) - self.loop.run_until_complete(asyncio.sleep(0)) - task.cancel() - with self.assertRaises(asyncio.CancelledError): - self.loop.run_until_complete(task) - self.assertTrue(stream.cancelled) - - -if __name__ == '__main__': - unittest.main()