How to use _trigger_event method in ATX

Best Python code snippet using ATX Github


Full Screen

1import roslib; roslib.load_manifest('smach_ros')2import rospy3import threading4import traceback5import smach6__all__ = ['MonitorState']7class MonitorState(smach.State):8 """9 A state that will check a given ROS topic with a condition function.10 """11 def __init__(self, topic, msg_type, cond_cb, max_checks=-1,input_keys = [],output_keys=[]):12 """State constructor13 @type topic string14 @param topic the topic to monitor15 @type msg_type a ROS message type16 @param msg_type determines the type of the monitored topic17 @type max_checks int18 @param max_checks the number of messages to receive and evaluate. If cond_cb returns False for any19 of them, the state will finish with outcome 'invalid'. If cond_cb returns True for 20 all of them, the outcome will be 'valid'21 22 """23 smach.State.__init__(24 self,25 outcomes=['valid','invalid','preempted'],26 input_keys = input_keys,27 output_keys = output_keys)28 self._topic = topic29 self._msg_type = msg_type30 self._cond_cb = cond_cb31 self._max_checks = max_checks32 self._n_checks = 033 self._trigger_event = threading.Event()34 def execute(self, ud):35 # If prempted before even getting a chance, give up.36 if self.preempt_requested():37 self.service_preempt()38 return 'preempted'39 self._n_checks = 040 self._trigger_event.clear()41 self._sub = rospy.Subscriber(self._topic, self._msg_type, self._cb, callback_args=ud)42 self._trigger_event.wait()43 self._sub.unregister()44 if self.preempt_requested():45 self.service_preempt()46 return 'preempted'47 if self._max_checks > 0 and self._n_checks >= self._max_checks:48 return 'valid'49 return 'invalid'50 def _cb(self,msg,ud) :51 try:52 if self._cond_cb(ud, msg):53 self._n_checks +=154 else:55 self._trigger_event.set()56 except Exception as e:57 rospy.logerr("Error thrown while executing condition callback %s: %s" % (str(self._cond_cb), e))58 self._trigger_event.set()59 60 if (self._max_checks > 0 and self._n_checks >= self._max_checks):61 self._trigger_event.set()62 def request_preempt(self):63 smach.State.request_preempt(self)...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:


You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run ATX automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?