Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
old_names = set([ ctrl.name for ctrl in old_lib.get_controllers() ])
changed_names = set.intersection(new_names, old_names)
deleted_names = old_names.difference(new_names)
new_names = new_names.difference(old_names)
for new_name in new_names:
new_elements.append(new_lib.get_controller(new_name))
for changed_name in changed_names:
changed_elements.append(new_lib.get_controller(changed_name))
for deleted_name in deleted_names:
deleted_elements.append(old_lib.get_controller(deleted_name))
evt = { "new" : new_elements, "change" : changed_elements,
"del" : deleted_elements }
self.fire_event(EventType("ElementsChanged"), evt)
if old_lib is not None:
for pool_ctrl in init_pool_ctrls:
pool_ctrl.re_init()
def _set_step_per_unit(self, step_per_unit, propagate=1):
self._step_per_unit = step_per_unit
if propagate:
self.fire_event(EventType("step_per_unit",
priority=propagate), step_per_unit)
# force ask controller for new position to send priority event
self.get_position(cache=False, propagate=2)
state = State.On
status = "Motor HW is ON"
if m.isInMotion():
state = State.Moving
status = "Motor HW is MOVING"
m.getCurrentUserPosition()
switchstate = 0
if m.hitLowerLimit():
switchstate |= MotorController.LowerLimitSwitch
state = State.Alarm
status = "Motor HW is in ALARM. Hit hardware lower limit switch"
if m.hitUpperLimit():
switchstate |= MotorController.UpperLimitSwitch
state = State.Alarm
status = "Motor HW is in ALARM. Hit hardware upper limit switch"
if state != State.Alarm and not m.hasPower():
state = State.Off
status = "Motor is powered off"
#self._log.info("StateOne(%s) = %s", axis, (state, status, switchstate))
return state, status, switchstate
def re_init(self):
self.set_state(State.Init, propagate=2)
status = "{0} is Initializing (temporarily unavailable)".format(self.name)
self.set_status(status, propagate=2)
manager = self.pool.ctrl_manager
old_e_ids = self._element_ids
old_p_e_ids = self._pending_element_ids
elem_axis = dict(self._element_axis)
for axis in elem_axis:
self.remove_axis(axis, propagate=0)
if self._lib_info is None:
mod_name = self.get_library_name()
else:
mod_name = self._lib_info.name
if self._ctrl_info is None:
def StateOne(self, axis):
#self._log.debug("StateOne(%d)", axis)
#raise Exception("Cannot StateOne %d" % axis)
idx = axis - 1
m = self.m[idx]
state = State.On
status = "Motor HW is ON"
if m.isInMotion():
state = State.Moving
status = "Motor HW is MOVING"
m.getCurrentUserPosition()
switchstate = 0
if m.hitLowerLimit():
switchstate |= MotorController.LowerLimitSwitch
state = State.Alarm
status = "Motor HW is in ALARM. Hit hardware lower limit switch"
if m.hitUpperLimit():
switchstate |= MotorController.UpperLimitSwitch
state = State.Alarm
status = "Motor HW is in ALARM. Hit hardware upper limit switch"
if state != State.Alarm and not m.hasPower():
state = State.Off
status = "Motor is powered off"
#self._log.info("StateOne(%s) = %s", axis, (state, status, switchstate))
return state, status, switchstate
ChannelView.Output, ChannelView.ValueRefEnabled,
ChannelView.ValueRefPattern):
data = qvalue
elif taurus_role == ChannelView.DataType:
if len(qvalue.strip()) == 0:
# empty strings are considered as unspecified data type
try:
ch_data.pop(key)
except KeyError:
pass # data_type key may not be there if not specified
return
else:
data = qvalue
elif taurus_role == ChannelView.PlotType:
data = PlotType[qvalue]
elif taurus_role == ChannelView.Normalization:
data = Normalization[qvalue]
elif taurus_role == ChannelView.PlotAxes:
data = [a for a in qvalue.split('|')]
elif taurus_role == ChannelView.Shape:
s = qvalue
try:
data = eval(s, {}, {})
if not isinstance(data, (tuple, list)):
raise ValueError
except:
from taurus.core.util.log import Logger
Logger(self.__class__.__name__).error('Invalid shape %s', s)
data = ()
else:
raise NotImplementedError('Unknown role')
ch_data[key] = data
def run_tango_server(tango_util=None, start_time=None):
try:
if tango_util is None:
tango_util = Util(sys.argv)
util = Util.instance()
SardanaServer.server_state = State.Init
util.server_init()
SardanaServer.server_state = State.Running
if start_time is not None:
import datetime
dt = datetime.datetime.now() - start_time
taurus.info("Ready to accept request in %s", dt)
else:
taurus.info("Ready to accept request")
util.server_run()
SardanaServer.server_state = State.Off
taurus.info("Exiting")
except DevFailed:
taurus.info("Exiting")
taurus.critical("Server exited with DevFailed", exc_info=1)
except KeyboardInterrupt:
taurus.info("Exiting")
taurus.critical("Interrupted by keyboard")
except Exception:
taurus.info("Exiting")
taurus.critical("Server exited with unforeseen exception", exc_info=1)
taurus.info("Exited")
def is_GetMacroEnv_allowed(self):
return self.get_state() in [Macro.Finished, Macro.Abort,
Macro.Exception]
class usenv(Macro):
"""Unsets the given environment variable"""
param_def = [
['environment_list',
ParamRepeat(
['env', Type.Env, None, 'Environment variable name'], min=1),
None, 'List of environment items to be removed'],
]
def run(self, env):
self.unsetEnv(env)
self.output("Success!")
class load_env(Macro):
""" Read environment variables from config_env.xml file"""
def run(self):
doc = etree.parse("config_env.xml")
root = doc.getroot()
for element in root:
if element.find("./name").text == "auto_filter":
self.output("Loading auto_filter variables:")
filter_max_elem = element.find(".//FilterMax")
if filter_max_elem is not None:
filter_max = filter_max_elem.text
self.setEnv("FilterMax", filter_max)
self.output("FilterMax loaded")
else:
self.output("FilterMax not found")
filter_min_elem = element.find(".//FilterMin")
['offon', Type.Boolean, None, 'Unset/Set logging'],
['mode', Type.Integer, -1, 'Mode: 0 append, 1 new file'],
]
def run(self, offon, mode):
if offon:
if mode == 1:
self.setEnv('LogMacroMode', True)
elif mode == 0:
self.setEnv('LogMacroMode', False)
self.setEnv('LogMacro', True)
else:
self.setEnv('LogMacro', False)
class repeat(Hookable, Macro):
"""This macro executes as many repetitions of a set of macros as
specified by nr parameter. The macros to be repeated can be
given as parameters or as body hooks.
If both are given first will be executed the ones given as
parameters and then the ones given as body hooks.
If nr has negative value, repetitions will be executed until you
stop repeat macro.
.. note::
The repeat macro has been included in Sardana
on a provisional basis. Backwards incompatible changes
(up to and including removal of the macro) may occur if
deemed necessary by the core developers."""
hints = {'allowsHooks': ('body',)}