Data processing inside a plan, involving externally-stored data (e.g. images)

example.py

from ophyd.sim import img as detector
from bluesky.plan_stubs import open_run, stage, unstage, close_run, trigger_and_read, subscribe, unsubscribe
from bluesky_live.bluesky_run import BlueskyRun, DocumentCache


def plan():
    """
    Take 10 shots with a detector and do some data processing on the result.

    Key points

    * This detector writes data in external files (temporary files in this
      case). The complexity comes from getting that data out so we can process
      it.
    * The processing here is happening "in band", within the loop of the plan
      logic and in a blocking fashion. For expensive data processing this
      should be moved into a separate process to avoid slowing down the data
      acquisition process and blocking its ability to (for example) monitor for
      Ctrl+C or other interruptions. But of course that adds yet more
      complexity, so this is suitable for light processing.
    """
    # We'll collect documents from the run in here. This is a callable that
    # accepts (name, doc) pairs, so we can just subscribe it.
    dc = DocumentCache()
    token = yield from subscribe("all", dc)

    yield from stage(detector)
    md = {}  # Put some metadata about plan parameters, intent, etc. in here.
    yield from open_run(md=md)
    # The BlueskyRun object provide an interfact to the documents that works
    # exactly like what you get back from databroker.v2, as in
    #
    #     run = catalog[-1]
    #
    # but unlike databroker the documents are in memory (in DocumentCache) not
    # coming from a database.
    run = BlueskyRun(dc)
    for i in range(10):
        yield from trigger_and_read([detector])
        data = run.primary.read()["img"]
        last_image = data[-1]  # most recent image
        # Do some computation...
        print("Total intensity", last_image.sum())
    yield from close_run()
    yield from unsubscribe(token)
    yield from unstage(detector)

if __name__ == "__main__":
    from bluesky import RunEngine

    RE = RunEngine()
    RE(plan())