How to PyTango

Intended audience: developers, Programming language: python

My list of short recipes for common tasks … please check official documentation first!

Before anything else

1import PyTango

Installation notes

The new PyTango, is now available to download from the Tango download page or PiPy

If you have already installed PyTango with pip you can simply update your PyTango version by doing:

1$ pip -U PyTango

1$ sudo apt-get update python-pytango

The documentation is available at official readthedocs website.

If you encounter problems installing or running this release, please report them back to the tango mailing list.

You can check out this version:

1$ svn co https://tango-cs.svn.sourceforge.net/svnroot/tango-cs/PyTango/tags/Release_7_1_3 PyTango-7.1.3`

You can check out the latest version:

1$ svn co https://tango-cs.svn.sourceforge.net/svnroot/tango-cs/PyTango/trunk PyTango-latest`

Nice PyTango dancing

Using the DeviceProxy object

Getting the polling buffer values

Only for polled attributes we can get the last N read values. the polling buffer depth is managed by the admin device.

1dp = PyTango.DeviceProxy('some/tango/device')
2dp.attribute_history('cpustatus',10)

Get/Set polled attributes

1def get_polled_attributes(dev_name):
2    dp = PyTango.DeviceProxy(dev_name)
3    attrs = dp.get_attribute_list()
4    periods = [(a,dp.get_attribute_poll_period(a)) for a in attrs]
5    return dict((a,p) for a,p in periods if p)
6
7[plc4.poll_attribute(a,5000) for k,v in periods if v]

Modify the polling of attributes

1 import re,PyTango
2 period = 10000
3 devs = PyTango.Database().get_device_exported('some/tango/devices*')
4 for dev in devs:
5     dp = PyTango.DeviceProxy(dev)
6     attrs = sorted([a for a in dp.get_attribute_list() if re.match('(Output|Temperature)_[0-9]$',a)])
7     [dp.poll_attribute(a,period) for a in attrs]
8     print('\n'.join(dp.polling_status()))

Events

Creating an event callback

1# The callback must be a callable or an object with a push_event(self,event) method

Configuring an event

1#From the client side
2#subscribe_event(attr_name, event_type, cb_or_queuesize, filters=[], stateless=False, extract_as=PyTango._PyTango.ExtractAs.Numpy)
3event_id = PyTango.DeviceProxy.subscribe_event(attributeName,PyTango.EventType.CHANGE_EVENT,callback_function,[],True)
4
5#From inside the device server
6self.set_change_event('State',True,True)

Device Server Internal Objects

Forcing in which host the device is exported

This environment variable must be set before launching the device:

1$ export OMNIORB_USEHOSTNAME=10.0.0.10

Creating a Device Server from ipython

Having defined your device in MyDS.py:

1from MyDS import *
2py = PyTango.PyUtil(['MyDS.py','InstanceName'])
3py.add_TgClass(MyDSClass,MyDS,'MyDS')
4U = PyTango.Util.instance()
5U.server_init()
6U.server_run()

Get the device server admin

NOT TESTED

1 U = PyTango.Util.instance()
2 U.get_dserver_device()

Modify internal polling

Note

It doesn’t work at init_device(); must be done later on in a hook method.

 1 U = PyTango.Util.instance()
 2 admin = U.get_dserver_device()
 3 dir(admin)
 4     [
 5         StartPolling
 6         StopPolling
 7         AddObjPolling
 8         RemObjPolling
 9         UpdObjPollingPeriod
10         DevPollStatus
11         PolledDevice
12     ]
13
14 polled_attrs = {}
15 for st in admin.DevPollStatus(name):
16     lines = st.split('\n')
17     try: polled_attrs[lines[0].split()[-1]]=lines[1].split()[-1]
18     except: pass
19
20 type_ = 'command' or 'attribute'
21 for aname in args:
22 if aname in polled_attrs:
23     admin.UpdObjPollingPeriod([[200],[name,type_,aname]])
24 else:
25     admin.AddObjPolling([[3000],[name,type_,aname]])

Get all polling attributes

The polling of the attributes is recorded in the property_device table of the tango database in the format of a list like [ATTR1,PERIOD1,ATTR2,PERIOD2,…]

The list of polled attributes can be accessed using this method of admin device:

