PyTango: Configure events using the DeviceProxy object

Events

Creating an event callback

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

def callback_function(*args): print args

(
  EventData(attr_name = 'tango://tbl0901.cells.es:10000/bl09/vc/all/state', attr_value = None, device = PyStateComposer(bl09/vc/all), 
      err = True, errors = (DevError(desc = 'Event channel is not responding anymore, maybe the server or event system is down', 
      origin = 'EventConsumer::KeepAliveThread()', reason = 'API_EventTimeout', 
      severity = PyTango._PyTango.ErrSeverity.ERR),), 
      event = 'change', 
      reception_date = TimeVal(tv_nsec = 0, tv_sec = 1391531528, tv_usec = 597630)),
)

Configuring an event

#From the client side
#subscribe_event(attr_name, event_type, cb_or_queuesize, filters=[], stateless=False, extract_as=PyTango._PyTango.ExtractAs.Numpy)
dp = PyTango.DeviceProxy(deviceName)
event_id = dp.subscribe_event(attributeName,PyTango.EventType.CHANGE_EVENT,callback_function,[],True)

#From inside the device server
self.set_change_event('State',True,True) 

dp.unsubscribe_event(event_id)

Pushing the event

Device Monitor / Test events from a device

$ devicemonitor mot01 state position

code:

#!/usr/bin/env python

""" 
usage: devicemonitor <device name> [ <attr name> ]*
""" 

import sys
import time
import PyTango

m = PyTango.DeviceProxy(sys.argv[1])

cb = PyTango.utils.EventCallBack()

if len(sys.argv) == 2:
    attrs = &quot;state&quot;,
else:
    attrs = map(str.strip, sys.argv[2:])

attrs = map(str.strip, attrs)

eids = [m.subscribe_event(attr, PyTango.EventType.CHANGE_EVENT, cb) for attr in attrs]

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    print "Finsihed monitoring"