Moved generic camera methods into BaseCamera class

This commit is contained in:
Joel Collins 2018-11-07 10:54:39 +00:00
parent 35c7d97db9
commit 7b23ca4b8a
4 changed files with 325 additions and 317 deletions

View file

@ -1,11 +1,8 @@
import time
import os
import io
import threading
import copy
import datetime
import uuid
from PIL import Image
import logging
try:
from greenlet import getcurrent as get_ident
@ -15,221 +12,22 @@ except ImportError:
except ImportError:
from _thread import get_ident
# Slightly crappy debugger logging
DEBUG = True
from .capture import StreamObject
def log(s):
if DEBUG:
print('DEBUG: {}'.format(s))
def last_entry(object_list: list):
if object_list: # If any images have been captured
return object_list[-1] # Return the latest captured image
else:
return None
class StreamObject(object):
def __init__(self, write_to_file: bool=None, filename: str=None, folder: str=None, fmt: str='file') -> None:
# Store a nice ID
self.id = uuid.uuid4().hex
log("Created {}".format(self.id))
# Create file name
iterator = 0
f_name = self.build_file_path(filename, folder, fmt)
while os.path.isfile(f_name): # While file already exists
iterator += 1 # Add a file name iterator
f_name = self.build_file_path(filename, folder, fmt, iterator=iterator) # Rebuild file name
self.file = f_name
# Byte stream properties
self.stream = io.BytesIO() # Byte stream that data will be written to
# Set default write target
self.write_to_file = write_to_file
if self.write_to_file is False:
log("Default target for {} set to 'stream'".format(self.id))
self.target = self.stream
else:
log("Default target for {} set to 'file'".format(self.id))
self.target = self.file
# Keep on disk after close by default
self.keep_on_disk = True
# Log if created by context manager
self.context_manager = False
# Object lock
self.locked = False # Is the StreamObject locked for writing? (Handled by StreamingCamera)
def __enter__(self):
log("Entering context for {}.\
Stored files will be cleaned up automatically.".format(self.id))
self.keep_on_disk = False # Flag file to be removed on close.
self.context_manager = True # Used in metadata
return self
def __exit__(self, *args):
log("Cleaning up {}".format(self.id))
self.close()
def build_file_path(self, filename: str, folder: str, fmt: str, iterator: int=0) -> str:
"""
Construct a full file path, based on filename, folder, and file format.
Defaults to datestamp. Iterator adds a numeric increment to the file name.
"""
if filename:
file_name = "{}.{}".format(filename, fmt)
else:
file_name_base = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
if iterator:
file_name_base = "{}_{}".format(file_name_base, iterator)
file_name = "{}.{}".format(file_name_base, fmt)
# Create folder and file
if folder:
if not os.path.exists(folder):
os.mkdir(folder)
file_path = os.path.join(folder, file_name)
else:
file_path = file_name
return file_path
def lock(self):
"""Set locked flag to True."""
self.locked = True
def unlock(self):
"""Set locked flag to False."""
self.locked = False
@property
def stream_exists(self, auto_rewind=True) -> bool:
"""Check if BytesIO stream is empty"""
if auto_rewind:
self.stream.seek(0) # Rewind the data bytes for reading
if self.stream.getvalue(): # If data stream contains data
self.stream.seek(0) # Rewind the data bytes for reading
return True
else:
self.stream.seek(0) # Rewind the data bytes for reading
return False
@property
def file_exists(self) -> bool:
"""Check if corresponding file exists"""
if os.path.isfile(self.file):
return True
else:
return False
@property
def metadata(self) -> dict:
d = {
'id': self.id,
'locked': self.locked,
'keep_on_disk': self.keep_on_disk,
'path': self.file,
'context_manager': self.context_manager,
}
# Get file path
if self.file_exists:
d['file'] = self.file
else:
d['file'] = None
# Check stream
if self.stream_exists:
d['stream'] = True
else:
d['stream'] = False
# Check if file was manually deleted
if self.keep_on_disk and not self.file_exists:
d['path'] = "{} (Deleted)".format(d['path'])
return d
@property
def data(self) -> io.BytesIO:
"""Return a byte string of the capture data."""
self.stream.seek(0) # Rewind the data bytes for reading
if self.stream_exists: # If data stream contains data
data = io.BytesIO(self.stream.getbuffer()) # Create a copy of the stream bytes
else: # If data stream is empty
if self.file_exists: # If data file exists
# TODO: Streamline this bit
log("No stream data. Opening from file {}".format(self.file))
with open(self.file, 'rb') as f:
d = io.BytesIO(f.read()) # Load bytes from file in disk
d.seek(0) # Rewind loaded stream
data = io.BytesIO(d.getbuffer()) # Create a copy of the stream bytes
else:
data = None
return data # Read and return bytes data
@property
def binary(self) -> bytes:
"""Return a byte string of the capture data."""
return self.data.getvalue()
def load_file(self) -> bool:
"""Load data stored on disk to the in-memory stream."""
if self.file_exists: # If data file exists
# TODO: Streamline this bit
with open(self.file, 'rb') as f:
self.stream = io.BytesIO(f.read()) # Load bytes from file in disk
self.stream.seek(0) # Rewind data bytes again
return True
else:
return False
def save_file(self) -> bool:
"""
Write the StreamObjects stream to a file.
"""
if self.stream_exists: # If there's a stream to save
with open(self.file, 'ab') as f: # Load file as bytes
log("Writing stream 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 delete_stream(self):
"""Clears the BytesIO stream of the StreamObject."""
self.stream = io.BytesIO()
def delete_file(self) -> bool:
"""
If the StreamObject has been saved to disk, deletes the file and returns True.
"""
if os.path.isfile(self.file):
log("Deleting file {}".format(self.file))
os.remove(self.file)
return True
else:
return False
def shunt(self):
"""Demote the StreamObject from being stored in memory."""
if not self.file_exists: # If file doesn't already exist
self.save_file() # Save stream to disk, if it exists
self.delete_stream() # Delete the stream from memory
def close(self):
"""Both clear the stream, and delete any associated on-disk data"""
log("Closing {}".format(self.id))
self.delete_stream()
if not self.keep_on_disk:
self.delete_file()
def entry_by_id(id: str, object_list: list):
found = None
for o in object_list:
if o.id == id:
found = o
return found
class CameraEvent(object):
@ -281,25 +79,51 @@ class BaseCamera(object):
frame = None # Current frame is stored here by background thread
last_access = 0 # Time of last client access to the camera
event = CameraEvent()
# CONTEXT MANAGER AND OBJECT CLEANUP
def __init__(self):
"""Base implementation of StreamingCamera."""
self.state = {} # Create dict for capture state
self.settings = {} # Create dict to store settings
# Capture data
self.images = []
self.videos = []
def __enter__(self):
"""Create camera on context enter."""
return self
def __exit__(self, exc_type, exc_value, traceback):
"""Close camera stream on context exit."""
self.close()
def close(self):
"""Handles closing the StreamingCamera"""
# Close all StreamObjects
for capture_list in [self.images, self.videos]:
for stream_object in capture_list:
stream_object.close()
# Stop worker thread
self.stop_worker()
# START AND STOP WORKER THREAD
def start_worker(self, timeout: int=5) -> bool:
"""Start the background camera thread if it isn't running yet."""
timeout_time = time.time() + timeout
self.last_access = time.time()
self.stop = False
if self.thread is None:
# start background frame thread
self.thread = threading.Thread(target=self._thread)
self.thread.start()
# wait until frames are available
log("Waiting for frames")
logging.info("Waiting for frames")
while self.get_frame() is None:
if time.time() > timeout_time:
raise TimeoutError("Timeout waiting for frames.")
@ -319,6 +143,8 @@ class BaseCamera(object):
time.sleep(0)
return True
# HANDLE STREAM FRAMES
def get_frame(self):
"""Return the current camera frame."""
self.last_access = time.time()
@ -333,16 +159,45 @@ class BaseCamera(object):
"""Generator that returns frames from the camera."""
raise RuntimeError('Must be implemented by subclasses.')
def close(self):
"""Handles closing the StreamingCamera"""
self.stop_worker()
# RETURNING CAPTURES
def __enter__(self):
return self
@property
def image(self):
"""Return the latest captured image"""
return last_entry(self.images)
def __exit__(self, exc_type, exc_value, traceback):
"""Close camera stream on context exit."""
self.close()
@property
def video(self):
"""Return the latest recorded video"""
return last_entry(self.videos)
def image_from_id(self, id):
"""Returns an image StreamObject with a matching ID"""
return entry_by_id(id, self.images)
def video_from_id(self, id):
"""Returns a video StreamObject with a matching ID"""
return entry_by_id(id, self.videos)
# CREATING NEW CAPTURES
def new_stream_object(self, stream_object, target_list: list, shunt_others: bool=True):
"""Add a new capture to a list of captures, and shunt all others"""
if shunt_others: # If shunting all older captures
for obj in target_list: # For each older capture
obj.shunt() # Shunt capture from memory to storage
target_list.append(stream_object)
return stream_object
def new_image(self, stream_object, shunt_others=True):
"""Add a new capture to the image list, and shunt all others"""
return self.new_stream_object(stream_object, self.images, shunt_others=shunt_others)
def new_video(self, stream_object, shunt_others=True):
"""Add a new capture to the video list, and shunt all others"""
return self.new_stream_object(stream_object, self.videos, shunt_others=shunt_others)
# WORKER THREAD
def _thread(self):
"""Camera background thread."""
@ -359,17 +214,17 @@ class BaseCamera(object):
# if there hasn't been any clients asking for frames in
# the last 10 seconds then stop the thread
if time.time() - self.last_access > 20:
self.frames_iterator.close()
break
self.frames_iterator.close()
break
try:
if self.stop is True:
self.frames_iterator.close()
break
except AttributeError:
pass
# Reset thread
self.thread = None
# Destroy any stored camera object (used especially for handling Pi cameras)

View file

@ -0,0 +1,213 @@
import uuid
import io
import os
import datetime
import copy
import logging
class StreamObject(object):
def __init__(self, write_to_file: bool=None, filename: str=None, folder: str=None, fmt: str='file') -> None:
# Store a nice ID
self.id = uuid.uuid4().hex
logging.info("Created {}".format(self.id))
# Create file name
iterator = 0
f_name = self.build_file_path(filename, folder, fmt)
while os.path.isfile(f_name): # While file already exists
iterator += 1 # Add a file name iterator
f_name = self.build_file_path(filename, folder, fmt, iterator=iterator) # Rebuild file name
self.file = f_name
# Byte stream properties
self.stream = io.BytesIO() # Byte stream that data will be written to
# Set default write target
self.write_to_file = write_to_file
if self.write_to_file is False:
logging.debug("Default target for {} set to 'stream'".format(self.id))
self.target = self.stream
else:
logging.debug("Default target for {} set to 'file'".format(self.id))
self.target = self.file
# Keep on disk after close by default
self.keep_on_disk = True
# Log if created by context manager
self.context_manager = False
# Object lock
self.locked = False # Is the StreamObject locked for writing? (Handled by StreamingCamera)
def __enter__(self):
logging.debug("Entering context for {}.\
Stored files will be cleaned up automatically.".format(self.id))
self.keep_on_disk = False # Flag file to be removed on close.
self.context_manager = True # Used in metadata
return self
def __exit__(self, *args):
logging.info("Cleaning up {}".format(self.id))
self.close()
def build_file_path(self, filename: str, folder: str, fmt: str, iterator: int=0) -> str:
"""
Construct a full file path, based on filename, folder, and file format.
Defaults to datestamp. Iterator adds a numeric increment to the file name.
"""
if filename:
file_name = "{}.{}".format(filename, fmt)
else:
file_name_base = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
if iterator:
file_name_base = "{}_{}".format(file_name_base, iterator)
file_name = "{}.{}".format(file_name_base, fmt)
# Create folder and file
if folder:
if not os.path.exists(folder):
os.mkdir(folder)
file_path = os.path.join(folder, file_name)
else:
file_path = file_name
return file_path
def lock(self):
"""Set locked flag to True."""
self.locked = True
def unlock(self):
"""Set locked flag to False."""
self.locked = False
@property
def stream_exists(self, auto_rewind=True) -> bool:
"""Check if BytesIO stream is empty"""
if auto_rewind:
self.stream.seek(0) # Rewind the data bytes for reading
if self.stream.getvalue(): # If data stream contains data
self.stream.seek(0) # Rewind the data bytes for reading
return True
else:
self.stream.seek(0) # Rewind the data bytes for reading
return False
@property
def file_exists(self) -> bool:
"""Check if corresponding file exists"""
if os.path.isfile(self.file):
return True
else:
return False
@property
def metadata(self) -> dict:
d = {
'id': self.id,
'locked': self.locked,
'keep_on_disk': self.keep_on_disk,
'path': self.file,
'context_manager': self.context_manager,
}
# Get file path
if self.file_exists:
d['file'] = self.file
else:
d['file'] = None
# Check stream
if self.stream_exists:
d['stream'] = True
else:
d['stream'] = False
# Check if file was manually deleted
if self.keep_on_disk and not self.file_exists:
d['path'] = "{} (Deleted)".format(d['path'])
return d
@property
def data(self) -> io.BytesIO:
"""Return a byte string of the capture data."""
self.stream.seek(0) # Rewind the data bytes for reading
if self.stream_exists: # If data stream contains data
data = io.BytesIO(self.stream.getbuffer()) # Create a copy of the stream bytes
else: # If data stream is empty
if self.file_exists: # If data file exists
# TODO: Streamline this bit
logging.info("No stream data. Opening from file {}".format(self.file))
with open(self.file, 'rb') as f:
d = io.BytesIO(f.read()) # Load bytes from file in disk
d.seek(0) # Rewind loaded stream
data = io.BytesIO(d.getbuffer()) # Create a copy of the stream bytes
else:
data = None
return data # Read and return bytes data
@property
def binary(self) -> bytes:
"""Return a byte string of the capture data."""
return self.data.getvalue()
def load_file(self) -> bool:
"""Load data stored on disk to the in-memory stream."""
if self.file_exists: # If data file exists
# TODO: Streamline this bit
with open(self.file, 'rb') as f:
self.stream = io.BytesIO(f.read()) # Load bytes from file in disk
self.stream.seek(0) # Rewind data bytes again
return True
else:
return False
def save_file(self) -> bool:
"""
Write the StreamObjects stream to a file.
"""
if self.stream_exists: # If there's a stream to save
with open(self.file, 'ab') as f: # Load file as bytes
logging.debug("Writing stream 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 delete_stream(self):
"""Clears the BytesIO stream of the StreamObject."""
self.stream = io.BytesIO()
def delete_file(self) -> bool:
"""
If the StreamObject has been saved to disk, deletes the file and returns True.
"""
if os.path.isfile(self.file):
logging.info("Deleting file {}".format(self.file))
os.remove(self.file)
return True
else:
return False
def shunt(self):
"""Demote the StreamObject from being stored in memory."""
if not self.file_exists: # If file doesn't already exist
self.save_file() # Save stream to disk, if it exists
self.delete_stream() # Delete the stream from memory
def close(self):
"""Both clear the stream, and delete any associated on-disk data"""
logging.info("Closing {}".format(self.id))
self.delete_stream()
if not self.keep_on_disk:
self.delete_file()

View file

@ -32,6 +32,7 @@ import yaml
import copy
import numpy as np
from PIL import Image
import logging
# Pi camera
import picamera
@ -43,7 +44,7 @@ from typing import Tuple
# Threading
import threading
from .base import BaseCamera, StreamObject, log
from .base import BaseCamera, StreamObject
# Richard's fix gain
from .set_picamera_gain import set_analog_gain, set_digital_gain
# Manage config
@ -53,29 +54,12 @@ HERE = os.path.abspath(os.path.dirname(__file__))
DEFAULT_CONFIG = os.path.join(HERE, 'config_picamera.yaml')
def last_entry(object_list: list):
if object_list: # If any images have been captured
return object_list[-1] # Return the latest captured image
else:
return None
def entry_by_id(id: str, object_list: list):
found = None
for o in object_list:
if o.id == id:
found = o
return found
class StreamingCamera(BaseCamera):
"""Raspberry Pi camera implementation of StreamingCamera."""
def __init__(self):
# Run BaseCamera init to start thread
"""Raspberry Pi camera implementation of StreamingCamera."""
# Run BaseCamera init
BaseCamera.__init__(self)
# Capture data
self.images = []
self.videos = []
# Camera settings
self.settings.update({
'video_resolution': (832, 624),
@ -96,55 +80,9 @@ class StreamingCamera(BaseCamera):
def initialisation(self):
"""Run any initialisation code when the frame iterator starts."""
pass
# HANDLE CONTEXT MANAGER AND FILE CLOSING
def close(self):
"""Close method called by BaseCamera __exit__"""
# Close all StreamObjects
for capture_list in [self.images, self.videos]:
for stream_object in capture_list:
stream_object.close()
self.stop_worker()
# RETURNING CAPTURES
@property
def image(self):
"""Return the latest captured image"""
return last_entry(self.images)
@property
def video(self):
"""Return the latest recorded video"""
return last_entry(self.videos)
def image_from_id(self, id):
"""Returns an image StreamObject with a matching ID"""
return entry_by_id(id, self.images)
def video_from_id(self, id):
"""Returns a video StreamObject with a matching ID"""
return entry_by_id(id, self.videos)
# CREATING NEW CAPTURES
def new_stream_object(self, stream_object, target_list: list, shunt_others: bool=True):
"""Add a new capture to a list of captures, and shunt all others"""
if shunt_others: # If shunting all older captures
for obj in target_list: # For each older capture
obj.shunt() # Shunt capture from memory to storage
target_list.append(stream_object)
return stream_object
def new_image(self, stream_object, shunt_others=True):
"""Add a new capture to the image list, and shunt all others"""
return self.new_stream_object(stream_object, self.images, shunt_others=shunt_others)
def new_video(self, stream_object, shunt_others=True):
"""Add a new capture to the video list, and shunt all others"""
return self.new_stream_object(stream_object, self.videos, shunt_others=shunt_others)
# HANDLING SETTINGS
# HANDLE SETTINGS
# TODO: Handle exceptions
# TODO: Have this take a dictionary (not a config file)
@ -162,8 +100,7 @@ class StreamingCamera(BaseCamera):
else:
config = load_config(config_path)
log("Applying config:")
log(config)
logging.debug(config)
# StreamingCamera settings
if 'video_resolution' in config:
@ -221,7 +158,7 @@ class StreamingCamera(BaseCamera):
new_fov = (centre[0] - size/2, centre[1] - size/2, size, size)
self.camera.zoom = new_fov
# LAUNCH ACTIONS
# LAUNCH ACTIONS
def start_preview(self) -> bool:
"""Start the onboard GPU camera preview."""
@ -282,14 +219,15 @@ class StreamingCamera(BaseCamera):
target_obj = target
# Start the camera video recording on port 2
log("Starting record at {}".format(self.settings['video_resolution']))
logging.info("Starting record at {}".format(self.settings['video_resolution']))
self.camera.start_recording(
target,
format=fmt,
splitter_port=2,
resize=self.settings['video_resolution'],
quality=quality)
logging.debug("Recording started successfully.")
# Update state dictionary
self.state['record_active'] = True
@ -303,9 +241,9 @@ class StreamingCamera(BaseCamera):
"""Stop the last started video recording on splitter port 2."""
# Stop the camera video recording on port 2
log("Stopping recording")
logging.info("Stopping recording")
self.camera.stop_recording(splitter_port=2)
log("Recording stopped")
logging.info("Recording stopped")
# Update state dictionary
self.state['record_active'] = False
@ -318,7 +256,7 @@ class StreamingCamera(BaseCamera):
resolution ((int, int)): Resolution to set the camera to,
after stopping recording.
"""
log("Pausing stream")
logging.debug("Pausing stream")
# If no resolution is specified, default to image_resolution
if not resolution:
resolution = self.settings['image_resolution']
@ -337,7 +275,7 @@ class StreamingCamera(BaseCamera):
resolution ((int, int)): Resolution to set the camera to,
before starting recording.
"""
log("Unpausing stream")
logging.debug("Unpausing stream")
if not resolution:
resolution = self.settings['video_resolution']
@ -389,7 +327,7 @@ class StreamingCamera(BaseCamera):
else:
target_obj = target
log("Capturing to {}".format(target))
logging.info("Capturing to {}".format(target))
if not use_video_port:
@ -432,7 +370,7 @@ class StreamingCamera(BaseCamera):
resolution = self.settings['video_resolution']
else:
resolution = self.settings['numpy_resolution']
if resize:
size = resize
else:
@ -440,10 +378,10 @@ class StreamingCamera(BaseCamera):
if not use_video_port:
self.pause_stream_for_capture(resolution=resolution)
log("Creating PiYUVArray")
logging.debug("Creating PiYUVArray")
with picamera.array.PiYUVArray(self.camera, size=size) as output:
log("Capturing to PiYUVArray from {}".format(self.camera))
logging.debug("Capturing to PiYUVArray from {}".format(self.camera))
self.camera.capture(
output,
@ -451,13 +389,13 @@ class StreamingCamera(BaseCamera):
format='yuv',
use_video_port=use_video_port)
log("Capturing complete")
logging.debug("Capturing complete")
if not use_video_port:
self.resume_stream_for_capture()
if rgb:
log("Converting to RGB")
logging.debug("Converting to RGB")
return output.rgb_array
else:
return output.array
@ -473,7 +411,6 @@ class StreamingCamera(BaseCamera):
"""
# Run this initialisation method
self.initialisation()
self.stream_method = 'PiCamera'
with picamera.PiCamera() as self.camera:
# Let camera warm up
@ -492,7 +429,7 @@ class StreamingCamera(BaseCamera):
format='mjpeg',
quality=self.settings['jpeg_quality'],
splitter_port=1)
while True:
# reset stream for next frame
self.stream.seek(0)

View file

@ -1,5 +1,5 @@
#!/usr/bin/env python
from openflexure_microscope.camera.pi import StreamingCamera, StreamObject, log
from openflexure_microscope.camera.pi import StreamingCamera, StreamObject
import os
import io
import sys
@ -12,6 +12,9 @@ from PIL import Image
import unittest
from pprint import pprint
import logging, sys
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
success_string = """
/O
| |