BMM camera ophyd integration

camera_device.py

import os
import uuid
import threading
import itertools

import requests
from ophyd import Device, Component, Signal, DeviceStatus
from ophyd.areadetector.filestore_mixins import resource_factory

# See for resource_factory docstring
# https://github.com/bluesky/ophyd/blob/b1d258a36c974013b6e3ac8ee7112ed876b7653a/ophyd/areadetector/filestore_mixins.py#L70-L112


class ExternalFileReference(Signal):
    """
    A pure software signal where a Device can stash a datum_id
    """
    def __init__(self, *args, shape, **kwargs):
        super().__init__(*args, **kwargs)
        self.shape = shape

    def describe(self):
        res = super().describe()
        res[self.name].update(dict(external="FILESTORE:", dtype="array", shape=self.shape))
        return res


class Camera(Device):
    # We initialize the shape to [] and update it below once we know the shape
    # of the array.
    image = Component(ExternalFileReference, value="", kind="normal", shape=[])

    def __init__(self, *args, root, **kwargs):
        super().__init__(*args, **kwargs)
        self._root = root
        # Use this lock to ensure that we only process one "trigger" at a time.
        # Generally bluesky should care of this, so this is just an extra
        # precaution.
        self._acquiring_lock = threading.Lock()
        self._counter = None  # set to an itertools.count object when staged
        self._asset_docs_cache = []
        self._SPEC = "BMM_CAMERA_1"

    def stage(self):
        # Set the filepath where will be saving images.
        self._rel_path_template = f"path/to/files/{uuid.uuid4()}_%d.ext"
        # Create a Resource document referring to this series of images that we
        # are about to take, and stash it in _asset_docs_cache.
        resource, self._datum_factory = resource_factory(
            self._SPEC, self._root, self._rel_path_template, {}, "posix")
        self._asset_docs_cache.append(("resource", resource))
        self._counter = itertools.count()
        return super().stage()

    def unstage(self):
        self._counter = None
        self._asset_docs_cache.clear()
        return super().unstage()

    def trigger(self):
        status = DeviceStatus(self)
        if self._counter is None:
            raise RuntimeError("Device must be staged before triggering.")
        i = next(self._counter)
        # Start a background thread to capture an image and write it to disk.
        thread = threading.Thread(target=self._capture, args=(status, i))
        thread.start()
        # Promptly return a status object, which will be marked "done" when the
        # capture completes.
        return status

    def _capture(self, status, i):
        "This runs on a background thread."
        try:
            if not self._acquiring_lock.acquire(timeout=0):
                raise RuntimeError("Cannot trigger, currently trigggering!")
            filename = os.path.join(self._root, self._rel_path_template % i)
            # Kick off requests, or subprocess, or whatever with the result
            # that a file is saved at `filename`.
            ...
            # Extract the shape of the array.
            shape = (512, 512)  # placeholder
            self.image.shape = shape
            # Compose a Datum document referring to this specific image, and
            # stash it in _asset_docs_cache.
            datum = self._datum_factory({"index": i})
            self._asset_docs_cache.append(("datum", datum))
            self.image.set(datum["datum_id"]).wait()
        except Exception as exc:
            status.set_exception(exc)
        else:
            status.set_finished()
        finally:
            self._acquiring_lock.release()

    def collect_asset_docs(self):
        "Yield the documents from our cache, and reset it."
        yield from self._asset_docs_cache
        self._asset_docs_cache.clear()



class BMM_JPEG_HANDLER:
    def __init__(self, resource_path):
        # resource_path is really a template string with a %d in it
        self._template = resource_path

    def __call__(self, index):
        import PIL, numpy
        filepath = self._template % index
        return numpy.asarray(PIL.Image.open(filepath))

This has been polished and added to the Bluesky Tutorials via bluesky/tutorials#107