-
Notifications
You must be signed in to change notification settings - Fork 76
/
device.py
764 lines (598 loc) · 26.1 KB
/
device.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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
import time as ttime
import logging
from collections import (OrderedDict, namedtuple)
from .ophydobj import (OphydObject, DeviceStatus)
from .utils import (TimeoutError, ExceptionBundle, set_and_wait)
logger = logging.getLogger(__name__)
class Component:
'''A descriptor representing a device component (or signal)
Unrecognized keyword arguments will be passed directly to the component
class initializer.
Parameters
----------
cls : class
Class of signal to create. The required signature of
`cls.__init__` is (if `suffix` is given)::
def __init__(self, pv_name, parent=None, **kwargs):
or (if suffix is None) ::
def __init__(self, parent=None, **kwargs):
The class may have a `wait_for_connection()` which is called
during the component instance creation.
suffix : str, optional
The PV suffix, which gets appended onto the device prefix to
generate the final PV that the instance component will bind to.
lazy : bool, optional
Lazily instantiate the signal. If False, the signal will be
instantiated upon component instantiation
trigger_value : any, optional
Mark as a signal to be set on trigger. The value is sent to the signal
at trigger time.
add_prefix : sequence, optional
Keys in the kwargs to prefix with the Device PV prefix during
creation of the component instance.
Defaults to ('suffix', 'write_pv', )
doc : str, optional
string to attach to component DvcClass.component.__doc__
'''
def __init__(self, cls, suffix=None, *, lazy=False, trigger_value=None,
add_prefix=None, doc=None, **kwargs):
self.attr = None # attr is set later by the device when known
self.cls = cls
self.kwargs = kwargs
self.lazy = lazy
self.suffix = suffix
self.doc = doc
self.trigger_value = trigger_value # TODO discuss
if add_prefix is None:
add_prefix = ('suffix', 'write_pv')
self.add_prefix = tuple(add_prefix)
def maybe_add_prefix(self, instance, kw, suffix):
"""Add prefix to a suffix if kw is in self.add_prefix
Parameters
----------
instance : Device
The instance to extract the prefix to maybe append to the
suffix from.
kw : str
The key of associated with the suffix. If this key is
self.add_prefix than prepend the prefix to the suffix and
return, else just return the suffix.
suffix : str
The suffix to maybe have something prepended to.
Returns
-------
str
"""
if kw in self.add_prefix:
return '{prefix}{suffix}'.format(prefix=instance.prefix,
suffix=suffix)
return suffix
def create_component(self, instance):
'''Create a component for the instance'''
kwargs = self.kwargs.copy()
kwargs['name'] = '{}_{}'.format(instance.name, self.attr)
for kw, val in list(kwargs.items()):
kwargs[kw] = self.maybe_add_prefix(instance, kw, val)
if self.suffix is not None:
pv_name = self.maybe_add_prefix(instance, 'suffix', self.suffix)
cpt_inst = self.cls(pv_name, parent=instance, **kwargs)
else:
cpt_inst = self.cls(parent=instance, **kwargs)
if self.lazy and hasattr(self.cls, 'wait_for_connection'):
cpt_inst.wait_for_connection()
return cpt_inst
def make_docstring(self, parent_class):
if self.doc is not None:
return self.doc
return '{} component with suffix {}'.format(self.attr, self.suffix)
def __get__(self, instance, owner):
if instance is None:
return self
if self.attr not in instance._signals:
instance._signals[self.attr] = self.create_component(instance)
return instance._signals[self.attr]
def __set__(self, instance, owner):
raise RuntimeError('Use .put()')
class FormattedComponent(Component):
'''A Component which takes a dynamic format string
This differs from Component in that the parent prefix is not automatically
added onto the Component suffix. Additionally, `str.format()` style strings
are accepted, allowing access to Device instance attributes:
>>> from ophyd import (Component as C, FormattedComponent as FC)
>>> class MyDevice(Device):
... # A normal component, where 'suffix' is added to prefix verbatim
... cpt = C(EpicsSignal, 'suffix')
... # A formatted component, where 'self' refers to the Device instance
... ch = FC(EpicsSignal, '{self.prefix}{self._ch_name}')
...
... def __init__(self, prefix, ch_name=None, **kwargs):
... self._ch_name = ch_name
... super().__init__(prefix, **kwargs)
>>> dev = MyDevice('prefix:', ch_name='some_channel', name='dev')
>>> print(dev.cpt.pvname)
prefix:suffix
>>> print(dev.ch.pvname)
prefix:some_channel
For additional documentation, refer to Component.
'''
def maybe_add_prefix(self, instance, kw, suffix):
if kw not in self.add_prefix:
return suffix
return suffix.format(self=instance)
class DynamicDeviceComponent:
'''An Device component that dynamically creates a OphyDevice
Parameters
----------
defn : OrderedDict
The definition of all attributes to be created, in the form of:
defn['attribute_name'] = (SignalClass, pv_suffix, keyword_arg_dict)
This will create an attribute on the sub-device of type `SignalClass`,
with a suffix of pv_suffix, which looks something like this:
parent.attribute_name = SignalClass(pv_suffix, **keyword_arg_dict)
Keep in mind that this is actually done in the metaclass creation, and
not exactly as written above.
clsname : str, optional
The name of the class to be generated
This defaults to {parent_name}{this_attribute_name.capitalize()}
doc : str, optional
The docstring to put on the dynamically generated class
'''
def __init__(self, defn, *, clsname=None, doc=None):
self.defn = defn
self.clsname = clsname
self.attr = None # attr is set later by the device when known
self.lazy = False
self.doc = doc
# TODO: component compatibility
self.trigger_value = None
self.attrs = list(defn.keys())
def make_docstring(self, parent_class):
if self.doc is not None:
return self.doc
return '{} dynamicdevice containing {}'.format(self.attr,
self.attrs)
def create_attr(self, attr_name):
cls, suffix, kwargs = self.defn[attr_name]
inst = Component(cls, suffix, **kwargs)
inst.attr = attr_name
return inst
def create_component(self, instance):
'''Create a component for the instance'''
clsname = self.clsname
if clsname is None:
# make up a class name based on the instance's class name
clsname = ''.join((instance.__class__.__name__,
self.attr.capitalize()))
# TODO: and if the attribute has any underscores, convert that to
# camelcase
docstring = self.doc
if docstring is None:
docstring = '{} sub-device'.format(clsname)
clsdict = OrderedDict(__doc__=docstring)
for attr in self.defn.keys():
clsdict[attr] = self.create_attr(attr)
attrs = set(self.defn.keys())
inst_read = set(instance.read_attrs)
if self.attr in inst_read:
# if the sub-device is in the read list, then add all attrs
read_attrs = attrs
else:
# otherwise, only add the attributes that exist in the sub-device
# to the read_attrs list
read_attrs = inst_read.intersection(attrs)
cls = type(clsname, (Device, ), clsdict)
return cls(instance.prefix, read_attrs=list(read_attrs),
name='{}_{}'.format(instance.name, self.attr),
parent=instance)
def __get__(self, instance, owner):
if instance is None:
return self
if self.attr not in instance._signals:
instance._signals[self.attr] = self.create_component(instance)
return instance._signals[self.attr]
def __set__(self, instance, owner):
raise RuntimeError('Use .put()')
class ComponentMeta(type):
'''Creates attributes for Components by inspecting class definition'''
@classmethod
def __prepare__(self, name, bases):
'''Prepare allows the class attribute dictionary to be ordered as
defined by the user'''
return OrderedDict()
def __new__(cls, name, bases, clsdict):
clsobj = super().__new__(cls, name, bases, clsdict)
RESERVED_ATTRS = ['name', 'parent', 'signal_names', '_signals',
'read_attrs', 'configuration_attrs', 'monitor_attrs',
'_sig_attrs', '_sub_devices']
for attr in RESERVED_ATTRS:
if attr in clsdict:
raise TypeError("The attribute name %r is reserved for "
"use by the Device class. Choose a different "
"name." % attr)
clsobj._sig_attrs = OrderedDict()
for base in reversed(bases):
if not hasattr(base, '_sig_attrs'):
continue
for attr, cpt in base._sig_attrs.items():
clsobj._sig_attrs[attr] = cpt
# map component classes to their attribute names from this class
for attr, value in clsdict.items():
if isinstance(value, (Component, DynamicDeviceComponent)):
clsobj._sig_attrs[attr] = value
for cpt_attr, cpt in clsobj._sig_attrs.items():
# Notify the component of their attribute name
cpt.attr = cpt_attr
# List Signal attribute names.
clsobj.signal_names = list(clsobj._sig_attrs.keys())
# The namedtuple associated with the device
clsobj._device_tuple = namedtuple(name + 'Tuple', clsobj.signal_names,
rename=True)
# Finally, create all the component docstrings
for cpt in clsobj._sig_attrs.values():
cpt.__doc__ = cpt.make_docstring(clsobj)
# List the attributes that are Devices (not Signals).
# This list is used by stage/unstage. Only Devices need to be staged.
clsobj._sub_devices = []
for attr, cpt in clsobj._sig_attrs.items():
if isinstance(cpt, Component) and not issubclass(cpt.cls, Device):
continue
clsobj._sub_devices.append(attr)
return clsobj
# These stub 'Interface' classes are the apex of the mro heirarchy for
# their respective methods. They make multiple interitance more
# forgiving, and let us define classes that customize these methods
# but are not full Devices.
class BlueskyInterface:
"""Classes that inherit from this can safely customize the
these methods without breaking mro."""
def __init__(self, *args, **kwargs):
# Subclasses can populate this with (signal, value) pairs, to be
# set by stage() and restored back by unstage().
self._staged = False
self.stage_sigs = OrderedDict()
self._original_vals = OrderedDict()
super().__init__(*args, **kwargs)
def trigger(self):
pass
def read(self):
return {}
def describe(self):
return {}
def stage(self):
"Prepare the device to be triggered."
if self._staged:
raise RuntimeError("Device is already stage. Unstage it first.")
self._staged = True
logger.debug("Staging %s", self.name)
# Read current values, to be restored by unstage()
original_vals = {sig: sig.get() for sig, _ in self.stage_sigs.items()}
# We will add signals and values from original_vals to
# self._original_vals one at a time so that
# we can undo our partial work in the event of an error.
# Apply settings.
try:
for sig, val in self.stage_sigs.items():
set_and_wait(sig, val)
# It worked -- now add it to this list of sigs to unstage.
self._original_vals[sig] = original_vals[sig]
# Call stage() on child devices.
for attr in self._sub_devices:
device = getattr(self, attr)
if hasattr(device, 'stage'):
device.stage()
except Exception:
logger.debug("An exception was raised while staging %s or "
"one of its children. Attempting to restore "
"original settings before re-raising the "
"exception.", self.name)
self.unstage()
raise
def unstage(self):
"""
Restore the device to 'standby'.
Multiple calls (without a new call to 'stage') have no effect.
"""
self._staged = False
logger.debug("Unstaging %s", self.name)
# Call unstage() on child devices.
for attr in self._sub_devices[::-1]:
device = getattr(self, attr)
if hasattr(device, 'unstage'):
device.unstage()
# Restore original values.
for sig, val in reversed(list(self._original_vals.items())):
set_and_wait(sig, val)
self._original_vals.pop(sig)
class GenerateDatumInterface:
"""Classes that inherit from this can safely customize the
`generate_datum` method without breaking mro. If used along with the
BlueskyInterface, inherit from this second."""
def generate_datum(self, key, timestamp):
pass
class Device(BlueskyInterface, OphydObject, metaclass=ComponentMeta):
"""Base class for device objects
This class provides attribute access to one or more Signals, which can be
a mixture of read-only and writable. All must share the same base_name.
Parameters
----------
prefix : str
The PV prefix for all components of the device
read_attrs : sequence of attribute names
the components to include in a normal reading (i.e., in ``read()``)
configuration_attrs : sequence of attribute names
the components to be read less often (i.e., in
``read_configuration()``) and to adjust via ``configure()``
name : str, optional
The name of the device
parent : instance or None
The instance of the parent device, if applicable
"""
SUB_ACQ_DONE = 'acq_done' # requested acquire
def __init__(self, prefix, *, read_attrs=None, configuration_attrs=None,
monitor_attrs=None, name=None, parent=None,
**kwargs):
# Store EpicsSignal objects (only created once they are accessed)
self._signals = {}
self.prefix = prefix
if self.signal_names and prefix is None:
raise ValueError('Must specify prefix if device signals are being '
'used')
if name is None:
name = prefix
super().__init__(name=name, parent=parent, **kwargs)
if read_attrs is None:
read_attrs = self.signal_names
if configuration_attrs is None:
configuration_attrs = []
if monitor_attrs is None:
monitor_attrs = []
self.read_attrs = list(read_attrs)
self.configuration_attrs = list(configuration_attrs)
self.monitor_attrs = list(monitor_attrs)
# Instantiate non-lazy signals
[getattr(self, attr) for attr, cpt in self._sig_attrs.items()
if not cpt.lazy]
def wait_for_connection(self, all_signals=False, timeout=2.0):
'''Wait for signals to connect
Parameters
----------
all_signals : bool, optional
Wait for all signals to connect (including lazy ones)
timeout : float or None
Overall timeout
'''
names = [attr for attr, cpt in self._sig_attrs.items()
if not cpt.lazy or all_signals]
# Instantiate first to kickoff connection process
signals = [getattr(self, name) for name in names]
t0 = ttime.time()
while timeout is None or (ttime.time() - t0) < timeout:
connected = [sig.connected for sig in signals]
if all(connected):
return
ttime.sleep(min((0.05, timeout / 10.0)))
unconnected = ', '.join(self._get_unconnected())
raise TimeoutError('Failed to connect to all signals: {}'
''.format(unconnected))
def _get_unconnected(self):
'''Yields all of the signal pvnames or prefixes that are unconnected
This recurses throughout the device hierarchy, only checking signals
that have already been instantiated.
'''
for attr, sig in self.get_instantiated_signals():
if sig.connected:
continue
if hasattr(sig, 'pvname'):
prefix = sig.pvname
else:
prefix = sig.prefix
yield '{} ({})'.format(attr, prefix)
def get_instantiated_signals(self, *, attr_prefix=None):
'''Yields all of the instantiated signals in a device hierarchy
Parameters
----------
attr_prefix : string, optional
The attribute prefix. If None, defaults to self.name
Yields
------
(fully_qualified_attribute_name, signal_instance)
'''
if attr_prefix is None:
attr_prefix = self.name
for attr, sig in self._signals.items():
# fully qualified attribute name from top-level device
full_attr = '{}.{}'.format(attr_prefix, attr)
if isinstance(sig, Device):
yield from sig.get_instantiated_signals(attr_prefix=full_attr)
else:
yield full_attr, sig
@property
def connected(self):
return all(signal.connected for name, signal in self._signals.items())
def __getattr__(self, name):
'''Get a component from a fully-qualified name
As a reminder, __getattr__ is only called if a real attribute doesn't
already exist, or a device component has yet to be instantiated.
'''
if '.' not in name:
try:
# Initial access of signal
cpt = self._sig_attrs[name]
return cpt.__get__(self, None)
except KeyError:
raise AttributeError(name)
attr_names = name.split('.')
try:
attr = getattr(self, attr_names[0])
except AttributeError:
raise AttributeError('{} of {}'.format(attr_names[0], name))
if len(attr_names) > 1:
sub_attr_names = '.'.join(attr_names[1:])
return getattr(attr, sub_attr_names)
return attr
def _read_attr_list(self, attr_list, *, config=False):
'''Get a 'read' dictionary containing attributes in attr_list'''
values = OrderedDict()
for attr in attr_list:
obj = getattr(self, attr)
if config:
values.update(obj.read_configuration())
values.update(obj.read())
return values
def read(self):
"""returns dictionary mapping names to (value, timestamp) pairs
To control which fields are included, adjust the ``read_attrs`` list.
"""
res = super().read()
res.update(self._read_attr_list(self.read_attrs))
return res
def read_configuration(self):
"""
returns dictionary mapping names to (value, timestamp) pairs
To control which fields are included, adjust the
``configuration_attrs`` list.
"""
return self._read_attr_list(self.configuration_attrs, config=True)
def _describe_attr_list(self, attr_list, *, config=False):
'''Get a 'describe' dictionary containing attributes in attr_list'''
desc = OrderedDict()
for attr in attr_list:
obj = getattr(self, attr)
if config:
desc.update(obj.describe_configuration())
desc.update(obj.describe())
return desc
def describe(self):
'''describe the read data keys' data types and other metadata'''
res = super().describe()
res.update(self._describe_attr_list(self.read_attrs))
return res
def describe_configuration(self):
'''describe the configuration data keys' data types/other metadata'''
return self._describe_attr_list(self.configuration_attrs, config=True)
@property
def trigger_signals(self):
names = [attr for attr, cpt in self._sig_attrs.items()
if cpt.trigger_value is not None]
return [getattr(self, name) for name in names]
def _done_acquiring(self, **kwargs):
'''Call when acquisition has completed.'''
self._run_subs(sub_type=self.SUB_ACQ_DONE,
success=True, **kwargs)
self._reset_sub(self.SUB_ACQ_DONE)
def trigger(self):
"""Start acquisition"""
signals = self.trigger_signals
if len(signals) > 1:
raise NotImplementedError('More than one trigger signal is not '
'currently supported')
status = DeviceStatus(self)
if not signals:
status._finished()
return status
acq_signal, = signals
self.subscribe(status._finished,
event_type=self.SUB_ACQ_DONE, run=False)
def done_acquisition(**ignored_kwargs):
# Keyword arguments are ignored here from the EpicsSignal
# subscription, as the important part is that the put completion
# has finished
self._done_acquiring()
acq_signal.put(1, wait=False, callback=done_acquisition)
return status
def stop(self):
'''Stop the Device and all (instantiated) subdevices'''
exc_list = []
for attr in self._sub_devices:
dev = getattr(self, attr)
if not dev.connected:
logger.debug('stop: device %s (%s) is not connected; '
'skipping', attr, dev)
continue
try:
dev.stop()
except ExceptionBundle as ex:
exc_list.extend([('{}.{}'.format(attr, sub_attr), ex)
for sub_attr, ex in ex.exceptions.items()])
except Exception as ex:
exc_list.append((attr, ex))
logger.error('Device %s (%s) stop failed', attr, dev,
exc_info=ex)
if exc_list:
exc_info = '\n'.join('{} raised {!r}'.format(attr, ex)
for attr, ex in exc_list)
raise ExceptionBundle('{} exception(s) were raised during stop: \n'
'{}'.format(len(exc_list), exc_info),
exceptions=dict(exc_list))
def get(self, **kwargs):
'''Get the value of all components in the device
Keyword arguments are passed onto each signal.get()
'''
values = {}
for attr in self.signal_names:
signal = getattr(self, attr)
values[attr] = signal.get(**kwargs)
return self._device_tuple(**values)
def put(self, dev_t, **kwargs):
'''Put a value to all components of the device
Keyword arguments are passed onto each signal.put()
Parameters
----------
dev_t : DeviceTuple or tuple
The device tuple with the value(s) to put (see get_device_tuple)
'''
if not isinstance(dev_t, self._device_tuple):
try:
dev_t = self._device_tuple(dev_t)
except TypeError as ex:
raise ValueError('{}\n\tDevice tuple fields: {}'
''.format(ex, self._device_tuple._fields))
for attr in self.signal_names:
value = getattr(dev_t, attr)
signal = getattr(self, attr)
signal.put(value, **kwargs)
@classmethod
def get_device_tuple(cls):
'''The device tuple type associated with an Device class
This is a tuple representing the full state of all components and
dynamic device sub-components.
'''
return cls._device_tuple
@property
def report(self):
# TODO
return {}
def configure(self, d):
'''Configure the device for something during a run
This default implementation allows the user to change any of the
`configuration_attrs`. Subclasses might override this to perform
additional input validation, cleanup, etc.
Parameters
----------
d : dict
The configuration dictionary. To specify the order that
the changes should be made, use an OrderedDict.
Returns
-------
(old, new) tuple of dictionaries
Where old and new are pre- and post-configure configuration states.
'''
old = self.read_configuration()
for key, val in d.items():
if key not in self.configuration_attrs:
# a little extra checking for a more specific error msg
if key not in self.signal_names:
raise ValueError("There is no signal named %s" % key)
else:
raise ValueError("%s is not one of the "
"configuration_fields, so it cannot be "
"changed using configure" % key)
set_and_wait(getattr(self, key), val)
new = self.read_configuration()
return old, new
def _repr_info(self):
yield ('prefix', self.prefix)
yield from super()._repr_info()
yield ('read_attrs', self.read_attrs)
yield ('configuration_attrs', self.configuration_attrs)
yield ('monitor_attrs', self.monitor_attrs)