Ophyd-async

[TeresiaOlsson]

As discussed on the maintainers' meeting today I will ask around and see if I find someone from the ophyd-async community who are willing to give us a short introduction to the features and how to make devices so we can see what we think.

In the meantime, I have some examples of devices I have written as part of learning how to use this package. I chose some I thought were the most simple ones since I just wanted to test the features.

At the moment I'm trying to figure out how to turn these devices into generic ones where I can have a factory which creates either EPICS or TANGO signals based on the configuration.

Masterclock:

This device implements the status feature when setting the device. @AsyncStatus.wrap returns a Status object which has the attribute success and done. In my tests I added the functionality that I can specify a ramp rate and done will only become True after a time based on that. I also added a tolerance check between the readback and setpoint which will raise an exception and make success False if the tolerance isn't meet.

I think it worked well and was easy to implement after you had understood the documentation because that is quite lacking...

import asyncio
from pydantic import BaseModel
from typing import Optional
from ophyd_async.core import (
    StandardReadable,
    AsyncStatus,
    CALCULATE_TIMEOUT,
    DEFAULT_TIMEOUT,
    CalculatableTimeout,
)

from ophyd_async.epics.core import epics_signal_rw

import logging

for handler in logging.root.handlers[:]:
    logging.root.removeHandler(handler)
logging.basicConfig(level=logging.DEBUG)


class MasterClockConfig(BaseModel):

    name: Optional[str] = 'masterclock'
    setpoint_pv: str
    readback_pv: str
    ramp_rate: float # Unit should be in kHz/s
    tolerance: float # Unit should be in Hz
    # TODO: Units needs to be generalised and handled better


class MasterClock(StandardReadable):

    def __init__(self, config: MasterClockConfig, prefix: Optional[str] = None) -> None:

        # TODO: this needs to be generalised so can also incude ca and pva.
        if prefix:
            self._setpoint_pv = prefix + ':' + config.setpoint_pv
            self._readback_pv = prefix + ':' + config.readback_pv
        else:
            self._setpoint_pv = config.setpoint_pv
            self._readback_pv = config.readback_pv

        self._ramp_rate = config.ramp_rate
        self._tolerance = config.tolerance

        with self.add_children_as_readables():
            self.frequency = epics_signal_rw(float, read_pv= self._readback_pv,
                                             write_pv= self._setpoint_pv, name='frequency')
        super().__init__(name=config.name)

    @property
    def setpoint_pv(self):
        return self._setpoint_pv

    @property
    def readback_pv(self):
        return self._readback_pv

    @property
    def ramp_rate(self):
        return self._ramp_rate

    @property
    def tolerances(self):
        return self._tolerance


    @AsyncStatus.wrap
    async def set(self, new_value: float, timeout: CalculatableTimeout = CALCULATE_TIMEOUT):
        logging.debug(f"set() called with new_value={new_value}")

        old_value = await self.frequency.get_value()
        logging.debug(f"Old value = {old_value}")

        # Calculate the time to wait based on the ramp rate
        change = abs(new_value - old_value)
        wait_time = change / self._ramp_rate
        logging.debug(f"Calculated wait time = {wait_time}")

        # Define the timeout based on the ramp rate
        if timeout == CALCULATE_TIMEOUT:
            timeout = wait_time + DEFAULT_TIMEOUT
            logging.debug(f"Calculated timeout = {timeout}")

        # Set the value
        await self.frequency.set(new_value)
        await asyncio.sleep(wait_time)

        # Compare so readback and setpoint is within some tolerance
        current_values = await self.frequency.locate()
        diff = abs(current_values['readback'] - current_values['setpoint'])

        logging.debug(f"Compare setpoint = {current_values['setpoint']} and readback = {current_values['readback']}")
        logging.debug(f"Absolute difference is = {diff}")

        if diff > self._tolerance:
            raise Exception('The difference between the readback and setpoint is larger than the tolerance.')

Tune:

Our tune PVs are in frequency so I made a tune device to test the way ophyd-async handles conversions and see if I could get it to convert to physics units.

I think it was very easy to use. Only thing is that you need to define a raw and a derived. I'm now thinking about how this device could be made generic so depending on if my tune PV/attribute returns frequency or tune there would be a factory which chooses which signal is raw and which is derived and also allows you to skip having derived signals in case you don't need them.

