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

How to use check_cancelled method in lisa

Best Python code snippet using lisa_python

demo_sat.py

Source:demo_sat.py

...101                "command_id": id,102                "level": "warning",103                "message": "Command is not running. Unable to cancel command."104            }]))105    def check_cancelled(self, id, gateway):106        if self.running_commands[str(id)]["cancel"]:107            # Raise an exception to immediately stop the command operations108            raise(CommandCancelledError(f"Command {id} Cancelled"))109        else:110            return111    async def command_callback(self, command, gateway):112        self.running_commands[str(command.id)] = {"cancel": False}113        try:114            if command.type == "ping":115                asyncio.ensure_future(gateway.complete_command(116                    command_id=command.id, output="pong"))117            elif command.type == "connect":118                """119                Simulates achieving an RF Lock with the spacecraft.120                """121                await asyncio.sleep(2)122                self.check_cancelled(id=command.id, gateway=gateway)123                asyncio.ensure_future(gateway.transmit_command_update(124                    command_id=command.id,125                    state="preparing_on_gateway",126                    dict={"status": "Pointing Antennas"}127                ))128                await asyncio.sleep(4)129                self.check_cancelled(id=command.id, gateway=gateway)130                asyncio.ensure_future(gateway.transmit_command_update(131                    command_id=command.id,132                    state="uplinking_to_system",133                    dict={"status": "Broadcasting Acquisition Signal"}134                ))135                await asyncio.sleep(4)136                self.check_cancelled(id=command.id, gateway=gateway)137                asyncio.ensure_future(gateway.transmit_command_update(138                    command_id=command.id,139                    state="acked_by_system",140                    dict={"status": "Received acknowledgement from Spacecraft"}141                ))142                await asyncio.sleep(3)143                self.check_cancelled(id=command.id, gateway=gateway)144                asyncio.ensure_future(gateway.complete_command(145                    command_id=command.id,146                    output="Link Established"147                ))148            elif command.type == "telemetry":149                """150                Begins telemetry beaconing. 2 modes: error and nominal151                Error sends data with low battery voltage and low uptime counter152                Nominal sends normal data that just varies slightly153                """154                self.telemetry.safemode = False155                if type(command.fields['duration']) != type(int()):156                    asyncio.ensure_future(gateway.fail_command(157                        command_id=command.id, errors=[158                            f"Duration type is invalid. Must be an int. Type: {type(command.fields['duration'])}"159                        ]))160                else:161                    await asyncio.sleep(2)162                    self.check_cancelled(id=command.id, gateway=gateway)163                    if command.fields['mode'] == "ERROR":164                        asyncio.ensure_future(self.telemetry.generate_telemetry(165                            duration=command.fields['duration'], gateway=gateway, type="ERROR"))166                    else:167                        asyncio.ensure_future(self.telemetry.generate_telemetry(168                            duration=command.fields['duration'], gateway=gateway, type="NOMINAL"))169                    await asyncio.sleep(2)170                    self.check_cancelled(id=command.id, gateway=gateway)171                    asyncio.ensure_future(gateway.complete_command(172                        command_id=command.id,173                        output=f"Started Telemetry Beacon in mode: {command.fields['mode']} for {command.fields['duration']} seconds."))174            elif command.type == "update_file_list":175                """176                Sends a dummy file list to Major Tom.177                """178                for i in range(1, randint(2, 4)):179                    self.file_list.append({180                        "name": f'Payload-Image-{(len(self.file_list)+1):04d}.png',181                        "size": randint(2000000, 3000000),182                        "timestamp": int(time.time() * 1000) + i*10,183                        "metadata": {"type": "image", "lat": (randint(-89, 89) + .0001*randint(0, 9999)), "lng": (randint(-179, 179) + .0001*randint(0, 9999))}184                    })185                self.check_cancelled(id=command.id, gateway=gateway)186                asyncio.ensure_future(gateway.update_file_list(187                    system=self.name, files=self.file_list))188                await asyncio.sleep(3)189                self.check_cancelled(id=command.id, gateway=gateway)190                asyncio.ensure_future(gateway.complete_command(191                    command_id=command.id,192                    output="Updated Remote File List"193                ))194            elif command.type == "error":195                """196                Always errors.197                """198                self.check_cancelled(id=command.id, gateway=gateway)199                asyncio.ensure_future(gateway.transmit_command_update(200                    command_id=command.id,201                    state="uplinking_to_system",202                    dict={203                        "status": "Uplinking Command"204                    }205                ))206                await asyncio.sleep(3)207                self.check_cancelled(id=command.id, gateway=gateway)208                asyncio.ensure_future(gateway.fail_command(209                    command_id=command.id, errors=["Command failed to execute."]))210            elif command.type == "spacecraft_error":211                """212                Makes the Spacecraft generate a Critical error event.213                """214                asyncio.ensure_future(gateway.transmit_command_update(215                    command_id=command.id,216                    state="uplinking_to_system",217                    dict={218                        "status": "Uplinking Command"219                    }220                ))221                await asyncio.sleep(1)222                self.check_cancelled(id=command.id, gateway=gateway)223                event = {224                    "system": self.name,225                    "type": "CRITICAL ERROR",226                    "level": "critical",227                    "message": "A Critical Error Occurred!",228                    "timestamp": int(time.time() * 1000)229                }230                asyncio.ensure_future(gateway.transmit_events(events=[event]))231                await asyncio.sleep(1)232                self.check_cancelled(id=command.id, gateway=gateway)233                asyncio.ensure_future(gateway.fail_command(234                    command_id=command.id, errors=["Command caused critical error"]))235            elif command.type == "safemode":236                """237                Simulates uplinking a safemode command, and the satellite confirming.238                """239                asyncio.ensure_future(gateway.transmit_command_update(240                    command_id=command.id,241                    state="transmitted_to_system",242                    dict={243                        "status": "Transmitted Safemode Command",244                        "payload": "0xFFFF"245                    }246                ))247                await asyncio.sleep(3)248                self.check_cancelled(id=command.id, gateway=gateway)249                self.telemetry.safemode = True250                await asyncio.sleep(3)251                self.check_cancelled(id=command.id, gateway=gateway)252                asyncio.ensure_future(gateway.complete_command(253                    command_id=command.id,254                    output="Spacecraft Confirmed Safemode"255                ))256            elif command.type == "uplink_file":257                """258                Simulates uplinking a file by going through the whole progress bar scenario259                """260                self.check_cancelled(id=command.id, gateway=gateway)261                asyncio.ensure_future(gateway.transmit_command_update(262                    command_id=command.id,263                    state="processing_on_gateway",264                    dict={265                        "status": "Downloading Staged File from Major Tom for Transmission"266                    }267                ))268                # Download file from Major Tom269                try:270                    self.check_cancelled(id=command.id, gateway=gateway)271                    filename, content = gateway.download_staged_file(272                        gateway_download_path=command.fields["gateway_download_path"])273                except Exception as e:274                    asyncio.ensure_future(gateway.fail_command(command_id=command.id, errors=[275                                          "File failed to download", f"Error: {traceback.format_exc()}"]))276                # Write file locally.277                with open(filename, "wb") as f:278                    f.write(content)279                # Delete file because we aren't actually doing anything with it.280                os.remove(filename)281                # Update Major Tom with progress as if we're uplinking the file to the spacecraft282                await asyncio.sleep(2)283                self.check_cancelled(id=command.id, gateway=gateway)284                asyncio.ensure_future(gateway.transmit_command_update(285                    command_id=command.id,286                    state="uplinking_to_system",287                    dict={288                        "status": "Transmitting File to Spacecraft",289                        "progress_1_current": 10,290                        "progress_1_max": 100,291                        "progress_1_label": "Percent Transmitted",292                        "progress_2_current": 0,293                        "progress_2_max": 100,294                        "progress_2_label": "Percent Acked"295                    }296                ))297                await asyncio.sleep(2)298                self.check_cancelled(id=command.id, gateway=gateway)299                asyncio.ensure_future(gateway.transmit_command_update(300                    command_id=command.id,301                    state="uplinking_to_system",302                    dict={303                        "status": "Transmitting File to Spacecraft",304                        "progress_1_current": 30,305                        "progress_2_current": 10306                    }307                ))308                await asyncio.sleep(2)309                self.check_cancelled(id=command.id, gateway=gateway)310                asyncio.ensure_future(gateway.transmit_command_update(311                    command_id=command.id,312                    state="uplinking_to_system",313                    dict={314                        "status": "Transmitting File to Spacecraft",315                        "progress_1_current": 50,316                        "progress_2_current": 30317                    }318                ))319                await asyncio.sleep(2)320                self.check_cancelled(id=command.id, gateway=gateway)321                asyncio.ensure_future(gateway.transmit_command_update(322                    command_id=command.id,323                    state="uplinking_to_system",324                    dict={325                        "status": "Transmitting File to Spacecraft",326                        "progress_1_current": 70,327                        "progress_2_current": 50328                    }329                ))330                await asyncio.sleep(2)331                self.check_cancelled(id=command.id, gateway=gateway)332                asyncio.ensure_future(gateway.transmit_command_update(333                    command_id=command.id,334                    state="uplinking_to_system",335                    dict={336                        "status": "Transmitting File to Spacecraft",337                        "progress_1_current": 90,338                        "progress_2_current": 70339                    }340                ))341                await asyncio.sleep(2)342                self.check_cancelled(id=command.id, gateway=gateway)343                asyncio.ensure_future(gateway.transmit_command_update(344                    command_id=command.id,345                    state="uplinking_to_system",346                    dict={347                        "status": "Transmitting File to Spacecraft",348                        "progress_1_current": 100,349                        "progress_2_current": 90350                    }351                ))352                await asyncio.sleep(2)353                self.check_cancelled(id=command.id, gateway=gateway)354                asyncio.ensure_future(gateway.transmit_command_update(355                    command_id=command.id,356                    state="uplinking_to_system",357                    dict={358                        "progress_1_current": 100,359                        "progress_2_current": 100360                    }361                ))362                await asyncio.sleep(2)363                self.check_cancelled(id=command.id, gateway=gateway)364                asyncio.ensure_future(gateway.complete_command(365                    command_id=command.id,366                    output=f"File {filename} Successfully Uplinked to Spacecraft"367                ))368            elif command.type == "downlink_file":369                """370                "Downlinks" an image file and uploads it to Major Tom.371                Ignores the filename argument, and always pulls the latest372                image from NASA's Epic cam.373                """374                await asyncio.sleep(1)375                self.check_cancelled(id=command.id, gateway=gateway)376                asyncio.ensure_future(gateway.transmit_command_update(377                    command_id=command.id,378                    state="downlinking_from_system",379                    dict={380                        "status": "Downlinking File from Spacecraft"381                    }382                ))383                await asyncio.sleep(3)384                self.check_cancelled(id=command.id, gateway=gateway)385                # Get the latest image of the earth from epic cam386                try:387                    # Get the image info and download url388                    url = "https://epic.gsfc.nasa.gov/api/natural"389                    r = requests.get(url)390                    if r.status_code != 200:391                        raise(RuntimeError(f"File Download Failed. Status code: {r.status_code}"))392                    # Retrieve necessary data from the response393                    images = json.loads(r.content)394                    latest_image = images[-1]395                    for field in latest_image:396                        logger.debug(f'{field}  :  {latest_image[field]}')397                    image_date = datetime.datetime.strptime(398                        latest_image["date"], "%Y-%m-%d %H:%M:%S")399                    api_filename = latest_image["image"] + ".png"400                    if command.fields["filename"] != "":401                        image_filename = command.fields["filename"]402                    else:403                        image_filename = api_filename404                    image_url = "https://epic.gsfc.nasa.gov/archive/natural" + \405                        image_date.strftime("/%Y/%m/%d") + "/png/" + api_filename406                    # Get the image itself407                    self.check_cancelled(id=command.id, gateway=gateway)408                    image_r = requests.get(image_url)409                    if image_r.status_code != 200:410                        raise(RuntimeError(411                            f"File Download Failed. Status code: {image_r.status_code}"))412                    # Write file to disk413                    self.check_cancelled(id=command.id, gateway=gateway)414                    with open(image_filename, "wb") as f:415                        f.write(image_r.content)416                    logger.info(f"Downloaded Image: {api_filename} as name {image_filename}")417                except RuntimeError as e:418                    asyncio.ensure_future(gateway.fail_command(command_id=command.id, errors=[419                                          "File failed to download", f"Error: {traceback.format_exc()}"]))420                # Update command in Major Tom421                await asyncio.sleep(2)422                self.check_cancelled(id=command.id, gateway=gateway)423                asyncio.ensure_future(gateway.transmit_command_update(424                    command_id=command.id,425                    state="processing_on_gateway",426                    dict={427                        "status": f'File: "{api_filename}" Downlinked, Validating'428                    }429                ))430                await asyncio.sleep(3)431                self.check_cancelled(id=command.id, gateway=gateway)432                asyncio.ensure_future(gateway.transmit_command_update(433                    command_id=command.id,434                    state="processing_on_gateway",435                    dict={436                        "status": f'"{api_filename}" is Valid, Uploading to Major Tom'437                    }438                ))439                # Upload file to Major Tom with Metadata440                self.check_cancelled(id=command.id, gateway=gateway)441                try:442                    gateway.upload_downlinked_file(443                        filename=image_filename,444                        filepath=image_filename,  # Same as the name since we stored it locally445                        system=self.name,446                        command_id=command.id,447                        content_type=image_r.headers["Content-Type"],448                        metadata=latest_image449                    )450                    await asyncio.sleep(2)451                    self.check_cancelled(id=command.id, gateway=gateway)452                    asyncio.ensure_future(gateway.complete_command(453                        command_id=command.id,454                        output=f'"{image_filename}" successfully downlinked from Spacecraft and uploaded to Major Tom'455                    ))456                except RuntimeError as e:457                    asyncio.ensure_future(gateway.fail_command(command_id=command.id, errors=[458                                          "Downlinked File failed to upload to Major Tom", f"Error: {traceback.format_exc()}"]))459                # Remove file now that it's uploaded so we don't fill the disk.460                os.remove(image_filename)461        except Exception as e:462            if type(e) == type(CommandCancelledError()):463                asyncio.ensure_future(gateway.cancel_command(command_id=command.id))464            else:465                asyncio.ensure_future(gateway.fail_command(...

