rahix/openflexure-microscope-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Microscope capture objects now just handle on-disk data

Browse source
This commit is contained in:
Joel Collins 2019-09-16 16:35:32 +01:00
parent a832dfb2f1
commit dc794d4df8
10 changed files with 125 additions and 311 deletions
Download patch file Download diff file Expand all files Collapse all files

8
docs/source/plugins/example/plugin.py
View file

@ -59,15 +59,15 @@ class MyPluginClass(MicroscopePlugin):
for _ in range(n_images):
# Create a data stream to capture to
capture_data = self.microscope.camera.new_image(
write_to_file=True, temporary=False
output = self.microscope.camera.new_image(
temporary=False
)
# Capture a still image from the Pi camera, into the data stream
self.microscope.camera.capture(capture_data, use_video_port=True)
self.microscope.camera.capture(output.file, use_video_port=True)
# Append the capture data to our list
capture_array.append(capture_data)
capture_array.append(output)
# Wait for 1 minute
time.sleep(60)

7
docs/source/plugins/hardware.rst
View file

@ -103,17 +103,16 @@ For example, a timelapse plugin may look like:
for _ in range(n_images):
# Create a data stream to capture to
capture_data = self.microscope.camera.new_image(
write_to_file=True,
output = self.microscope.camera.new_image(
temporary=False)
# Capture a still image from the Pi camera, into the data stream
self.microscope.camera.capture(
capture_data,
output.file,
use_video_port=True)
# Append the capture data to our list
capture_array.append(capture_data)
capture_array.append(output)
# Wait for 1 minute
time.sleep(60)

71
openflexure_microscope/api/app.py
View file

@ -43,7 +43,7 @@ DEFAULT_LOGFILE = os.path.join(USER_CONFIG_DIR, "openflexure_microscope.log")
if (__name__ == "__main__") or (not is_gunicorn):
# If imported, but not by gunicorn
print("Letting sys handle logs")
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logging.basicConfig(level=logging.DEBUG)
else:
# Direct standard Python logging to file and console
root = logging.getLogger()
@ -66,10 +66,33 @@ else:
# Create a dummy microscope object, with no hardware attachments
api_microscope = Microscope()
# Rebuild the capture list
# TODO: Offload to a thread?
stored_image_list = build_captures_from_exif()
# Initialise camera
logging.debug("Creating camera object...")
try:
api_camera = PiCameraStreamer()
except Exception as e:
logging.error(e)
logging.warning("No valid camera hardware found. Falling back to mock camera!")
api_camera = MockStreamer()
# Initialise stage
logging.debug("Creating stage object...")
try:
api_stage = SangaStage()
except Exception as e:
logging.error(e)
logging.warning("No valid stage hardware found. Falling back to mock stage!")
api_stage = MockStage()
# Attach devices to microscope
logging.debug("Attaching devices to microscope...")
api_microscope.attach(api_camera, api_stage)
# Restore loaded capture array to camera object
logging.debug("Restoring captures...")
api_microscope.camera.images = build_captures_from_exif(api_microscope.camera.paths["default"])
logging.debug("Microscope successfully attached!")
# Generate API URI based on version from filename
def uri(suffix, api_version, base=None):
@ -89,48 +112,8 @@ CORS(app, resources=r"/api/*")
# Make errors more API friendly
handler = JSONExceptionHandler(app)
# After app starts, but before first request, attach hardware to global microscope
@app.before_first_request
def attach_microscope():
# Create the microscope object globally (common to all spawned server threads)
global api_microscope, stored_image_list
logging.debug("First request made. Populating microscope with hardware...")
# Initialise camera
logging.debug("Creating camera object...")
try:
api_camera = PiCameraStreamer()
except Exception as e:
logging.error(e)
logging.warning("No valid camera hardware found. Falling back to mock camera!")
api_camera = MockStreamer()
# Initialise stage
logging.debug("Creating stage object...")
try:
api_stage = SangaStage()
except Exception as e:
logging.error(e)
logging.warning("No valid stage hardware found. Falling back to mock stage!")
api_stage = MockStage()
# Attach devices to microscope
logging.debug("Attaching devices to microscope...")
api_microscope.attach(api_camera, api_stage)
# Restore loaded capture array to camera object
logging.debug("Restoring captures...")
if stored_image_list:
api_microscope.camera.images = stored_image_list
logging.debug("Microscope successfully attached!")
# WEBAPP ROUTES
# API ROUTES
# Base routes
base_blueprint = blueprints.base.construct_blueprint(api_microscope)
app.register_blueprint(base_blueprint, url_prefix=uri("", "v1"))

4
openflexure_microscope/api/v1/blueprints/camera/capture.py
View file

@ -123,11 +123,11 @@ class ListAPI(MicroscopeView):
# Explicitally acquire lock (prevents empty files being created if lock is unavailable)
with self.microscope.camera.lock:
output = self.microscope.camera.new_image(
write_to_file=True, temporary=temporary, filename=filename
temporary=temporary, filename=filename
)
self.microscope.camera.capture(
output, use_video_port=use_video_port, resize=resize, bayer=bayer
output.file, use_video_port=use_video_port, resize=resize, bayer=bayer
)
metadata.update(

47
openflexure_microscope/camera/base.py
View file

@ -1,17 +1,22 @@
# -*- coding: utf-8 -*-
import time
import os
import shutil
import threading
import datetime
import logging
from abc import ABCMeta, abstractmethod
from .capture import CaptureObject, BASE_CAPTURE_PATH, TEMP_CAPTURE_PATH
from .capture import CaptureObject
from openflexure_microscope.utilities import entry_by_id
from openflexure_microscope.lock import StrictLock
BASE_CAPTURE_PATH = os.path.join(os.path.expanduser("~"), "micrographs")
TEMP_CAPTURE_PATH = os.path.join(BASE_CAPTURE_PATH, "tmp")
def last_entry(object_list: list):
"""Return the last entry of a list, if the list contains items."""
if object_list: # If any images have been captured
@ -121,11 +126,11 @@ class BaseCamera(metaclass=ABCMeta):
self.stream_timeout_enabled = False
self.state = {}
# TODO: Load/save these to config
self.paths = {
"image": BASE_CAPTURE_PATH,
"video": BASE_CAPTURE_PATH,
"image_tmp": TEMP_CAPTURE_PATH,
"video_tpm": TEMP_CAPTURE_PATH,
"default": BASE_CAPTURE_PATH,
"temp": TEMP_CAPTURE_PATH
}
# Capture data
@ -157,10 +162,22 @@ class BaseCamera(metaclass=ABCMeta):
for capture_list in [self.images, self.videos]:
for stream_object in capture_list:
stream_object.close()
# Empty temp directory
self.clear_tmp()
# Stop worker thread
self.stop_worker()
logging.info("Closed {}".format(self))
def clear_tmp(self):
"""
Removes all files in the temporary capture directories
"""
if os.path.isdir(self.paths["temp"]):
logging.info("Clearing {}...".format(self.paths["temp"]))
shutil.rmtree(self.paths["temp"])
logging.debug("Cleared {}.".format(self.paths["temp"]))
# START AND STOP WORKER THREAD
def start_worker(self, timeout: int = 5) -> bool:
@ -244,7 +261,6 @@ class BaseCamera(metaclass=ABCMeta):
def new_image(
self,
write_to_file: bool = True,
temporary: bool = True,
filename: str = None,
folder: str = "",
@ -252,10 +268,9 @@ class BaseCamera(metaclass=ABCMeta):
):
"""
Create a new image capture object. Adds to the image list, and shunt all others.
Create a new image capture object.
Args:
write_to_file (bool): Should the StreamObject write to a file, or an in-memory byte stream.
temporary (bool): Should the data be deleted after session ends.
Creating the capture with a content manager sets this to true.
filename (str): Name of the stored file. Defaults to timestamp.
@ -270,16 +285,14 @@ class BaseCamera(metaclass=ABCMeta):
filename = "{}.{}".format(filename, fmt)
# Generate folder
base_folder = self.paths["image_tmp"] if temporary else self.paths["image"]
base_folder = self.paths["temp"] if temporary else self.paths["default"]
folder = os.path.join(base_folder, folder)
# Generate file path
filepath = os.path.join(folder, filename)
# Create capture object
output = CaptureObject(
write_to_file=write_to_file, temporary=temporary, filepath=filepath
)
output = CaptureObject(filepath=filepath)
# Update capture list
shunt_captures(self.images)
@ -289,7 +302,6 @@ class BaseCamera(metaclass=ABCMeta):
def new_video(
self,
write_to_file: bool = True,
temporary: bool = False,
filename: str = None,
folder: str = "",
@ -297,10 +309,9 @@ class BaseCamera(metaclass=ABCMeta):
):
"""
Create a new video capture object. Adds to the image list, and shunt all others.
Create a new video capture object.
Args:
write_to_file (bool): Should the StreamObject write to a file, or an in-memory byte stream.
temporary (bool): Should the data be deleted after session ends.
Creating the capture with a content manager sets this to true.
filename (str): Name of the stored file. Defaults to timestamp.
@ -315,16 +326,14 @@ class BaseCamera(metaclass=ABCMeta):
filename = "{}.{}".format(filename, fmt)
# Generate folder
base_folder = self.paths["video_tmp"] if temporary else self.paths["video"]
base_folder = self.paths["temp"] if temporary else self.paths["default"]
folder = os.path.join(base_folder, folder)
# Generate file path
filepath = os.path.join(folder, filename)
# Create capture object
output = CaptureObject(
write_to_file=write_to_file, temporary=temporary, filepath=filepath
)
output = CaptureObject(filepath=filepath)
# Update capture list
shunt_captures(self.videos)

207
openflexure_microscope/camera/capture.py
View file

@ -12,32 +12,11 @@ import atexit
from openflexure_microscope.camera import piexif
from openflexure_microscope.camera.piexif._exceptions import InvalidImageDataError
"""
Attributes:
BASE_CAPTURE_PATH (str): Base path to store all captures
TEMP_CAPTURE_PATH (str): Base path to store all temporary captures (automatically emptied)
"""
PIL_FORMATS = ["JPG", "JPEG", "PNG", "TIF", "TIFF"]
EXIF_FORMATS = ["JPG", "JPEG", "TIF", "TIFF"]
THUMBNAIL_SIZE = (200, 150)
BASE_CAPTURE_PATH = os.path.join(os.path.expanduser("~"), "micrographs")
TEMP_CAPTURE_PATH = os.path.join(BASE_CAPTURE_PATH, "tmp")
# TODO: Move these methods to a camera utilities module?
def clear_tmp():
"""
Removes all files in the ``TEMP_CAPTURE_PATH`` directory
"""
global TEMP_CAPTURE_PATH
if os.path.isdir(TEMP_CAPTURE_PATH):
logging.info("Clearing {}...".format(TEMP_CAPTURE_PATH))
shutil.rmtree(TEMP_CAPTURE_PATH)
logging.debug("Cleared {}.".format(TEMP_CAPTURE_PATH))
def pull_usercomment_dict(filepath):
"""
@ -68,11 +47,11 @@ def make_file_list(directory, formats):
return files
def build_captures_from_exif():
global BASE_CAPTURE_PATH, EXIF_FORMATS
def build_captures_from_exif(capture_path):
global EXIF_FORMATS
logging.debug("Reloading captures from {}...".format(BASE_CAPTURE_PATH))
files = make_file_list(BASE_CAPTURE_PATH, EXIF_FORMATS)
logging.debug("Reloading captures from {}...".format(capture_path))
files = make_file_list(capture_path, EXIF_FORMATS)
captures = []
for f in files:
@ -119,27 +98,20 @@ def capture_from_exif(path, exif_dict):
class CaptureObject(object):
"""
StreamObject used to store and process capture data, and metadata.
StreamObject used to store and process on-disk capture data, and metadata.
Attributes:
timestring (str): Timestring of capture creation time
temporary (bool): Mark the capture as temporary, to be deleted as the server closes,
or as resources are required
_metadata (dict): Dictionary of custom metadata to be included in metadata file
tags (list): List of tags. Essentially just as extra custom metadata field, but useful for quick organisation
bytestream (:py:class:`io.BytesIO`): Byte bytestream that data will be written to
filefolder (str): Folder in which the capture file will be stored
filename (str): Full name of the capture file
basename (str): Filename of the capture, without a file extension
format (str): Format of the capture data
Notes:
Captures cannot be stored in a lower-level directory than BASE_CAPTURE_PATH.
"""
def __init__(
self, write_to_file: bool = False, temporary: bool = False, filepath: str = ""
) -> None:
def __init__(self, filepath) -> None:
"""Create a new StreamObject, to manage capture data."""
# Store a nice ID
@ -147,14 +119,8 @@ class CaptureObject(object):
logging.debug("Created StreamObject {}".format(self.id))
self.timestring = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
# Keep on disk after close by default
self.temporary = temporary
# Create file name. Default to UUID
if not filepath:
self.file = self.build_file_path()
else:
self.file = filepath
self.file = filepath
self.split_file_path(self.file)
# Dictionary for storing custom metadata
@ -163,54 +129,11 @@ class CaptureObject(object):
# List for storing tags
self.tags = []
# Byte bytestream properties
self.bytestream = io.BytesIO()
# Set default write target
if not write_to_file:
self.stream = self.bytestream
else:
self.stream = self.file
# Log if created by context manager
self.context_manager = False
# Thumbnail (populated only for PIL captures)
self.thumb_bytes = None
def __enter__(self):
"""Create StreamObject in context, to auto-clean disk data."""
logging.debug(
"Entering context for {}. Stored files will be cleaned up automatically regardless of location.".format(
self.id
)
)
self.temporary = True # Flag file to be removed on close.
self.context_manager = True # Used in metadata
logging.info("Rebuilding as a temporary capture...")
self.build_file_path()
return self
def __exit__(self, *args):
"""Exit StreamObject, and auto-clean disk data."""
logging.info("Cleaning up {}".format(self.id))
self.close()
def build_file_path(self):
"""
Construct a full file path, based on filename, folder, and file format.
Defaults to UUID.
"""
global TEMP_CAPTURE_PATH, BASE_CAPTURE_PATH
if self.temporary:
base_dir = TEMP_CAPTURE_PATH
else:
base_dir = BASE_CAPTURE_PATH
return os.path.join(
base_dir, self.filename
) # Full file name by joining given folder to given name
def open(self, mode):
return open(self.file, mode)
def split_file_path(self, filepath):
"""
@ -230,19 +153,7 @@ class CaptureObject(object):
os.makedirs(self.filefolder)
@property
def stream_exists(self, auto_rewind=True) -> bool:
"""Check if capture data BytesIO bytestream exists in memory."""
if auto_rewind:
self.bytestream.seek(0) # Rewind the data bytes for reading
if self.bytestream.getvalue(): # If data bytestream contains data
self.bytestream.seek(0) # Rewind the data bytes for reading
return True
else:
self.bytestream.seek(0) # Rewind the data bytes for reading
return False
@property
def file_exists(self) -> bool:
def exists(self) -> bool:
"""Check if capture data file exists on disk."""
if os.path.isfile(self.file):
return True
@ -293,7 +204,7 @@ class CaptureObject(object):
"""
global EXIF_FORMATS
if self.format.upper() in EXIF_FORMATS and self.file_exists:
if self.format.upper() in EXIF_FORMATS and self.exists:
logging.debug("Writing exif data to capture file")
# Extract current Exif data
exif_dict = piexif.load(self.file)
@ -331,15 +242,6 @@ class CaptureObject(object):
"""
return yaml.dump(self.metadata, default_flow_style=False)
@property
def exists(self) -> bool:
"""
Check if either an in-memory byte stream or on-disk file of capture data exists.
If False, the capture data cannot be obtained.
"""
return self.stream_exists or self.file_exists
@property
def state(self) -> dict:
"""
@ -347,13 +249,7 @@ class CaptureObject(object):
"""
# Create basic state dictionary
d = {"path": self.file, "temporary": self.temporary, "metadata": self.metadata}
# Check bytestream
if self.stream_exists:
d["bytestream"] = True
else:
d["bytestream"] = False
d = {"path": self.file, "metadata": self.metadata}
# Combined availability of data
if self.exists:
@ -367,27 +263,17 @@ class CaptureObject(object):
def data(self) -> io.BytesIO:
"""
Return a byte string of the capture data.
If the capture data exists in-memory, this will be loaded. If the data only exists
on disk, this will automatically fall back to loading from disk.
"""
self.bytestream.seek(0) # Rewind the data bytes for reading
if self.stream_exists: # If data bytestream contains data
if self.exists: # If data file exists
logging.info("Opening from file {}".format(self.file))
with open(self.file, "rb") as f:
d = io.BytesIO(f.read()) # Load bytes from file
d.seek(0) # Rewind loaded bytestream
# Create a copy of the bytestream bytes
logging.debug("STREAM EXISTS")
data = io.BytesIO(self.bytestream.getbuffer())
else: # If data bytestream is empty
if self.file_exists: # If data file exists
logging.info("Opening from file {}".format(self.file))
with open(self.file, "rb") as f:
d = io.BytesIO(f.read()) # Load bytes from file
d.seek(0) # Rewind loaded bytestream
# Create a copy of the bytestream bytes
data = io.BytesIO(d.getbuffer())
else:
data = None
data = io.BytesIO(d.getbuffer())
else:
data = None
return data # Read and return bytes data
@ -418,41 +304,13 @@ class CaptureObject(object):
data = io.BytesIO(self.thumb_bytes.getbuffer())
return data
def load_file(self) -> bool:
"""Load data stored on disk to the in-memory bytestream."""
if self.file_exists: # If data file exists
with open(self.file, "rb") as f:
self.bytestream = io.BytesIO(f.read()) # Load bytes from file
self.bytestream.seek(0) # Rewind data bytes again
return True
else:
return False
def save_file(self) -> bool:
"""Write the StreamObjects bytestream to a file."""
if self.stream_exists: # If there's a bytestream to save
with open(self.file, "ab") as f: # Load file as bytes
logging.debug("Writing bytestream to file {}".format(self.file))
f.seek(0, 0) # Seek to the start of the file
f.write(self.binary) # Write data bytes to file
return True
else:
return False
def save(self) -> None:
"""Write stream to file, and save/update metadata file"""
# Try to save the file (only succeeds if an unsaved stream exists)
self.save_file()
# If a stream OR file exists, save the metadata file
if self.exists:
self.save_metadata()
def delete_stream(self):
"""Clear the BytesIO bytestream of the StreamObject."""
self.bytestream = io.BytesIO()
def delete_file(self) -> bool:
def delete(self) -> bool:
"""If the StreamObject has been saved, delete the file."""
if os.path.isfile(self.file):
@ -463,25 +321,10 @@ class CaptureObject(object):
else:
return False
def delete(self):
"""Entirely delete all capture data."""
logging.info("Deleting {}".format(self.id))
self.delete_stream()
self.delete_file()
def shunt(self):
"""Demote the StreamObject from being stored in memory."""
if not self.file_exists: # If file doesn't already exist
self.save() # Save bytestream to disk, if it exists
self.delete_stream() # Delete the bytestream from memory
# TODO: Remove
pass
def close(self):
"""Both clear the bytestream, and delete any associated on-disk data."""
logging.info("Closing {}".format(self.id))
self.delete_stream()
# Delete the file from disk if temporary
if self.temporary:
self.delete()
atexit.register(clear_tmp)
# TODO: Remove
pass

21
openflexure_microscope/camera/mock.py
View file

@ -90,24 +90,18 @@ class MockStreamer(BaseCamera):
Args:
config (dict): Dictionary of config parameters.
"""
paused_stream = False
logging.debug("PiCameraStreamer: Applying config:")
logging.debug("MockStreamer: Applying config:")
logging.debug(config)
with self.lock:
# Apply valid config params to Picamera object
# Apply valid config params to camera object
if not self.state["record_active"]: # If not recording a video
# PiCameraStreamer parameters
for key, value in config.items(): # For each provided setting
if hasattr(self, key):
setattr(self, key, value)
# If stream was paused to update config, unpause
if paused_stream:
logging.info("Resuming stream.")
else:
raise Exception(
"Cannot update camera config while recording is active."
@ -135,7 +129,7 @@ class MockStreamer(BaseCamera):
Start a new video recording, writing to a output object.
Args:
output (CaptureObject/str): Output object to write data bytes to.
output: String or file-like object to write capture data to
fmt (str): Format of the capture.
quality (int): Video recording quality.
@ -167,7 +161,7 @@ class MockStreamer(BaseCamera):
Target object can be overridden for development purposes.
Args:
output (CaptureObject/str): Output object to write data bytes to.
output: String or file-like object to write capture data to
use_video_port (bool): Capture from the video port used for streaming. Lower resolution, faster.
fmt (str): Format of the capture.
resize ((int, int)): Resize the captured image.
@ -175,7 +169,12 @@ class MockStreamer(BaseCamera):
"""
with self.lock:
logging.warning("Capture not implemented in mock camera")
if isinstance(output, str):
output = open(output, 'wb')
output.write(self.stream.getvalue())
output.close()
# HANDLE STREAM FRAMES

22
openflexure_microscope/camera/pi.py
View file

@ -296,7 +296,7 @@ class PiCameraStreamer(BaseCamera):
Start a new video recording, writing to a output object.
Args:
output (CaptureObject/str): Output object to write data bytes to.
output: String or file-like object to write capture data to
fmt (str): Format of the capture.
quality (int): Video recording quality.
@ -308,18 +308,11 @@ class PiCameraStreamer(BaseCamera):
# Start recording method only if a current recording is not running
if not self.state["record_active"]:
# If output is a StreamObject
if isinstance(output, CaptureObject):
# Set target to capture stream
output_stream = output.stream
else:
output_stream = output
# Start the camera video recording on port 2
logging.info("Recording to {}".format(output))
self.camera.start_recording(
output_stream,
output,
format=fmt,
splitter_port=2,
resize=self.stream_resolution,
@ -453,7 +446,7 @@ class PiCameraStreamer(BaseCamera):
Target object can be overridden for development purposes.
Args:
output (CaptureObject/str): Output object to write data bytes to.
output: String or file-like object to write capture data to
use_video_port (bool): Capture from the video port used for streaming. Lower resolution, faster.
fmt (str): Format of the capture.
resize ((int, int)): Resize the captured image.
@ -461,13 +454,6 @@ class PiCameraStreamer(BaseCamera):
"""
with self.lock:
# If output is a StreamObject
if isinstance(output, CaptureObject):
# Set target to capture stream
output_stream = output.stream
else:
output_stream = output
logging.info("Capturing to {}".format(output))
# Set resolution and stop stream recording if necessary
@ -475,7 +461,7 @@ class PiCameraStreamer(BaseCamera):
self.stop_stream_recording()
self.camera.capture(
output_stream,
output,
format=fmt,
quality=100,
resize=resize,

4
openflexure_microscope/plugins/default/scan/plugin.py
View file

@ -71,12 +71,12 @@ class ScanPlugin(MicroscopePlugin):
# Create output object
output = self.microscope.camera.new_image(
write_to_file=True, temporary=temporary, filename=filename, folder=folder
temporary=temporary, filename=filename, folder=folder
)
# Capture
self.microscope.camera.capture(
output, use_video_port=use_video_port, resize=resize, bayer=bayer
output.file, use_video_port=use_video_port, resize=resize, bayer=bayer
)
# Affix metadata

45
tests/test_camera.py
View file

@ -28,9 +28,9 @@ class TestCaptureMethods(unittest.TestCase):
camera.wait_for_camera()
# Capture to a context (auto-deletes files when done)
with camera.new_image(write_to_file=False) as output:
with camera.new_image() as output:
camera.capture(output, use_video_port=use_video_port, resize=resize)
camera.capture(output.file, use_video_port=use_video_port, resize=resize)
# Ensure file deletion fails and returns False
self.assertFalse(output.delete_file())
@ -87,7 +87,7 @@ class TestCaptureMethods(unittest.TestCase):
# Capture
output = camera.capture(
camera.new_image(write_to_file=True),
camera.new_image().file,
use_video_port=use_video_port,
resize=resize,
)
@ -176,34 +176,29 @@ class TestRecordMethods(unittest.TestCase):
"""Tests recording videos to BytesIO stream, and to file on disk."""
global camera
for write_to_file in [True, False, None]:
# Wait for camera
camera.wait_for_camera()
logging.debug("\nWRITE TO FILE: {}".format(write_to_file))
with camera.new_video() as output:
# Wait for camera
camera.wait_for_camera()
# Start recording
camera.start_recording(output)
with camera.new_video(write_to_file=write_to_file) as output:
# Record for 2 seconds
time.sleep(2)
# Stop recording
camera.stop_recording()
# Start recording
camera.start_recording(output)
# Check stream
self.assertTrue(isinstance(output.data, io.IOBase))
self.assertTrue(isinstance(output.binary, (bytes, bytearray)))
# 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)
# 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
# Log path
temp_path = output.file
# Check file got deleted on __exit__
self.assertFalse(os.path.isfile(temp_path))