Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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_Elements_allowed(self, req_type):
return True
return SardanaServer.server_state == State.Running
def _on_measurement_group_changed(self, event_source, event_type, event_value):
# during server startup and shutdown avoid processing element
# creation events
if SardanaServer.server_state != State.Running:
return
timestamp = time.time()
name = event_type.name
name = name.replace('_', '')
multi_attr = self.get_device_attr()
attr = multi_attr.get_attr_by_name(name)
quality = AttrQuality.ATTR_VALID
priority = event_type.priority
error = None
if name == "state":
event_value = self.calculate_tango_state(event_value)
elif name == "status":
event_value = self.calculate_tango_status(event_value)
elif name == "acquisitionmode":
def is_Elements_allowed(self, req_type):
return SardanaServer.server_state == State.Running
def _on_motor_changed(self, event_source, event_type, event_value):
# during server startup and shutdown avoid processing element
# creation events
if SardanaServer.server_state != State.Running:
return
timestamp = time.time()
name = event_type.name.lower()
if name == "w_position" and not self.in_write_position:
self.debug("Storing dial set point: %s",
self.motor.dial_position.w_value)
self.set_write_dial_position_to_db()
return
try:
attr = self.get_attribute_by_name(name)
except DevFailed:
return
def is_Elements_allowed(self, req_type):
return True
return SardanaServer.server_state == State.Running
def _on_pseudo_counter_changed(self, event_source, event_type,
event_value):
# during server startup and shutdown avoid processing element
# creation events
if SardanaServer.server_state != State.Running:
return
timestamp = time.time()
name = event_type.name.lower()
attr_name = name
# TODO: remove this condition when Data attribute will be substituted
# by ValueBuffer
if name == "valuebuffer":
attr_name = "data"
try:
attr = self.get_attribute_by_name(attr_name)
except DevFailed:
return
quality = AttrQuality.ATTR_VALID
def on_zerod_changed(self, event_source, event_type, event_value):
# during server startup and shutdown avoid processing element
# creation events
if SardanaServer.server_state != State.Running:
return
timestamp = time.time()
quality = AttrQuality.ATTR_VALID
priority = event_type.priority
value = None
error = None
name = event_type.name.lower()
attr_name = name
attr = self.get_device_attr().get_attr_by_name(attr_name)
if name == "state":
value = self.calculate_tango_state(event_value)
elif name == "status":
value = self.calculate_tango_status(event_value)
def _on_ior_changed(self, event_source, event_type, event_value):
# during server startup and shutdown avoid processing element
# creation events
if SardanaServer.server_state != State.Running:
return
timestamp = time.time()
name = event_type.name.lower()
if name == "w_value" and not self.in_write_value:
self.debug("Storing value set point: %s", self.ior.value.w_value)
self.set_write_value_to_db()
return
multi_attr = self.get_device_attr()
try:
attr = multi_attr.get_attr_by_name(name)
except DevFailed:
return