Personal tools
You are here: Home Members srubio's Home HowToS How To PyTango

How To PyTango

MOVED TO SITES.GOOGLE

 

 

Before anything else

import PyTango


Device Server Internal Objects

Creating a Device Server from ipython

1: from AlbaPLC import *
2: py = PyTango.PyUtil(['AlbaPLC.py','SR04_VC'])
3: py.add_TgClass(AlbaPLCClass,AlbaPLC,'AlbaPLC')
4: U = PyTango.Util.instance()
5: U.server_init()
6: U.server_run()

Get the device server admin

NOT TESTED

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

Modify internal polling

NOT TESTED

U = PyTango.Util.instance()
admin = U.get_dserver_device()
dir(admin)
[
StartPolling
StopPolling
AddObjPolling
RemObjPolling
UpdObjPollingPeriod
DevPollStatus
PolledDevice
]

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:

dp = PyTango.DeviceProxy('dserver/AlbaPLC/id22')
polled_attrs = [a.split('\n')[0].split(' ')[-1] for a in dp.DevPollStatus('ID22/EPS/PLC-01')]

Get the device class object from the device itself

self.get_device_class()

Get the devices inside a Device Server

    def get_devs_in_server(self,MyClass=None):
        """
        Method for getting a dictionary with all the devices running in this server
        """
        MyClass = MyClass or DynamicDS
        if not hasattr(MyClass,'_devs_in_server'):
            MyClass._devs_in_server = {} #This dict will keep an access to the class objects instantiated in this Tango server
        if not MyClass._devs_in_server:
            U = PyTango.Util.instance()
            for klass in U.get_class_list():
                for dev in U.get_device_list_by_class(klass.get_name()):
                    if isinstance(dev,DynamicDS):
                        MyClass._devs_in_server[dev.get_name()]=dev
        return MyClass._devs_in_server

Identify each attribute inside read_attr_hardware()

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

Using Database Object

import PyTango
db = PyTango.Database()

Get all tango devices matching a given name

Note: this will return both exported and unexported devices, for just exported use PyTango.Database().get_device_exported('target*')

import PyTango
db = PyTango.Database()

target = '*/*vc*/*'
assert target.count('/')==2, 'target must be a device name'
td,tf,tm = target.split('/')

results = []
domains = db.get_device_domain(target)
for d in domains:
    families = db.get_device_family(d+'/'+tf+'/'+tm)
    for f in families:
        members = db.get_device_member((d+'/'+f+'/'+tm))
        for m in members:
            results.append('/'.join((d,f,m)))

In [30]:all([ 'vc' in s.lower() for s in results])
Out[30]:True

Get all servers of a given class

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

Differences between DB methods:

  • get_instance_name_list(exec_name): return names of instances
  • get_server_list(): returns list of all executable/instance
  • get_server_name_list(): return names of all executables

Get all devices of a server or a given class

The command is:

  • db.get_device_class_list(server_name): return ['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:

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

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

Get property values for a list of devices

  • db.get_device_property_list(device_name,'*') : returns list of available properties
  • db.get_device_property(device_name,[property_name]) : return {property_name : value}
prop_names = db.get_device_property_list(device_name)
   ['property1','property2']
dev_props = db.get_device_property(device_name,prop_names)
   {'property1':'first_value' , 'property2':'second_value' }

Get the server for a given devic

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

Set property values for a list of devices

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

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

Get Starter Level configuration for a list of servers

[(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

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

Useful constants and enums

In [31]:PyTango.ArgType.values
Out[31]:
{0: PyTango._PyTango.ArgType.DevVoid,
 1: PyTango._PyTango.ArgType.DevBoolean,
 2: PyTango._PyTango.ArgType.DevShort,
 3: PyTango._PyTango.ArgType.DevLong,
 4: PyTango._PyTango.ArgType.DevFloat,
 5: PyTango._PyTango.ArgType.DevDouble,
 6: PyTango._PyTango.ArgType.DevUShort,
 7: PyTango._PyTango.ArgType.DevULong,
 8: PyTango._PyTango.ArgType.DevString,
 9: PyTango._PyTango.ArgType.DevVarCharArray,
 10: PyTango._PyTango.ArgType.DevVarShortArray,
 11: PyTango._PyTango.ArgType.DevVarLongArray,
 12: PyTango._PyTango.ArgType.DevVarFloatArray,
 13: PyTango._PyTango.ArgType.DevVarDoubleArray,
 14: PyTango._PyTango.ArgType.DevVarUShortArray,
 15: PyTango._PyTango.ArgType.DevVarULongArray,
 16: PyTango._PyTango.ArgType.DevVarStringArray,
 17: PyTango._PyTango.ArgType.DevVarLongStringArray,
 18: PyTango._PyTango.ArgType.DevVarDoubleStringArray,
 19: PyTango._PyTango.ArgType.DevState,
 20: PyTango._PyTango.ArgType.ConstDevString,
 21: PyTango._PyTango.ArgType.DevVarBooleanArray,
 22: PyTango._PyTango.ArgType.DevUChar,
 23: PyTango._PyTango.ArgType.DevLong64,
 24: PyTango._PyTango.ArgType.DevULong64,
 25: PyTango._PyTango.ArgType.DevVarLong64Array,
 26: PyTango._PyTango.ArgType.DevVarULong64Array}

In [30]:PyTango.AttrWriteType.values
Out[30]:
{0: PyTango._PyTango.AttrWriteType.READ,
 1: PyTango._PyTango.AttrWriteType.READ_WITH_WRITE,
 2: PyTango._PyTango.AttrWriteType.WRITE,
 3: PyTango._PyTango.AttrWriteType.READ_WRITE}

In [29]:PyTango.AttrWriteType.values[3] is PyTango.READ_WRITE
Out[29]:True

Using Tango Groups

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

import PyTango

server_name = 'VacuumController/AssemblyArea'
group = PyTango.Group(server_name)
devs = [d for d in PyTango.Database().get_device_class_list(server_name) if '/' in d and 'dserver' not in d]
for d in devs:
    group.add(d)

answers = group.command_inout('Status',[])
for reply in answers:
    print 'Device %s Status is:' % reply.dev_name()
    print reply.get_data()

Passing Arguments to Device command_inout

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

In [25]:api.manager.command_inout('UpdateSnapComment',[[40],['provant,provant...']])

Using asynchronous commands

cid = self.modbus.command_inout_asynch(command,arr_argin)
while True:
    self.debug('Waiting for asynchronous answer ...')
    threading.Event().wait(0.1)
    #time.sleep(0.1)
    try:
        result = self.modbus.command_inout_reply(cid)
        self.debug('Received: %s' % result)
        break
    except PyTango.DevFailed,e:
        self.debug('Received DevFailed: %s' %e)
        if e.args[0]['reason'] != 'API_AsynReplyNotArrived':
        raise Exception,'Weird exception received!: %s' % e

Setting Attribute Config

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

Porting device servers to PyTango7

Serial , Modbus , !PyPLC, !AlbaPLC ported to Tango7 and !SuSE 11, installed in ictlael01. The changes done have been:

  • C++ : Replace Device_3Impl with Device_4Impl
  • Python : Replace Device_3Impl with Device_4Impl, PyDeviceClass? with DeviceClass? and PyUtil? with Util. The old devices have been tagged as Tango6 inside svn/branches/tango_ds.Tango6 NOTE: AlbaPLC crashes with Seg fault if the device is not defined in DB.

Simplify changes by adding this line

if 'PyUtil' not in dir(PyTango): 
    PyTango.PyDeviceClass = PyTango.DeviceClass
    PyTango.PyUtil = PyTango.Util
Document Actions