Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __preprocessResult(self, result):
"""decodes the given output from a macro in order to be able to send to
the result channel as a sequence
:param out: output value
:return: the output as a sequence of strings
"""
if result is None:
return ()
if is_non_str_seq(result):
result = map(str, result)
else:
result = (str(result),)
return result
inside_str = False
pars.append(par)
par = ''
else:
inside_str = True
elif c == ' ':
if inside_str:
par += c
else:
pars.append(par)
par = ''
else:
par += c
if par: pars.append(par)
return pars
elif is_non_str_seq(parameters):
return parameters
def is_chunk(type_, obj):
if not is_non_str_seq(obj):
return False
if type_ == ElementType.CTExpChannel:
return True
elif type_ == ElementType.OneDExpChannel:
# empty list is also considered as chunk
if len(obj) == 0 or not is_number(obj[0]):
return True
elif type_ == ElementType.TwoDExpChannel:
# empty list is also considered as chunk
if len(obj) == 0 or not is_number(obj[0][0]):
return True
return False
#kwargs like 'synch' are ignored in this re-implementation
if self._spock_state != TaurusSWDevState.Running:
print "Unable to run macro: No connection to door '%s'" % self.getSimpleName()
raise Exception("Unable to run macro: No connection")
if xml is None:
xml = self.getRunningXML()
kwargs['synch'] = True
try:
return BaseDoor._runMacro(self, xml, **kwargs)
except KeyboardInterrupt:
self.write('\nCtrl-C received: Stopping... ')
self.block_lines = 0
self.command_inout("StopMacro")
self.writeln("Done!")
except PyTango.DevFailed, e:
if is_non_str_seq(e.args) and \
not isinstance(e.args, (str, unicode)):
reason, desc = e.args[0].reason, e.args[0].desc
macro_obj = self.getRunningMacro()
if reason == 'MissingParam':
print "Missing parameter:", desc
print macro_obj.getInfo().doc
elif reason == 'WrongParam':
print "Wrong parameter:", desc
print macro_obj.getInfo().doc
elif reason == 'UnkownParamObj':
print "Unknown parameter:", desc
elif reason == 'MissingEnv':
print "Missing environment:", desc
elif reason in ('API_CantConnectToDevice', 'API_DeviceNotExported'):
self._updateState(self._old_sw_door_state, TaurusSWDevState.Shutdown, silent=True)
print "Unable to run macro: No connection to door '%s'" % self.getSimpleName()
pars.append(par)
par = ''
else:
inside_str = True
elif c == ' ':
if inside_str:
par += c
else:
pars.append(par)
par = ''
else:
par += c
if par:
pars.append(par)
return pars
elif is_non_str_seq(parameters):
return parameters
if raw_param_repeat is None:
raw_param_repeat = []
len_rep = len(raw_param_repeat)
if min_rep and len_rep < min_rep:
msg = 'Found %d repetitions of param %s, min is %d' % \
(len_rep, name, min_rep)
raise MissingRepeat, msg
if max_rep and len_rep > max_rep:
msg = 'Found %d repetitions of param %s, max is %d' % \
(len_rep, name, max_rep)
raise SupernumeraryRepeat(msg)
# repeat params with only one member and only one repetition value are
# allowed - encapsulate it in list and try to decode anyway;
# for the moment this only works for non XML decoding but could be
# extended in the future to support XML as well
if not is_non_str_seq(raw_param_repeat)\
and not isinstance(raw_param_repeat, etree._Element):
raw_param_repeat = [raw_param_repeat]
for raw_repeat in raw_param_repeat:
if len(param_type) > 1:
repeat = []
for i, member_type in enumerate(param_type):
try:
member_raw = raw_repeat[i]
except IndexError:
member_raw = None
member = self.decodeNormal(member_raw, member_type)
repeat.append(member)
else:
# if the repeat parameter is composed of just one member
# do not encapsulate it in list and pass directly the item
if isinstance(raw_repeat, etree._Element):