satellite.py

Source:satellite.py

...98                "command_id": id,99                "level": "warning",100                "message": "Command is not running. Unable to cancel command."101            }]))102    def check_cancelled(self, id, gateway):103        if self.running_commands[str(id)]["cancel"]:104            # Raise an exception to immediately stop the command operations105            raise(CommandCancelledError(f"Command {id} Cancelled"))106        else:107            return108    async def command_callback(self, command, gateway):109        self.running_commands[str(command.id)] = {"cancel": False}110        try:111            if command.type == "ping":112                ser.write("send") # Add command text for ping here.113                pong = ser.readline() # See if response114                if "blah" in pong: # Fill in response for ping here.115                    asyncio.ensure_future(gateway.complete_command(116                        command_id=command.id, output="pong"))117            elif command.type == "update_file_list":118                """119                Sends a file to Major Tom.120                """121                ser.write("") # Fill in relevant command here122                ser.readline() # Fill in required size for a file. Adjust wait times etc.123                for i in range(1, randint(2, 4)):124                    self.file_list.append({125                        "name": f'Payload-Image-{(len(self.file_list)+1):04d}.png',126                        "size": randint(2000000, 3000000), # fill in real file format later127                        "timestamp": int(time.time() * 1000) + i*10,128                        "metadata": {"type": "image", "lat": (randint(-89, 89) + .0001*randint(0, 9999)), "lng": (randint(-179, 179) + .0001*randint(0, 9999))}129                        # Fill in real lat/long later!130                    })131                self.check_cancelled(id=command.id, gateway=gateway)132                asyncio.ensure_future(gateway.update_file_list(133                    system=self.name, files=self.file_list))134                await asyncio.sleep(3)135                self.check_cancelled(id=command.id, gateway=gateway)136                asyncio.ensure_future(gateway.complete_command(137                    command_id=command.id,138                    output="Updated Remote File List"139                ))140            elif command.type == "error":141                """142                Always errors. Used for testing143                """144                self.check_cancelled(id=command.id, gateway=gateway)145                asyncio.ensure_future(gateway.transmit_command_update(146                    command_id=command.id,147                    state="uplinking_to_system",148                    dict={149                        "status": "Uplinking Command"150                    }151                ))152                await asyncio.sleep(3)153                self.check_cancelled(id=command.id, gateway=gateway)154                asyncio.ensure_future(gateway.fail_command(155                    command_id=command.id, errors=["Command failed to execute."]))156            elif command.type == "state":157                """158                Sends changes in satellite state to the satellite. 159                """160                state = command.fields["gateway_download_path"]161                asyncio.ensure_future(gateway.transmit_command_update(162                    command_id=command.id,163                    state="transmitted_to_system",164                    dict={165                        "status": "Transmitted State Command",166                        "payload": state167                    }168                ))169                await asyncio.sleep(3)170                self.check_cancelled(id=command.id, gateway=gateway)171                self.telemetry.state = True172                ser.write(state) # Is there a state change awk. bit? If so then can confirm here!173                asyncio.ensure_future(gateway.complete_command(174                    command_id=command.id,175                    output="Spacecraft Confirmed In State " + state176                ))177            elif command.type == "actuate":178                """179                Sends changes in satellite state to the satellite. 180                """181                dipole = command.fields["gateway_download_path"]182                asyncio.ensure_future(gateway.transmit_command_update(183                    command_id=command.id,184                    state="transmitted_to_system",185                    dict={186                        "status": "Transmitted Actuate Command to " + dipole,187                        "payload": dipole188                    }189                ))190                await asyncio.sleep(3)191                self.check_cancelled(id=command.id, gateway=gateway)192                self.telemetry.state = True193                ser.write(dipole) # Add the correct commanding format194                asyncio.ensure_future(gateway.complete_command(195                    command_id=command.id,196                    output="Spacecraft Confirmed Dipole" + dipole197                ))198            elif command.type == "picture":199                """200                Sends changes in satellite state to the satellite. 201                """202                asyncio.ensure_future(gateway.transmit_command_update(203                    command_id=command.id,204                    state="transmitted_to_system",205                    dict={206                        "status": "Transmitted Picture Command",207                    }208                ))209                await asyncio.sleep(3)210                self.check_cancelled(id=command.id, gateway=gateway)211                self.telemetry.state = True212                ser.write("take picture") # add command name!213                asyncio.ensure_future(gateway.complete_command(214                    command_id=command.id,215                    output="Spacecraft Confirmed Picture"216                ))217            elif command.type == "uplink_file":218                """219                Uplink a file!220                """221                self.check_cancelled(id=command.id, gateway=gateway)222                asyncio.ensure_future(gateway.transmit_command_update(223                    command_id=command.id,224                    state="processing_on_gateway",225                    dict={226                        "status": "Downloading Staged File from Major Tom for Transmission"227                    }228                ))229                # Download file from Major Tom230                try:231                    self.check_cancelled(id=command.id, gateway=gateway)232                    filename, content = gateway.download_staged_file(233                        gateway_download_path=command.fields["gateway_download_path"])234                except Exception as e:235                    asyncio.ensure_future(gateway.fail_command(command_id=command.id, errors=[236                                          "File failed to download", f"Error: {traceback.format_exc()}"]))237                # Write file locally.238                with open(filename, "wb") as f:239                    f.write(content)240                ser.write(filename) # Again, not sure if this is the right format for the radio yet - need info.241                # Need to add a specific approach to changing the code on the satellite remotely.242                await asyncio.sleep(2)243                self.check_cancelled(id=command.id, gateway=gateway)244                asyncio.ensure_future(gateway.complete_command(245                    command_id=command.id,246                    output=f"File {filename} Successfully Uplinked to Spacecraft"247                ))248            elif command.type == "downlink_file":249                """250                "Downlinks" an image file and uploads it to Major Tom.251                """252                await asyncio.sleep(1)253                self.check_cancelled(id=command.id, gateway=gateway)254                asyncio.ensure_future(gateway.transmit_command_update(255                    command_id=command.id,256                    state="downlinking_from_system",257                    dict={258                        "status": "Downlinking File from Spacecraft"259                    }260                ))261                await asyncio.sleep(3)262                self.check_cancelled(id=command.id, gateway=gateway)263                ser.write("Pls give file.")264                file = ser.readline()  # eat the file265                # Update command in Major Tom266                await asyncio.sleep(2)267                self.check_cancelled(id=command.id, gateway=gateway)268                asyncio.ensure_future(gateway.transmit_command_update(269                    command_id=command.id,270                    state="processing_on_gateway",271                    dict={272                        "status": f'File: "{file}" Downlinked, Validating'273                    }274                ))275                await asyncio.sleep(3)276                self.check_cancelled(id=command.id, gateway=gateway)277                asyncio.ensure_future(gateway.transmit_command_update(278                    command_id=command.id,279                    state="processing_on_gateway",280                    dict={281                        "status": f'"{file}" is Valid, Uploading to Major Tom'282                    }283                ))284                # Upload file to Major Tom with Metadata285                self.check_cancelled(id=command.id, gateway=gateway)286                try:287                    gateway.upload_downlinked_file(288                        filename=file,289                        filepath=file,  # Same as the name since we stored it locally290                        system=self.name,291                        command_id=command.id,292                    )293                    await asyncio.sleep(2)294                    self.check_cancelled(id=command.id, gateway=gateway)295                    asyncio.ensure_future(gateway.complete_command(296                        command_id=command.id,297                        output=f'"{file}" successfully downlinked from Spacecraft and uploaded to Major Tom'298                    ))299                except RuntimeError as e:300                    asyncio.ensure_future(gateway.fail_command(command_id=command.id, errors=[301                                          "Downlinked File failed to upload to Major Tom",302                                          f"Error: {traceback.format_exc()}"]))303                # Remove file now that it's uploaded so we don't fill the disk.304                os.remove(file)305        except Exception as e:306            if type(e) == type(CommandCancelledError()):307                asyncio.ensure_future(gateway.cancel_command(command_id=command.id))308            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.