Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# tigger + trigger mode
count = Cpt(EpicsSignal, '.CNT', trigger_value=1, kind=Kind.omitted)
count_mode = Cpt(EpicsSignal, '.CONT', string=True, kind=Kind.config)
# delay from triggering to starting counting
delay = Cpt(EpicsSignal, '.DLY', kind=Kind.config)
auto_count_delay = Cpt(EpicsSignal, '.DLY1', kind=Kind.config)
time = Cpt(EpicsSignal, '.T')
freq = Cpt(EpicsSignal, '.FREQ', kind=Kind.config)
preset_time = Cpt(EpicsSignal, '.TP', kind=Kind.config)
auto_count_time = Cpt(EpicsSignal, '.TP1', kind=Kind.config)
update_rate = Cpt(EpicsSignal, '.RATE', kind=Kind.omitted)
auto_count_update_rate = Cpt(EpicsSignal, '.RAT1', kind=Kind.omitted)
egu = Cpt(EpicsSignal, '.EGU', kind=Kind.config)
def match_names(self):
for s in self.channels.component_names:
getattr(self.channels, s).match_name()
def select_channels(self, chan_names):
'''Select channels based on the EPICS name PV
Parameters
----------
chan_names : Iterable[str] or None
The names (as reported by the channel.chname signal)
def _validate_kind(self, val):
if isinstance(val, str):
val = getattr(Kind, val.lower())
if Kind.normal & val:
val = val | Kind.config
return super()._validate_kind(val)
def _validate_kind(self, val):
if isinstance(val, str):
return Kind[val.lower()]
return Kind(val)
auto_count_delay = Cpt(EpicsSignal, '.DLY1', kind=Kind.config)
# the data
channels = DDCpt(_scaler_fields(EpicsSignalRO, 'chan', '.S', range(1, 33),
kind=Kind.hinted))
names = DDCpt(_scaler_fields(EpicsSignal, 'name', '.NM', range(1, 33),
kind=Kind.config))
time = Cpt(EpicsSignal, '.T', kind=Kind.config)
freq = Cpt(EpicsSignal, '.FREQ', kind=Kind.config)
preset_time = Cpt(EpicsSignal, '.TP', kind=Kind.config)
auto_count_time = Cpt(EpicsSignal, '.TP1', kind=Kind.config)
presets = DDCpt(_scaler_fields(EpicsSignal, 'preset', '.PR', range(1, 33),
kind=Kind.omitted))
gates = DDCpt(_scaler_fields(EpicsSignal, 'gate', '.G', range(1, 33),
kind=Kind.omitted))
update_rate = Cpt(EpicsSignal, '.RATE', kind=Kind.config)
auto_count_update_rate = Cpt(EpicsSignal, '.RAT1', kind=Kind.config)
egu = Cpt(EpicsSignal, '.EGU', kind=Kind.config)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.stage_sigs.update([('count_mode', 0)])
class ScalerChannel(Device):
name=name, parent=parent, **kwargs)
if self.__class__ is PVPositioner:
raise TypeError('PVPositioner must be subclassed with the correct '
'signals set in the class definition.')
self._egu = egu
if limits is not None:
self._limits = tuple(limits)
else:
self._limits = None
if self.readback is not None:
self.readback.subscribe(self._pos_changed)
self.readback.kind = Kind.hinted
elif self.setpoint is not None:
self.setpoint.subscribe(self._pos_changed)
else:
raise ValueError('A setpoint or a readback must be specified')
if self.done is None and not self.put_complete:
msg = ('PVPositioner {} is mis-configured. A "done" Signal must be'
' provided or use PVPositionerPC (which uses put completion'
' to determine when motion has completed).'
''.format(self.name))
raise ValueError(msg)
if self.done is not None:
self.done.subscribe(self._move_changed)
else:
# If there is not a `move_changed` signal, indicate that the
def __init__(self, defn, *, clsname=None, doc=None, kind=Kind.normal,
default_read_attrs=None, default_configuration_attrs=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
if isinstance(default_read_attrs, collections.Iterable):
default_read_attrs = tuple(default_read_attrs)
if isinstance(default_configuration_attrs, collections.Iterable):
default_configuration_attrs = tuple(default_configuration_attrs)
self.default_read_attrs = default_read_attrs
self.default_configuration_attrs = default_configuration_attrs
self.kind = _ensure_kind(kind)
# TODO: component compatibility
self.trigger_value = None
def read_attrs(self, val):
self.__attr_list_helper(val, Kind.normal, Kind.hinted, 'read_attrs')
def __init__(self, cls, suffix=None, *, lazy=False, trigger_value=None,
add_prefix=None, doc=None, kind=Kind.normal, **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
self.kind = _ensure_kind(kind)
if add_prefix is None:
add_prefix = ('suffix', 'write_pv')
self.add_prefix = tuple(add_prefix)
def __init__(self, *, name=None, attr_name='', parent=None, labels=None,
kind=None):
if labels is None:
labels = set()
self._ophyd_labels_ = set(labels)
if kind is None:
kind = Kind.normal
self.kind = kind
super().__init__()
# base name and ref to parent, these go with properties
if name is None:
name = ''
self._attr_name = attr_name
if not isinstance(name, str):
raise ValueError("name must be a string.")
self._name = name
self._parent = parent
self.subscriptions = {getattr(self, k)
for k in dir(type(self))
if (k.startswith('SUB') or
count_mode = Cpt(EpicsSignal, '.CONT', string=True, kind=Kind.config)
# delay from triggering to starting counting
delay = Cpt(EpicsSignal, '.DLY', kind=Kind.config)
auto_count_delay = Cpt(EpicsSignal, '.DLY1', kind=Kind.config)
time = Cpt(EpicsSignal, '.T')
freq = Cpt(EpicsSignal, '.FREQ', kind=Kind.config)
preset_time = Cpt(EpicsSignal, '.TP', kind=Kind.config)
auto_count_time = Cpt(EpicsSignal, '.TP1', kind=Kind.config)
update_rate = Cpt(EpicsSignal, '.RATE', kind=Kind.omitted)
auto_count_update_rate = Cpt(EpicsSignal, '.RAT1', kind=Kind.omitted)
egu = Cpt(EpicsSignal, '.EGU', kind=Kind.config)
def match_names(self):
for s in self.channels.component_names:
getattr(self.channels, s).match_name()
def select_channels(self, chan_names):
'''Select channels based on the EPICS name PV
Parameters
----------
chan_names : Iterable[str] or None
The names (as reported by the channel.chname signal)
of the channels to select.
If *None*, select all channels named in the EPICS scaler.
'''