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

How to use wait_up method in localstack

Best Python code snippet using localstack_python

measure_scan.py

Source:measure_scan.py

...114    rotate_stage = None115    ##USB3 - THORLABS116    try:117        rotate_stage = HDR50(serial_port="/dev/ttyUSB3", serial_number="40106754", home=True, swap_limit_switches=True)118        rotate_stage.wait_up()119    except:120        print(colored('Error in connecting to Thorlabs Motor!', 'red'))121    ##USB2 - ORIENTAL MOTORS122    try:123        r_stage = AZD_AD(port="/dev/ttyUSB2")124    except:125        print(colored('Error in connecting to Oriental Motor!', 'red'))126127    rotate_stage.move_relative(-90)128    rotate_stage.wait_up()129130    r_stage.moveToHome(slave_address)131    time.sleep(5)132    print(colored("Motor setup finished", 'green'))133    LD = PMX70_1A('10.25.123.249')134    LD.connect_instrument()135    LD.set_volt_current(voltage, 0.02)136    #Warm up LD137    print('Warm up LD (10 min)')138    for i in tqdm(range(600)):139        time.sleep(1)140    return rotate_stage, r_stage, LD141142def setup_top_devices(rotate_slave_address, r_slave_address, voltage):143    print(colored("Setting up motors...", 'green'))144    stage = None145    ##USB2 - ORIENTAL MOTORS146    try:147        stage = AZD_AD(port="/dev/ttyUSB2")148    except:149        print(colored('Error in connecting to Oriental Motor!', 'red'))150151    stage.moveToHome(rotate_slave_address)152    time.sleep(5)153    stage.moveToHome(r_slave_address)154    time.sleep(5)155    print(colored("Motor setup finished", 'green'))156    LD = PMX70_1A('10.25.123.249')157    LD.connect_instrument()158    LD.set_volt_current(voltage, 0.02)159    #Warm up LD160    print('Warm up LD (10 min)')161    for i in tqdm(range(600)):162        time.sleep(1)163    return stage164165#############################################################################166167def measure_brscan(dir_sig, dir_ref, degg, nevents, voltage,168                    theta_step, theta_max, theta_scan_points,169                    r_step, r_max, r_scan_points,170                    mtype='stamp', measure_side='bottom'):171    print('brscan')172    slave_address = 1173    rotate_stage, r_stage, LD = setup_bottom_devices(slave_address, voltage)174    ##initialize reference settings175    reference_pmt_channel = 1176    scope = setup_reference(reference_pmt_channel)177178    for theta_point in theta_scan_points:179        print(r'-- $\theta$:' + f'{theta_point} --')180        try:181                        r_scan_points, mtype=mtype, forward_backward='forward')182        ##when finished, return motor to home183        r_stage.moveToHome(slave_address)184        print('r_stage homing')185        time.sleep(20)186187        # measure_r_steps(dir_sig, degg, nevents, r_stage, slave_address, theta_point+180, r_step, 188        #                 r_scan_points, mtype=mtype, forward_backward='backward') 189190        # r_stage.moveToHome(slave_address)191        # print('r_stage homing')192        # time.sleep(20)193        194        reference_pmt_file = os.path.join(dir_ref, f'ref_{theta_point}.hdf5')195        measure_reference(reference_pmt_file, scope, reference_pmt_channel)196197        rotate_stage.move_relative(theta_step)198        rotate_stage.wait_up()199    rotate_stage.move_relative(-270)200    rotate_stage.wait_up()201202203204def measure_bzscan(dir_sig, dir_ref, degg, nevents, voltage,205                    theta_step, theta_max, theta_scan_points,206                    z_step, z_max, z_scan_points,207                    mtype='stamp', measure_side='bottom'):208    print('bzscan')209    slave_address = 2210    rotate_stage, r_stage, LD = setup_bottom_devices(slave_address, voltage)211    ##initialize reference settings212    reference_pmt_channel = 1213    scope = setup_reference(reference_pmt_channel)214215    for theta_point in theta_scan_points:216217        print(r'-- $\theta$:' + f'{theta_point} --')218        measure_r_steps(dir_sig, degg, nevents, r_stage, slave_address, theta_point, z_step, 219                        z_scan_points, mtype=mtype, forward_backward='forward')220        reference_pmt_file = os.path.join(dir_ref, f'ref_{theta_point}.hdf5')221        measure_reference(reference_pmt_file, scope, reference_pmt_channel)222223        r_stage.moveToHome(slave_address)224        print('r_stage homing')225        time.sleep(10)226        rotate_stage.move_relative(theta_step)227        rotate_stage.wait_up()228    rotate_stage.move_relative(-270)229    rotate_stage.wait_up()230231232233def measure_trscan(dir_sig, dir_ref, degg, nevents, voltage,234                    theta_step, theta_max, theta_scan_points,235                    r_step, r_max, r_scan_points,236                    mtype='stamp', measure_side='top'):237    print('trscan')238    rotate_slave_address = 5239    r_slave_address = 3240    stage = setup_top_devices(rotate_slave_address, r_slave_address, voltage)241    ##initialize reference settings242    reference_pmt_channel = 1243    scope = setup_reference(reference_pmt_channel)
...

