feat(flight controller): Support configurable FC baudrate

Closes #538

Author: amilcarlucas

ardupilot_methodic_configurator/__main__.py

@@ -79,7 +79,7 @@ def create_argument_parser() -> argparse.ArgumentParser:
 
 
 def connect_to_fc_and_set_vehicle_type(args: argparse.Namespace) -> tuple[FlightController, str]:
-    flight_controller = FlightController(args.reboot_time)
+    flight_controller = FlightController(reboot_time=args.reboot_time, baudrate=args.baudrate)
 
     error_str = flight_controller.connect(args.device, log_errors=False)
     if error_str:

ardupilot_methodic_configurator/backend_flightcontroller.py

@@ -60,6 +60,10 @@ def close(self) -> None:
         pass
 
 
+DEFAULT_BAUDRATE: int = 115200
+DEFAULT_REBOOT_TIME: int = 7
+
+
 class FlightController:
     """
     A class to manage the connection and parameters of a flight controller.
@@ -71,13 +75,14 @@ class FlightController:
 
     """
 
-    def __init__(self, reboot_time: int) -> None:
+    def __init__(self, reboot_time: int = DEFAULT_REBOOT_TIME, baudrate: int = DEFAULT_BAUDRATE) -> None:
         """Initialize the FlightController communication object."""
         # warn people about ModemManager which interferes badly with ArduPilot
         if os_path.exists("/usr/sbin/ModemManager"):
             logging_warning(_("You should uninstall ModemManager as it conflicts with ArduPilot"))
 
         self.__reboot_time = reboot_time
+        self.__baudrate = baudrate
         self.__connection_tuples: list[tuple[str, str]] = []
         self.discover_connections()
         self.master: Union[mavutil.mavlink_connection, None] = None  # pyright: ignore[reportGeneralTypeIssues]
@@ -167,7 +172,9 @@ def connect(
                     self.__connection_tuples.insert(-1, (self.comport.device, getattr(self.comport, "description", "")))
             else:
                 return _("No serial ports found. Please connect a flight controller and try again.")
-        return self.__create_connection_with_retry(progress_callback=progress_callback, log_errors=log_errors)
+        return self.__create_connection_with_retry(
+            progress_callback=progress_callback, baudrate=self.__baudrate, log_errors=log_errors
+        )
 
     def __request_banner(self) -> None:
         """Request banner information from the flight controller."""
@@ -219,11 +226,12 @@ def __request_message(self, message_id: int) -> None:
                 0,
             )
 
