import ophyd
from ophyd import EpicsSignal
# ophyd.set_cl('caproto')
from bluesky import RunEngine
import bluesky.plan_stubs as bps
from bluesky.utils import ts_msg_hook
from bluesky.suspenders import SuspendBoolHigh
from functools import partial
ev = EpicsSignal('simple:A', name='a')
class PauseInterupt(SuspendBoolHigh):
def __call__(self, value, **kwargs):
"""Make the class callable so that we can
pass it off to the ophyd callback stack.
This expects the massive blob that comes from ophyd
"""
with self._lock:
if self._should_suspend(value):
self._tripped = True
loop = self.RE._loop
# this does dirty things with internal state
if (self.RE is not None):
print('entered right condition')
cb = partial(
self.RE.request_pause, defer=False)
if self.RE.state.is_running:
print('about to pause')
loop.call_soon_threadsafe(cb)
self._SuspenderBase__make_event()
elif self._should_resume(value):
self._tripped = False
self._SuspenderBase__set_event()
RE = RunEngine()
RE.msg_hook = ts_msg_hook
def manual_plan():
yield from bps.sleep(.3)
print('go flip the PV manually')
yield from bps.sleep(100)
Demo of how this works: