wip

potato

wip

update board detection

wip

tests pass

Author: Williangalvani

core/services/ardupilot_manager/api/v1/routers/index.py

@@ -60,7 +60,7 @@ async def wrapper(*args: Tuple[Any], **kwargs: dict[str, Any]) -> Any:
 # all the CRUD operations, we gonna keep ones that have less than 2 endpoints in the index router.
 
 
-async def target_board(board_name: Optional[str]) -> FlightController:
+async def target_board(board_name: Optional[str], board_id: Optional[int] = None) -> FlightController:
     """Returns the board that should be used to perform operations on.
 
     Most of the API routes that have operations related to board management will give the option to perform those
@@ -72,6 +72,10 @@ async def target_board(board_name: Optional[str]) -> FlightController:
     """
     if board_name is not None:
         try:
+            if board_id is not None:
+                return next(
+                    board for board in await autopilot.available_boards(True) if board.ardupilot_board_id == board_id
+                )
             return next(board for board in await autopilot.available_boards(True) if board.name == board_name)
         except StopIteration as error:
             raise ValueError("Chosen board not available.") from error
@@ -151,8 +155,13 @@ async def get_firmware_vehicle_type() -> Any:
     summary="Retrieve dictionary of available firmwares versions with their respective URL.",
 )
 @index_to_http_exception
-async def get_available_firmwares(vehicle: Vehicle, board_name: Optional[str] = None) -> Any:
-    return autopilot.get_available_firmwares(vehicle, (await target_board(board_name)))
+async def get_available_firmwares(
+    vehicle: Vehicle,
+    board_name: Optional[str] = None,
+    board_id: Optional[int] = None,
+    firmware: Optional[str] = "Ardupilot",
+) -> Any:
+    return autopilot.get_available_firmwares(vehicle, await target_board(board_name, board_id), firmware)
 
 
 @index_router_v1.post("/install_firmware_from_url", summary="Install firmware for given URL.")

core/services/ardupilot_manager/autopilot_manager.py

@@ -656,8 +656,10 @@ async def update_endpoints(self, endpoints_to_update: Set[Endpoint]) -> None:
         self._save_current_endpoints()
         await self.mavlink_manager.restart()
 
-    def get_available_firmwares(self, vehicle: Vehicle, board: FlightController) -> List[Firmware]:
-        return self.firmware_manager.get_available_firmwares(vehicle, board)
+    def get_available_firmwares(
+        self, vehicle: Vehicle, board: FlightController, firmware_name: Optional[str] = "Ardupilot"
+    ) -> List[Firmware]:
+        return self.firmware_manager.get_available_firmwares(vehicle, board, firmware_name)
 
     async def install_firmware_from_file(
         self, firmware_path: pathlib.Path, board: FlightController, default_parameters: Optional[Parameters] = None

core/services/ardupilot_manager/firmware/FirmwareDownload.py

@@ -21,7 +21,7 @@
     NoCandidate,
     NoVersionAvailable,
 )
-from typedefs import FirmwareFormat, FlightController, Platform, PlatformType, Vehicle
+from typedefs import Firmware, FirmwareFormat, FlightController, Platform, PlatformType, Vehicle
 
 # TODO: This should be not necessary
 # Disable SSL verification
@@ -105,7 +105,7 @@ def download_manifest(self) -> bool:
 
         return True
 
-    def _find_version_item(self, **args: str) -> List[Dict[str, Any]]:
+    def _find_version_item(self, **args: str) -> List[Firmware]:
         """Find version objects in the manifest that match the specific case of **args
 
         The arguments should follow the same name described in the dictionary inside the manifest
@@ -117,6 +117,7 @@ def _find_version_item(self, **args: str) -> List[Dict[str, Any]]:
         Returns:
             List[Dict[str, Any]]: A list of firmware items that match the arguments.
         """
+        logger.info(f"Searching for version item with args: {args}")
         if not self._manifest and not self.download_manifest():
             raise ManifestUnavailable("Manifest file is not available. Cannot use it to find firmware candidates.")
 
@@ -125,39 +126,45 @@ def _find_version_item(self, **args: str) -> List[Dict[str, Any]]:
         # Make sure that the item matches all args value
         for item in self._manifest["firmware"]:
             for key, value in args.items():
-                real_key = key.replace("_", "-").lower()
-                if real_key not in item or item[real_key].lower() != value.lower():
+                real_key = key.replace("_", "-").lower() if key != "board_id" else "board_id"
+                if real_key not in item or str(item[real_key]).lower() != str(value).lower():
                     break
             else:
                 found_version_item.append(item)
 
-        return found_version_item
+        return [
+            Firmware(
+                board_id=item["board_id"] if "board_id" in item else None,
+                platform=item["platform"],
+                name=item["mav-firmware-version-type"],
+                url=item["url"],
+            )
+            for item in found_version_item
+        ]
 
     @temporary_cache(timeout_seconds=3600)
-    def get_available_versions(self, vehicle: Vehicle, board: FlightController) -> List[str]:
+    def get_available_versions(
+        self, vehicle: Vehicle, board: FlightController, firmware_name: Optional[str] = "Ardupilot"
+    ) -> List[Firmware]:
         """Get available firmware versions for the specific plataform and vehicle
 
         Args:
             vehicle (Vehicle): Desired vehicle.
             board (FlightController): Desired Flight Controller.
+            firmware (Optional[str]): Desired firmware ("Ardupilot" or "PX4").
 
         Returns:
             List[str]: List of available versions that match the specific desired configuration.
         """
-        available_versions: List[str] = []
-
+        logger.info(f"Getting available versions for {vehicle=}, {board=}, {firmware_name=}")
+        logger.error("implement px4 stuff!", firmware_name)
         if not self._manifest_is_valid():
             raise InvalidManifest("Manifest file is invalid. Cannot use it to find available versions.")
-
-        platform = board.platform.value if board.platform != Platform.GenericSerial else board.name
-        platform_type = board.platform.type if board.platform != Platform.GenericSerial else PlatformType.Serial
-        items = self._find_version_item(vehicletype=vehicle.value, platform=platform)
-
-        for item in items:
-            if item["format"] == FirmwareDownloader._supported_firmware_formats[platform_type]:
-                available_versions.append(item["mav-firmware-version-type"])
-
-        return available_versions
+        if board.ardupilot_board_id is None:
+            if board.platform != Platform.GenericSerial:
+                return self._find_version_item(vehicletype=vehicle.value, platform=board.platform.value)
+            return self._find_version_item(vehicletype=vehicle.value, platform=board.name)
+        return self._find_version_item(vehicletype=vehicle.value, board_id=str(board.ardupilot_board_id))
 
     @temporary_cache(timeout_seconds=3600)
     def get_download_url(self, vehicle: Vehicle, board: FlightController, version: str = "") -> str:
@@ -178,21 +185,21 @@ def get_download_url(self, vehicle: Vehicle, board: FlightController, version: s
         if not versions:
             raise NoVersionAvailable(f"Could not find available firmware versions for {board}/{vehicle}.")
 
-        if version and version not in versions:
+        if version and not any(version == found_version.name for found_version in versions):
             raise NoVersionAvailable(f"Version {version} was not found for {board}/{vehicle}.")
 
-        firmware_format = FirmwareDownloader._supported_firmware_formats[board.platform.type]
+        firmware_format = FirmwareDownloader._supported_firmware_formats[board.platform.type].value
 
         # Autodetect the latest supported version.
         # For .apj firmwares (e.g. Pixhawk), we use the latest STABLE version while for the others (e.g. SITL and
         # Navigator) we use latest BETA. Specially on this development phase of the BlueOS/navigator, using
         # the BETA release allow us to track and fix introduced bugs faster.
         if not version:
             if firmware_format == FirmwareFormat.APJ:
-                supported_versions = [version for version in versions if "STABLE" in version]
+                supported_versions = [version for version in versions if "STABLE" in version.name]
                 newest_version: Optional[str] = None
                 for supported_version in supported_versions:
-                    semver_version = supported_version.split("-")[1]
+                    semver_version = supported_version.name.split("-")[1]
                     if not newest_version or Version(newest_version) < Version(semver_version):
                         newest_version = semver_version
                 if not newest_version:
@@ -219,7 +226,7 @@ def get_download_url(self, vehicle: Vehicle, board: FlightController, version: s
 
         item = items[0]
         logger.debug(f"Downloading following firmware: {item}")
-        return str(item["url"])
+        return item.url
 
     def download(self, vehicle: Vehicle, board: FlightController, version: str = "") -> pathlib.Path:
         """Download a specific firmware that matches the arguments.