-    def __create_connection_with_retry(
+    def __create_connection_with_retry(  # pylint: disable=too-many-arguments, too-many-positional-arguments
         self,
         progress_callback: Union[None, Callable[[int, int], None]],
         retries: int = 3,
         timeout: int = 5,
+        baudrate: int = DEFAULT_BAUDRATE,
         log_errors: bool = True,
     ) -> str:
         """
@@ -239,6 +247,7 @@ def __create_connection_with_retry(
                                                     of the connection attempt. Default is None.
             retries (int, optional): The number of retries before giving up. Default is 3.
             timeout (int, optional): The timeout in seconds for each connection attempt. Default is 5.
+            baudrate (int, optional): The baud rate for the connection. Default is DEFAULT_BAUDRATE.
             log_errors (bool): log errors.
 
         Returns:
@@ -248,11 +257,15 @@ def __create_connection_with_retry(
         """
         if self.comport is None or self.comport.device == "test":  # FIXME for testing only pylint: disable=fixme
             return ""
-        logging_info(_("Will connect to %s"), self.comport.device)
+        logging_info(_("Will connect to %s @ %u baud"), self.comport.device, baudrate)
         try:
             # Create the connection
             self.master = mavutil.mavlink_connection(
-                device=self.comport.device, timeout=timeout, retries=retries, progress_callback=progress_callback
+                device=self.comport.device,
+                baud=baudrate,
+                timeout=timeout,
+                retries=retries,
+                progress_callback=progress_callback,
             )
             logging_debug(_("Waiting for MAVLink heartbeat"))
             if not self.master:
@@ -522,7 +535,7 @@ def reset_and_reconnect(
             reset_progress_callback(current_step, sleep_time)
 
         # Reconnect to the flight controller
-        return self.__create_connection_with_retry(connection_progress_callback)
+        return self.__create_connection_with_retry(connection_progress_callback, baudrate=self.__baudrate)
 
     @staticmethod
     def __list_serial_ports() -> list[serial.tools.list_ports_common.ListPortInfo]:
@@ -603,6 +616,12 @@ def put_progress_callback(completion: float) -> None:
 
     @staticmethod
     def add_argparse_arguments(parser: ArgumentParser) -> ArgumentParser:
+        parser.add_argument(
+            "--baudrate",
+            type=int,
+            default=DEFAULT_BAUDRATE,
+            help=_("MAVLink serial connection baudrate to the flight controller. Default is %(default)s"),
+        )
         parser.add_argument(  # type: ignore[attr-defined]
             "--device",
             type=str,
@@ -620,7 +639,7 @@ def add_argparse_arguments(parser: ArgumentParser) -> ArgumentParser:
             min=5,
             max=50,
             action=CheckRange,
-            default=7,
+            default=DEFAULT_REBOOT_TIME,
             help=_("Flight controller reboot time. Default is %(default)s"),
         )
         return parser

ardupilot_methodic_configurator/frontend_tkinter_connection_selection.py

@@ -285,8 +285,8 @@ def main() -> None:
     )
     # pylint: enable=duplicate-code
 
-    flight_controller = FlightController(args.reboot_time)  # Initialize your FlightController instance
-    result = flight_controller.connect(device=args.device)  # Connect to the flight controller
+    flight_controller = FlightController(reboot_time=args.reboot_time, baudrate=args.baudrate)
+    result = flight_controller.connect(device=args.device)
     if result:
         logging_warning(result)
         window = ConnectionSelectionWindow(flight_controller, result)

ardupilot_methodic_configurator/frontend_tkinter_parameter_editor.py

@@ -961,7 +961,7 @@ def argument_parser() -> Namespace:
 
     logging_basicConfig(level=logging_getLevelName(args.loglevel), format="%(asctime)s - %(levelname)s - %(message)s")
 
-    fc = FlightController(args.reboot_time)
+    fc = FlightController(reboot_time=args.reboot_time, baudrate=args.baudrate)
     filesystem = LocalFilesystem(
         args.vehicle_dir, args.vehicle_type, "", args.allow_editing_template_files, args.save_component_to_system_templates
     )

tests/test__main__.py

@@ -52,7 +52,7 @@ def test_connect_to_fc_and_read_parameters(self, mock_flight_controller) -> None
         mock_fc.info.vendor = "vendor"
         mock_fc.info.firmware_type = "type"
 
-        args = argparse.Namespace(device="test_device", vehicle_type="", reboot_time=5)
+        args = argparse.Namespace(device="test_device", vehicle_type="", reboot_time=5, baudrate=115200)
         flight_controller, vehicle_type = connect_to_fc_and_set_vehicle_type(args)
         assert flight_controller == mock_fc
         assert vehicle_type == "quad"
@@ -115,7 +115,7 @@ def test_connect_to_fc_with_explicit_vehicle_type(self, mock_flight_controller)
         mock_fc.info.flight_sw_version_and_type = "v1.0"
 
         # Set an explicit vehicle type
-        args = argparse.Namespace(device="test_device", vehicle_type="plane", reboot_time=5)
+        args = argparse.Namespace(device="test_device", vehicle_type="plane", reboot_time=5, baudrate=115200)
         flight_controller, vehicle_type = connect_to_fc_and_set_vehicle_type(args)
 
         assert flight_controller == mock_fc
@@ -131,7 +131,7 @@ def test_connect_to_fc_with_connection_error(self, mock_flight_controller) -> No
             mock_window_instance = MagicMock()
             mock_window.return_value = mock_window_instance
 
-            args = argparse.Namespace(device="test_device", vehicle_type="", reboot_time=5)
+            args = argparse.Namespace(device="test_device", vehicle_type="", reboot_time=5, baudrate=115200)
             connect_to_fc_and_set_vehicle_type(args)
 
             # Verify ConnectionSelectionWindow was created with the right parameters
@@ -198,6 +198,7 @@ def test_backup_fc_parameters(self, mock_param_editor, mock_local_filesystem, mo
             # Just calling connect_to_fc_and_set_vehicle_type to get flight_controller
             args = argparse.Namespace(
                 device="test_device",
+                baudrate=115200,
                 vehicle_type="",
                 reboot_time=5,
                 n=0,

tests/test_backend_flightcontroller.py

@@ -17,26 +17,26 @@
 
 
 def test_add_connection() -> None:
-    fc = FlightController(reboot_time=7)
+    fc = FlightController(reboot_time=7, baudrate=115200)
     assert fc.add_connection("test_connection") is True
     assert fc.add_connection("test_connection") is False
     assert fc.add_connection("") is False
 
 
 def test_discover_connections() -> None:
-    fc = FlightController(reboot_time=7)
+    fc = FlightController(reboot_time=7, baudrate=115200)
     fc.discover_connections()
     assert len(fc.get_connection_tuples()) > 0
 
 
 def test_connect() -> None:
-    fc = FlightController(reboot_time=7)
+    fc = FlightController(reboot_time=7, baudrate=115200)
     result = fc.connect(device="test")
     assert result == ""
 
 
 def test_disconnect() -> None:
-    fc = FlightController(reboot_time=7)
+    fc = FlightController(reboot_time=7, baudrate=115200)
     fc.connect(device="test")
     fc.disconnect()
     assert fc.master is None
@@ -48,7 +48,7 @@ def test_disconnect() -> None:
     side_effect=lambda x: {"param1": Par(1, x), "param2": Par(2, x)},
 )
 def test_download_params(mock_load_param_file_into_dict, mock_file) -> None:
-    fc = FlightController(reboot_time=7)
+    fc = FlightController(reboot_time=7, baudrate=115200)
     fc.connect(device="test")
     with patch("ardupilot_methodic_configurator.backend_flightcontroller.open", mock_file):
         params, _ = fc.download_params()
@@ -63,7 +63,7 @@ def test_download_params(mock_load_param_file_into_dict, mock_file) -> None:
     side_effect=lambda x: {"param1": Par(1, x), "param2": Par(2, x)},
 )
 def test_set_param(mock_load_param_file_into_dict, mock_file) -> None:
-    fc = FlightController(reboot_time=7)
+    fc = FlightController(reboot_time=7, baudrate=115200)
     fc.connect(device="test")
     fc.set_param("TEST_PARAM", 1.0)
     with patch("ardupilot_methodic_configurator.backend_flightcontroller.open", mock_file):
@@ -73,22 +73,22 @@ def test_set_param(mock_load_param_file_into_dict, mock_file) -> None:
 
 
 def test_reset_and_reconnect() -> None:
-    fc = FlightController(reboot_time=7)
+    fc = FlightController(reboot_time=7, baudrate=115200)
     fc.connect(device="test")
     result = fc.reset_and_reconnect()
     assert result == ""
 
 
 def test_upload_file() -> None:
-    fc = FlightController(reboot_time=7)
+    fc = FlightController(reboot_time=7, baudrate=115200)
     fc.connect(device="test")
     result = fc.upload_file("local.txt", "remote.txt")
     # Assuming the mock environment always returns False for upload_file
     assert result is False
 
 
 def test_get_connection_tuples() -> None:
-    fc = FlightController(reboot_time=7)
+    fc = FlightController(reboot_time=7, baudrate=115200)
     fc.add_connection("test_connection")
     connections = fc.get_connection_tuples()
     assert ("test_connection", "test_connection") in connections
@@ -100,7 +100,7 @@ def test_get_connection_tuples() -> None:
     side_effect=lambda x: {"param1": Par(1, x), "param2": Par(2, x)},
 )
 def test_set_param_and_verify(mock_load_param_file_into_dict, mock_file) -> None:
