diff --git a/openflexure_microscope/camera/mock.py b/openflexure_microscope/camera/mock.py index 9b6103af..180d8a2f 100644 --- a/openflexure_microscope/camera/mock.py +++ b/openflexure_microscope/camera/mock.py @@ -8,7 +8,7 @@ from __future__ import division import io import time import numpy as np -from PIL import Image +from PIL import Image, ImageFont, ImageDraw import logging @@ -20,16 +20,12 @@ from .base import BaseCamera, CaptureObject # MAIN CLASS class MockStreamer(BaseCamera): - def __init__(self): # Run BaseCamera init BaseCamera.__init__(self) # Store state of PiCameraStreamer - self.state.update({ - 'stream_active': False, - 'record_active': False - }) + self.state.update({"stream_active": False, "record_active": False}) # Update config properties self.image_resolution = (1312, 976) @@ -41,9 +37,25 @@ class MockStreamer(BaseCamera): # Create an empty stream self.stream = io.BytesIO() + # Generate an initial dummy image + self.generate_new_dummy_image() + # Start streaming self.start_worker() + def generate_new_dummy_image(self): + # Create a dummy image to serve in the stream + image = Image.new( + "RGB", + (self.stream_resolution[0], self.stream_resolution[1]), + color=(0, 0, 0), + ) + + draw = ImageDraw.Draw(image) + draw.text((20, 70), "Camera disconnected") + + image.save(self.stream, format="JPEG") + def initialisation(self): """Run any initialisation code when the frame iterator starts.""" pass @@ -60,10 +72,10 @@ class MockStreamer(BaseCamera): """ conf_dict = { - 'stream_resolution': self.stream_resolution, - 'image_resolution': self.image_resolution, - 'numpy_resolution': self.numpy_resolution, - 'jpeg_quality': self.jpeg_quality + "stream_resolution": self.stream_resolution, + "image_resolution": self.image_resolution, + "numpy_resolution": self.numpy_resolution, + "jpeg_quality": self.jpeg_quality, } return conf_dict @@ -78,8 +90,6 @@ class MockStreamer(BaseCamera): Args: config (dict): Dictionary of config parameters. """ - # TODO: Include timing and batching logic when applying PiCamera settings - paused_stream = False logging.debug("PiCameraStreamer: Applying config:") logging.debug(config) @@ -87,7 +97,7 @@ class MockStreamer(BaseCamera): with self.lock: # Apply valid config params to Picamera object - if not self.state['record_active']: # If not recording a video + if not self.state["record_active"]: # If not recording a video # PiCameraStreamer parameters for key, value in config.items(): # For each provided setting @@ -101,9 +111,10 @@ class MockStreamer(BaseCamera): else: raise Exception( - "Cannot update camera config while recording is active.") + "Cannot update camera config while recording is active." + ) - def set_zoom(self, zoom_value: float = 1.) -> None: + def set_zoom(self, zoom_value: float = 1.0) -> None: """ Change the camera zoom, handling re-centering and scaling. """ @@ -119,12 +130,7 @@ class MockStreamer(BaseCamera): """Stop the on board GPU camera preview.""" logging.warning("GPU preview not implemented in mock camera") - - def start_recording( - self, - output, - fmt: str = 'h264', - quality: int = 15): + def start_recording(self, output, fmt: str = "h264", quality: int = 15): """Start recording. Start a new video recording, writing to a output object. @@ -142,19 +148,19 @@ class MockStreamer(BaseCamera): # Start recording method only if a current recording is not running logging.warning("Recording not implemented in mock camera") - def stop_recording(self): """Stop the last started video recording on splitter port 2.""" with self.lock: logging.warning("Recording not implemented in mock camera") def capture( - self, - output, - fmt: str = 'jpeg', - use_video_port: bool = False, - resize: Tuple[int, int] = None, - bayer: bool = True): + self, + output, + fmt: str = "jpeg", + use_video_port: bool = False, + resize: Tuple[int, int] = None, + bayer: bool = True, + ): """ Capture a still image to a StreamObject. @@ -172,20 +178,11 @@ class MockStreamer(BaseCamera): with self.lock: logging.warning("Capture not implemented in mock camera") - - def gen_img(self): - imarray = np.random.rand(self.stream_resolution[1], self.stream_resolution[0], 3) * 255 - im = Image.fromarray(imarray.astype('uint8')).convert('L') - im.save(self.stream, format="JPEG") - # HANDLE STREAM FRAMES def frames(self): """ Create generator that returns frames from the camera. - - Records video from port 1 to a byte stream, - and iterates sequential frames. """ # Run this initialisation method self.initialisation() @@ -196,14 +193,7 @@ class MockStreamer(BaseCamera): # While the iterator is not closed try: while True: - # reset stream for next frame - self.stream.seek(0) - self.stream.truncate() - # to stream, read the new frame time.sleep(1 / self.framerate * 0.1) - - # yield the result to be read - self.gen_img() frame = self.stream.getvalue() # ensure the size of package is right