from pydantic import BaseModel
from typing import Optional, TypedDict
from ophyd_async.core import StandardReadable, derived_signal_r, Transform, DerivedSignalFactory
from ophyd_async.epics.core import epics_signal_r


class Tune1DConfig(BaseModel):
    name: Optional[str] = 'tune'
    read_pv: str
    settling_time: Optional[float] = 0

    
class TuneRaw(TypedDict):
    frequency: float

class TuneDerived(TypedDict):
    tune: float
              
class TuneTransform(Transform):
    
    revolution_frequency: float
    
    def derived_to_raw(self, *, tune: float) -> TypedDict:
        return TuneRaw(
            frequency = tune*self.revolution_frequency*1e-3
            )
    
    def raw_to_derived(self, *, frequency: float) -> TypedDict:
        return TuneDerived(
            tune = frequency*1e3/self.revolution_frequency
            )    


class Tune1D(StandardReadable):
    
    def __init__(self, config: Tune1DConfig, prefix: Optional[str] = None) -> None:
        
        self._settling_time = config.settling_time
        
        self._read_pv = f"{prefix}:{config.read_pv}" if prefix else config.read_pv
        
        revolution_frequency = 6.25e6 # This should not be hardcoded but read from somewhere else.
        
        frequency = epics_signal_r(float, self._read_pv, name='frequency')
                      
        self._factory = DerivedSignalFactory(
                TuneTransform,
                frequency=frequency,
                revolution_frequency=revolution_frequency,
            )
              
        tune = self._factory.derived_signal_r(float,name='tune')
        
        with self.add_children_as_readables():
            self.frequency = frequency
            self.tune = tune
            
        super().__init__(name=config.name)

Ophyd-async also has the feature that you can add a format specifier to signals to specify under which conditions they should be read. For example, if you have signals which are purely for configuration and you want them to only be read once and not for every step during a measurement you could add a format specifier and they will be read by read_configuration and not read().

Here is an example for how I use these devices directly without needing the runengine. I will need an event loop and await values since they are doing asynchronous calls. However, some IDEs and jupyter notebooks already provide an event loop so if you run it there you don't need to create it. But you still need to do await. No way around that without some wrapper.

async def main():
    await masterclock.connect()

    current_values = await masterclock.frequency.locate()
    masterclock_setpoint = current_values['setpoint']

    status = masterclock.set(masterclock_setpoint+2)

    print(status.done)
    print(status.success)

    try:
        await status  # waits for completion; exception will propagate if you want it
    except Exception as e:
        print("Exception caught while awaiting status:", e)

    print(status.done)
    print(status.success)

asyncio.run(main())

[TeresiaOlsson]

@Sulimankhail Maybe you can add your examples too? I think your examples where using the declarative way of defining devices which is an interesting option since it might simplify configuration if we did something similar for the pyAML devices.

Here is a comparison of the declarative vs procedural approach:
https://blueskyproject.io/ophyd-async/main/explanations/declarative-vs-procedural.html

[Sulimankhail]

Instead of copying and pasting the code, I will put the link to my code:
Tango implementation is in tango/main branch
https://github.com/python-accelerator-middle-layer/accml/blob/tango/main/examples/tune/tune_response_measurement.py
Epics implementation of ophyd-async is under epic/main branch
https://github.com/python-accelerator-middle-layer/accml/blob/epics/main/examples/tune/tune_response_measurement.py

[TeresiaOlsson]

In case you want to see more examples there are some in the ophyd-async repository. For beamline elements but still.

TANGO: https://github.com/bluesky/ophyd-async/tree/main/src%2Fophyd_async%2Ftango%2Fdemo

EPICS: https://github.com/bluesky/ophyd-async/tree/main/src%2Fophyd_async%2Fepics%2Fdemo

As I understand it ophyd-async uses this for the signals which can include the TANGO host https://tango-controls.readthedocs.io/projects/rfc/en/latest/16/TangoResourceLocator.html#tango-resource-locator-trl-specification and that's why they don't need a separate step to initialise the control system. Same for the EPICS prefix.

[JeanLucPons]

To avoid await you need to use gather() as in this ophyd example

old_position, units, precision, velocity = await asyncio.gather(
            self.setpoint.get_value(),
            self.units.get_value(),
            self.precision.get_value(),
            self.velocity.get_value(),
        )