-    fc = FlightController(reboot_time=7)
+    fc = FlightController(reboot_time=7, baudrate=115200)
     fc.connect(device="test")
     fc.set_param("TEST_PARAM", 1.0)
     with patch("ardupilot_methodic_configurator.backend_flightcontroller.open", mock_file):
@@ -111,15 +111,15 @@ def test_set_param_and_verify(mock_load_param_file_into_dict, mock_file) -> None
 
 
 def test_download_params_via_mavftp() -> None:
-    fc = FlightController(reboot_time=7)
+    fc = FlightController(reboot_time=7, baudrate=115200)
     fc.connect(device="test")
     params, default_params = fc.download_params_via_mavftp()
     assert isinstance(params, dict)
     assert isinstance(default_params, dict)
 
 
 def test_auto_detect_serial() -> None:
-    fc = FlightController(reboot_time=7)
+    fc = FlightController(reboot_time=7, baudrate=115200)
     serial_ports = fc._FlightController__auto_detect_serial()  # pylint: disable=protected-access
     assert isinstance(serial_ports, list)
 
@@ -136,34 +136,34 @@ def test_list_network_ports() -> None:
 
 
 def test_request_banner() -> None:
-    fc = FlightController(reboot_time=7)
+    fc = FlightController(reboot_time=7, baudrate=115200)
     fc.connect(device="test")
     fc._FlightController__request_banner()  # pylint: disable=protected-access
     # Since we cannot verify in the mock environment, we will just ensure no exceptions are raised
 
 
 def test_receive_banner_text() -> None:
-    fc = FlightController(reboot_time=7)
+    fc = FlightController(reboot_time=7, baudrate=115200)
     fc.connect(device="test")
     banner_text = fc._FlightController__receive_banner_text()  # pylint: disable=protected-access
     assert isinstance(banner_text, list)
 
 
 def test_request_message() -> None:
-    fc = FlightController(reboot_time=7)
+    fc = FlightController(reboot_time=7, baudrate=115200)
     fc.connect(device="test")
     fc._FlightController__request_message(1)  # pylint: disable=protected-access
     # Since we cannot verify in the mock environment, we will just ensure no exceptions are raised
 
 
 def test_create_connection_with_retry() -> None:
-    fc = FlightController(reboot_time=7)
+    fc = FlightController(reboot_time=7, baudrate=115200)
     result = fc._FlightController__create_connection_with_retry(progress_callback=None, retries=1, timeout=1)  # pylint: disable=protected-access
     assert result == ""
 
 
 def test_process_autopilot_version() -> None:
-    fc = FlightController(reboot_time=7)
+    fc = FlightController(reboot_time=7, baudrate=115200)
     fc.connect(device="test")
     banner_msgs = ["ChibiOS: 123", "ArduPilot"]
     result = fc._FlightController__process_autopilot_version(None, banner_msgs)  # pylint: disable=protected-access