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.
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.
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 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.)
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).
databroker.temp()
, establishing a temporary, disposable yet fully-functional databroker for just this tutorial session.Links
databroker.temp()
- https://blueskyproject.io/databroker/v1/api.htmlimport logging
import threading
import time
import ophyd
import bluesky
import bluesky.plans
import databroker
logger = logging.getLogger()
RE = bluesky.RunEngine({})
db = databroker.temp()
RE.subscribe(db.insert)
In Bluesky, a Flyer is an ophyd.Device
that meets the Fly-able interface, which has three methods:
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.
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.
This example (which does absolutely nothing) meets the bare minimum requirement.
Links
ophyd.Device
- https://blueskyproject.io/ophyd/device-overview.htmlophyd.status.Status
- https://blueskyproject.io/ophyd/generated/ophyd.status.Status.htmlDeviceStatus()
- https://blueskyproject.io/ophyd/generated/ophyd.status.DeviceStatus.htmlclass BareMinimumFlyer(ophyd.Device):
def kickoff(self):
kickoff_status = ophyd.DeviceStatus(self)
kickoff_status._finished(success=True)
return kickoff_status
def complete(self):
complete_status = ophyd.DeviceStatus(self)
complete_status._finished(success=True)
return complete_status
def collect(self):
yield {'data':{}, 'timestamps':{}, 'time':time.time()}
def describe_collect(self):
return {self.name: {}}
flyer = BareMinimumFlyer(name="flyer")
print(flyer.complete())
print(list(flyer.collect()))
# if this next step succeeds, it's proof that we did this right!
RE(bluesky.plans.fly([flyer]))
Only three documents
were emitted from the RunEngine
: start
, descriptor
, and stop
since we generated no data content.
list(db[-1].documents())
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.
def __init__(self, *args, **kwargs):
super().__init__('', parent=None, **kwargs)
self.kickoff_status = None
self.complete_status = None
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()
:
thread = threading.Thread(target=self.my_activity, daemon=True)
thread.start()
Link
epics.ca.CAThread
- https://github.com/NSLS-II/tutorial-docker/pull/7#discussion_r184905960The basic outline of my_activity()
is:
def my_activity(self):
# tell the *controller* to do its work
# once started, we notify by updating the status object
self.kickoff_status._finished(success=True)
# wait for for it to tell us it is done
# after the wait, we declare victory
self.complete_status._finished(success=True)
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.
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.
We've also added some diagnostic reporting (calls to logger.info(...)
) to build out the next example:
class MyFlyer(ophyd.Device):
"""
starting template for a Flyer that we understand
"""
def __init__(self, *args, **kwargs):
super().__init__('', parent=None, **kwargs)
self.complete_status = None
def my_activity(self):
"""
start the "fly scan" here, could wait for completion
It's OK to use blocking calls here
since this is called in a separate thread
from the Bluesky RunEngine.
"""
logger.info("activity()")
if self.complete_status is None:
logger.info("leaving activity() - not complete")
return
# TODO: do the activity here
# once started, we notify by updating the status object
self.kickoff_status._finished(success=True)
# TODO: wait for completion
# after the wait, we declare victory
self.complete_status._finished(success=True)
logger.info("activity() complete. status = " + str(self.complete_status))
def kickoff(self):
"""
Start this Flyer
"""
logger.info("kickoff()")
self.kickoff_status = ophyd.DeviceStatus(self)
self.complete_status = ophyd.DeviceStatus(self)
thread = threading.Thread(target=self.my_activity, daemon=True)
thread.start()
return self.kickoff_status
def complete(self):
"""
Wait for flying to be complete
"""
logger.info("complete()")
if self.complete_status is None:
raise RuntimeError("No collection in progress")
return self.complete_status
def collect(self):
"""
Retrieve the data
"""
logger.info("collect()")
self.complete_status = None
yield {'data':{}, 'timestamps':{}, 'time':time.time()}
def describe_collect(self):
"""
Describe details for ``collect()`` method
"""
logger.info("describe_collect()")
return {self.name: {}}
ifly = MyFlyer(name="ifly")
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.
create an instance of the Flyer
flyer = MyFlyer(name="flyer")
verify that kickoff()
returns a status that is "Done"
status = flyer.kickoff() status.done
verify that complete()
returns a status that is "Done"
status = flyer.complete() status.done
verify that describe_collect()
returns a dictionary
d = flyer.describe_collect() d
verify that collect()
returns a generator
g = flyer.collect() g
verify that generator is a list of data dictionaries
list(g)
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):
ifly.describe_collect()
list(ifly.collect())
Now, run this fly scan:
RE(bluesky.plans.fly([ifly]))
db[-1].stream_names
db[-1].table("ifly")
Still only three documents
were emitted from the RunEngine
: start
, descriptor
, and stop
.
list(db[-1].documents())
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.
For the most trivial case, we'll return a single number (1.2345
) as the result of the first working Flyer.
In the describe_collect()
method, we create a dictionary that describes the data to be collected:
d = dict(
source = "fictional",
dtype = "number",
shape = []
)
return {
'ifly': {
"x": d
}
}
Then, in the collect()
method, add the actual data collection code:
t = time.time()
d = dict(
time=t,
data=dict(x=1.2345),
timestamps=dict(x=t)
)
yield d
class MyFlyer(ophyd.Device):
"""
build a Flyer that we understand
"""
def __init__(self, *args, **kwargs):
super().__init__('', parent=None, **kwargs)
self.complete_status = None
def my_activity(self):
"""
start the "fly scan" here, could wait for completion
It's OK to use blocking calls here
since this is called in a separate thread
from the Bluesky RunEngine.
"""
logger.info("activity()")
if self.complete_status is None:
logger.info("leaving activity() - not complete")
return
# TODO: do the activity here
# once started, we notify by updating the status object
self.kickoff_status._finished(success=True)
# TODO: wait for completion
# after the wait, we declare victory
self.complete_status._finished(success=True)
logger.info("activity() complete. status = " + str(self.complete_status))
def kickoff(self):
"""
Start this Flyer
"""
logger.info("kickoff()")
self.kickoff_status = ophyd.DeviceStatus(self)
self.complete_status = ophyd.DeviceStatus(self)
thread = threading.Thread(target=self.my_activity, daemon=True)
thread.start()
return self.kickoff_status
def complete(self):
"""
Wait for flying to be complete
"""
logger.info("complete()")
if self.complete_status is None:
raise RuntimeError("No collection in progress")
return self.complete_status
def describe_collect(self):
"""
Describe details for ``collect()`` method
"""
logger.info("describe_collect()")
d = dict(
source = "fictional",
dtype = "number",
shape = []
)
return {
'ifly': {
"x": d
}
}
def collect(self):
"""
Start this Flyer
"""
logger.info("collect()")
self.complete_status = None
t = time.time()
d = dict(
time=t,
data=dict(x=1.2345),
timestamps=dict(x=t)
)
yield d
As before, create a new instance of the revised MyFlyer
class.
ifly = MyFlyer(name="ifly")
print('output from describe_collect() : ', ifly.describe_collect())
print("list output from collect() : ", list(ifly.collect()))
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.
RE(bluesky.plans.fly([ifly]))
We next query the last scan in the databroker and show a table of the stream from collect()
:
h = db[-1]
h.table(h.stream_names[0])
Now, there is one more document type emitted: event
. Only one event
since there is only one data point.
list(db[-1].documents())
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.
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.
class MyFlyer(ophyd.Device):
"""
a Flyer that we understand that reports 1-D array of data
"""
def __init__(self, *args, **kwargs):
super().__init__('', parent=None, **kwargs)
self.complete_status = None
self.t0 = 0
def my_activity(self):
"""
start the "fly scan" here, could wait for completion
It's OK to use blocking calls here
since this is called in a separate thread
from the Bluesky RunEngine.
"""
logger.info("activity()")
if self.complete_status is None:
logger.info("leaving activity() - not complete")
return
# TODO: do the activity here
# once started, we notify by updating the status object
self.kickoff_status._finished(success=True)
# TODO: wait for completion
# after the wait, we declare victory
self.complete_status._finished(success=True)
logger.info("activity() complete. status = " + str(self.complete_status))
def kickoff(self):
"""
Start this Flyer
"""
logger.info("kickoff()")
self.kickoff_status = ophyd.DeviceStatus(self)
self.complete_status = ophyd.DeviceStatus(self)
self.t0 = time.time()
thread = threading.Thread(target=self.my_activity, daemon=True)
thread.start()
return self.kickoff_status
def complete(self):
"""
Wait for flying to be complete
"""
logger.info("complete()")
if self.complete_status is None:
raise RuntimeError("No collection in progress")
return self.complete_status
def describe_collect(self):
"""
Describe details for ``collect()`` method
"""
logger.info("describe_collect()")
d = dict(
source = "elapsed time, s",
dtype = "number",
shape = (1,)
)
return {
self.name: {
"x": d
}
}
def collect(self):
"""
Start this Flyer
"""
logger.info("collect()")
self.complete_status = None
for _ in range(5):
t = time.time()
x = t - self.t0 # data is elapsed time since kickoff()
d = dict(
time=t,
data=dict(x=x),
timestamps=dict(x=t)
)
yield d
ifly = MyFlyer(name="ifly")
print(ifly.describe_collect())
list(ifly.collect())
Again, not much information from running this flyer, except that it succeeds and a uuid is returned.
RE(bluesky.plans.fly([ifly]))
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.
h = db[-1]
h.stream_names[0]
h.table(h.stream_names[0])
We generated 5 data points, there are 5 event
documents to match.
list(db[-1].documents())
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.