{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Flyer Basics\n", "\n", "We want to understand how to build a *Flyer* in \"Bluesky\" to support various types of fly scans and remote data loggers. The data about Flyers is spread about the standard documentation. We need some clarity and a few examples that build complexity incrementally.\n", "\n", "The basic notion of a Flyer is that it directs an *external controller* (we'll call it the *controller* here) to perform some data collection process. Usually, a *controller* is used to collect data at rates beyond the capabilities of Bluesky plans and the RunEngine. Examples could be waveforms from a storage oscilloscope or a continuous motion scan of a diffractometer.\n", "\n", "This notebook will show the basic requirements for a Flyer and build a simple working example you can use a template for your own work. (Our inspiration to create this [basic flyer tutorial](https://github.com/prjemian/ipython_mintvm/issues/3#issuecomment-385058577) was the goal of operating various fly scans at the APS, such as the USAXS fly scan, from Bluesky. The examples we found before we started this project quickly became too instrument-specific to serve as tutorials.)\n", "## Python imports and definitions \n", "\n", "Here are the full set of packages to imported. The first block are Python standard packages, then come the ophyd, Bluesky, and databroker packages. Just the parts we plan on using here. Since this is also a tutorial, we will not rename imports or use other such shortcuts in the documentation (the online code has some shortcuts).\n", "\n", "* Create a logger instance in case we want to investigate internal details as our code runs.\n", "* Create an instance of the Bluesky RunEngine.\n", "* Create an instance of the databroker using `databroker.temp()`, establishing a temporary, disposable yet fully-functional databroker for just this tutorial session.\n", "* Subscribe that databroker instance to the RunEngine so it receives the document stream each time we run a plan.\n", "\n", "Links\n", "\n", "* USAXS fly scan - https://github.com/APS-USAXS/ipython-usaxs/issues/3\n", "* `databroker.temp()` - https://blueskyproject.io/databroker/v1/api.html\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import logging\n", "import threading\n", "import time\n", "\n", "import ophyd\n", "import bluesky\n", "import bluesky.plans\n", "import databroker\n", "\n", "logger = logging.getLogger()\n", "RE = bluesky.RunEngine({})\n", "db = databroker.temp()\n", "RE.subscribe(db.insert)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Bare Minimum Requirements for a Flyer \n", "\n", "In Bluesky, a *Flyer* is an `ophyd.Device` that meets the *Fly-able interface*, which has three methods:\n", "\n", "1. Kickoff - begin accumulating data\n", "1. Complete - Bluesky tells the Flyer that Bluesky is ready to receive data\n", "1. Collect - the device provides the data to Bluesky\n", "\n", "The first two methods must return an instance of `ophyd.status.Status` (a.k.a. a *status* object). For our code, we choose a specialized version of `Status()`, the `DeviceStatus()` object, based on the explanation provided in discussion.\n", "\n", "The `collect()` method requires a companion `describe_collect()` that informs the RunEngine what kind of data to expect from `collect()`. If no timestamp information is provided from the *controller*, then do as we show here and use the workstation's clock to provide a timestamp for the event.\n", "\n", "This example (which does absolutely nothing) meets the bare minimum requirement.\n", "\n", "Links\n", "\n", "* *Flyer* - https://blueskyproject.io/bluesky/async.html?highlight=flyer#flying\n", "* `ophyd.Device` - https://blueskyproject.io/ophyd/device-overview.html\n", "* *Fly-able interface* - https://blueskyproject.io/ophyd/architecture.html#fly-able-interface\n", "* returns - https://blueskyproject.io/bluesky/hardware.html#kickoff\n", "* `ophyd.status.Status` - https://blueskyproject.io/ophyd/generated/ophyd.status.Status.html\n", "* `DeviceStatus()` - https://blueskyproject.io/ophyd/generated/ophyd.status.DeviceStatus.html\n", "* discussion - https://github.com/NSLS-II/tutorial-docker/pull/7#discussion_r185054364" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "class BareMinimumFlyer(ophyd.Device):\n", "\n", " def kickoff(self):\n", " kickoff_status = ophyd.DeviceStatus(self)\n", " kickoff_status._finished(success=True)\n", " return kickoff_status\n", "\n", " def complete(self):\n", " complete_status = ophyd.DeviceStatus(self)\n", " complete_status._finished(success=True)\n", " return complete_status\n", "\n", " def collect(self):\n", " yield {'data':{}, 'timestamps':{}, 'time':time.time()}\n", " \n", " def describe_collect(self):\n", " return {self.name: {}}\n", "\n", "\n", "flyer = BareMinimumFlyer(name=\"flyer\")\n", "print(flyer.complete())\n", "print(list(flyer.collect()))\n", "\n", "# if this next step succeeds, it's proof that we did this right!\n", "RE(bluesky.plans.fly([flyer]))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Only three `documents` were emitted from the `RunEngine`: `start`, `descriptor`, and `stop` since we generated no data content." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "list(db[-1].documents())" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Flyer : a starting template \n", "\n", "The `BareMinimumFlyer` is a good start to use a Flyer but we'll need to add a few more things to make a good template. The first thing to do is to make the status objects known to any method of the class. We'll prepend both of them with `self.` and describe in the code comments what is happening. In the constructor (`__init__()`), we set both to `None`, the value we expect when not kicked off or *flying*, respectively. Since we **need** a constructor, we must remember to call the constructor of the superclass as well or our `ophyd.Device` will not work correctly.\n", "\n", " def __init__(self, *args, **kwargs):\n", " super().__init__('', parent=None, **kwargs)\n", " self.kickoff_status = None\n", " self.complete_status = None\n", "\n", "Our *controller* could take some time to signal that it is finished (seconds to minutes, at least). We need a way to detect this completion. We can do that either by polling the PV or by subscribing to a callback on the completion event. We'll make that choice when we implement the actual activity. Here, intend to wait in a polling loop. Since the polling loop is an activity that does not return until the *controller* is done, we must do that waiting in a thread (consider an [alternative suggestion to use `epics.ca.CAThread` separate from that of the RunEngine. (We do not want to block the RunEngine thread so it is free to respond to other activities, such as data from other streams or the user inerface.) So, we run `my_activity()` in a separate thread that is called from `kickoff()`:\n", "\n", " thread = threading.Thread(target=self.my_activity, daemon=True)\n", " thread.start()\n", "\n", "Link\n", "\n", "* `epics.ca.CAThread` - https://github.com/NSLS-II/tutorial-docker/pull/7#discussion_r184905960\n", "\n", "The basic outline of `my_activity()` is:\n", "\n", " def my_activity(self):\n", " # tell the *controller* to do its work\n", " # once started, we notify by updating the status object\n", " self.kickoff_status._finished(success=True)\n", "\n", " # wait for for it to tell us it is done \n", " # after the wait, we declare victory\n", " self.complete_status._finished(success=True)\n", "\n", "The waiting step will *block the thread* in which `my_activity()` is running but that's OK since this is separate from the RunEngine's thread.\n", "\n", "Within `my_activity()`, we've left two comments starting with `# TODO: ` marking where we need to add code that will complete the specifics of our final project. Since this tutorial develops a starting template for that project (and others), we leave these comments as-is.\n", "\n", "We've also added some diagnostic reporting (calls to `logger.info(...)`) to build out the next example:" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "class MyFlyer(ophyd.Device):\n", " \"\"\"\n", " starting template for a Flyer that we understand\n", " \"\"\"\n", "\n", " def __init__(self, *args, **kwargs):\n", " super().__init__('', parent=None, **kwargs)\n", " self.complete_status = None\n", "\n", " def my_activity(self):\n", " \"\"\"\n", " start the \"fly scan\" here, could wait for completion\n", " \n", " It's OK to use blocking calls here \n", " since this is called in a separate thread\n", " from the Bluesky RunEngine.\n", " \"\"\"\n", " logger.info(\"activity()\")\n", " if self.complete_status is None:\n", " logger.info(\"leaving activity() - not complete\")\n", " return\n", " \n", " # TODO: do the activity here\n", "\n", " # once started, we notify by updating the status object\n", " self.kickoff_status._finished(success=True)\n", "\n", " # TODO: wait for completion\n", " \n", " # after the wait, we declare victory\n", " self.complete_status._finished(success=True)\n", " logger.info(\"activity() complete. status = \" + str(self.complete_status))\n", "\n", " def kickoff(self):\n", " \"\"\"\n", " Start this Flyer\n", " \"\"\"\n", " logger.info(\"kickoff()\")\n", " self.kickoff_status = ophyd.DeviceStatus(self)\n", " self.complete_status = ophyd.DeviceStatus(self)\n", " \n", " thread = threading.Thread(target=self.my_activity, daemon=True)\n", " thread.start()\n", "\n", " return self.kickoff_status\n", "\n", " def complete(self):\n", " \"\"\"\n", " Wait for flying to be complete\n", " \"\"\"\n", " logger.info(\"complete()\")\n", " if self.complete_status is None:\n", " raise RuntimeError(\"No collection in progress\")\n", "\n", " return self.complete_status\n", "\n", " def collect(self):\n", " \"\"\"\n", " Retrieve the data\n", " \"\"\"\n", " logger.info(\"collect()\")\n", " self.complete_status = None\n", " yield {'data':{}, 'timestamps':{}, 'time':time.time()}\n", " \n", " def describe_collect(self):\n", " \"\"\"\n", " Describe details for ``collect()`` method\n", " \"\"\"\n", " logger.info(\"describe_collect()\")\n", " return {self.name: {}}\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "ifly = MyFlyer(name=\"ifly\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Diagnostics \n", "\n", "When building a `Flyer`, it is useful to have some diagnostics in place. Already, we have been using some of these, including printing interim messages via calls to `logger.info(...)`. Another useful diagnostic step is to call each of the methods individually to make sure they are acting as expected.\n", "\n", "1. create an instance of the `Flyer`\n", "\n", " flyer = MyFlyer(name=\"flyer\")\n", "\n", "1. verify that `kickoff()` returns a status that is \"Done\"\n", "\n", " status = flyer.kickoff()\n", " status.done\n", "\n", "1. verify that `complete()` returns a status that is \"Done\"\n", "\n", " status = flyer.complete()\n", " status.done\n", "\n", "1. verify that `describe_collect()` returns a dictionary\n", "\n", " d = flyer.describe_collect()\n", " d\n", "\n", "1. verify that `collect()` returns a generator\n", "\n", " g = flyer.collect()\n", " g\n", "\n", "1. verify that generator is a list of data dictionaries\n", "\n", " list(g)\n", "\n", "\n", "Apply some of those steps here (we'll skip testing the `ifly.complete()` method when not flying since it raises a `RuntimeError` exception if data collection is not in progress):" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "ifly.describe_collect()" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "list(ifly.collect())" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Now, run this fly scan:" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "RE(bluesky.plans.fly([ifly]))" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "db[-1].stream_names" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "db[-1].table(\"ifly\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Still only three `documents` were emitted from the `RunEngine`: `start`, `descriptor`, and `stop`." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "list(db[-1].documents())" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## First working Flyer - trivial data \n", "\n", "To collect data, we need to modify both the `collect()` *and* the `describe_collect()` methods. Bluesky needs to know what kind of data to expect from this Flyer, so that it can generate the correct `descriptor` document.\n", "\n", "For the *most* trivial case, we'll return a single number (`1.2345`) as the result of the first working Flyer.\n", "\n", "In the `describe_collect()` method, we create a dictionary that describes the data to be collected:\n", "\n", " d = dict(\n", " source = \"fictional\",\n", " dtype = \"number\",\n", " shape = []\n", " )\n", " return {\n", " 'ifly': {\n", " \"x\": d\n", " }\n", " }\n", "\n", "Then, in the `collect()` method, add the actual data collection code:\n", "\n", " t = time.time()\n", " d = dict(\n", " time=t,\n", " data=dict(x=1.2345),\n", " timestamps=dict(x=t)\n", " )\n", " yield d\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "class MyFlyer(ophyd.Device):\n", " \"\"\"\n", " build a Flyer that we understand\n", " \"\"\"\n", "\n", " def __init__(self, *args, **kwargs):\n", " super().__init__('', parent=None, **kwargs)\n", " self.complete_status = None\n", "\n", " def my_activity(self):\n", " \"\"\"\n", " start the \"fly scan\" here, could wait for completion\n", " \n", " It's OK to use blocking calls here \n", " since this is called in a separate thread\n", " from the Bluesky RunEngine.\n", " \"\"\"\n", " logger.info(\"activity()\")\n", " if self.complete_status is None:\n", " logger.info(\"leaving activity() - not complete\")\n", " return\n", " \n", " # TODO: do the activity here\n", "\n", " # once started, we notify by updating the status object\n", " self.kickoff_status._finished(success=True)\n", "\n", " # TODO: wait for completion\n", " \n", " # after the wait, we declare victory\n", " self.complete_status._finished(success=True)\n", " logger.info(\"activity() complete. status = \" + str(self.complete_status))\n", "\n", " def kickoff(self):\n", " \"\"\"\n", " Start this Flyer\n", " \"\"\"\n", " logger.info(\"kickoff()\")\n", " self.kickoff_status = ophyd.DeviceStatus(self)\n", " self.complete_status = ophyd.DeviceStatus(self)\n", " \n", " thread = threading.Thread(target=self.my_activity, daemon=True)\n", " thread.start()\n", "\n", " return self.kickoff_status\n", "\n", " def complete(self):\n", " \"\"\"\n", " Wait for flying to be complete\n", " \"\"\"\n", " logger.info(\"complete()\")\n", " if self.complete_status is None:\n", " raise RuntimeError(\"No collection in progress\")\n", "\n", " return self.complete_status\n", "\n", " def describe_collect(self):\n", " \"\"\"\n", " Describe details for ``collect()`` method\n", " \"\"\"\n", " logger.info(\"describe_collect()\")\n", " d = dict(\n", " source = \"fictional\",\n", " dtype = \"number\",\n", " shape = []\n", " )\n", " return {\n", " 'ifly': {\n", " \"x\": d\n", " }\n", " }\n", "\n", " def collect(self):\n", " \"\"\"\n", " Start this Flyer\n", " \"\"\"\n", " logger.info(\"collect()\")\n", " self.complete_status = None\n", " t = time.time()\n", " d = dict(\n", " time=t,\n", " data=dict(x=1.2345),\n", " timestamps=dict(x=t)\n", " )\n", " yield d" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "As before, create a new instance of the *revised* `MyFlyer` class." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "ifly = MyFlyer(name=\"ifly\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "print('output from describe_collect() : ', ifly.describe_collect())\n", "print(\"list output from collect() : \", list(ifly.collect()))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Running this flyer with the RunEngine seems anticlimactic but the lack of exceptions tells us that it ran and we get a UUID at the end." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "RE(bluesky.plans.fly([ifly]))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "We next query the last scan in the databroker and show a table of the stream from `collect()`:" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "h = db[-1]\n", "h.table(h.stream_names[0])" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Now, there is one more document type emitted: `event`. Only one `event` since there is only one data point." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "list(db[-1].documents())" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Flyer that \"collects\" 1-D array data \n", "\n", "We continue to expand the capabilities of our working example of a Flyer. Now, we wish to \"collect\" a 1-D array of data. We'll manufacture that data ourselves to keep the code simple.\n", "\n", "Here, we generate an array of 5 numbers, each representing the elapsed time (in seconds) since the scan started (in `kickoff()`). We record the start time with `self.t0 = time.time()`. Also, `time.time()` is used to generate the timestamps in the data events. In a real case, we *might* get timestamps from the *controller* or we'd have to make it up ourselves, just like this example." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "class MyFlyer(ophyd.Device):\n", " \"\"\"\n", " a Flyer that we understand that reports 1-D array of data\n", " \"\"\"\n", "\n", " def __init__(self, *args, **kwargs):\n", " super().__init__('', parent=None, **kwargs)\n", " self.complete_status = None\n", " self.t0 = 0\n", "\n", " def my_activity(self):\n", " \"\"\"\n", " start the \"fly scan\" here, could wait for completion\n", " \n", " It's OK to use blocking calls here \n", " since this is called in a separate thread\n", " from the Bluesky RunEngine.\n", " \"\"\"\n", " logger.info(\"activity()\")\n", " if self.complete_status is None:\n", " logger.info(\"leaving activity() - not complete\")\n", " return\n", " \n", " # TODO: do the activity here\n", "\n", " # once started, we notify by updating the status object\n", " self.kickoff_status._finished(success=True)\n", "\n", " # TODO: wait for completion\n", " \n", " # after the wait, we declare victory\n", " self.complete_status._finished(success=True)\n", " logger.info(\"activity() complete. status = \" + str(self.complete_status))\n", "\n", " def kickoff(self):\n", " \"\"\"\n", " Start this Flyer\n", " \"\"\"\n", " logger.info(\"kickoff()\")\n", " self.kickoff_status = ophyd.DeviceStatus(self)\n", " self.complete_status = ophyd.DeviceStatus(self)\n", " self.t0 = time.time()\n", " \n", " thread = threading.Thread(target=self.my_activity, daemon=True)\n", " thread.start()\n", "\n", " return self.kickoff_status\n", "\n", " def complete(self):\n", " \"\"\"\n", " Wait for flying to be complete\n", " \"\"\"\n", " logger.info(\"complete()\")\n", " if self.complete_status is None:\n", " raise RuntimeError(\"No collection in progress\")\n", "\n", " return self.complete_status\n", "\n", " def describe_collect(self):\n", " \"\"\"\n", " Describe details for ``collect()`` method\n", " \"\"\"\n", " logger.info(\"describe_collect()\")\n", " d = dict(\n", " source = \"elapsed time, s\",\n", " dtype = \"number\",\n", " shape = (1,)\n", " )\n", " return {\n", " self.name: {\n", " \"x\": d\n", " }\n", " }\n", "\n", " def collect(self):\n", " \"\"\"\n", " Start this Flyer\n", " \"\"\"\n", " logger.info(\"collect()\")\n", " self.complete_status = None\n", " for _ in range(5):\n", " t = time.time()\n", " x = t - self.t0 # data is elapsed time since kickoff()\n", " d = dict(\n", " time=t,\n", " data=dict(x=x),\n", " timestamps=dict(x=t)\n", " )\n", " yield d\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "ifly = MyFlyer(name=\"ifly\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "print(ifly.describe_collect())" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "list(ifly.collect())" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Again, not much information from running this flyer, except that it succeeds and a uuid is returned." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "RE(bluesky.plans.fly([ifly]))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "More information is obtained by asking the databroker about the most recent scan. Here's the data table from our stream of data events. (The stream is named \"ifly\" as set from `self.name` in the `describe_collect()` method.) We'll reference that last scan as `h` to reduce the amount of typing." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "h = db[-1]\n", "h.stream_names[0]" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "h.table(h.stream_names[0])" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "We generated 5 data points, there are 5 `event` documents to match." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "list(db[-1].documents())" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Conclusion\n", "\n", "Now you have seen the basic steps in creating a `Flyer`. You have also been shown a few diagnostic tools to help you investigate your code development." ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.7.7-final" } }, "nbformat": 4, "nbformat_minor": 4 }