How To PyTango
MOVED TO SITES.GOOGLE
-
- Before anything else
- Device Server Internal Objects
-
Using Database Object
- Get all tango devices matching a given name
- Get all servers of a given class
- Get all devices of a server or a given class
- Get all devices of a given class from the database
- Get property values for a list of devices
- Get the server for a given devic
- Set property values for a list of devices
- Get Starter Level configuration for a list of servers
- Set Memorized Value for an Attribute
- Useful constants and enums
- Using Tango Groups
- Passing Arguments to Device command_inout
- Using asynchronous commands
- Setting Attribute Config
- Porting device servers to PyTango7
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