327 lines
9.5 KiB
Python
327 lines
9.5 KiB
Python
from io import BytesIO
|
|
from typing import List, Optional, Union
|
|
from uuid import UUID
|
|
|
|
from flask import abort, redirect, request, send_file, url_for
|
|
from labthings import Schema, fields, find_component
|
|
from labthings.marshalling import marshal_with, use_args
|
|
from labthings.utilities import description_from_view
|
|
from labthings.views import PropertyView, View
|
|
from marshmallow import pre_dump
|
|
|
|
from openflexure_microscope.api.utilities import get_bool
|
|
from openflexure_microscope.captures import CaptureObject
|
|
|
|
# SCHEMAS
|
|
|
|
|
|
class InstrumentSchema(Schema):
|
|
id = fields.UUID()
|
|
configuration = fields.Dict()
|
|
settings = fields.Dict()
|
|
state = fields.Dict()
|
|
|
|
|
|
class ImageSchema(Schema):
|
|
id = fields.UUID()
|
|
time = fields.String(format="date")
|
|
format = fields.String()
|
|
name = fields.String()
|
|
tags = fields.List(fields.String())
|
|
annotations = fields.Dict(keys=fields.Str(), values=fields.Str())
|
|
|
|
|
|
class CaptureMetadataSchema(Schema):
|
|
# Full dataset dictionary will change depending on the type of
|
|
# dataset, so we can't make a specific schema in this case.
|
|
dataset = fields.Dict()
|
|
# Nested schema for Image data
|
|
image = fields.Nested(ImageSchema())
|
|
# Nested schema for instrument data
|
|
instrument = fields.Nested(InstrumentSchema())
|
|
|
|
|
|
class BasicDatasetSchema(Schema):
|
|
id = fields.UUID()
|
|
name = fields.String()
|
|
type = fields.String()
|
|
|
|
|
|
class CaptureSchema(ImageSchema):
|
|
"""
|
|
Schema containing only basic attributes required
|
|
for interacting with a capture. Additional attributes
|
|
are returned by using FullCaptureSchema
|
|
"""
|
|
|
|
# We need dataset information in the capture array
|
|
# so that client applications can sort data into folders
|
|
# without the server having to do a tonne of file IO
|
|
dataset = fields.Nested(BasicDatasetSchema())
|
|
file = fields.String(
|
|
data_key="path", description="Path of file on microscope device"
|
|
)
|
|
# No need to make a schema for links as we only ever
|
|
# create the dictionary right here in `generate_links`
|
|
links = fields.Dict()
|
|
|
|
@pre_dump
|
|
def generate_links(self, data: Union[dict, CaptureObject], **_):
|
|
if isinstance(data, dict):
|
|
capture_id: Optional[Union[str, UUID]] = data.get("id")
|
|
capture_name: Optional[str] = data.get("name")
|
|
else:
|
|
capture_id = data.id
|
|
capture_name = data.name
|
|
|
|
links = {
|
|
"self": {
|
|
"href": url_for(CaptureView.endpoint, id_=capture_id, _external=True),
|
|
"mimetype": "application/json",
|
|
**description_from_view(CaptureView),
|
|
}
|
|
if CaptureView.endpoint
|
|
else {},
|
|
"tags": {
|
|
"href": url_for(CaptureTags.endpoint, id_=capture_id, _external=True),
|
|
"mimetype": "application/json",
|
|
**description_from_view(CaptureTags),
|
|
}
|
|
if CaptureTags.endpoint
|
|
else {},
|
|
"annotations": {
|
|
"href": url_for(
|
|
CaptureAnnotations.endpoint, id_=capture_id, _external=True
|
|
),
|
|
"mimetype": "application/json",
|
|
**description_from_view(CaptureAnnotations),
|
|
}
|
|
if CaptureAnnotations.endpoint
|
|
else {},
|
|
"download": {
|
|
"href": url_for(
|
|
CaptureDownload.endpoint,
|
|
id_=capture_id,
|
|
filename=capture_name,
|
|
_external=True,
|
|
),
|
|
"mimetype": "image/jpeg",
|
|
**description_from_view(CaptureDownload),
|
|
}
|
|
if CaptureDownload.endpoint
|
|
else {},
|
|
}
|
|
|
|
if isinstance(data, dict):
|
|
data["links"] = links
|
|
else:
|
|
setattr(data, "links", links)
|
|
|
|
return data
|
|
|
|
|
|
class FullCaptureSchema(CaptureSchema):
|
|
"""
|
|
Capture schema including metadata. We exclude this by default
|
|
since it can become huge due to complex settings including
|
|
lens shading tables and CSM matrices.
|
|
"""
|
|
|
|
metadata = fields.Nested(CaptureMetadataSchema())
|
|
|
|
|
|
# VIEWS
|
|
|
|
|
|
class CaptureList(PropertyView):
|
|
tags = ["captures"]
|
|
schema = CaptureSchema(many=True)
|
|
|
|
def get(self):
|
|
"""
|
|
List all image captures
|
|
"""
|
|
microscope = find_component("org.openflexure.microscope")
|
|
image_list: List[CaptureObject] = microscope.captures.images.values()
|
|
return image_list
|
|
|
|
|
|
CAPTURE_ID_PARAMETER = {
|
|
"name": "id_",
|
|
"in": "path",
|
|
"description": "The unique ID of the capture",
|
|
"required": True,
|
|
"schema": {"type": "string"},
|
|
"example": "eeae7ae9-0c0d-45a4-9ef2-7b84bb67a1d1",
|
|
}
|
|
|
|
|
|
class CaptureView(View):
|
|
tags = ["captures"]
|
|
parameters = [CAPTURE_ID_PARAMETER]
|
|
|
|
@marshal_with(FullCaptureSchema())
|
|
def get(self, id_):
|
|
"""
|
|
Description of a single image capture
|
|
"""
|
|
microscope = find_component("org.openflexure.microscope")
|
|
capture_obj: Optional[CaptureObject] = microscope.captures.images.get(id_)
|
|
|
|
if not capture_obj:
|
|
return abort(404) # 404 Not Found
|
|
|
|
return capture_obj
|
|
|
|
get.responses = {404: {"description": "Capture object was not found"}}
|
|
|
|
def delete(self, id_):
|
|
"""
|
|
Delete a single image capture
|
|
"""
|
|
microscope = find_component("org.openflexure.microscope")
|
|
capture_obj: Optional[CaptureObject] = microscope.captures.images.get(id_)
|
|
|
|
if not capture_obj:
|
|
return abort(404) # 404 Not Found
|
|
|
|
# Delete the capture file
|
|
capture_obj.delete()
|
|
# Delete from capture list
|
|
del microscope.captures.images[id_]
|
|
|
|
return "", 204
|
|
|
|
|
|
class CaptureDownload(View):
|
|
tags = ["captures"]
|
|
responses = {
|
|
200: {"content": {"image/jpeg": {}}, "description": "Image data in JPEG format"}
|
|
}
|
|
parameters = [
|
|
CAPTURE_ID_PARAMETER,
|
|
{
|
|
"name": "filename",
|
|
"in": "path",
|
|
"description": "The filename of the downloaded image.",
|
|
"required": False,
|
|
"schema": {"type": "string"},
|
|
"example": "myimage.jpeg",
|
|
}
|
|
]
|
|
|
|
def get(self, id_, filename: Optional[str]):
|
|
"""
|
|
Image data for a single image capture
|
|
"""
|
|
microscope = find_component("org.openflexure.microscope")
|
|
capture_obj: Optional[CaptureObject] = microscope.captures.images.get(id_)
|
|
|
|
if not capture_obj:
|
|
return abort(404) # 404 Not Found
|
|
|
|
thumbnail: bool = get_bool(request.args.get("thumbnail", ""))
|
|
|
|
# If no filename is specified, redirect to the capture's currently set filename
|
|
if not filename:
|
|
return redirect(
|
|
url_for(
|
|
"DownloadAPI",
|
|
id=id_,
|
|
filename=capture_obj.name,
|
|
thumbnail=thumbnail,
|
|
),
|
|
code=307,
|
|
)
|
|
|
|
# Download the image data using the requested filename
|
|
if thumbnail:
|
|
img: Optional[BytesIO] = capture_obj.thumbnail
|
|
else:
|
|
img = capture_obj.data
|
|
|
|
# If we can't get any data, return 404
|
|
if not img:
|
|
return abort(404) # 404 Not Found
|
|
|
|
return send_file(img, mimetype="image/jpeg")
|
|
|
|
|
|
class CaptureTags(View):
|
|
tags = ["captures"]
|
|
parameters = [CAPTURE_ID_PARAMETER]
|
|
|
|
def get(self, id_):
|
|
"""
|
|
Get tags associated with a single image capture
|
|
"""
|
|
microscope = find_component("org.openflexure.microscope")
|
|
capture_obj: Optional[CaptureObject] = microscope.captures.images.get(id_)
|
|
|
|
if not capture_obj:
|
|
return abort(404) # 404 Not Found
|
|
|
|
return capture_obj.tags
|
|
|
|
@use_args(fields.List(fields.String(), required=True))
|
|
def put(self, args, id_):
|
|
"""
|
|
Add tags to a single image capture
|
|
"""
|
|
microscope = find_component("org.openflexure.microscope")
|
|
capture_obj: Optional[CaptureObject] = microscope.captures.images.get(id_)
|
|
|
|
if not capture_obj:
|
|
return abort(404) # 404 Not Found
|
|
|
|
capture_obj.put_tags(args)
|
|
|
|
return capture_obj.tags
|
|
|
|
@use_args(fields.List(fields.String(), required=True))
|
|
def delete(self, args, id_):
|
|
"""
|
|
Delete tags from a single image capture
|
|
"""
|
|
microscope = find_component("org.openflexure.microscope")
|
|
capture_obj: Optional[CaptureObject] = microscope.captures.images.get(id_)
|
|
|
|
if not capture_obj:
|
|
return abort(404) # 404 Not Found
|
|
|
|
for tag in args:
|
|
capture_obj.delete_tag(str(tag))
|
|
|
|
return capture_obj.tags
|
|
|
|
|
|
class CaptureAnnotations(View):
|
|
tags = ["captures"]
|
|
parameters = [CAPTURE_ID_PARAMETER]
|
|
|
|
def get(self, id_):
|
|
"""
|
|
Get annotations associated with a single image capture
|
|
"""
|
|
microscope = find_component("org.openflexure.microscope")
|
|
capture_obj: Optional[CaptureObject] = microscope.captures.images.get(id_)
|
|
|
|
if not capture_obj:
|
|
return abort(404) # 404 Not Found
|
|
|
|
return capture_obj.annotations
|
|
|
|
@use_args(fields.Dict())
|
|
def put(self, args, id_):
|
|
"""
|
|
Update metadata for a single image capture
|
|
"""
|
|
microscope = find_component("org.openflexure.microscope")
|
|
capture_obj = microscope.captures.images.get(id_)
|
|
|
|
if not capture_obj:
|
|
return abort(404) # 404 Not Found
|
|
|
|
capture_obj.put_annotations(args)
|
|
|
|
return capture_obj.annotations
|