api.py

Source:api.py

...24    def __init__(self, vm):25        assert(isinstance(vm, VM))26        self.vm = vm27        self.base_path = ''28    def wait_up(self):29        if not self.vm._api_up:30            log = logging.getLogger(__name__)31            self.vm.wait_ip()32            '''33            requests.exceptions.ConnectionError occures if VM is not up yet (stdout/err are not redirected to file, and34            we only blindly set vm._ip to expected static ip.35            '''36            iimax = 5037            for ii in range(1, iimax):38                try:39                    # dummy request, just to wait on service up40                    uri = 'http://%s:%d' % (self.vm._ip, settings.OSV_API_PORT)41                    uri += '/os/uptime'42                    resp = requests.get(uri)43                    self.vm._api_up = True44                    return45                except requests.exceptions.ConnectionError:46                    log.debug('API wait_up requests.exceptions.ConnectionError %d/%d, uri %s', ii, iimax, uri)47                    sleep(0.1)48    def uri(self):49        return 'http://%s:%d' % (self.vm._ip, settings.OSV_API_PORT) + self.base_path50    # kwargs is there only to pass in timeout for requests.get51    def http_get(self, params=None, path_extra='', **kwargs):52        self.wait_up()53        url_all = self.uri() + path_extra54        if params:55            url_all += '?' + urlencode(params)56        resp = requests.get(url_all, **kwargs)57        if resp.status_code != 200:58            raise ApiResponseError('HTTP call failed', resp)59        return resp.content60    # OSv uses data encoded in URL manytimes (more often than POST data).61    def http_post(self, params=None, data=None, path_extra='', **kwargs):62        log = logging.getLogger(__name__)63        self.wait_up()64        url_all = self.uri() + path_extra65        if params:66            url_all += '?' + urlencode(params)67        ## log.debug('http_post %s, data "%s"', url_all, str(data))68        resp = requests.post(url_all, data, **kwargs)69        if resp.status_code != 200:70            raise ApiResponseError('HTTP call failed', resp)71        return resp.content72    def http_put(self, params=None, data=None, path_extra='', **kwargs):73        self.wait_up()74        url_all = self.uri() + path_extra75        if params:76            url_all += '?' + urlencode(params)77        resp = requests.put(url_all, data, **kwargs)78        if resp.status_code != 200:79            raise ApiResponseError('HTTP call failed', resp)80        return resp.content81    def http_delete(self, path_extra='', **kwargs):82        self.wait_up()83        resp = requests.delete(self.uri() + path_extra, **kwargs)84        if resp.status_code != 200:85            raise ApiResponseError('HTTP call failed', resp)86        return resp.content87# line = 'key=value', returns key, value88def env_var_split(line):89    ii = line.find('=')90    kk = line[:ii]91    vv = line[ii+1:]92    return kk, vv93class EnvAll(BaseApi):94    def __init__(self, vm):95        BaseApi.__init__(self, vm)96        self.base_path = '/env'...