1dp = PyTango.DeviceProxy('dserver/myServerClass/id22')
2polled_attrs = [a.split('\n')[0].split(' ')[-1] for a in dp.DevPollStatus('domain/family/member-01')]

Get the device class object from the device itself

1self.get_device_class()

Get the devices inside a Device Server

 1def get_devs_in_server(self,MyClass=None):
 2     """
 3     Method for getting a dictionary with all the devices running in this server
 4     """
 5     MyClass = MyClass or type(self) or DynamicDS
 6     if not hasattr(MyClass,'_devs_in_server'):
 7         MyClass._devs_in_server = {} #This dict will keep an access to the class objects instantiated in this Tango server
 8     if not MyClass._devs_in_server:
 9         U = PyTango.Util.instance()
10         for klass in U.get_class_list():
11             for dev in U.get_device_list_by_class(klass.get_name()):
12                 if isinstance(dev,DynamicDS):
13                     MyClass._devs_in_server[dev.get_name()]=dev
14     return MyClass._devs_in_server

Identify each attribute inside read_attr_hardware()

 1def read_attr_hardware(self,data):
 2    self.debug("In DynDS::read_attr_hardware()")
 3    try:
 4        attrs = self.get_device_attr()
 5        for d in data:
 6            a_name = attrs.get_attr_by_ind(d).get_name()
 7            if a_name in self.dyn_attrs:
 8                self.lock.acquire() #This lock will be released at the end of read_dyn_attr
 9                self.myClass.DynDev=self #VITAL: It tells the admin class which device attributes are going to be read
10                self.lock_acquired += 1
11        self.debug('DynamicDS::read_attr_hardware(): lock acquired %d times'%self.lock_acquired)
12    except Exception,e:
13        self.last_state_exception = 'Exception in read_attr_hardware: %s'%str(e)
14        self.error('Exception in read_attr_hardware: %s'%str(e))

Device server logging (using Tango logs)

1$ Device_4Impl.
2$
3$ debug_stream ( str )
4$ info_stream ( str )
5$ warning_stream ( str )
6$ error_stream ( str )
7$ fatal_stream ( str )

Or use fandango.Logger object instead

Adding dynamic attributes to a device

1 self.add_attribute(
2     PyTango.Attr( #or PyTango.SpectrumAttr
3         new_attr_name,PyTango.DevArg.DevState,PyTango.AttrWriteType.READ, #or READ_WRITE
4         #max_size or dyntype.dimx #If Spectrum
5         ),
6     self.read_new_attribute, #(attr)
7     None, #self.write_new_attribute #(attr)
8     self.is_new_attribute_allowed, #(request_type)
9     )

Using Database Object

1 import PyTango
2 db = PyTango.Database()

Register a new device server

1 dev = 'SR%02d/VC/ALL'%sector
2 klass = 'PyStateComposer'
3 server = klass+'/'+dev.replace('/','_')
4
5 di = PyTango.DbDevInfo()
6 di.name,di._class,di.server = device,klass,server
7 db.add_device(di)

Remove “empty” servers from database

1 tango = PyTango.Database()
2 [tango.delete_server(s)
3     for s in tango.get_server_list()
4     if all(d.lower().startswith('dserver') for d in tango.get_device_class_list(s))
5 ]

Force unexport of a failing server

You can check using db object if a device is still exported after killed

1$ bool(db.import_device('dserver/HdbArchiver/11').exported)
2$ True

You can unexport this device or server with the following call:

1 db.unexport_server('HdbArchiver/11')

It would normally allow you to restart the server again.

Get all servers of a given class

1 class_name = 'Modbus'
2 list_of_names = ['/'.join((class_name,name)) for name in db.get_instance_name_list(class_name)]

Differences between DB methods:

1get_instance_name_list(exec_name): return names of **instances**
2get_server_list(): returns list of all **executable/instance**
3get_server_name_list(): return names of all **executables**

Get all devices of a server or a given class

The command is:

1 db.get_device_class_list(server_name): return
2 ['device/name/family','device_class']*num_of_devs_in_server

The list returned includes the admin server (dserver/exec_name/instance) that must be pruned from the result:

1 list_of_devs = [dev for dev in db.get_device_class_list(server_name) if '/' in dev and not dev.startswith('dserver')]

Get all devices of a given class from the database

