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

How to use get_connection_status method in localstack

Best Python code snippet using localstack_python

module_connect.py

Source:module_connect.py

...23    self.last_seen       = datetime.now()24# ---25def display_connection_status ():26  print()27  connection_status = get_connection_status()28  if connection_status['State'] == 'connected':29    print('Connected to ' + connection_status['SSID'] + ' (' + connection_status['BSSID'] + ')')30  else:31    print('Disconnected.')32# ---33def get_connection_status ():34  result = {}35  output = str(check_output('netsh wlan show interface wi-fi'))36  for line in output.split('\\r\\n'):37    pair = line.split(' : ')38    if len(pair) == 2:39      result[pair[0].strip()] = pair[1].strip()40  return result41# ---42def get_network_list ():43  networks = {}44  # Get all nearby networks using netsh45  netsh_output = str(check_output('netsh wlan show networks mode=bssid'))46  # For each SSID47  for result in netsh_output.split('\\r\\nSSID')[1:]:48    # Split the result into key/value pairs49    pairs = result.split('\\r\\n')50    SSID = pairs[0].split(' : ')[1].strip()51    network_type = pairs[1].split(' : ')[1].strip()52    authentication = pairs[2].split(' : ')[1].strip()53    encryption = pairs[3].split(' : ')[1].strip()54    # For each BSSID55    for sub_result in result.split('BSSID')[1:]:56      # Split the result into key/value pairs57      sub_pairs = sub_result.split('\\r\\n')58      BSSID = sub_pairs[0].split(' : ')[1].strip()59      signal = sub_pairs[1].split(' : ')[1].strip()60      protocol = sub_pairs[2].split(' : ')[1].strip()61      channel = sub_pairs[3].split(' : ')[1].strip()62      # Record the network details63      networks[BSSID] = Network(BSSID, SSID, network_type, authentication, encryption, protocol, channel)64      networks[BSSID].update_signal_strength(signal)65  return networks66# ---67def autocomplete_network (SSID):68  for network in get_network_list().values():69    if network.SSID.lower().startswith(SSID.lower()):70      return network71# ---72def connect_to_network (name):73  print()74  # Are we already connected?75  connection_status = get_connection_status()76  if connection_status['State'] == 'connected':77    print('Already connected to ' + connection_status['SSID'] + ' (' + connection_status['BSSID'] + ')')78    if confirm('Disconnect?'):79      disconnect_from_network()80      print()81    else:82      return83  # Is there a nearby network with this SSID?84  target_network = autocomplete_network(name)85  if not target_network:86    print('Could not locate ' + name)87    return88  print('Found network ' + target_network.SSID + ' (' + target_network.BSSID + ')')89  print('Security: ' + target_network.authentication + ' (' + target_network.encryption + ')' + NL)90  # Does a profile exist for this SSID?91  for profile_SSID, profile_name in get_profile_list().items():92    if profile_SSID.lower() == target_network.SSID.lower():93      if confirm('Use existing profile?'):94        # Attempt to connect using the existing profile95        print()96        attempt_connection(profile_name)97        return98      else:99        print(NL + 'This will overwrite the existing profile.')100        if confirm('Are you sure?'):101          # Delete the existing profile102          delete_profile(profile_name)103          print()104        else:105          return106  # Generate a new profile107  profile = generate_profile(target_network)108  # Add the network password, if required109  if target_network.authentication not in ['None', 'Open']:110    password = input('Password required: ')111    profile = profile.format(password)112  # Attempt to connect using the new profile113  command_result = add_profile(profile)114  if not command_result['Success']:115    print(command_result['Output'])116    return117  attempt_connection(target_network.SSID)118  # ---119def attempt_connection (profile_name):120  print('Sending connection request...')121  command_result = try_call(['netsh', 'wlan', 'connect', profile_name])122  if not command_result['Success']:123    print(command_result['Output'])124  try:125    # Check if the request was successful126    for i in range(5):127      sleep(1)128      if get_connection_status()['State'] == 'connected':129        print('Connected.')130        return131  except KeyboardInterrupt:132    pass133  print('Request timed out.' + NL)134  print('This may mean the network is no longer available or the network password is incorrect')135  if confirm('Delete associated profile?'):136    delete_profile(profile_name)137# ---138def disconnect_from_network ():139  print()140  connection_status = get_connection_status()141  if connection_status['State'] == 'disconnected':142    print('Not connected to a network.')143    return144  command_result = try_call('netsh wlan disconnect')145  if command_result['Success']:146    print('Disconnected from ' + connection_status['SSID'])147  else:148    print('Disconnect failed.')...

