Connect Bluesky with EPICS

From APS Python Training for Bluesky Data Acquisition.

Objective:

Connect Bluesky with EPICS Process Variable(s).

Fundamentals

In Bluesky, the ophyd package is used to connect with the underlying control system (EPICS, in this case). The fundamental structure is the EpicsSignal which connects a single EPICS Process Variable (PV) with a single python object. Behind the scenes, the connection is made using the PyEpics package and the EPICS Channel Access protocol. Other information, when available (such as units, limits, displayed precision, data type, enumeration labels), is obtained from EPICS.

ophyd class	description
EpicsSignal	establish a read/write connection with an EPICS PV, monitor and update the object in the background
EpicsSignalRO	read-only version of EpicsSignal

Note that EpicsSignal & EpicsSignalRO have many additional configuration features beyond what is described here. Consult the ophyd documentation or source code for more details.

EpicsSignal

Start by importing the support from ophyd.

from ophyd import EpicsSignal

There is an EPICS PV (gp:gp:bit1) with which we can connect. This PV is a single bit. Connect with it, naming the python object as bit1 (for convenience, we make an ioc prefix variable, just in case someone has an EPICS IOC with a different prefix).

In addition to the EPICS PV, it is required to add a name keyword so the Python object can be constructed with the object name. By convention, use the same name as on the left side of the = sign.

ioc = "gp:"
bit1 = EpicsSignal(f"{ioc}gp:bit1", name="bit1")
print(f"{bit1.connected = }")

bit1.connected = False

Immediately after creating the connection, we tested if the Python object had fully connected with EPICS (usually this takes a short time but it is not instantaneous). As shown, the object has not connected with EPICS yet.

When humans interact with a Jupyter notebook, the EPICS PV connection usually happens within the time it takes to move to the next cell and execute it. But when these PV connections are executed by a single program and the object is needed right away, it may be necessary to wait for the connection to complete before proceeding.

bit1.wait_for_connection()
print(f"{bit1.connected = }")

bit1.connected = True

Print the value of the Python object using its get() method:

print(f"{bit1.get() = }")

bit1.get() = 0

There may be labels associated with the two different values this PV can take. Show them with the object’s enum_strs property:

print(f"{bit1.enum_strs = }")

bit1.enum_strs = ('off', 'on')

We can set this with either numbers (0 or 1) or with text (off or on) and we can show this as either numbers or text.

bit1.put(1)
print(f"{bit1.get() = }")
print(f"{bit1.get(as_string=False) = }")
print(f"{bit1.get(as_string=True) = }")

bit1.put("off")
print(f"{bit1.get() = }")
print(f"{bit1.get(as_string=False) = }")
print(f"{bit1.get(as_string=True) = }")

bit1.get() = 1
bit1.get(as_string=False) = 1
bit1.get(as_string=True) = 'on'
bit1.get() = 0
bit1.get(as_string=False) = 0
bit1.get(as_string=True) = 'off'

We can also make the text representation the default by adding the string=True keyword when we create the connection:

bit1 = EpicsSignal(f"{ioc}gp:bit1", name="bit1", string=True)
bit1.wait_for_connection()
print(f"{bit1.get() = }")

bit1.get() = 'off'

The read() method (used by data acquisition) shows both value and the timestamp that value was received from EPICS. The timestamp is recorded by Python and is absolute number of seconds since January 1, 1970 GMT.

bit1.read()

{'bit1': {'value': 'off', 'timestamp': 1629399463.264405}}

EpicsSignalRO

The gp:UPTIME PV tells us how long the IOC has been running. This PV contains information that we cannot change since it reports a value that only the IOC knows. Create an uptime object with the EpicsSignalRO class from ophyd:

from ophyd import EpicsSignalRO
uptime =  EpicsSignalRO(f"{ioc}UPTIME", name="uptime")
uptime.wait_for_connection()
print(f"{uptime.get() = }")

uptime.get() = '04:47:37'

Device and Component

We may have a group of PVs that are related in some way. It is possible to organize the Python objects into a larger structure called an ophyd.Device where each of the EpicsSignal objects is an ophyd.Component. Import the ophyd structures next:

from ophyd import Component, Device

Build a structure that associates these PVs and (hypothetical) uses:

PV	how it is used	attribute
gp:gp:bit1	on/off control	enable
gp:gp:float1	desired temperature	setpoint
gp:gp:float1.EGU	temperature units	setpoint_units
gp:gp:text1	short label	label

To group these PVs together, a custom Python class must be created using Device as a base class and Component for each of the PVs.

To make this class more useful, we omit the IOC prefix (the first gp:) from our class. We’ll use the IOC prefix later, when we create the Python object for this structure.

class MyGroup(Device):
    enable = Component(EpicsSignal, "gp:bit1")
    setpoint = Component(EpicsSignal, "gp:float1")
    units = Component(EpicsSignal, "gp:float1.EGU")
    label = Component(EpicsSignal, "gp:text1")

Use this structure to connect with all the PVs (at once) using the same type of command as with EpicsSignal above.

thing = MyGroup(ioc, name="thing")
thing.wait_for_connection()

Show all the values at once using the read() method.

thing.read()

OrderedDict([('thing_enable', {'value': 0, 'timestamp': 1629399463.264405}),
             ('thing_setpoint',
              {'value': 521.7482, 'timestamp': 1629399413.336128}),
             ('thing_units',
              {'value': 'Rankine', 'timestamp': 1629399413.336128}),
             ('thing_label', {'value': '', 'timestamp': 631152000.0})])

Set the temperature units and set point:

thing.setpoint.put(521.7482)
thing.units.put("Rankine")
thing.read()

OrderedDict([('thing_enable', {'value': 0, 'timestamp': 1629399463.264405}),
             ('thing_setpoint',
              {'value': 521.7482, 'timestamp': 1629399463.519874}),
             ('thing_units',
              {'value': 'Rankine', 'timestamp': 1629399463.519874}),
             ('thing_label', {'value': '', 'timestamp': 631152000.0})])

print(f"set point: {thing.setpoint.get():.2f} {thing.units.get()}")

set point: 521.75 Rankine

Device structures can be nested to make even more complex structures. For example:

class Stack(Device):
    basic = Component(MyGroup, "")
    reading = Component(EpicsSignal, "gp:float2")
    abstract = Component(EpicsSignal, "gp:longtext2", string=True)

stack = Stack(ioc, name="stack")
stack.wait_for_connection()
stack.read()

OrderedDict([('stack_basic_enable',
              {'value': 0, 'timestamp': 1629399463.264405}),
             ('stack_basic_setpoint',
              {'value': 521.7482, 'timestamp': 1629399463.519874}),
             ('stack_basic_units',
              {'value': 'Rankine', 'timestamp': 1629399463.519874}),
             ('stack_basic_label', {'value': '', 'timestamp': 631152000.0}),
             ('stack_reading', {'value': 0.0, 'timestamp': 631152000.0}),
             ('stack_abstract', {'value': '', 'timestamp': 631152000.0})])