db.py

Source:db.py

1from dataclasses import dataclass2from random import choices, randint3from string import ascii_letters, digits4from typing import Optional5from uuid import UUID, uuid46def randword():7    return "".join(choices(ascii_letters, k=randint(12, 18)))8@dataclass9class Route:10    id: UUID11    number: str12    stops: list["Stop"]13    def __lt__(self, other):14        return len(self.number) < len(other.number) or self.number < other.number15@dataclass16class Stop:17    id: UUID18    name: str19    wait_down: int = 020    wait_up: int = 021    bus_down: Optional[str] = None22    bus_up: Optional[str] = None23    @property24    def time(self):25        bus_down = self.bus_down or "âââ"26        bus_up = self.bus_up or "âââ"27        return (28            f"{bus_down}  â  {self.wait_down: >2}  |  {self.wait_up: >2}  â  {bus_up}"29        )30routes: list[Route] = []31for i in range(60):32    number = str(randint(1, 99))33    if i == 0:34        number = str(41)35    if any(map(lambda route: route.number == number, routes)):36        continue37    stops: list[Stop] = []38    for _ in range(randint(20, 30)):39        prev_down = stops[-1].wait_down if stops else 040        prev_up = stops[-1].wait_up if stops else 041        if delta_up := randint(0, 6):42            up = prev_up + delta_up43            bus_up = None44        else:45            up = 046            bus_up = "".join(choices(digits, k=3))47        if (delta_down := randint(1, 6)) > prev_down:48            down = randint(15, 60)49            if stops:50                stops[-1].bus_down = "".join(choices(digits, k=3))51        else:52            down = prev_down - delta_down53        stop = Stop(uuid4(), randword(), down, up, bus_up=bus_up)54        stops.append(stop)55    routes.append(Route(uuid4(), number=number, stops=stops))56routes.sort()57def get_route(id: str) -> Route:58    for route in routes:59        if str(route.id) == id:60            return route...

RedWhiteBlueSnoring.py

Source:RedWhiteBlueSnoring.py

1# Carl Ferkinhoff2# starting from from adafruit example3# https://learn.adafruit.com/welcome-to-circuitpython/creating-and-editing-code4#5import time6from adafruit_circuitplayground import cp7# set up the (red) LED8#cp.pixels.brightness = 0.19#cp.pixels.fill((50,50,50))10 11while True:12    length = 15 #sets the length of the snore13    j=014    wait_down = .0115    wait_up = .0116    while (j<length):17        cp.pixels[0]=(150-j, 0, 0)18        cp.pixels[9]=(150-j, 0, 0)19        cp.pixels[1]=(0, 0,15-j)20        cp.pixels[8]=(0, 0,15-j)21        cp.pixels[2]=(15-j,15-j,15-j)22        cp.pixels[7]=(15-j,15-j,15-j)23        cp.pixels[3]=(0, 0,15-j)24        cp.pixels[6]=(0, 0,15-j)25        cp.pixels[4]=(150-j, 0, 0)26        cp.pixels[5]=(150-j, 0, 0)27        j = j+128        time.sleep(wait_down)29    time.sleep(.01)30    while (j>0):31        cp.pixels[0]=(50-j, 0, 0)32        cp.pixels[9]=(50-j, 0, 0)33        cp.pixels[1]=(0, 0,255-j)34        cp.pixels[8]=(0, 0,255-j)35        cp.pixels[2]=(15-j,15-j,15-j)36        cp.pixels[7]=(15-j,15-j,15-j)37        cp.pixels[3]=(0, 0,255-j)38        cp.pixels[6]=(0, 0,255-j)39        cp.pixels[4]=(50-j, 0, 0)40        cp.pixels[5]=(50-j, 0, 0)41        j = j-1...

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.