#!/usr/bin/env python from openflexure_microscope.camera.pi import StreamingCamera, CaptureObject import os import io import time import numpy as np from PIL import Image import unittest import logging import sys logging.basicConfig(stream=sys.stderr, level=logging.DEBUG) class TestCaptureMethods(unittest.TestCase): def test_still_capture(self): """Tests capturing still images to a BytesIO stream.""" global camera for use_video_port in [True, False]: for resize in [None, (640, 480)]: # Wait for camera camera.wait_for_camera() # Capture to a context (auto-deletes files when done) with camera.new_image(write_to_file=False) as output: camera.capture( output, use_video_port=use_video_port, resize=resize ) # Ensure file deletion fails and returns False self.assertFalse(output.delete_file()) # Ensure capture not stored to file self.assertFalse(os.path.isfile(output.file)) # BEFORE DELETE: Ensure StreamObject 'stream' has # a valid BytesIO object and byte string self.assertTrue(isinstance( output.data, io.IOBase )) self.assertTrue(isinstance( output.binary, (bytes, bytearray) )) # Save capture to file output.save_file() # Check file got saved self.assertTrue(os.path.isfile(output.file)) # Delete file output.delete_file() # Check file got deleted self.assertFalse(os.path.isfile(output.file)) # AFTER DELETE: Ensure StreamObject 'stream' has # a valid BytesIO object and byte string self.assertTrue(isinstance(output.data, io.IOBase)) self.assertTrue(isinstance( output.binary, (bytes, bytearray) )) # Create a PIL image from stream image = Image.open(output.data) # Ensure a valid PIL image was created self.assertTrue(isinstance(image, Image.Image)) # Calculate expected dimensions if resize: dims = resize else: if use_video_port: dims = camera.config['video_resolution'] else: dims = camera.config['image_resolution'] # Ensure PIL image size matches expected size print(image.size, dims) self.assertTrue(image.size == dims) def test_still_store(self): """Tests capturing still images to a file on disk.""" global camera for use_video_port in [True, False]: for resize in [None, (640, 480)]: # Wait for camera camera.wait_for_camera() # Capture output = camera.capture( camera.new_image(write_to_file=True), use_video_port=use_video_port, resize=resize) # Check file got saved self.assertTrue(os.path.isfile(output.file)) statinfo = os.stat(output.file) self.assertTrue(statinfo.st_size > 0) # Ensure StreamObject 'stream' has # a valid BytesIO object and byte string self.assertTrue(isinstance(output.data, io.IOBase)) self.assertTrue(isinstance(output.binary, (bytes, bytearray))) # Ensure file deletion completes and returns True self.assertTrue(output.delete_file()) # Check file got deleted self.assertFalse(os.path.isfile(output.file)) class TestUnencodedMethods(unittest.TestCase): def test_yuv_array(self): """Tests capturing unencoded YUV data to a Numpy array.""" global camera for use_video_port in [True, False]: for resize in [None, (640, 480)]: print("{}, {}, {}".format(id, resize, use_video_port)) # Wait for camera camera.wait_for_camera() # Capture RGB array yuv = camera.yuv( use_video_port=use_video_port, resize=resize) # Ensure capture output is a valid numpy array self.assertTrue(isinstance(yuv, np.ndarray)) # Calculate expected dimensions if resize: dims = resize else: if use_video_port: dims = camera.config['video_resolution'] else: dims = camera.config['numpy_resolution'] # Ensure array shape matches expected dimensions self.assertTrue(yuv.shape == (dims[1], dims[0], 3)) def test_rgb_array(self): """Tests capturing unencoded YUV/RGB data to a Numpy array.""" global camera for use_video_port in [True, False]: for resize in [None, (640, 480)]: print("{}, {}, {}".format(id, resize, use_video_port)) # Wait for camera camera.wait_for_camera() # Capture RGB array rgb = camera.array( use_video_port=use_video_port, resize=resize) # Ensure capture output is a valid numpy array self.assertTrue(isinstance(rgb, np.ndarray)) # Calculate expected dimensions if resize: dims = resize else: if use_video_port: dims = camera.config['video_resolution'] else: dims = camera.config['numpy_resolution'] # Ensure array shape matches expected dimensions self.assertTrue(rgb.shape == (dims[1], dims[0], 3)) class TestRecordMethods(unittest.TestCase): def test_video_record(self): """Tests recording videos to BytesIO stream, and to file on disk.""" global camera for write_to_file in [True, False, None]: logging.debug("\nWRITE TO FILE: {}".format(write_to_file)) # Wait for camera camera.wait_for_camera() with camera.new_video( write_to_file=write_to_file ) as output: # Start recording camera.start_recording(output) # Record for 2 seconds time.sleep(2) # Stop recording camera.stop_recording() # Check stream self.assertTrue(isinstance(output.data, io.IOBase)) self.assertTrue(isinstance(output.binary, (bytes, bytearray))) # Check file if write_to_file: statinfo = os.stat(output.file) self.assertTrue(statinfo.st_size > 0) # Log path temp_path = output.file # Check file got deleted on __exit__ self.assertFalse(os.path.isfile(temp_path)) time.sleep(0.25) def test_video_store(self): """Tests recording videos to file on disk, without context manager.""" global camera # Wait for camera camera.wait_for_camera() # Start recording output = camera.start_recording(camera.new_video()) # Record for 2 seconds time.sleep(2) # Stop recording camera.stop_recording() # Check file statinfo = os.stat(output.file) self.assertTrue(statinfo.st_size > 0) # Ensure file deletion completes and returns True self.assertTrue(output.delete_file()) # Check file got deleted self.assertFalse(os.path.isfile(output.file)) class TestThreadStarting(unittest.TestCase): def test_restarting_stream(self): """Tests that a capture call restarts the camera worker thread.""" global camera # Wait for camera camera.wait_for_camera() # Force-stop the camera worker thread (should return True) self.assertTrue(camera.stop_worker()) # Check Pi camera has been disconnected self.assertTrue(camera.camera) self.assertFalse(camera.thread) # Restart worker thread self.assertTrue(camera.start_worker()) self.assertTrue(camera.camera) self.assertTrue(camera.thread) self.assertIsInstance(camera.stream, io.IOBase) if __name__ == '__main__': with StreamingCamera() as camera: suites = [ unittest.TestLoader().loadTestsFromTestCase(TestCaptureMethods), unittest.TestLoader().loadTestsFromTestCase(TestUnencodedMethods), unittest.TestLoader().loadTestsFromTestCase(TestThreadStarting), unittest.TestLoader().loadTestsFromTestCase(TestRecordMethods), ] alltests = unittest.TestSuite(suites) result = unittest.TextTestRunner(verbosity=2).run(alltests)