openflexure-microscope-server/openflexure_microscope/api/v2/views/actions/camera.py

import io
import logging
from typing import Dict, Optional, Tuple

from flask import send_file
from labthings import Schema, fields, find_component
from labthings.views import ActionView

from openflexure_microscope.api.v2.views.captures import CaptureSchema


class CaptureResizeSchema(Schema):
    width = fields.Integer(example=640, required=True)
    height = fields.Integer(example=480, required=True)


class BasicCaptureArgs(Schema):
    use_video_port = fields.Boolean(load_default=False)
    bayer = fields.Boolean(
        load_default=False, metadata={"description": "Include raw bayer data in capture"}
    )
    resize = fields.Nested(CaptureResizeSchema(), required=False)


class FullCaptureArgs(BasicCaptureArgs):
    filename = fields.String(example="MyFileName")
    temporary = fields.Boolean(load_default=False, metadata={"description": "Delete capture on shutdown"})
    annotations = fields.Dict(load_default={}, example={"Client": "SwaggerUI"})
    tags = fields.List(fields.String, load_default=[], example=["docs"])


class CaptureAPI(ActionView):
    """
    Create a new image capture. 
    """

    args = FullCaptureArgs()
    schema = CaptureSchema()

    def post(self, args):
        """
        Create a new capture
        """
        microscope = find_component("org.openflexure.microscope")

        resize_dict: Optional[Dict[str, int]] = args.get("resize", None)
        if resize_dict:
            resize: Optional[Tuple[int, int]] = (
                int(resize_dict["width"]),
                int(resize_dict["height"]),
            )  # Convert dict to tuple
        else:
            resize = None

        # Explicitally acquire lock (prevents empty files being created if lock is unavailable)
        with microscope.camera.lock:
            return microscope.capture(
                filename=args.get("filename"),
                temporary=args.get("temporary"),
                use_video_port=args.get("use_video_port"),
                resize=resize,
                bayer=args.get("bayer"),
                annotations=args.get("annotations"),
                tags=args.get("tags"),
            )


class RAMCaptureAPI(ActionView):
    """Take a non-persistent image capture."""

    args = BasicCaptureArgs()
    responses = {
        200: {
            "content": {"image/jpeg": {}},
            "description": "A JPEG image, representing the capture",
        }
    }

    def post(self, args):
        """
        Take a non-persistant image capture.
        """
        microscope = find_component("org.openflexure.microscope")

        resize_dict: Optional[Dict[str, int]] = args.get("resize", None)
        if resize_dict:
            resize: Optional[Tuple[int, int]] = (
                int(resize_dict["width"]),
                int(resize_dict["height"]),
            )  # Convert dict to tuple
        else:
            resize = None

        # Open a BytesIO stream to be destroyed once request has returned
        with microscope.camera.lock, io.BytesIO() as stream:

            microscope.camera.capture(
                stream,
                use_video_port=args.get("use_video_port"),
                resize=resize,
                bayer=args.get("bayer"),
            )

            stream.seek(0)

            return send_file(io.BytesIO(stream.getbuffer()), mimetype="image/jpeg")


class GPUPreviewStartAPI(ActionView):
    """
    Start the onboard GPU preview.
    Optional "window" parameter can be passed to control the position and size of the preview window,
    in the format ``[x, y, width, height]``.
    """

    args = {"window": fields.List(fields.Integer, load_default=[], example=[0, 0, 640, 480])}

    def post(self, args):
        """
        Start the onboard GPU preview.
        """
        microscope = find_component("org.openflexure.microscope")

        # Get window argument from request
        window_arg = args.get("window")
        logging.debug(window_arg)

        # Default to no window
        fullscreen: bool = True
        window: Optional[Tuple[int, int, int, int]] = None

        # If request argument is well formed, use that
        if len(window_arg) == 4:
            fullscreen = False
            window = (int(w) for w in window_arg)

        microscope.camera.start_preview(fullscreen=fullscreen, window=window)
        return microscope.state


class GPUPreviewStopAPI(ActionView):
    def post(self):
        """
        Stop the onboard GPU preview.
        """
        microscope = find_component("org.openflexure.microscope")
        microscope.camera.stop_preview()
        return microscope.state