1 import operator
2 list_of_devs = reduce(operator.add,(list(dev for dev in db.get_device_class_list(n) \
3     if '/' in dev and not dev.startswith('dserver')) for n in \
4     ('/'.join((class_name,instance)) for instance in db.get_instance_name_list(class_name)) \
5     ))

Get property values for a list of devices

1 db.get_device_property_list(device_name,'*') : returns list of
2 available properties
3 db.get_device_property(device_name,[property_name]) : return
4 {property_name : value}

1 prop_names = db.get_device_property_list(device_name)
2     ['property1','property2']
3 dev_props = db.get_device_property(device_name,prop_names)
4     {'property1':'first_value' , 'property2':'second_value' }

Get the history (last ten values) of a property

1 [ph.get_value().value_string for ph in tango.get_device_property_history('some/alarms/device','AlarmsList')]
2
3 [['MyAlarm:a/gauge/controller/Pressure>1e-05', 'TempAlarm:a/nice/device/Temperature_Max > 130'],

Get the server for a given device

1 >>> print db.get_server_list('Databaseds/*')
2 ['DataBaseds/2']
3 >>> print db.get_device_name('DataBaseds/2','DataBase')
4 ['sys/database/2']
5 >>> db_dev=PyTango.DeviceProxy('sys/database/2')
6 >>> print db_dev.command_inout('DbImportDevice','et/wintest/01')
7 ([0, 2052], ['et/wintest/01', 'IOR:0100000017000xxxxxx', '4',
8 'WinTest/manu', 'PCTAUREL.esrf.fr', 'WinTest'])

Get the Info of a not running device (exported, host, server)

1 def get_device_info(dev):
2     vals = PyTango.DeviceProxy('sys/database/2').DbGetDeviceInfo(dev)
3     di = dict((k,v) for k,v in zip(('name','ior','level','server','host','started','stopped'),vals[1]))
4     di['exported'],di['PID'] = vals[0]
5     return di

Set property values for a list of devices

Attention , Tango property values are always inserted as lists! {property_name : [ property_value ]}

1 prop_name,prop_value = 'Prop1','Value1'
2 [db.put_device_property(dev,{prop_name:[prop_value]}) for dev in list_of_devs]

Get Starter Level configuration for a list of servers

1 [(si.name,si.mode,si.level) for si in [db.get_server_info(s) for s in list_of_servers]]

Set Memorized Value for an Attribute

1 db.get_device_attribute_property('tcoutinho/serial/01/Baudrate',['__value'])
2 db.put_device_attribute_property('tcoutinho/serial/01/Baudrate',{'__value':VALUE})

Useful constants and enums

 1 In [31]:PyTango.ArgType.values
 2 Out[31]:
 3 {0: PyTango._PyTango.ArgType.DevVoid,
 4  1: PyTango._PyTango.ArgType.DevBoolean,
 5  2: PyTango._PyTango.ArgType.DevShort,
 6  3: PyTango._PyTango.ArgType.DevLong,
 7  4: PyTango._PyTango.ArgType.DevFloat,
 8  5: PyTango._PyTango.ArgType.DevDouble,
 9  6: PyTango._PyTango.ArgType.DevUShort,
10  7: PyTango._PyTango.ArgType.DevULong,
11  8: PyTango._PyTango.ArgType.DevString,
12  9: PyTango._PyTango.ArgType.DevVarCharArray,
13  10: PyTango._PyTango.ArgType.DevVarShortArray,
14  11: PyTango._PyTango.ArgType.DevVarLongArray,
15  12: PyTango._PyTango.ArgType.DevVarFloatArray,
16  13: PyTango._PyTango.ArgType.DevVarDoubleArray,
17  14: PyTango._PyTango.ArgType.DevVarUShortArray,
18  15: PyTango._PyTango.ArgType.DevVarULongArray,
19  16: PyTango._PyTango.ArgType.DevVarStringArray,
20  17: PyTango._PyTango.ArgType.DevVarLongStringArray,
21  18: PyTango._PyTango.ArgType.DevVarDoubleStringArray,
22  19: PyTango._PyTango.ArgType.DevState,
23  20: PyTango._PyTango.ArgType.ConstDevString,
24  21: PyTango._PyTango.ArgType.DevVarBooleanArray,
25  22: PyTango._PyTango.ArgType.DevUChar,
26  23: PyTango._PyTango.ArgType.DevLong64,
27  24: PyTango._PyTango.ArgType.DevULong64,
28  25: PyTango._PyTango.ArgType.DevVarLong64Array,
29  26: PyTango._PyTango.ArgType.DevVarULong64Array}
30
31 In [30]:PyTango.AttrWriteType.values
32 Out[30]:
33 {0: PyTango._PyTango.AttrWriteType.READ,
34  1: PyTango._PyTango.AttrWriteType.READ_WITH_WRITE,
35  2: PyTango._PyTango.AttrWriteType.WRITE,
36  3: PyTango._PyTango.AttrWriteType.READ_WRITE}
37
38 In [29]:PyTango.AttrWriteType.values[3] is PyTango.READ_WRITE
39 Out[29]:True

Using Tango Groups

This example uses PyTangoGroup to read the status of all devices in a Device Server

 1 import PyTango
 2
 3 server_name = 'VacuumController/AssemblyArea'
 4 group = PyTango.Group(server_name)
 5 devs = [d for d in PyTango.Database().get_device_class_list(server_name) if '/' in d and 'dserver' not in d]
 6 for d in devs:
 7     group.add(d)
 8
 9 answers = group.command_inout('Status',[])
10 for reply in answers:
11     print 'Device %s Status is:' % reply.dev_name()
12     print reply.get_data()

About Exceptions

Be aware that I’m not sure about all of this:

1 try:
2     #reason,desc(ription),origin
3     PyTango.Except.throw_exception("TimeWAITBetweenRetries",
4                  "Last communication failed at %s, waiting %s millis"%(time.ctime(self.last_failed),self.ErrorTimeWait),
5                  inspect.currentframe().f_code.co_name)
6 except PyTango.DevFailed,e:
7     if e.args[0]['reason']!='API_AsynReplyNotArrived':
8         PyTango.Except.re_throw_exception(e,"DevFailed Exception",str(e),inspect.currentframe().f_code.co_name)

Passing Arguments to Device command_inout

When type of Arguments is special like DevVarLongStringArray the introduction of arguments is something like:

1 api.manager.command_inout('UpdateSnapComment',[[40],['provant,provant...']])

Using asynchronous commands

 1 cid = self.modbus.command_inout_asynch(command,arr_argin)
 2 while True:
 3     self.debug('Waiting for asynchronous answer ...')
 4     threading.Event().wait(0.1)
 5     #time.sleep(0.1)
 6     try:
 7         result = self.modbus.command_inout_reply(cid)
 8         self.debug('Received: %s' % result)
 9         break
10     except PyTango.DevFailed,e:
11         self.debug('Received DevFailed: %s' %e)
12         if e.args[0]['reason'] != 'API_AsynReplyNotArrived':
13            raise Exception,'Weird exception received!: %s' % e

Setting Attribute Config

 1 for server in astor.values():
 2     for dev in server.get_device_list():
 3         dp = server.get_proxy(dev)
 4         attrs = dp.get_attribute_list()
 5         if dev.rsplit('/')[-1].lower() not in [a.lower() for a in attrs]: continue
 6         conf = dp.get_attribute_config(dev.rsplit('/')[-1])
 7         conf.format = "%1.1e"
 8         conf.unit = "mbar"
 9         conf.label = "%s-Pressure"%dev
10         print 'setting config for %s/%s' % (dev,conf.name)
11         dp.set_attribute_config(conf)

Porting device servers to PyTango

The changes to easily port PyTango devices are:

  • C++ : Replace Device_3Impl with Device_4Impl

  • Python : Replace Device_3Impl with Device_4Impl, PyDeviceClass with DeviceClass and PyUtil with Util.

If you are quite lazy you can add this at the beginning of your $Class.py file (and be still parseable by Pogo):

1 import PyTango
2 if 'PyUtil' not in dir(PyTango):
3     PyTango.Device_3Impl = PyTango.Device_4Impl
4     PyTango.PyDeviceClass = PyTango.DeviceClass
5     PyTango.PyUtil = PyTango.Util

Simplify changes by adding this line

1 if 'PyUtil' not in dir(PyTango):
2 PyTango.PyDeviceClass = PyTango.DeviceClass
3 PyTango.PyUtil = PyTango.Util