KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _start_async method in localstack

Best Python code snippet using localstack_python

robot.py

Source:robot.py

...67	def theta(self):68		return self.pose.theta69	#end def70		71	def _start_async(self):72		if self._thread != None:73			return74		self._thread = threading.Thread(target=self._async_publish)75		self._async = True76		self._thread.start()77	#end def78	79	def _join_async(self):80		if self._thread == None:81			return82		self._async = False83		self._thread.join(1000)84		self._thread = None85	#end def86	def _async_publish(self):87		while(self._async):88			self.velocity_publisher.publish(self.speed)89			self.rate.sleep()90	#end def91	def _pose_callback(self, pose):92		self.pose = pose93		self.pose.x = round(self.pose.x, 4)94		self.pose.y = round(self.pose.y, 4)95		# print '\tturtle: ({:0.4f}, {:0.4f}, {:0.4f})'.format(self.pose.x, self.pose.y, self.pose.theta)96	#end def97	def stop(self):98		self._join_async()99		self.speed.linear.x=0100		self.speed.linear.y=0101		self.speed.linear.z=0102		self.speed.angular.x = 0103		self.speed.angular.y = 0104		self.speed.angular.z = 0105		self.velocity_publisher.publish(self.speed)106	#end def107	def advance(self, speed=LINEAR_SPEED):108		self.speed.angular.z = 0109		self.speed.linear.x = speed110		# self.velocity_publisher.publish(self.speed)111		self._start_async()112	#end def113	def spin(self, speed=ANGULAR_SPEED):114		#Converting from degrees to radians115		speed = 10*speed*PI/180.0116		self.speed.angular.z = speed117		self.speed.linear.x = 0118		# self.velocity_publisher.publish(self.speed)119		self._start_async()120	#end def121		122	def turn(self, speed=ANGULAR_SPEED):123		self.rotate(180, speed)124	#end def125	def rotate(self, degrees, speed=ANGULAR_SPEED):126		self._join_async()127		#Converting from degrees to radians128		speed = 10*speed*PI/180.0129		angle = self.pose.theta + degrees*PI/180.0130		step = abs(speed * 1.0/self._rate)131		diff = angle - self.pose.theta132		self.speed.angular.z = math.copysign(speed, diff)133		i = 0134		while(abs(diff) > 0.01 ):135			if i > 200: return136			i+=1137			diff = angle - self.pose.theta138			if abs(diff) < step:139				self.speed.angular.z = speed * diff/step140			self.velocity_publisher.publish(self.speed)141			self.rate.sleep()142		self.stop()143	#end def144	def step(self):145		self._join_async()146		self.speed.linear.x = LINEAR_SPEED147		self.velocity_publisher.publish(self.speed)148		for i in range(1, self._rate):149			self.rate.sleep()150		self.stop()151	#end def152	def circle(self, linear=LINEAR_SPEED, angular=ANGULAR_SPEED):153		self.speed.linear.x = linear154		self.speed.angular.z = 10*angular*PI/180.0155		self.velocity_publisher.publish(self.speed)156		self._start_async()157	#end def158	def idle(self):159		self.rate.sleep()160	#end def161#end class162if __name__ == '__main__':163	robot = Robot()164	robot.rotate(90)165	robot.rotate(-90)166	robot.turn()167	robot.step()168	robot.spin()169	time.sleep(5)170	robot.circle()...

aio_manager.py

Source:aio_manager.py

...3class AioManager:4    AIOSESSION = None5    @classmethod6    def start(cls, func, **kwargs):7        asyncio.run(cls._start_async(func, **kwargs))8    @classmethod9    async def _start_async(cls, func, **kwargs):10        try:11            cls.AIOSESSION = aiohttp.ClientSession(raise_for_status=True)12            await func(**kwargs)13        finally:14            if cls.AIOSESSION:...

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.