352 lines
11 KiB
Python
352 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
|
import time
|
|
import os
|
|
import shutil
|
|
import threading
|
|
import datetime
|
|
import logging
|
|
|
|
import threading
|
|
|
|
from abc import ABCMeta, abstractmethod
|
|
from collections import OrderedDict
|
|
|
|
from .capture import CaptureObject, build_captures_from_exif
|
|
from openflexure_microscope.utilities import entry_by_uuid
|
|
from labthings.core.lock import StrictLock
|
|
from labthings.core.event import ClientEvent
|
|
|
|
from openflexure_microscope.paths import data_file_path
|
|
|
|
BASE_CAPTURE_PATH = data_file_path("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
|
|
return object_list[-1] # Return the latest captured image
|
|
else:
|
|
return None
|
|
|
|
|
|
def generate_basename():
|
|
"""Return a default filename based on the capture datetime"""
|
|
return datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
|
|
|
|
|
|
def generate_numbered_basename(obj_list: list) -> str:
|
|
initial_basename = generate_basename()
|
|
basename = initial_basename
|
|
# Handle clashing
|
|
iterator = 1
|
|
while basename in [obj.basename for obj in obj_list]:
|
|
basename = initial_basename + "_{}".format(iterator)
|
|
iterator += 1
|
|
|
|
return basename
|
|
|
|
|
|
class BaseCamera(metaclass=ABCMeta):
|
|
"""
|
|
Base implementation of StreamingCamera.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.thread = None
|
|
self.camera = None
|
|
|
|
self.lock = StrictLock(timeout=1, name="Camera")
|
|
|
|
self.frame = None
|
|
self.last_access = 0
|
|
self.event = ClientEvent()
|
|
self.stop = False # Used to indicate that the stream loop should break
|
|
|
|
self.stream_timeout = 20
|
|
self.stream_timeout_enabled = False
|
|
|
|
self.stream_active = False
|
|
self.record_active = False
|
|
|
|
self.paths = {"default": BASE_CAPTURE_PATH, "temp": TEMP_CAPTURE_PATH}
|
|
|
|
# Capture data
|
|
self.images = OrderedDict()
|
|
self.videos = OrderedDict()
|
|
|
|
@property
|
|
@abstractmethod
|
|
def configuration(self):
|
|
"""The current camera configuration."""
|
|
pass
|
|
|
|
@property
|
|
@abstractmethod
|
|
def state(self):
|
|
"""The current read-only camera state."""
|
|
pass
|
|
|
|
@property
|
|
def settings(self):
|
|
return self.read_settings()
|
|
|
|
@abstractmethod
|
|
def update_settings(self, config: dict):
|
|
"""Update settings from a config dictionary"""
|
|
with self.lock:
|
|
# Apply valid config params to camera object
|
|
for key, value in config.items(): # For each provided setting
|
|
if hasattr(self, key): # If the instance has a matching property
|
|
setattr(self, key, value) # Set to the target value
|
|
|
|
@abstractmethod
|
|
def read_settings(self) -> dict:
|
|
"""Return the current settings as a dictionary"""
|
|
return {"paths": self.paths}
|
|
|
|
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):
|
|
"""Close the BaseCamera and all attached StreamObjects."""
|
|
logging.info("Closing {}".format(self))
|
|
# Close all StreamObjects
|
|
for capture_list in [self.images.values(), self.videos.values()]:
|
|
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"]))
|
|
|
|
def rebuild_captures(self):
|
|
self.images = build_captures_from_exif(self.paths["default"])
|
|
|
|
# RETURNING CAPTURES
|
|
|
|
@property
|
|
def image(self):
|
|
"""Return the latest captured image."""
|
|
return last_entry(self.images.values())
|
|
|
|
@property
|
|
def video(self):
|
|
"""Return the latest recorded video."""
|
|
return last_entry(self.videos.values())
|
|
|
|
def image_from_id(self, image_id):
|
|
"""Return an image StreamObject with a matching ID."""
|
|
logging.warning("image_from_id is deprecated. Access captures as a dictionary.")
|
|
return entry_by_uuid(image_id, self.images.values())
|
|
|
|
def video_from_id(self, video_id):
|
|
"""Return a video StreamObject with a matching ID."""
|
|
logging.warning("video_from_id is deprecated. Access captures as a dictionary.")
|
|
return entry_by_uuid(video_id, self.videos.values())
|
|
|
|
# CREATING NEW CAPTURES
|
|
|
|
def new_image(
|
|
self,
|
|
temporary: bool = True,
|
|
filename: str = None,
|
|
folder: str = "",
|
|
fmt: str = "jpeg",
|
|
):
|
|
|
|
"""
|
|
Create a new image capture object.
|
|
|
|
Args:
|
|
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.
|
|
folder (str): Name of the folder in which to store the capture.
|
|
fmt (str): Format of the capture.
|
|
"""
|
|
|
|
# Generate file name
|
|
if not filename:
|
|
filename = generate_numbered_basename(self.images.values())
|
|
logging.debug(filename)
|
|
filename = "{}.{}".format(filename, fmt)
|
|
|
|
# Generate folder
|
|
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(filepath=filepath)
|
|
# Insert a temporary tag if temporary
|
|
if temporary:
|
|
output.put_tags(["temporary"])
|
|
|
|
# Update capture list
|
|
capture_key = str(output.id)
|
|
logging.debug(f"Adding image {output} with key {capture_key}")
|
|
self.images[capture_key] = output
|
|
|
|
return output
|
|
|
|
def new_video(
|
|
self,
|
|
temporary: bool = False,
|
|
filename: str = None,
|
|
folder: str = "",
|
|
fmt: str = "h264",
|
|
):
|
|
|
|
"""
|
|
Create a new video capture object.
|
|
|
|
Args:
|
|
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.
|
|
folder (str): Name of the folder in which to store the capture.
|
|
fmt (str): Format of the capture.
|
|
"""
|
|
# TODO: Remove the redundancy here
|
|
|
|
# Generate file name
|
|
if not filename:
|
|
filename = generate_numbered_basename(self.videos.values())
|
|
logging.debug(filename)
|
|
filename = "{}.{}".format(filename, fmt)
|
|
|
|
# Generate folder
|
|
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(filepath=filepath)
|
|
# Insert a temporary tag if temporary
|
|
if temporary:
|
|
output.put_tags(["temporary"])
|
|
|
|
# Update capture list
|
|
capture_key = str(output.id)
|
|
logging.debug(f"Adding video {output} with key {capture_key}")
|
|
self.videos[capture_key] = output
|
|
|
|
return output
|
|
|
|
# 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 not self.stream_active:
|
|
# Spawn a greenlet to handle stream
|
|
# start background frame thread
|
|
self.thread = threading.Thread(target=self._thread)
|
|
self.thread.daemon = True
|
|
self.thread.start()
|
|
|
|
# wait until frames are available
|
|
logging.info("Waiting for frames")
|
|
while self.get_frame() is None:
|
|
if time.time() > timeout_time:
|
|
raise TimeoutError("Timeout waiting for frames.")
|
|
else:
|
|
time.sleep(0.1)
|
|
return True
|
|
|
|
def stop_worker(self, timeout: int = 5) -> bool:
|
|
"""Flag worker thread for stop. Waits for thread close or timeout."""
|
|
logging.debug("Stopping worker thread")
|
|
timeout_time = time.time() + timeout
|
|
|
|
if self.stream_active:
|
|
self.stop = True
|
|
self.thread.join() # Wait for stream thread to exit
|
|
logging.debug("Waiting for stream thread to exit.")
|
|
|
|
while self.stream_active:
|
|
if time.time() > timeout_time:
|
|
logging.debug("Timeout waiting for worker thread close.")
|
|
raise TimeoutError("Timeout waiting for worker thread close.")
|
|
else:
|
|
time.sleep(0.1)
|
|
return True
|
|
|
|
# HANDLE STREAM FRAMES
|
|
|
|
def get_frame(self):
|
|
"""Return the current camera frame."""
|
|
self.last_access = time.time()
|
|
|
|
# wait for a signal from the camera thread
|
|
self.event.wait()
|
|
self.event.clear()
|
|
|
|
return self.frame
|
|
|
|
@abstractmethod
|
|
def frames(self):
|
|
"""Create generator that returns frames from the camera."""
|
|
pass
|
|
|
|
# WORKER THREAD
|
|
|
|
def _thread(self):
|
|
"""Camera background thread."""
|
|
self.frames_iterator = self.frames()
|
|
logging.debug("Entering worker thread.")
|
|
|
|
self.stream_active = True
|
|
|
|
for frame in self.frames_iterator:
|
|
self.frame = frame
|
|
self.event.set() # send signal to clients
|
|
|
|
# Handle timeout
|
|
if (
|
|
self.stream_timeout_enabled
|
|
and ( # If using timeout
|
|
time.time() - self.last_access > self.stream_timeout
|
|
)
|
|
and not self.preview_active # And GPU preview is not active
|
|
):
|
|
self.frames_iterator.close()
|
|
break
|
|
|
|
try:
|
|
if self.stop is True:
|
|
logging.debug("Worker thread flagged for stop.")
|
|
self.frames_iterator.close()
|
|
break
|
|
|
|
except AttributeError:
|
|
pass
|
|
|
|
logging.debug("BaseCamera worker thread exiting...")
|
|
# Set stream_activate state
|
|
self.stream_active = False
|