Ophyd.v2: async get_value() calls don't work with ipython await syntax

Overview

When working with an ipython terminal, testing connections to PVs with Ophyd devices fails when using the await syntax. This is because the await keyword tries to run the coroutine in ipython's native event loop. This can be fixed by making await map any coroutines to the bluesky event loop instead.

In general it could be useful to have this patched so anyone wanting to poke their ophyd objects don't have to write bluesky plans to check their PVs and read values from them. But am happy to have my mind changed on this if people think users shouldn't be interacting with devices this way.

Steps to Reproduce the Issue

You will need a virtualenv or container with aioca, ipython, bluesky and ophyd installed. Additionally, access to some Motor PV for this particular example. Then, the following can be run in an ipython terminal:

In [1]: from bluesky import RunEngine
   ...: from ophyd.v2.core import DeviceCollector
   ...: from ophyd_epics_devices.motor import Motor

In [2]: RE = RunEngine()

In [3]: with DeviceCollector():
   ...:     rot_motor = Motor("BL46P-MO-MAP-01:STAGE:A")
   ...: 

In [4]: await rot_motor.acceleration.get_value()
---------------------------------------------------------------------------
CancelledError                            Traceback (most recent call last)
File ./venv/lib/python3.9/site-packages/aioca/_catools.py:685, in caget(pv, datatype, format, count)
    684 channel = get_channel(pv)
--> 685 await channel.wait()
    687 # A count of zero will be treated by EPICS in a version dependent manner,
    688 # either returning the entire waveform (equivalent to count=-1) or a data
    689 # dependent waveform length.

File ./venv/lib/python3.9/site-packages/aioca/_catools.py:247, in Channel.wait(self)
    246 """Waits for the channel to become connected if not already connected."""
--> 247 await self.__connect_event.wait()

File ./venv/lib/python3.9/site-packages/aioca/_catools.py:53, in ValueEvent.wait(self)
     52 if not self._event.is_set():
---> 53     await self._event.wait()
     54 if isinstance(self.value, Exception):

File /dls_sw/apps/python/anaconda/4.6.14/64/envs/python3.9/lib/python3.9/asyncio/locks.py:226, in Event.wait(self)
    225 try:
--> 226     await fut
    227     return True

CancelledError: 

During handling of the above exception, another exception occurred:

CancelledError                            Traceback (most recent call last)
File /dls_sw/apps/python/anaconda/4.6.14/64/envs/python3.9/lib/python3.9/asyncio/tasks.py:489, in wait_for(fut, timeout, loop)
    488 try:
--> 489     fut.result()
    490 except exceptions.CancelledError as exc:

CancelledError: 

The above exception was the direct cause of the following exception:

TimeoutError                              Traceback (most recent call last)
File ./venv/lib/python3.9/site-packages/aioca/_catools.py:141, in ca_timeout(awaitable, name, timeout)
    140 try:
--> 141     result = await asyncio.wait_for(awaitable, timeout)
    142 except asyncio.TimeoutError as e:

File /dls_sw/apps/python/anaconda/4.6.14/64/envs/python3.9/lib/python3.9/asyncio/tasks.py:491, in wait_for(fut, timeout, loop)
    490 except exceptions.CancelledError as exc:
--> 491     raise exceptions.TimeoutError() from exc
    492 else:

TimeoutError: 

The above exception was the direct cause of the following exception:

CANothing                                 Traceback (most recent call last)
Cell In [4], line 1
----> 1 await rot_motor.acceleration.get_value()

File ./venv/lib/python3.9/site-packages/ophyd/v2/epics.py:105, in EpicsSignalR.get_value(self, cached)
    103     return self._value
    104 else:
--> 105     return await self.read_channel.get_value()

File ./venv/lib/python3.9/site-packages/ophyd/v2/_channelca.py:160, in ChannelCa.get_value(self)
    158 async def get_value(self) -> T:
    159     assert self.ca_datatype is not None, f"{self.source} not connected yet"
--> 160     value = await caget(self.pv, datatype=self.ca_datatype)
    161     return self._converter.from_ca(value)

File ./venv/lib/python3.9/site-packages/aioca/_catools.py:113, in maybe_throw.<locals>.call_wrapper(*args, **kwargs)
    111 @functools.wraps(async_function)
    112 async def call_wrapper(*args, **kwargs):
--> 113     return await throw_wrapper(*args, **kwargs)

File ./venv/lib/python3.9/site-packages/aioca/_catools.py:95, in maybe_throw.<locals>.throw_wrapper(pv, timeout, throw, *args, **kwargs)
     93 awaitable = ca_timeout(async_function(pv, *args, **kwargs), pv, timeout)
     94 if throw:
---> 95     return await awaitable
     96 else:
     97     # We catch all the expected exceptions, converting them into
     98     # CANothing() objects as appropriate.  Any unexpected exceptions
     99     # will be raised anyway, which seems fair enough!
    100     try:

File ./venv/lib/python3.9/site-packages/aioca/_catools.py:143, in ca_timeout(awaitable, name, timeout)
    141         result = await asyncio.wait_for(awaitable, timeout)
    142     except asyncio.TimeoutError as e:
--> 143         raise CANothing(name, cadef.ECA_TIMEOUT) from e
    144 else:
    145     result = await awaitable

CANothing: BL46P-MO-MAP-01:STAGE:A.ACCL: User specified timeout on IO operation expired

How to fix the issue

One solution, is to just use bluesky plans, i.e.:

In [1]: from bluesky import RunEngine
   ...: from ophyd.v2.core import DeviceCollector
   ...: from ophyd_epics_devices.motor import Motor

In [2]: import bluesky.plan_stubs as bps

In [3]: RE = RunEngine()

In [4]: with DeviceCollector():
   ...:     rot_motor = Motor("BL46P-MO-MAP-01:STAGE:A")
   ...: 

In [5]: def plan():
   ...:     accel = yield from bps.rd(rot_motor.acceleration)
   ...:     print(accel)
   ...: 

In [6]: RE(plan(), lambda name, doc: print())
0.1
Out[6]: ()

Another, is to map the await functionality of ipython onto the call_in_bluesky_event_loop function:

In [1]: from bluesky.run_engine import call_in_bluesky_event_loop
   ...: from IPython import get_ipython
   ...: 
   ...: get_ipython().run_line_magic("autoawait", "call_in_bluesky_event_loop")

In [2]: from bluesky import RunEngine
   ...: from ophyd.v2.core import DeviceCollector
   ...: from ophyd_epics_devices.motor import Motor

In [3]: RE = RunEngine()

In [4]: with DeviceCollector():
   ...:     rot_motor = Motor("BL46P-MO-MAP-01:STAGE:A")
   ...: 

In [5]: await rot_motor.acceleration.get_value()
Out[5]: 0.1

credit to @coretl for discovering the second solution. I don't know which is best, but thought for any future users this would be a good note to have.

bluesky / ophyd