PrinterControl.py

Source:PrinterControl.py

1import requests2import time3base_url = 'http://octopi.local/api/'4apikey = {'apikey': 'C74BC62010F3417FBC5EEBA6A1BDF0BF'}5def get_connection_status():6    r = requests.get(base_url + 'connection', params=apikey)7    return r.json()["current"]["state"]8def connect_to_printer():9    command_data = {'command': 'connect', 'baudrate': 115200, 'printerProfile': 'prusa_i3_mk2'}10    r = requests.post(base_url + 'connection', json=command_data, params=apikey)11    return r12def upload_stl(filename):13    file = {'file': (filename, open(filename, 'rb'), 'application/octet-stream')}14    r = requests.post(base_url + 'files/local', files=file, params=apikey)15    return r16def slice_and_select(filename):17    command_parameters = {'command': 'slice',18                          'position': {'x': 125, 'y': 105},19                          'printerProfile': 'prusa_i3_mk2',20                          'profile': 'original_prusa_i3_mk2_0_15_pla_normal',21                          'select': True}22    r = requests.post(base_url + 'files/local/' + filename, json=command_parameters, params=apikey)23    return r24def get_print_job_info():25    return requests.get(base_url + 'job', params=apikey).json()26def start_print_job():27    start_command = {'command': 'start'}28    return requests.post(base_url + 'job', json=start_command, params=apikey)29def print_stl_file(filename):30    # Check current printer status and try to connect if necessary31    current_status = get_connection_status()32    print('Current printer connection status: ' + current_status)33    if current_status == 'Closed':34        print('Connecting to printer:')35        connect_to_printer()36        current_status = get_connection_status()37        print('\t' + current_status)38        while current_status != 'Operational':39            time.sleep(2)40            current_status = get_connection_status()41            print('\t' + current_status)42    # Upload stl file to OctoPrint43    file_upload_response = upload_stl(filename)44    if file_upload_response.json()['done']:45        print('File "' + filename + '" uploaded successfully')46    else:47        print('Failed to upload ' + filename)48        exit()49    # Slice stl file to gcode50    slice_and_select(filename)51    print('Preparing file for printing; please wait...')52    # Retrieve print info once slicing is complete53    print_job_info = get_print_job_info()54    while(print_job_info['job']['estimatedPrintTime'] == None):...

dbcon.py

Source:dbcon.py

...11    def __init__(self):12        ...13    def connect_to_db(self):14        ...15    def get_connection_status(self):16        ...17class GenericConnector:18    dbname = None19    username = None20    connection_str = None21    def __init__(self):22        self.connected = False23    def connect_to_db(self):24        self.connected = True25        logger.info(f"Connected to {self.dbname} with user {self.username} and"26                    f"connection details '{self.connection_str}'.")27        return self.dbname28    def get_connection_status(self):29        pass30class ConcreteSourceDBConnector(GenericConnector):31    dbname = 'my_source_database'32    username = 'my_user'33    connection_str = 'my_source_connection_string'34    def get_connection_status(self):35        if not self.connected:36            raise ConcreteDBConnectionError37        else:38            logger.info("Successfully connected to Source Database")39class ConcreteTargetDBConnector(GenericConnector):40    dbname = 'my_target_database'41    username = 'my_user'42    connection_str = 'my_target_connection_string'43    def get_connection_status(self):44        if not self.connected:45            raise ConcreteDBConnectionError46        else:...

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.