-
Notifications
You must be signed in to change notification settings - Fork 76
/
callbacks.py
76 lines (59 loc) · 2.2 KB
/
callbacks.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
import logging
from .utils import (DisconnectedError, TimeoutError)
logger = logging.getLogger(__name__)
class UidPublish:
'''Publishes current run start document UID to a given signal
Processed on every start/end document.
Note: If used with an EpicsSignal, it's recommended to use a waveform in
place of a stringin record on the EPICS side, as the start document UID
will be published both on run start and run completion. A stringin record
will only process on change and monitor events will only be received on run
start.
Sample EPICS record definition:
record(waveform, "$(Sys)$(Dev)UID-I") {
# Using waveform here as it always reprocesses, so you'll get a
# monitor event on start/stop of the run
field(DESC, "Last run UID")
field(FTVL, "STRING")
field(MPST, "Always")
info(autosaveFields_pass0, "VAL")
}
Parameters
----------
uid_signal : Signal
The signal to publish to
raise_if_disconnected : bool, optional
Fail if the UID signal is disconnected
put_kw : kwargs, optional
Keyword arguments to send to uid_signal.put()
'''
def __init__(self, signal, raise_if_disconnected=False, **put_kw):
self._uid = None
self.last_start = None
self.uid_signal = signal
self.put_kw = put_kw
self.raise_if_disconnected = raise_if_disconnected
@property
def uid(self):
'''The uid of the last run'''
return self._uid
@uid.setter
def uid(self, uid):
self._uid = uid
if uid is None:
uid = ''
try:
self.uid_signal.put(uid, **self.put_kw)
except (DisconnectedError, TimeoutError):
logger.error('UID signal disconnected. Is the IOC running?')
if self.raise_if_disconnected:
raise
def clear(self):
'''Clear the run uid'''
self.uid = None
def __call__(self, name, doc):
'''Bluesky callback with document info'''
if name == 'start':
self.last_start = doc
if self.last_start and name in ('start', 'stop'):
self.uid = self.last_start['uid']