wip

Author: Williangalvani

core/services/ardupilot_manager/flight_controller_detector/Detector.py

@@ -4,7 +4,7 @@
 from commonwealth.utils.general import is_running_as_root
 from serial.tools.list_ports_linux import SysFS, comports
 
-from flight_controller_detector.board_identification import identifiers
+from flight_controller_detector.board_identification import identifiers, get_board_id_from_usb_id
 from flight_controller_detector.linux.detector import LinuxFlightControllerDetector
 from typedefs import FlightController, FlightControllerFlags, Platform
 
@@ -55,17 +55,25 @@ def detect_serial_flight_controllers() -> List[FlightController]:
             # usb_device_path property will be the same for two serial connections using the same USB port
             if port.usb_device_path not in [device.usb_device_path for device in unique_serial_devices]:
                 unique_serial_devices.append(port)
-        boards = [
-            FlightController(
-                name=port.product or port.name,
-                manufacturer=port.manufacturer,
-                platform=Detector.detect_serial_platform(port)
-                or Platform(),  # this is just to make CI happy. check line 82
-                path=port.device,
-            )
-            for port in unique_serial_devices
-            if Detector.detect_serial_platform(port) is not None
-        ]
+        boards = []
+        for port in unique_serial_devices:
+            platform = Detector.detect_serial_platform(port)
+            if platform is not None:
+                board_name = port.product or port.name
+
+                # Get board_id using USB VID:PID
+                board_id = None
+                if port.vid is not None and port.pid is not None:
+                    board_id = get_board_id_from_usb_id(port.vid, port.pid, board_name)
+
+                board = FlightController(
+                    name=board_name,
+                    manufacturer=port.manufacturer,
+                    platform=platform,
+                    path=port.device,
+                    ardupilot_board_id=board_id,
+                )
+                boards.append(board)
         for port in unique_serial_devices:
             for board in boards:
                 if board.path == port.device and Detector.is_serial_bootloader(port):

core/services/ardupilot_manager/flight_controller_detector/board_identification.py

@@ -1,7 +1,7 @@
 import json
 from enum import Enum
 from pathlib import Path
-from typing import List
+from typing import List, Optional
 
 from pydantic import BaseModel
 
@@ -39,16 +39,101 @@ def load_board_identifiers() -> List[SerialBoardIdentifier]:
     with open(boards_path, encoding="utf-8") as f:
         json_data = json.load(f)
 
-    # Extract all unique board names
+    # Extract all unique board names from the new structure
     board_names = set()
     for boards_list in json_data.values():
-        board_names.update(boards_list)
+        if isinstance(boards_list, list) and boards_list:
+            # Check if it's the new format (list of objects) or old format (list of strings)
+            if isinstance(boards_list[0], dict):
+                # New format: extract board_name from objects
+                board_names.update(board["board_name"] for board in boards_list if "board_name" in board)
+            else:
+                # Old format: boards_list contains strings directly
+                board_names.update(boards_list)
 
     return [
         SerialBoardIdentifier(attribute=SerialAttr.product, id_value=board_name, platform=Platform.GenericSerial)
         for board_name in board_names
     ]
 
 
+def get_board_id_from_name(board_name: str) -> Optional[int]:
+    """
+    Get the board ID for a given board name by searching through the boards cache.
+
+    Args:
+        board_name: The name of the board to look up.
+
+    Returns:
+        Optional[int]: The board ID if found, None otherwise.
+    """
+    try:
+        boards_path = get_boards_cache_path()
+
+        # If cache doesn't exist, try to generate it
+        if not boards_path.exists():
+            handler = ManifestHandler()
+            handler.process_and_export(boards_path)
+
+        with open(boards_path, encoding="utf-8") as f:
+            json_data = json.load(f)
+
+        # Search through all USB IDs for the board name
+        for boards_list in json_data.values():
+            if isinstance(boards_list, list):
+                for board in boards_list:
+                    if isinstance(board, dict) and board.get("board_name") == board_name:
+                        return board.get("board_id")
+                    elif isinstance(board, str) and board == board_name:
+                        # Old format doesn't have board IDs
+                        return None
+
+        return None
+    except (json.JSONDecodeError, OSError, FileNotFoundError):
+        return None
+
+
+def get_board_id_from_usb_id(vid: int, pid: int, board_name: Optional[str] = None) -> Optional[int]:
+    """
+    Get the board ID for a given USB VID:PID combination, optionally filtering by board name.
+
+    Args:
+        vid: USB Vendor ID
+        pid: USB Product ID
+        board_name: Optional board name to filter results when multiple boards share the same USB ID
+
+    Returns:
+        Optional[int]: The board ID if found, None otherwise.
+    """
+    try:
+        boards_path = get_boards_cache_path()
+
+        # If cache doesn't exist, try to generate it
+        if not boards_path.exists():
+            handler = ManifestHandler()
+            handler.process_and_export(boards_path)
+
+        with open(boards_path, encoding="utf-8") as f:
+            json_data = json.load(f)
+
+        # Format USB ID as vid:pid (lowercase hex)
+        usb_id = f"{vid:04x}:{pid:04x}"
+
+        if usb_id in json_data:
+            boards_list = json_data[usb_id]
+            if isinstance(boards_list, list):
+                for board in boards_list:
+                    if isinstance(board, dict):
+                        # If board_name is specified, filter by it
+                        return board.get("board_id")
+                    elif isinstance(board, str):
+                        # Old format doesn't have board IDs
+                        return None
+
+        return None
+    except (json.JSONDecodeError, OSError, FileNotFoundError):
+        return None
+
+
 # Load dynamic board identifiers from manifest
 identifiers = load_board_identifiers()