Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _fill_channel_data(self, channel, channel_data):
"""Fill channel default values for the given channel dictionary"""
name = channel.name
ctype = channel.get_type()
full_name = channel.full_name
# choose ndim
ndim = None
if ctype == ElementType.CTExpChannel:
ndim = 0
elif ctype == ElementType.PseudoCounter:
ndim = 0
elif ctype == ElementType.ZeroDExpChannel:
ndim = 0
elif ctype == ElementType.OneDExpChannel:
ndim = 1
elif ctype == ElementType.TwoDExpChannel:
ndim = 2
elif ctype == ElementType.External:
config = channel.get_config()
if config is not None:
ndim = int(config.data_format)
elif ctype == ElementType.IORegister:
ndim = 0
if ctype != ElementType.External and channel.is_referable():
value_ref_enabled = channel_data.get('value_ref_enabled', False)
channel_data['value_ref_enabled'] = value_ref_enabled
value_ref_pattern = channel_data.get('value_ref_pattern', '')
channel_data['value_ref_pattern'] = value_ref_pattern
elif 'value_ref_enabled' in channel_data or 'value_ref_pattern' in \
channel_data:
if self._value_ref_compat:
def __init__(self, **kwargs):
kwargs['manager'] = kwargs.pop('macro_server')
kwargs['elem_type'] = ElementType.MacroLibrary
SardanaLibrary.__init__(self, **kwargs)
def __init__(self, **kwargs):
self._physical_elements = []
self._in_start_move = False
kwargs['elem_type'] = ElementType.MotorGroup
PoolGroupElement.__init__(self, **kwargs)
on_change = self.on_change
self._position = Position(self, listeners=on_change)
attr.events.ch_event.abs_change = "5.0"
cfg.append(attr)
except:
pass
try:
if get_tango_version_number() < 80000:
attr = elem_proxy.get_attribute_config_ex("limit_switches")[0]
attr.events.ch_event.abs_change = "1"
cfg.append(attr)
except:
pass
elif elem_type == ElementType.CTExpChannel:
attr = elem_proxy.get_attribute_config_ex("value")[0]
attr.events.ch_event.abs_change = "1.0"
cfg.append(attr)
elif elem_type == ElementType.ZeroDExpChannel:
attr = elem_proxy.get_attribute_config_ex("value")[0]
attr.events.ch_event.abs_change = "1.0"
cfg.append(attr)
elif elem_type == ElementType.PseudoMotor:
attr = elem_proxy.get_attribute_config_ex("position")[0]
attr.events.ch_event.abs_change = "1"
cfg.append(attr)
elif elem_type == ElementType.PseudoCounter:
attr = elem_proxy.get_attribute_config_ex("value")[0]
attr.events.ch_event.abs_change = "1"
cfg.append(attr)
elif elem_type == ElementType.IORegister:
attr = elem_proxy.get_attribute_config_ex("value")[0]
attr.events.ch_event.abs_change = "1"
cfg.append(attr)
elem_proxy.set_attribute_config(cfg)
def __init__(self, **kwargs):
self._parent_instrument = kwargs.pop("parent")
self._instrument_class = kwargs.pop("klass")
if self._parent_instrument:
self._parent_instrument = weakref.ref(self._parent_instrument)
self._child_instruments = {}
self._elements = {}
kwargs['elem_type'] = ElementType.Instrument
super(PoolInstrument, self).__init__(**kwargs)
def _fill_channel_data(self, channel, channel_data):
"""Fill channel default values for the given channel dictionary"""
name = channel.name
ctype = channel.get_type()
full_name = channel.full_name
# choose ndim
ndim = None
if ctype == ElementType.CTExpChannel:
ndim = 0
elif ctype == ElementType.PseudoCounter:
ndim = 0
elif ctype == ElementType.ZeroDExpChannel:
ndim = 0
elif ctype == ElementType.OneDExpChannel:
ndim = 1
elif ctype == ElementType.TwoDExpChannel:
ndim = 2
elif ctype == ElementType.External:
config = channel.get_config()
if config is not None:
ndim = int(config.data_format)
elif ctype == ElementType.IORegister:
ndim = 0
if ctype != ElementType.External and channel.is_referable():
value_ref_enabled = channel_data.get('value_ref_enabled', False)
channel_data['value_ref_enabled'] = value_ref_enabled
value_ref_pattern = channel_data.get('value_ref_pattern', '')
channel_data['value_ref_pattern'] = value_ref_pattern
elif 'value_ref_enabled' in channel_data or 'value_ref_pattern' in \
try:
if get_tango_version_number() < 80000:
attr = elem_proxy.get_attribute_config_ex("limit_switches")[0]
attr.events.ch_event.abs_change = "1"
cfg.append(attr)
except:
pass
elif elem_type == ElementType.CTExpChannel:
attr = elem_proxy.get_attribute_config_ex("value")[0]
attr.events.ch_event.abs_change = "1.0"
cfg.append(attr)
elif elem_type == ElementType.ZeroDExpChannel:
attr = elem_proxy.get_attribute_config_ex("value")[0]
attr.events.ch_event.abs_change = "1.0"
cfg.append(attr)
elif elem_type == ElementType.PseudoMotor:
attr = elem_proxy.get_attribute_config_ex("position")[0]
attr.events.ch_event.abs_change = "1"
cfg.append(attr)
elif elem_type == ElementType.PseudoCounter:
attr = elem_proxy.get_attribute_config_ex("value")[0]
attr.events.ch_event.abs_change = "1"
cfg.append(attr)
elif elem_type == ElementType.IORegister:
attr = elem_proxy.get_attribute_config_ex("value")[0]
attr.events.ch_event.abs_change = "1"
cfg.append(attr)
elem_proxy.set_attribute_config(cfg)
return
# Determine which writtable attributes have default value and apply
self.dict_extra['properties_desc'] = self.ctrl_properties_descriptions
self.ctrl_attributes = ctrl_attrs = CaselessDict()
for k, v in klass.ctrl_attributes.items():
ctrl_attrs[k] = DataInfo.toDataInfo(k, v)
self.axis_attributes = axis_attrs = CaselessDict()
for k, v in klass.ctrl_extra_attributes.items(): # old member
axis_attrs[k] = DataInfo.toDataInfo(k, v)
for k, v in klass.axis_attributes.items():
axis_attrs[k] = DataInfo.toDataInfo(k, v)
self.types = types = self.__build_types()
self.type_names = map(ElementType.whatis, types)
if ElementType.PseudoMotor in types:
self.motor_roles = tuple(klass.motor_roles)
self.pseudo_motor_roles = tuple(klass.pseudo_motor_roles)
if len(self.pseudo_motor_roles) == 0:
self.pseudo_motor_roles = (klass.__name__,)
self.dict_extra['motor_roles'] = self.motor_roles
self.dict_extra['pseudo_motor_roles'] = self.pseudo_motor_roles
if ElementType.PseudoCounter in types:
self.counter_roles = tuple(klass.counter_roles)
self.pseudo_counter_roles = tuple(klass.pseudo_counter_roles)
if len(self.pseudo_counter_roles) == 0:
self.pseudo_counter_roles = (klass.__name__,)
self.dict_extra['counter_roles'] = self.counter_roles
self.dict_extra['pseudo_counter_roles'] = self.pseudo_counter_roles
if ElementType.IORegister in types:
def __init__(self, **kwargs):
self._data_source = None
self._timer = None
kwargs['elem_type'] = ElementType.TwoDExpChannel
PoolTimerableChannel.__init__(self, **kwargs)
def __init__(self, full_name, name=None, macro_path=None,
environment_db=None, recorder_path=None):
# dict
# key - device name (case insensitive)
# value - Pool object representing the device name
self._pools = CaselessDict()
self._max_parallel_macros = self.MaxParalellMacros
self._path_id = None
MSContainer.__init__(self)
MSObject.__init__(self, full_name=full_name, name=name, id=InvalidId,
macro_server=self, elem_type=ElementType.MacroServer)
registerExtensions()
self._type_manager = TypeManager(self)
self._environment_manager = EnvironmentManager(self,
environment_db=environment_db)
self._macro_manager = MacroManager(self, macro_path=macro_path)
self._recorder_manager = RecorderManager(self,
recorder_path=recorder_path)