Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def input(self, input_data=None):
# parent process
data_type = input_data.get('data_type', 'String')
if isinstance(data_type, (str, unicode)):
ms = genutils.get_macro_server()
interfaces = ms.getInterfaces()
if data_type in interfaces:
input_data['data_type'] = [ elem.name for elem in interfaces[data_type].values() ]
self._conn.send(input_data)
ret = self._conn.recv()
return ret
def macro_fn(parameter_s='', name=macro_name):
parameters = split_macro_parameters(parameter_s)
door = genutils.get_door()
door.runMacro(macro_name, parameters, synch=True)
macro = door.getLastRunningMacro()
if macro is not None: # maybe none if macro was aborted
return macro.getResult()
def _addMacro(self, macro_info):
macro_name = str(macro_info.name)
# IPython < 1 magic commands have different API
if genutils.get_ipython_version_list() < [1, 0]:
def macro_fn(shell, parameter_s='', name=macro_name):
parameters = split_macro_parameters(parameter_s)
door = genutils.get_door()
door.runMacro(macro_name, parameters, synch=True)
macro = door.getLastRunningMacro()
if macro is not None: # maybe none if macro was aborted
return macro.getResult()
else:
def macro_fn(parameter_s='', name=macro_name):
parameters = split_macro_parameters(parameter_s)
door = genutils.get_door()
door.runMacro(macro_name, parameters, synch=True)
macro = door.getLastRunningMacro()
if macro is not None: # maybe none if macro was aborted
return macro.getResult()
def macro_fn(parameter_s='', name=macro_name):
door = genutils.get_door()
ms = genutils.get_macro_server()
params_def = ms.getMacroInfoObj(name).parameters
parameters = split_macro_parameters(parameter_s, params_def)
door.runMacro(macro_name, parameters, synch=True)
macro = door.getLastRunningMacro()
if macro is not None: # maybe none if macro was aborted
return macro.getResult()
macro_fn.__name__ = macro_name
macro_fn.__doc__ = macro_info.doc + "\nWARNING: do not rely on the" \
" file path below\n"
# register magic command
genutils.expose_magic(macro_name, macro_fn)
self._local_magic[macro_name] = macro_fn
return macro_info
import os
import ctypes
import PyTango
from taurus.core import TaurusEventType, TaurusSWDevState, TaurusDevState
from sardana.sardanautils import is_pure_str, is_non_str_seq
from sardana.spock import genutils
from sardana.util.parser import ParamParser
from sardana.spock.inputhandler import SpockInputHandler, InputHandler
from sardana import sardanacustomsettings
CHANGE_EVTS = TaurusEventType.Change, TaurusEventType.Periodic
if genutils.get_gui_mode() == 'qt':
from sardana.taurus.qt.qtcore.tango.sardana.macroserver import QDoor, QMacroServer
BaseDoor = QDoor
BaseMacroServer = QMacroServer
BaseGUIViewer = object
else:
from sardana.taurus.core.tango.sardana.macroserver import BaseDoor, BaseMacroServer
BaseGUIViewer = object
RUNNING_STATE = TaurusDevState.Ready
class GUIViewer(BaseGUIViewer):
def __init__(self, door=None):
BaseGUIViewer.__init__(self)
def plot(self):
try:
import sps
except Exception:
print('sps module not available. No plotting')
return
try:
import pylab
except Exception:
print("pylab not available (try running 'spock -pylab'). "
"No plotting")
return
door = genutils.get_door()
try:
env = dict(door.getEnvironmentObj().read().value)
except Exception as e:
print('Unable to read environment. No plotting')
print(str(e))
return
program = door.getNormalName().replace('/', '').replace('_', '')
try:
array = env['ActiveMntGrp'].replace(
'/', '').replace('_', '').upper() + "0D"
array_ENV = '%s_ENV' % array
except:
print('ActiveMntGrp not defined. No plotting')
return
def macro_fn(shell, parameter_s='', name=macro_name):
door = genutils.get_door()
ms = genutils.get_macro_server()
params_def = ms.getMacroInfoObj(name).parameters
parameters = split_macro_parameters(parameter_s, params_def)
door.runMacro(macro_name, parameters, synch=True)
macro = door.getLastRunningMacro()
if macro is not None: # maybe none if macro was aborted
return macro.getResult()
else:
def _removeMacro(self, macro_info):
macro_name = macro_info.name
genutils.unexpose_magic(macro_name)
del self._local_magic[macro_name]
def _updateState(self, old_sw_state, new_sw_state, silent=False):
user_ns = genutils.get_ipapi().user_ns
if new_sw_state == RUNNING_STATE:
user_ns['DOOR_STATE'] = ""
else:
user_ns['DOOR_STATE'] = " (OFFLINE)"
if not self.isConsoleReady():
self._spock_state = new_sw_state
return
ss = self._spock_state
if ss is not None and ss != new_sw_state and not silent:
if ss == RUNNING_STATE:
self.write_asynch(
"\nConnection to door '%s' was lost.\n" % self.getSimpleName())
elif new_sw_state == RUNNING_STATE:
self.write_asynch("\nConnection to the door (%s) has "
def __init__(self):
# don't call super __init__ on purpose
self._input = genutils.spock_input