Improve handling of timeouts and cameras that stop responding

Add a cancel_all_and_flush() function to clear out and cancel any jobs
on the camera queue. This will allow the stop() function to operate
when it would otherwise have got queued behind a job that had hung.

Trying to access the result of any such cancelled job will result in a
CancelledError.

The wait parameter of all the "capture" functions is extended so that
you can also supply an integer, being the time in seconds after which
it will produce a TimeoutError if it has not finished.

Signed-off-by: David Plowman <[email protected]>

picamera2/__init__.py

@@ -1,10 +1,12 @@
 import os
+from concurrent.futures import TimeoutError
 
 import libcamera
 
 from .configuration import CameraConfiguration, StreamConfiguration
 from .controls import Controls
 from .converters import YUV420_to_RGB
+from .job import CancelledError
 from .metadata import Metadata
 from .picamera2 import Picamera2, Preview
 from .platform import Platform, get_platform

picamera2/job.py

@@ -1,4 +1,4 @@
-from concurrent.futures import Future
+from concurrent.futures import CancelledError, Future
 
 
 class Job:
@@ -77,3 +77,11 @@ def get_result(self, timeout=None):
         if necessary for the job to complete.
         """
         return self._future.result(timeout=timeout)
+
+    def cancel(self):
+        """Mark this job as cancelled, so that requesting the result raises a CancelledError.
+
+        User code should not call this because it won't unschedule the job, i.e. remove it
+        from the job queue. Use Picamera2.cancel_all_and_flush() to cancel and clear all jobs.
+        """
+        self._future.set_exception(CancelledError)

picamera2/picamera2.py

@@ -1171,6 +1171,20 @@ def start(self, config=None, show_preview=False) -> None:
             self.start_preview(show_preview)
         self.start_()
 
+    def cancel_all_and_flush(self) -> None:
+        """
+        Clear the camera system queue of pending jobs and cancel them.
+
+        Depending on what was happening at the time, this may leave the camera system in
+        an indeterminate state. This function is really only intended for tidying up
+        after an operation has unexpectedly timed out (for example, the camera cable has
+        become dislodged) so that the camera can be closed.
+        """
+        with self.lock:
+            for job in self._job_list:
+                job.cancel()
+            self._job_list = []
+
     def stop_(self, request=None) -> None:
         """Stop the camera.
 
@@ -1309,9 +1323,18 @@ def dispatch_functions(self, functions, wait, signal_function=None, immediate=Fa
         When there are multiple items each will be processed on a separate
         trip round the event loop, meaning that a single operation could stop and restart the
         camera and the next operation would receive a request from after the restart.
+
+        The wait parameter should be one of:
+            True - wait as long as necessary for the operation to compelte
+            False - return immediately, giving the caller a "job" they can wait for
+            None - default, if a signal_function was given do not wait, otherwise wait as long as necessary
+            a number - wait for this number of seconds before raising a "timed out" error.
         """
         if wait is None:
             wait = signal_function is None
+        timeout = wait
+        if timeout is True:
+            timeout = None
         with self.lock:
             only_job = not self._job_list
             job = Job(functions, signal_function)
@@ -1323,7 +1346,7 @@ def dispatch_functions(self, functions, wait, signal_function=None, immediate=Fa
             # stop commands, for which no requests are needed).
             if only_job and (self.completed_requests or immediate):
                 self._run_process_requests()
-        return job.get_result() if wait else job
+        return job.get_result(timeout=timeout) if wait else job
 
     def set_frame_drops_(self, num_frames):
         """Only for use within the camera event loop before calling drop_frames_."""  # noqa
@@ -1484,9 +1507,9 @@ def capture_request_and_stop_(self):
         return self.dispatch_functions(functions, wait, signal_function, immediate=True)
 
     @contextlib.contextmanager
-    def captured_request(self, flush=None):
+    def captured_request(self, wait=None, flush=None):
         """Capture a completed request using the context manager which guarantees its release."""
-        request = self.capture_request(flush=flush)
+        request = self.capture_request(wait=wait, flush=flush)
         try:
             yield request
         finally:

tests/test_list.txt

@@ -83,3 +83,4 @@ tests/qt_gl_preview_test.py
 tests/stop_slow_framerate.py
 tests/allocator_test.py
 tests/allocator_leak_test.py
+tests/wait_cancel_test.py

tests/wait_cancel_test.py

@@ -0,0 +1,53 @@
+#!/usr/bin/python3
+
+import time
+
+from picamera2 import CancelledError, Picamera2, TimeoutError
+
+# At 2 fps should take over 3s to see the first frame.
+controls = {'FrameRate': 2}
+
+with Picamera2() as picam2:
+    config = picam2.create_preview_configuration(controls=controls)
+    picam2.start(config)
+    t0 = time.monotonic()
+
+    # Test that we time out correctly, and that we can cancel everything so
+    # that we stop quickly.
+    try:
+        array = picam2.capture_array(wait=1.0)
+    except TimeoutError:
+        print("Timed out")
+    else:
+        print("ERROR: operation did not time out")
+
+    t1 = time.monotonic()
+    if t1 - t0 > 2.0:
+        print("ERROR: time out appears to have taken too long")
+
+    picam2.cancel_all_and_flush()
+    picam2.stop()
+    t2 = time.monotonic()
+    print("Stopping took", t2 - t1, "seconds")
+    if t2 - t1 > 0.1:
+        print(f"ERROR: stopping took too long ({t2-t1} seconds)")
+
+with Picamera2() as picam2:
+    config = picam2.create_preview_configuration(controls=controls)
+    picam2.start(config)
+    t0 = time.monotonic()
+
+    # Test that we can cancel a job and get a correct CancelledError.
+    job = picam2.capture_array(wait=False)
+    picam2.cancel_all_and_flush()
+
+    try:
+        array = job.get_result()
+    except CancelledError:
+        print("Job was cancelled")
+    else:
+        print("ERROR: job was not cancelled")
+
+    t1 = time.monotonic()
+    if t1 - t0 > 0.5:
+        print("ERROR: job took too long to cancel")