Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
for a in ci.plot_axes:
if a == '':
# skip plots against moveables in scans that do not involve
# them e.g. time scans
if len(ref_moveables) == 0:
continue
plotAxes.append(ref_moveables[i])
i += 1
else:
plotAxes.append(a)
try:
value_ref_enabled = ci.value_ref_enabled
except AttributeError:
value_ref_enabled = False
# create the ColumnDesc object
column = ColumnDesc(name=ci.full_name,
label=ci.label,
dtype=ci.data_type,
shape=ci.shape,
instrument=instrumentFullName,
source=ci.source,
output=ci.output,
conditioning=ci.conditioning,
normalization=ci.normalization,
plot_type=ci.plot_type,
plot_axes=plotAxes,
data_units=ci.unit,
value_ref_enabled=value_ref_enabled)
data_desc.append(column)
counters.append(column.name)
try:
counters.remove(master['full_name'])
plot_axes=plotAxes,
data_units=ci.unit)
data_desc.append(column)
counters.append(column.name)
try:
counters.remove(master['full_name'])
except ValueError:
# timer may be disabled
pass
env['counters'] = counters
for extra_column in self._extra_columns:
data_desc.append(extra_column.getColumnDesc())
# add extra columns
data_desc += self._extrainfodesc
data_desc.append(ColumnDesc(name='timestamp',
label='dt', dtype='float64'))
env['datadesc'] = data_desc
# set the data compression default
try:
env['DataCompressionRank'] = self.macro.getEnv(
'DataCompressionRank')
except UnknownEnv:
env['DataCompressionRank'] = -1
# set the sample information
# @todo: use the instrument API to get this info
try:
env['SampleInfo'] = self.macro.getEnv('SampleInfo')
except UnknownEnv:
"""Expected keywords are:
- model (str, mandatory): represents data source (ex.: a/b/c/d)
- label (str, mandatory): column label
- name (str, optional): unique name (defaults to model)
- shape (seq, optional): data shape
- dtype (numpy.dtype, optional): data type
- instrument (str, optional): full instrument name"""
self._label = kwargs['label']
self._model = kwargs['model']
if not kwargs.has_key('dtype'):
kwargs['dtype'] = self.getType()
if not kwargs.has_key('shape'):
kwargs['shape'] = self.getShape()
if not kwargs.has_key('name'):
kwargs['name'] = self._model
self._column = ColumnDesc(**kwargs)
:param elements: (list) list of tuples of label,src for the
elements to read (can be pool elements or Taurus
attribute names).
:return: (list) a list of :class:`ColumnDesc`,
each including a "pre_scan_value" attribute with the read
value for that attr
"""
manager = self.macro.getManager()
all_elements_info = manager.get_elements_with_interface('Element')
ret = []
for src, label in elements:
try:
if src in all_elements_info:
ei = all_elements_info[src]
column = ColumnDesc(name=ei.full_name,
label=label,
instrument=ei.instrument,
source=ei.source)
else:
column = ColumnDesc(name=src,
label=label,
source=src)
# @Fixme: Tango-centric. It should work for any Taurus
# Attribute
v = PyTango.AttributeProxy(column.source).read().value
column.pre_scan_value = v
column.shape = np.shape(v)
column.dtype = getattr(v, 'dtype', np.dtype(type(v))).name
ret.append(column)
except Exception:
'''reads the current values of the given elements
:param elements: (list) list of tuples of label,src for the elements to read
(can be pool elements or Taurus attribute names).
:return: (list) a list of :class:`ColumnDesc`, each including a
"pre_scan_value" attribute with the read value for that attr
'''
manager = self.macro.getManager()
all_elements_info = manager.get_elements_with_interface('Element')
ret = []
for src, label in elements:
try:
if src in all_elements_info:
ei = all_elements_info[src]
column = ColumnDesc(name=ei.full_name,
label=label,
instrument=ei.instrument,
source=ei.source)
else:
column = ColumnDesc(name=src,
label=label,
source=src)
v = PyTango.AttributeProxy(column.source).read().value #@Fixme: Tango-centric. It should work for any Taurus Attribute
column.pre_scan_value = v
column.shape = np.shape(v)
column.dtype = getattr(v, 'dtype', np.dtype(type(v))).name
ret.append(column)
except:
self.macro.warning('Error taking pre-scan snapshot of %s (%s)', label, src)
self.debug('Details:', exc_info=1)
:param elements: (list) list of tuples of label,src for the
elements to read (can be pool elements or Taurus
attribute names).
:return: (list) a list of :class:`ColumnDesc`,
each including a "pre_scan_value" attribute with the read
value for that attr
"""
manager = self.macro.getManager()
all_elements_info = manager.get_elements_with_interface('Element')
ret = []
for src, label in elements:
try:
if src in all_elements_info:
ei = all_elements_info[src]
column = ColumnDesc(name=ei.full_name,
label=label,
instrument=ei.instrument,
source=ei.source)
else:
column = ColumnDesc(name=src,
label=label,
source=src)
# @Fixme: Tango-centric. It should work for any Taurus
# Attribute
v = PyTango.AttributeProxy(column.source).read().value
column.pre_scan_value = v
column.shape = np.shape(v)
column.dtype = getattr(v, 'dtype', np.dtype(type(v))).name
ret.append(column)
except Exception:
instrumentFullName = self.macro.findObjs(instrument, type_class=Type.Instrument)[0].getFullName()
except InterruptException:
raise
except:
instrumentFullName = ''
#substitute the axis placeholder by the corresponding moveable.
plotAxes = []
i = 0
for a in ci.plot_axes:
if a == '':
plotAxes.append(ref_moveables[i])
i += 1
else: plotAxes.append(a)
#create the ColumnDesc object
column = ColumnDesc(name=ci.full_name,
label=ci.label,
dtype=ci.data_type,
shape=ci.shape,
instrument=instrumentFullName,
source=ci.source,
output=ci.output,
conditioning=ci.conditioning,
normalization=ci.normalization,
plot_type=ci.plot_type,
plot_axes=plotAxes,
data_units=ci.unit)
data_desc.append(column)
counters.append(column.name)
counters.remove(master['full_name'])
env['counters'] = counters
:return: (list) a list of :class:`ColumnDesc`, each including a
"pre_scan_value" attribute with the read value for that attr
'''
manager = self.macro.getManager()
all_elements_info = manager.get_elements_with_interface('Element')
ret = []
for src, label in elements:
try:
if src in all_elements_info:
ei = all_elements_info[src]
column = ColumnDesc(name=ei.full_name,
label=label,
instrument=ei.instrument,
source=ei.source)
else:
column = ColumnDesc(name=src,
label=label,
source=src)
v = PyTango.AttributeProxy(column.source).read().value #@Fixme: Tango-centric. It should work for any Taurus Attribute
column.pre_scan_value = v
column.shape = np.shape(v)
column.dtype = getattr(v, 'dtype', np.dtype(type(v))).name
ret.append(column)
except:
self.macro.warning('Error taking pre-scan snapshot of %s (%s)', label, src)
self.debug('Details:', exc_info=1)
return ret
data_units=ci.unit,
value_ref_enabled=value_ref_enabled)
data_desc.append(column)
counters.append(column.name)
try:
counters.remove(master['full_name'])
except ValueError:
# timer may be disabled
pass
env['counters'] = counters
for extra_column in self._extra_columns:
data_desc.append(extra_column.getColumnDesc())
# add extra columns
data_desc += self._extrainfodesc
data_desc.append(ColumnDesc(name='timestamp',
label='dt', dtype='float64'))
env['datadesc'] = data_desc
# set the data compression default
try:
env['DataCompressionRank'] = self.macro.getEnv(
'DataCompressionRank')
except UnknownEnv:
env['DataCompressionRank'] = -1
# set the sample information
# @todo: use the instrument API to get this info
try:
env['SampleInfo'] = self.macro.getEnv('SampleInfo')
except UnknownEnv: