Updated all log strings to new format
This commit is contained in:
Joel Collins
parent
36e837d374
commit
9f5252194a
35 changed files with 121 additions and 715 deletions
|
|
@ -1,266 +0,0 @@
|
|||
#!/usr/bin/env python
|
||||
import io
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import unittest
|
||||
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
|
||||
from openflexure_microscope.camera.pi import CaptureObject, PiCameraStreamer
|
||||
|
||||
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() 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.stream_resolution
|
||||
else:
|
||||
dims = camera.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().file,
|
||||
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.stream_resolution
|
||||
else:
|
||||
dims = camera.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.stream_resolution
|
||||
else:
|
||||
dims = camera.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
|
||||
|
||||
# Wait for camera
|
||||
camera.wait_for_camera()
|
||||
|
||||
with camera.new_video() 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
|
||||
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)
|
||||
|
||||
# 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 PiCameraStreamer() 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)
|
||||
Loading…
Add table
Add a link
Reference in a new issue