Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use list_queues method in localstack

Best Python code snippet using localstack_python

CustomController.py

Source:CustomController.py

1"""e-puck_touch_obstacles controller."""2# You may need to import some classes of the controller module. Ex:3#  from controller import Robot, Motor, DistanceSensor4from controller import Robot5from datetime import datetime6from datetime import timedelta7import sys8TIME_STEP = 649MAX_SPEED = 6.2810# create the Robot instance.11robot = Robot()12# get the time step of the current world.13timestep = int(robot.getBasicTimeStep())14# You should insert a getDevice-like function in order to get the15# instance of a device of the robot. Something like:16# initialize devices17ps = []18psNames = [19    'ps0', 'ps1', 'ps2', 'ps3',20    'ps4', 'ps5', 'ps6', 'ps7'21]22leds = []23ledNames = [24    'led0', 'led1', 'led2', 'led3', 'led4',25    'led5', 'led6', 'led7', 'led8', 'led9'26]27for i in range(len(ledNames)):28    leds.append(robot.getLED(ledNames[i]))29for i in range(8):30    ps.append(robot.getDistanceSensor(psNames[i]))31    ps[i].enable(TIME_STEP)32    33leftMotor = robot.getMotor('left wheel motor')34rightMotor = robot.getMotor('right wheel motor')35leftMotor.setPosition(float('inf'))36rightMotor.setPosition(float('inf'))37def lightAllLeds(num):38    diff = qtd_blocos_moveis_encontrados if num == 0 else 039    for k in range(3):40        for i in range(len(leds) - diff):41            leds[i].set(num)42        43#leftMotor.setVelocity(0.1 * MAX_SPEED)44#rightMotor.setVelocity(0.1 * MAX_SPEED)45#  motor = robot.getMotor('motorname')46#  ds = robot.getDistanceSensor('dsname')47#  ds.enable(timestep)48list_queues = []49for i in range(8):50    list_queues.append([])51    52avgArray = []53for i in range(8):54    avgArray.append([])55time_to_stop_turning = datetime.now()56qtd_blocos_moveis_encontrados = 057    58# Main loop:59# - perform simulation steps until Webots is stopping the controller60while robot.step(timestep) != -1:61    # Read the sensors:62    psValues = []63    for i in range(8):64        psValues.append(ps[i].getValue())65    # Enter here functions to read sensor data, like:66    # detect obstacles67    non_solid_sensor_value = 80068    sensor_value = 70069    min_percentage = 0.0570    right_obstacle = psValues[0] > sensor_value or psValues[1] > sensor_value or psValues[2] > sensor_value71    left_obstacle = psValues[5] > sensor_value or psValues[6] > sensor_value or psValues[7] > sensor_value72    73    for i in range(len(list_queues)):74        list_queues[i].append(psValues[i])75        #print('list_queues[{}] len({}) {}'.format(i, len(list_queues[i]), list_queues[i]))76        if len(list_queues[i]) > 50:77            list_queues[i].pop(0)78        79    #print(list_queues)80    #print('=========================================')81    82    #  val = ds.getValue()83    for i in range(len(psValues)):84        psValue = psValues[i]85        if psValue > non_solid_sensor_value and not ( right_obstacle or left_obstacle) and not (time_to_stop_turning > datetime.now()):86            qtd_blocos_moveis_encontrados += 1;87            lightAllLeds(1)88            print('{} collided with non solid object in sensor {}'.format(datetime.now(), i))89    #print(psValues)90    # Process sensor data here.91    # initialize motor speeds at 50% of MAX_SPEED.92    percentage = 193    front_sensor = max([psValues[0], psValues[1], psValues[6], psValues[7]])94    if front_sensor >= sensor_value:95        percentage = min_percentage96    elif front_sensor <= 100:97        percentage = 198    else:99        valor_imaginario = front_sensor - 100100        resultado = (valor_imaginario * 100)/sensor_value101        #print('resultado {}'.format(resultado))102        percentage = min_percentage + (1 - min_percentage) * (resultado/100)103    #print(percentage)104        105        106    # Vetor de mÃ©dias107    for i in range(len(list_queues)):108        avgArray[i].append(sum(list_queues[i]) / len(list_queues[i]))109        #print('list_queues[{}] len({}) {}'.format(i, len(list_queues[i]), list_queues[i]))110        if len(avgArray[i]) > 100:111            avgArray[i].pop(0)112        113    varArray = [0] * len(list_queues)114    for i in range(len(avgArray)):115        avg = sum(avgArray[i]) / len(avgArray[i])116        #print(avg)117        var = sum((x-avg)**2 for x in avgArray[i]) / len(avgArray[i])118        #print(var)119        varArray[i] = var120        121    #print(varArray)122    maior = 0123    for avg in avgArray:124        maior = max(maior, max(avg))125    #print(maior)126    127    stuck = max(varArray) < 1 and maior > 100128    if stuck:129        time_to_stop_turning = datetime.now() + timedelta(0, 2)130        #print('{} {} collision========================'.format(datetime.now(), time_to_stop_turning))131    132    # Carrin n anda atÃ© os calculos das mÃ©dias terminarem133    if not max(varArray) > 0:134        percentage = 0135        136    leftSpeed  = percentage * MAX_SPEED137    rightSpeed = percentage * MAX_SPEED138    # modify speeds according to obstacles139    #if time_to_stop_turning > datetime.now():140        #print('yup turning')141    if left_obstacle or stuck or time_to_stop_turning > datetime.now():142        # turn right143        leftSpeed  = percentage * MAX_SPEED144        rightSpeed = -percentage * MAX_SPEED145    elif right_obstacle:146        # turn left147        leftSpeed  = -percentage * MAX_SPEED148        rightSpeed = percentage * MAX_SPEED149        #lightAllLeds(1)150    else:151        lightAllLeds(0)152        153    lightAllLeds(0)154    # write actuators inputs155    leftMotor.setVelocity(leftSpeed)156    rightMotor.setVelocity(rightSpeed)157    158    #print(psValues)159    160    # Enter here functions to send actuator commands, like:161    #  motor.setPosition(10.0)162    pass...