I'm wondering if there are some array (list of signal) definitions in ophyd asynch that can gather a list of ..._signal_r ?

[TeresiaOlsson]

You mean like the DeviceVector?

https://blueskyproject.io/ophyd-async/v0.2.0/user/generated/ophyd_async.core.DeviceVector.html

You can also do read if you want to read all attributes on a device. As I understand it, it would read all children of subdevices too in case you have nested devices.

[JeanLucPons]

Ok
So let's continue the discussion here.

• How i can choose the backend I want ? I don't want to be based on tango asyncio (used in ophyd async) which is too slow.
• Do you think that end users will be happy to play with await, futures and all associated async programming stuff ?
• I don't see any Device here (EpicsDevice, TangoDevice??) , what do you call Device level ?

[TeresiaOlsson]

You want synchronous calls? I thought you also wanted asynchronous calls for TANGO? Or you somehow want the calls to be asynchronous in the backend and synchronous in pyAML?

Yes. I mean classic python calls that does not requires an asyncio even loop. And asynchronous gevent like calls. Not asyncio which is dedicated to full asynch programing. For Tango we have a fast c++ event loop (from omniORB) available from pytango that we would like to use.

Okay. That sounds similar to what MML currently is doing. At least at MAX IV they have asynchronous calls that the users don't seem to notice. Then we somehow need to implement that for EPICS too or it will not be possible to write applications that behave in the same way for the different control systems. That is what I meant by "wrap it". I have seen that pytango has this functionality and I thought there then perhaps could be a way to implement the same for EPICS by copying how it's done there.

[TeresiaOlsson]

In pyaml an Element is just a base class that is instantiated dynamically. Nothing prevents to create your own Element that overrides what you want.

But how do you then write applications that can be shared between labs?

[JeanLucPons]

But how do you then write applications that can be shared between labs?

Simply by following pyaml standards. If you want extra features, of course, it will not be portable.
For instance if you want to use bluesky, then it won't be usable at ESRF.

[TeresiaOlsson]

But how do you then write applications that can be shared between labs?

Simply by following pyaml standards. If you want extra features, of course, it will not be portable. For instance if you want to use bluesky, then it won't be usable at ESRF.

Okay, so pyAML will have protocols?

[JeanLucPons]

I would rather call that interface.
But yes and our work should consists in defining interfaces. There are lots of work here.

For instance for your tune object, you should follow the simple interface defined by @gubaidulinvadim.

@property
def tune(self) -> ReadFloatArray:

that return a vector of [qx,qy]
That was already discussed and agreed that we want a betatron tune (transverse only).
For synchrotron tune, an other device should be defined.

If you want to extend the interface to:

@property
def tune(self) -> ReadFloatArray:
@property
def tune_x(self) -> ReadFloatScalar:
@property
def tune_y(self) -> ReadFloatScalar:

This could be discussed and would be example of a pyaml constructive discussion.

[gubaidulinvadim]

It would be good to list here what functionality of ophyd-async devices people want to use. This will be a good first step for a discussion.

[TeresiaOlsson]

Okay. Here is the list of the things I like:

• The architecture using signal backends since it makes a clear interface for how to expand to other control systems and for new developers to easily understand at what level control system dependent code should be. I think it would be nice to also implement different simulators as signal backends because then all our backends would be handled the same. Maybe based on the already existing soft signal backend: https://blueskyproject.io/ophyd-async/v0.5.0/generated/ophyd_async.core.SoftSignalBackend.html

• The possibility to make nested devices and read all levels in one go. This allows to make complex devices by building them up using subdevices. We could for example use it for our families/arrays and in that way read signal, device and family level in the same way.

• The Status object so we can control execution of applications based on if a command was finished and successful and not have to have any sleeps in application scripts.

• StandardReadableFormat which allows you to use format specifiers to say that some signals are for configuration only and are only read when doing read_configuration and not read etc.

• Possibility to add stage and unstage functions for devices that need some special setup when switching applications.

• Derived signals and transforms for our conversions. More information: https://blueskyproject.io/ophyd-async/main/how-to/derive-one-signal-from-others.html

• Saving and loading settings of devices for our save-restore needs. More details: https://blueskyproject.io/ophyd-async/main/how-to/store-and-retrieve.html

There are likely more features that would be useful for us. I'm still exploring the package.