Commit 3960cd67 authored by Milan Prica's avatar Milan Prica

As currently deployed at FERMI.

parents
NAME = iris-srv
MAIN = iris_diaphragm.py
DIRNAME = $(NAME:-srv=)
MODNAME = $(MAIN:.py=)
PY_FILES += $(wildcard src/*.py)
default: bin ${PY_FILES}
@cp ${PY_FILES} bin/${DIRNAME}
@echo "#!/usr/bin/env python\nimport sys\nsys.path.append(sys.path[0]+'/${DIRNAME}')\nfrom ${MODNAME} import main\nif __name__ == '__main__':\n main()\n" > bin/${NAME}
@chmod +x bin/${NAME} bin/${DIRNAME}/${MAIN}
bin:
@test -d $@ || mkdir -p $@/${DIRNAME}
clean:
@rm -fr bin/ src/*~
.PHONY: clean
# Iris diaphragm Tango device server.
The device server controls the iris aperture and runs on top of a smaract positioner. Allows for simple commands like open and close and translates
the movements of the motors into aperture value.
#!/usr/bin/env python
# -*- coding:utf-8 -*-
##############################################################################
## license :
##============================================================================
##
## File : IrisDiaphragm.py
##
## This file is part of Tango device class.
##
## Tango is free software: you can redistribute it and/or modify
## it under the terms of the GNU Lesser General Public License as published by
## the Free Software Foundation, either version 3 of the License, or
## (at your option) any later version.
##
## Tango is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
## GNU Lesser General Public License for more details.
##
## You should have received a copy of the GNU Lesser General Public License
## along with Tango. If not, see <http://www.gnu.org/licenses/>.
##
##
## $Author : sci.comp$
##
## $Revision : $
##
## $Date : $
##
## $HeadUrl : $
##############################################################################
__all__ = ["IrisDiaphragm", "IrisDiaphragmClass", "main"]
__docformat__ = 'restructuredtext'
import PyTango
import sys
class IrisDiaphragm (PyTango.Device_4Impl):
def __init__(self,cl, name):
PyTango.Device_4Impl.__init__(self,cl,name)
self.debug_stream("In __init__()")
IrisDiaphragm.init_device(self)
def delete_device(self):
self.debug_stream("In delete_device()")
self.stop_polling()
del self.ctrl_proxy
def init_device(self):
self.debug_stream("In init_device()")
self.status_string = "Initializing..."
self.get_device_properties(self.get_device_class())
self.attr_Aperture_read = 0.0
self.attr_StepSize_read = 0.0
try:
# ctrl_proxy device to SmarAct MCS Controller
self.ctrl_proxy = PyTango.DeviceProxy(self.SmaractMCSDevice)
if self.ctrl_proxy.state() != PyTango.DevState.ON:
self.set_state(PyTango.DevState.FAULT)
self.status_string = "ctrl_proxy device not in ON state. INIT failed."
return
except PyTango.DevFailed:
exctype, value = sys.exc_info()[:2]
print "Failed with exception ! ", exctype
for err in value:
print " reason", err.reason
print " description", err.desc
print " origin", err.origin
print " severity", err.severity
self.set_state(PyTango.DevState.FAULT)
self.status_string = "INIT failed, ctrl_proxy proxy failed."
self.debug_stream(self.status_string)
return
self.status_string = "Device initialized."
self.set_status(self.status_string)
self.set_state(PyTango.DevState.ON)
def always_executed_hook(self):
# self.debug_stream("In always_excuted_hook()")
self.set_state(self.ctrl_proxy.state())
self.set_status(self.ctrl_proxy.status())
#-----------------------------------------------------------------------------
# Utility methods
#-----------------------------------------------------------------------------
def write_position(self, aperture):
""" write Position attribute on the controller Tango device."""
new_pos = aperture * self.ConversionFactor
try:
self.ctrl_proxy.write_attribute("Position", new_pos)
except PyTango.DevFailed:
exctype, value = sys.exc_info()[:2]
self.status_string = "Failed with exception " + str(exctype)
self.error_stream("Failed with exception " + str(exctype))
for err in value:
self.error_stream(" reason " + str(err.reason))
self.error_stream(" description" + str(err.desc))
self.error_stream(" origin" + str(err.origin))
self.error_stream(" severity" + str(err.severity))
def read_position(self):
""" read Position attribute on the controller Tango device."""
try:
value = self.ctrl_proxy.read_attribute("Position").value
self.attr_Aperture_read = value / self.ConversionFactor
return self.attr_Aperture_read
except PyTango.DevFailed:
exctype, value = sys.exc_info()[:2]
self.status_string = "Failed with exception " + str(exctype)
self.error_stream("Failed with exception " + str(exctype))
for err in value:
self.error_stream(" reason " + str(err.reason))
self.error_stream(" description" + str(err.desc))
self.error_stream(" origin" + str(err.origin))
self.error_stream(" severity" + str(err.severity))
def execute_command(self, cmd_name):
""" Execute the given command on the Tango device. """
try:
result = self.ctrl_proxy.command_inout(cmd_name)
return result
except PyTango.DevFailed:
exctype, value = sys.exc_info()[:2]
self.status_string = "Failed with exception " + str(exctype)
self.error_stream("Failed with exception " + str(exctype))
for err in value:
self.error_stream(" reason " + str(err.reason))
self.error_stream(" description" + str(err.desc))
self.error_stream(" origin" + str(err.origin))
self.error_stream(" severity" + str(err.severity))
#-----------------------------------------------------------------------------
# read/write attribute methods
#-----------------------------------------------------------------------------
def read_Aperture(self, attr):
# self.debug_stream("In read_Aperture()")
attr.set_value(self.read_position())
def write_Aperture(self, attr):
# self.debug_stream("In write_Aperture()")
new_aperture = attr.get_write_value()
self.write_position(new_aperture)
def is_Aperture_allowed(self, req_type):
return self.get_state() in (PyTango.DevState.ON,)
def read_StepSize(self, attr):
# self.debug_stream("In read_StepSize()")
attr.set_value(self.attr_StepSize_read)
def write_StepSize(self, attr):
# self.debug_stream("In write_StepSize()")
step_size = attr.get_write_value()
self.attr_StepSize_read = step_size
#-----------------------------------------------------------------------------
# command methods
#-----------------------------------------------------------------------------
def On(self):
"""
:param :
:type: PyTango.DevVoid
:return:
:rtype: PyTango.DevVoid """
#self.debug_stream("In On()")
return
def Off(self):
"""
:param :
:type: PyTango.DevVoid
:return:
:rtype: PyTango.DevVoid """
# self.debug_stream("In Off()")
self.set_state(PyTango.DevState.OFF)
def StepUp(self):
"""
:param :
:type: PyTango.DevVoid
:return:
:rtype: PyTango.DevVoid """
# self.debug_stream("In StepUp()")
if self.attr_StepSize_read == 0:
self.status_string = 'Step size is not set.'
self.set_status(self.status_string)
return
new_aperture = self.attr_Aperture_read + self.attr_StepSize_read
self.write_position(new_aperture)
def is_StepUp_allowed(self):
return self.get_state() in (PyTango.DevState.ON,)
def StepDown(self):
"""
:param :
:type: PyTango.DevVoid
:return:
:rtype: PyTango.DevVoid """
# self.debug_stream("In StepDown()")
if self.attr_StepSize_read == 0:
self.status_string = 'Step size is not set.'
self.set_status(self.status_string)
return
new_aperture = self.attr_Aperture_read - self.attr_StepSize_read
self.write_position(new_aperture)
def is_StepDown_allowed(self):
return self.get_state() in (PyTango.DevState.ON,)
def Open(self):
"""
:param :
:type: PyTango.DevVoid
:return:
:rtype: PyTango.DevVoid """
# self.debug_stream("In Open()")
self.execute_command("GoToHighLimit")
def is_Open_allowed(self):
return self.get_state() in (PyTango.DevState.ON,)
def Close(self):
"""
:param :
:type: PyTango.DevVoid
:return:
:rtype: PyTango.DevVoid """
# self.debug_stream("In Close()")
self.execute_command("GoToLowLimit")
def is_Close_allowed(self):
return self.get_state() in (PyTango.DevState.ON,)
class IrisDiaphragmClass(PyTango.DeviceClass):
def dyn_attr(self, dev_list):
return
# Class Properties
class_property_list = {
}
# Device Properties
device_property_list = {
'SmaractMCSDevice':
[PyTango.DevString,
"Tango Device that controls the physical device.",
[]],
'ConversionFactor':
[PyTango.DevDouble,
"Degrees of rotary positioner for 1 mm of aperture.",
[]],
}
# Command definitions
cmd_list = {
'On':
[[PyTango.DevVoid, "none"],
[PyTango.DevVoid, "none"]],
'Off':
[[PyTango.DevVoid, "none"],
[PyTango.DevVoid, "none"]],
'Open':
[[PyTango.DevVoid, "none"],
[PyTango.DevVoid, "none"]],
'Close':
[[PyTango.DevVoid, "none"],
[PyTango.DevVoid, "none"]],
'StepUp':
[[PyTango.DevVoid, "none"],
[PyTango.DevVoid, "none"]],
'StepDown':
[[PyTango.DevVoid, "none"],
[PyTango.DevVoid, "none"]],
}
# Attribute definitions
attr_list = {
'Aperture':
[[PyTango.DevDouble,
PyTango.SCALAR,
PyTango.READ_WRITE],
{
'max value': "22.0",
'min value': "0.0",
'description': "Iris diaphragm aperture in mm."
}],
'StepSize':
[[PyTango.DevDouble,
PyTango.SCALAR,
PyTango.READ_WRITE],
{
'max value': "11.0",
'min value': "0.0",
'description': "Aperture step size in mm.",
'Memorized':"true"
}],
}
def main():
try:
py = PyTango.Util(sys.argv)
py.add_class(IrisDiaphragmClass, IrisDiaphragm, 'IrisDiaphragm')
U = PyTango.Util.instance()
U.server_init()
U.server_run()
except PyTango.DevFailed, e:
print '-------> Received a DevFailed exception:', e
except Exception,e:
print '-------> An unforeseen exception occured....', e
if __name__ == '__main__':
main()
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment