Get the device server admin
U = PyTango.Util.instance()
U.get_dserver_device()
Get the attribute list
The MultiAttribute object is not iterable, but a workaround like this works:
U.server_init()
dev = U.get_device_by_name('test/sim/tonto3')
attrs = dev.get_device_attr().get_attribute_list()
attrlist = ['VAR3', 'State', 'Status']
Get all polled attributes
The polling of attributes is recorded in the property_device table of the tango database in the format of a list like[ATTR1,PERIOD1,ATTR2,PERIOD2,...]
It can be obtained with:
U = PyTango.Util.instance()
dev = U.get_device_by_name('test/sim/srj')
l = dev.get_polled_attr()
Out[82]: ['a', '60000\x00', 'poll', '30000']
import re
d = dict(zip(l[::2],map(lambda s:re.split('[^0-9]',s)[0],l[1::2])))
{'a': '60000', 'poll': '30000'}
Disable State update and/or Attribute reading when using Attribute Config/Quality
If an attribute has Alarm range configured every State() call from a client will trigger a read_attribute() and setState=ALARM if the quality is ALARM.
To avoid this, these two methods must be overriden:
def dev_state(self): #Tango>=7
return self.get_state()
def State(self): #Tango<=6
return self.get_state()
Get/Set events/attribute configuration
The events configuration for an attribute can be get as:
#Do not omit the first line as dev object may change after resetting event values
dev = U.get_device_by_name('test/test/test-03')
ac = dev.get_attribute_config_3('Variable')[0]
print ac.event_prop.ch_event.abs_change
Out[292]: 'Not specified'
And modified:
ac.event_prop.ch_event.abs_change = '0.033'
dev.set_attribute_config_3(ac)
set_change_event
Type: instancemethod
Base Class: <type 'instancemethod'>
String Form: <bound method PySignalSimulator.set_change_event of PySignalSimulator(test/test/test-03)>
Namespace: Interactive
Docstring:
set_change_event(self, attr_name, implemented, detect=True) -> None
Set an implemented flag for the attribute to indicate that the server fires
change events manually, without the polling to be started.
If the detect parameter is set to true, the criteria specified for the
change event are verified and the event is only pushed if they are fullfilled.
If detect is set to false the event is fired without any value checking!
Parameters :
- attr_name : (str) attribute name
- implemented : (bool) True when the server fires change events manually.
- detect : (bool) Triggers the verification of the change event properties
when set to true. Default value is true.
Return : None
To send events only when polling from jive:
set_change_event('attr_name',False,True)
To send events manually but only if configured in jive:
set_change_event('attr_name',True,True)
To have only manual-events:
set_change_event('attr_name',True,False)
The event configuration for the attribute can be got using:
ac = self.get_attribute_config_3(aname)[0]
try: cabs = float(ac.event_prop.ch_event.abs_change)
except: cabs = 0
try: crel = float(ac.event_prop.ch_event.rel_change)
except: crel = 0
push_change_event
The standard call will use the attribute,value,time,quality arguments:
...
if self.check_attribute_events(aname) and self.check_changed_event(aname,result):
self.info('>'*80)
self.info('Pushing %s event!: %s(%s)'%(aname,type(result),result))
self.push_change_event(aname,
getattr(result,'value',result),
getattr(result,'time',time.time()),
getattr(result,'quality',PyTango.AttrQuality.ATTR_VALID))
Be careful!, the push_change_event must be called BEFORE calling attr.set_value(...):
def read_CurrentTime(self, attr=None):
print('In read_CurrentTime(%s,%s)'%(attr,self.UseEvents))
t = time.time()
self.push_change_event('CurrentTime',t)
attr.set_value(t)
return attr
The full description of the method:
push_change_event(self, attr_name) -> None
push_change_event(self, attr_name, except) -> None
push_change_event(self, attr_name, data, dim_x = 1, dim_y = 0) -> None
push_change_event(self, attr_name, str_data, data) -> None
push_change_event(self, attr_name, data, time_stamp, quality, dim_x = 1, dim_y = 0) -> None
push_change_event(self, attr_name, str_data, data, time_stamp, quality) -> None
Push a change event for the given attribute name.
The event is pushed to the notification daemon.
Parameters:
- attr_name : (str) attribute name
- data : the data to be sent as attribute event data. Data must be compatible with the
attribute type and format.
for SPECTRUM and IMAGE attributes, data can be any type of sequence of elements
compatible with the attribute type
- str_data : (str) special variation for DevEncoded data type. In this case 'data' must also
be a str.
- except: (DevFailed) Instead of data, you may want to send an exception.
- dim_x : (int) the attribute x length. Default value is 1
- dim_y : (int) the attribute y length. Default value is 0
- time_stamp : (double) the time stamp
- quality : (AttrQuality) the attribute quality factor
Throws : DevFailed If the attribute data type is not coherent.
Get all 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 type(self) or DynamicDS
if not hasattr(MyClass,'_devs_in_server'):
MyClass._devs_in_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
Modify internal polling
U = PyTango.Util.instance()
dev = U.get_device_by_name('test/sim/srj')
NOTE:It doesn't work at init_device(); must be done later on in a hook method
U = PyTango.Util.instance()
admin = U.get_dserver_device()
dir(admin)
[
StartPolling
StopPolling
AddObjPolling
RemObjPolling
UpdObjPollingPeriod
DevPollStatus
PolledDevice
]
polled_attrs = {}
for st in admin.DevPollStatus(name):
lines = st.split('\n')
try: polled_attrs[lines[0].split()[-1]]=lines[1].split()[-1]
except: pass
type_ = 'command' or 'attribute'
for aname in args:
if aname in polled_attrs:
admin.UpdObjPollingPeriod([[200],[name,type_,aname]])
else:
admin.AddObjPolling([[3000],[name,type_,aname]])
Get the device class object from the device itself
self.get_device_class()
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()
self.myClass.DynDev=self
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))
Device server logging (using Tango logs)
PyTango.
<blockquote>
Device_4Impl.
<blockquote>
debug_stream ( str )
info_stream ( str )
warning_stream ( str )
error_stream ( str )
fatal_stream ( str )
</blockquote>
</blockquote>
Or use fandango.Logger object instead ...
Adding dynamic attributes to a device
NOTE: Always use a static bound method (e.g. staticmethod(self.read_dyn_attr)) as a read method! , if you use lambdas you'll get cross-reading between devices.
NOTE: dyn_attr() has to be called always AFTER init_device() has finished; to do this a dyn_attr(dev_list) hook is called in the DeviceClass once all init_device have finished.
...
def dyn_attr(self):
...
self.add_attribute(
PyTango.Attr( #or PyTango.SpectrumAttr
new_attr_name,PyTango.DevArg.DevState,PyTango.AttrWriteType.READ, #or READ_WRITE
#max_size or dyntype.dimx #If Spectrum
),
self.read_new_attribute, #(attr)
None, #self.write_new_attribute #(attr)
self.is_new_attribute_allowed, #(request_type)
)
...
class MyDeviceClass(PyTango.DeviceClass): #It fails using PyDeviceClass or PyUtil!!!
...
def dyn_attr(self,dev_list):
print ("In PyAlarmClass.dyn_attr(%s)"%dev_list)
for dev in dev_list:
dev.dyn_attr()
...
Setting up manually the attribute configuration (and default polling)
attr_list = {
'A':
[[PyTango.DevBoolean,
PyTango.SCALAR,
PyTango.READ],
{
'label':"Lalala",
'unit':"U",
'format':"%d",
'max value':5,
'min value':-5,
'max alarm':3,
'min alarm':0,
'max warning':2,
'min warning':1,
'delta time':0.1,
'delta val':0.1,
'Memorized':"true"
} ],
}
Setting Polling period:
attr_list = {
'A':
[[PyTango.DevBoolean,
PyTango.SCALAR,
PyTango.READ],
{
'Polling period':2000,
} ],
Setting Memorized:
'Frequency':
[[PyTango.DevDouble,
PyTango.SCALAR,
PyTango.WRITE],
{
'unit':"Hz",
'description':"This attribute defines the frequency between steps during the gradual change",
'Memorized':"true",
} ],
or, using dynamic attributes and setting it programatically:
attrib,format,unit = PyTango.Attr('Output_%d'%((i)),PyTango.DevShort, PyTango.READ),'%1.1f','%'
print 'Creating attribute %s ...'%attrib
props = PyTango.UserDefaultAttrProp(); props.set_format(format); props.set_unit(unit)
attrib.set_default_properties(props)
rfun = (lambda s,a,index=i: self.outputAttr(index,a))
self.add_attribute(attrib,rfun,None,(lambda s,req_type,index=i: True))
Methods for read/write attributes, is_attribute_allowed()
#------------------------------------------------------------------
# Read ProtectSetPoint attribute
#------------------------------------------------------------------
def read_IProtectSetPoints(self, attr):
self.debug( "In "+ self.get_name()+ "::read_IProtectSetPoint()")
# Add your own code here
#setpoints = lambda d:','.join(s[3:] for s in map(astor.proxies[d].SendCommand,['#171?\n\r','#172?\n\r']))
protect1 = self.readCommand('HV1 IProtect',long)
protect2 = self.readCommand('HV2 IProtect',long)
attr_IProtectSetPoint_read = [protect1,protect2]
attr.set_value(attr_IProtectSetPoint_read, len(attr_IProtectSetPoint_read))
#---- PressureSetPoints attribute State Machine -----------------
def is_IProtectSetPoints_allowed(self, req_type):
if self.get_state() in [PyTango.DevState.ON,PyTango.DevState.ALARM]:
# End of Generated Code
# Re-Start of Generated Code
if req_type == PyTango.AttReqType.WRITE_REQ:
return False
return True
#------------------------------------------------------------------
# Write ProtectSetPoint attribute
#------------------------------------------------------------------
def write_IProtectSetPoints(self, attr):
self.debug( "In "+ self.get_name()+ "::write_IProtectSetPoint()")
data=[]
attr.get_write_value(data)
# Add your own code here
#[proxies['%s22/vc/ipct-%02d'%(d,i)].SendCommand('#1711.0E-07') for d,i,j in (('fe',1,1),('fe',1,2),('bl',1,1),('bl',1,2),('bl',2,1))]
if len(data)!=2:
PyTango.Except.throw_exception('WrongDataLenght','Data length should be equal to 2',
'write_IProtectSetPoint')
data = map(int,data)
self.SendCommand([
self.packMultiGauge(1,self.HighVoltageCommands['Iprotect'],'%05d'%data[0]),
self.packMultiGauge(2,self.HighVoltageCommands['Iprotect'],'%05d'%data[1])
])
No comments:
Post a Comment