sqs_barrel.py

Source:sqs_barrel.py

...23        'list_queues'24    ])25    def __init__(self, oil, **kwargs):26        super().__init__(oil, **kwargs)27    def list_queues(self):28        if self.cache.get('list_queues'):29            return self.cache['list_queues']30        items = {}31        for region, client in self.clients.items():32            items[region] = []33            response = client.list_queues()34            items[region].extend(response['QueueUrls'])35        return items36    def get_queue_attributes(self):37        if self.cache.get('get_queue_attributes'):38            return self.cache['get_queue_attributes']39        items = {}40        for region, client in self.clients.items():41            items[region] = {}42            for queue in self.tap('list_queues')[region]:43                response = client.get_queue_attributes(44                    QueueUrl=queue45                )46                items[region][queue] = response['Attributes']47        return items

list_queues.py

Source:list_queues.py

...16# Create SQS client17sqs = boto3.client('sqs')1819# List SQS queues20response = sqs.list_queues()2122print(response['QueueUrls'])2324# snippet-end:[sqs.python.list_queues.complete]25# snippet-comment:[These are tags for the AWS doc team's sample catalog. Do not remove.]26# snippet-sourcedescription:[list_queues.py demonstrates how to list your current list of queues.]27# snippet-keyword:[Python]28# snippet-keyword:[AWS SDK for Python (Boto3)]29# snippet-keyword:[Code Sample]30# snippet-keyword:[Amazon Simple Queue Service]31# snippet-service:[sqs]32# snippet-sourcetype:[full-example]33# snippet-sourcedate:[2018-12-26]34# snippet-sourceauthor:[jschwarzwalder (AWS)]

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.