How to use _trigger_event method in ATX

Best Python code snippet using ATX

monitor_state.py

Source:monitor_state.py

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)...

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.