chore(ty): type safety improvements using ty

Author: amilcarlucas

ardupilot_methodic_configurator/backend_flightcontroller.py

@@ -289,8 +289,7 @@ def reset_and_reconnect(
             logging_warning(_("Cannot reset flight controller: not connected"))
             return ""
         # Issue a reset
-        # Type ignore needed because MavlinkConnection is a Union including object fallback
-        self.master.reboot_autopilot()  # type: ignore[union-attr]
+        self.master.reboot_autopilot()
         logging_info(_("Reset command sent to ArduPilot."))
         time_sleep(0.3)  # Short delay for command to be sent
 

ardupilot_methodic_configurator/backend_flightcontroller_commands.py

@@ -122,9 +122,9 @@ def send_command_and_wait_ack(  # pylint: disable=too-many-arguments,too-many-po
 
         try:
             # Send the command
-            self.master.mav.command_long_send(  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue]
-                self.master.target_system,  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue]
-                self.master.target_component,  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue]
+            self.master.mav.command_long_send(  # pyright: ignore[reportAttributeAccessIssue]
+                self.master.target_system,  # pyright: ignore[reportAttributeAccessIssue]
+                self.master.target_component,  # pyright: ignore[reportAttributeAccessIssue]
                 command,
                 0,  # confirmation
                 param1,
@@ -139,7 +139,7 @@ def send_command_and_wait_ack(  # pylint: disable=too-many-arguments,too-many-po
             # Wait for acknowledgment
             start_time = time_time()
             while time_time() - start_time < timeout:
-                msg = self.master.recv_match(  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue]
+                msg = self.master.recv_match(  # pyright: ignore[reportAttributeAccessIssue]
                     type="COMMAND_ACK", blocking=False
                 )
                 if msg and msg.command == command:
@@ -305,9 +305,9 @@ def test_all_motors(self, nr_of_motors: int, throttle_percent: int, timeout_seco
 
         for i in range(nr_of_motors):
             # MAV_CMD_DO_MOTOR_TEST command for all motors
-            self.master.mav.command_long_send(  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue]
-                self.master.target_system,  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue]
-                self.master.target_component,  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue]
+            self.master.mav.command_long_send(  # pyright: ignore[reportAttributeAccessIssue]
+                self.master.target_system,  # pyright: ignore[reportAttributeAccessIssue]
+                self.master.target_component,  # pyright: ignore[reportAttributeAccessIssue]
                 mavutil.mavlink.MAV_CMD_DO_MOTOR_TEST,
                 0,  # confirmation
                 param1=i + 1,  # motor number (1-based)
@@ -482,7 +482,7 @@ def get_battery_status(self) -> tuple[Union[tuple[float, float], None], str]:
 
         try:
             # Try to get real telemetry data
-            battery_status = self.master.recv_match(  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue]
+            battery_status = self.master.recv_match(  # pyright: ignore[reportAttributeAccessIssue]
                 type="BATTERY_STATUS", blocking=False, timeout=self.BATTERY_STATUS_TIMEOUT
             )
             if battery_status:

ardupilot_methodic_configurator/backend_flightcontroller_connection.py

@@ -203,7 +203,7 @@ def disconnect(self) -> None:
         """Close the connection to the flight controller."""
         if self.master is not None:
             with contextlib.suppress(Exception):
-                self.master.close()  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue]
+                self.master.close()  # pyright: ignore[reportAttributeAccessIssue]
             self.master = None
         self.info.reset()
 
@@ -368,7 +368,7 @@ def _create_mavlink_connection(  # pylint: disable=too-many-arguments, too-many-
         timeout: int = 5,
         retries: int = 3,
         progress_callback: Union[None, Callable[[int, int], None]] = None,
-    ) -> mavutil.mavlink_connection:  # pyright: ignore[reportGeneralTypeIssues]
+    ) -> Union["MavlinkConnection", None]:
         """
         Factory method for creating MAVLink connections.
 
@@ -410,7 +410,7 @@ def _detect_vehicles_from_heartbeats(self, timeout: int) -> dict[tuple[int, int]
         while time_time() - start_time < timeout:
             try:
                 m = (
-                    self.master.recv_match(  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue]
+                    self.master.recv_match(  # pyright: ignore[reportAttributeAccessIssue]
                         type="HEARTBEAT", blocking=False
                     )
                     if self.master
@@ -483,7 +483,7 @@ def _retrieve_autopilot_version_and_banner(self, timeout: int) -> str:
         # Request AUTOPILOT_VERSION message
         self._request_message(mavutil.mavlink.MAVLINK_MSG_ID_AUTOPILOT_VERSION)
         m = (
-            self.master.recv_match(  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue]
+            self.master.recv_match(  # pyright: ignore[reportAttributeAccessIssue]
                 type="AUTOPILOT_VERSION", blocking=True, timeout=timeout
             )
             if self.master
@@ -498,9 +498,9 @@ def _request_banner(self) -> None:
         if self.master is not None:
             # Note: Don't wait for ACK here as banner requests are fire-and-forget
             # and we handle the response via STATUS_TEXT messages
-            self.master.mav.command_long_send(  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue]
-                self.master.target_system,  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue]
-                self.master.target_component,  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue]
+            self.master.mav.command_long_send(  # pyright: ignore[reportAttributeAccessIssue]
+                self.master.target_system,  # pyright: ignore[reportAttributeAccessIssue]
+                self.master.target_component,  # pyright: ignore[reportAttributeAccessIssue]
                 mavutil.mavlink.MAV_CMD_DO_SEND_BANNER,
                 0,
                 0,
@@ -523,7 +523,7 @@ def _receive_banner_text(self) -> list[str]:
         start_time = time_time()
         banner_msgs: list[str] = []
         while self.master:
-            msg = self.master.recv_match(  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue]
+            msg = self.master.recv_match(  # pyright: ignore[reportAttributeAccessIssue]
                 type="STATUSTEXT", blocking=False
             )
             if msg:
@@ -551,7 +551,7 @@ def _request_message(self, message_id: int) -> None:
             system_id = int(self.info.system_id) if self.info.system_id else 0
             component_id = int(self.info.component_id) if self.info.component_id else 0
 
-            self.master.mav.command_long_send(  # type: ignore[union-attr] # pyright: ignore[reportAttributeAccessIssue]
+            self.master.mav.command_long_send(  # pyright: ignore[reportAttributeAccessIssue]
                 system_id,
                 component_id,
                 mavutil.mavlink.MAV_CMD_REQUEST_MESSAGE,
@@ -870,7 +870,7 @@ def baudrate(self) -> int:
 
     def set_master_for_testing(
         self,
-        master: Optional[mavutil.mavlink_connection],  # pyright: ignore[reportGeneralTypeIssues]
+        master: Optional["MavlinkConnection"],
     ) -> None:
         """
         Set the MAVLink connection for testing purposes.

ardupilot_methodic_configurator/backend_flightcontroller_factory_mavftp.py

@@ -8,14 +8,15 @@
 SPDX-License-Identifier: GPL-3.0-or-later
 """
 
-from typing import Optional, Union
-
-from pymavlink import mavutil
+from typing import TYPE_CHECKING, Optional, Union
 
 from ardupilot_methodic_configurator.backend_mavftp import MAVFTP
 
+if TYPE_CHECKING:
+    from ardupilot_methodic_configurator.backend_flightcontroller_protocols import MavlinkConnection
+
 
-def create_mavftp(master: Union[mavutil.mavlink_connection, None]) -> MAVFTP:  # pyright: ignore[reportGeneralTypeIssues]
+def create_mavftp(master: Union["MavlinkConnection", None]) -> MAVFTP:
     """
     Factory function for creating MAVFTP instances.
 
@@ -38,7 +39,7 @@ def create_mavftp(master: Union[mavutil.mavlink_connection, None]) -> MAVFTP:  #
 
 
 def create_mavftp_safe(
-    master: Union[mavutil.mavlink_connection, None],  # pyright: ignore[reportGeneralTypeIssues]
+    master: Union["MavlinkConnection", None],
 ) -> Optional[MAVFTP]:  # pyright: ignore[reportGeneralTypeIssues]
     """
     Factory function for creating MAVFTP instances with safe error handling.

ardupilot_methodic_configurator/backend_flightcontroller_factory_mavlink.py

@@ -8,10 +8,13 @@
 SPDX-License-Identifier: GPL-3.0-or-later
 """
 
-from typing import Optional, Protocol
+from typing import TYPE_CHECKING, Optional, Protocol
 
 from pymavlink import mavutil
 
+if TYPE_CHECKING:
+    from ardupilot_methodic_configurator.backend_flightcontroller_protocols import MavlinkConnection
+
 
 class MavlinkConnectionFactory(Protocol):  # pylint: disable=too-few-public-methods
     """Protocol for creating MAVLink connections."""
@@ -23,7 +26,7 @@ def create(  # pylint: disable=too-many-arguments, too-many-positional-arguments
         timeout: float = 5.0,
         retries: int = 3,
         progress_callback: Optional[object] = None,
-    ) -> Optional[mavutil.mavlink_connection]:  # pyright: ignore[reportGeneralTypeIssues]
+    ) -> Optional["MavlinkConnection"]:
         """Create a MAVLink connection."""
         ...  # pylint: disable=unnecessary-ellipsis
 
@@ -38,10 +41,10 @@ def create(  # pylint: disable=too-many-arguments, too-many-positional-arguments
         timeout: float = 5.0,
         retries: int = 3,
         progress_callback: Optional[object] = None,
-    ) -> Optional[mavutil.mavlink_connection]:  # pyright: ignore[reportGeneralTypeIssues]
+    ) -> Optional["MavlinkConnection"]:
         """Create connection using actual PyMAVLink library."""
         try:
-            return mavutil.mavlink_connection(
+            return mavutil.mavlink_connection(  # pyright: ignore[reportReturnType]
                 device=device,
                 baud=baudrate,
                 timeout=timeout,

ardupilot_methodic_configurator/backend_flightcontroller_files.py

@@ -15,14 +15,15 @@
 from logging import warning as logging_warning
 from typing import TYPE_CHECKING, Callable, ClassVar, Optional, Union
 
-from pymavlink import mavutil
-
 from ardupilot_methodic_configurator import _
 from ardupilot_methodic_configurator.backend_flightcontroller_factory_mavftp import create_mavftp_safe
 from ardupilot_methodic_configurator.data_model_flightcontroller_info import FlightControllerInfo
 
 if TYPE_CHECKING:
-    from ardupilot_methodic_configurator.backend_flightcontroller_protocols import FlightControllerConnectionProtocol
+    from ardupilot_methodic_configurator.backend_flightcontroller_protocols import (
+        FlightControllerConnectionProtocol,
+        MavlinkConnection,
+    )
 
 # Conditionally import MAVFTP if available
 try:
@@ -66,7 +67,7 @@ def __init__(
         self._connection_manager: FlightControllerConnectionProtocol = connection_manager
 
     @property
-    def master(self) -> Optional[mavutil.mavlink_connection]:  # pyright: ignore[reportGeneralTypeIssues]
+    def master(self) -> Optional["MavlinkConnection"]:
         """Get master connection."""
         return self._connection_manager.master
 

ardupilot_methodic_configurator/backend_flightcontroller_params.py

@@ -18,8 +18,6 @@
 from time import time as time_time
 from typing import TYPE_CHECKING, Any, Callable, Optional, Union
 
-from pymavlink import mavutil
-
 from ardupilot_methodic_configurator import _
 from ardupilot_methodic_configurator.backend_flightcontroller_connection import DEVICE_FC_PARAM_FROM_FILE
 from ardupilot_methodic_configurator.backend_flightcontroller_factory_mavftp import create_mavftp
@@ -28,7 +26,10 @@
 
 # Type hint for connection manager to avoid circular imports
 if TYPE_CHECKING:
-    from ardupilot_methodic_configurator.backend_flightcontroller_protocols import FlightControllerConnectionProtocol
+    from ardupilot_methodic_configurator.backend_flightcontroller_protocols import (
+        FlightControllerConnectionProtocol,
+        MavlinkConnection,
+    )
 
 
 class FlightControllerParams:
@@ -70,7 +71,7 @@ def __init__(
         self.fc_parameters: dict[str, float] = fc_parameters if fc_parameters is not None else {}
 
     @property
-    def master(self) -> Optional[mavutil.mavlink_connection]:  # pyright: ignore[reportGeneralTypeIssues]
+    def master(self) -> Optional["MavlinkConnection"]:
         """Get master connection."""
         return self._connection_manager.master
 

ardupilot_methodic_configurator/backend_flightcontroller_protocols.py

@@ -33,7 +33,6 @@
 from ardupilot_methodic_configurator.data_model_par_dict import ParDict
 
 # Type alias for MAVLink connection to avoid type checker issues
-# pymavlink.mavutil.mavlink_connection is actually a function that returns various connection types
 # We define MavlinkConnection as a protocol-like type to represent any MAVLink connection object
 if TYPE_CHECKING:
     # During type checking, import the actual mavutil module for better type hints
@@ -43,12 +42,13 @@
     # Use a union of known connection types for better type safety
     # Note: mavutil.mavlink_connection() returns different types based on the connection string
     MavlinkConnection = Union[
-        mavutil.mavserial,
-        mavutil.mavudp,
-        mavutil.mavtcp,
-        mavutil.mavtcpin,
-        mavutil.mavmcast,
-        object,  # Fallback for other connection types
+        mavutil.mavserial,  # a serial mavlink port
+        mavutil.mavudp,  # a UDP mavlink socket
+        mavutil.mavtcp,  # a TCP mavlink socket
+        mavutil.mavtcpin,  # a TCP input mavlink socket
+        mavutil.mavmcast,  # a UDP multicast mavlink socket
+        mavutil.mavwebsocket,  # Mavlink WebSocket server, single client only
+        mavutil.mavwebsocket_client,  # client using WebSocket over TCP with WS and WSS support
     ]
 else:
     # At runtime, we don't need the actual types

ardupilot_methodic_configurator/backend_mavftp.py

@@ -217,6 +217,49 @@ def __setattr__(self, name, value) -> None:
             return
         raise AttributeError
 
+    def command(self, args: list[str]) -> None:
+        """
+        Parse and apply 'set NAME VALUE' style arguments to update a setting.
+
+        args is expected to be [NAME, VALUE].
+        """
+        if len(args) != 2:
+            logging.error("Usage: ftp set <SETTING_NAME> <VALUE>")
+            return
+
+        name, value_str = args
+
+        if name not in self._vars:
+            logging.error("Unknown FTP setting '%s'", name)
+            return
+
+        setting = self._vars[name]
+        s_type = setting.type
+
+        # Only numeric (int/float) setting types are supported here, as documented.
+        if s_type not in (int, float):
+            logging.error(
+                "FTP setting '%s' has unsupported type '%s' for CLI update (only int and float are supported)",
+                name,
+                getattr(s_type, "__name__", str(s_type)),
+            )
+            return
+
+        try:
+            # Convert string to the appropriate type (int or float)
+            converted = s_type(value_str)
+        except (ValueError, TypeError):
+            logging.error(
+                "Invalid value '%s' for FTP setting '%s' (expected %s)",
+                value_str,
+                name,
+                getattr(s_type, "__name__", str(s_type)),
+            )
+            return
+
+        setting.value = converted
+        logging.info("FTP setting '%s' set to %s", name, converted)
+
 
 class MAVFTPReturn:
     """The result of a MAVFTP operation."""
@@ -589,7 +632,7 @@ def __handle_open_ro_reply(self, op, _m) -> MAVFTPReturn:
     def __check_read_finished(self) -> bool:
         """Check if download has completed."""
         # logging.debug("FTP: check_read_finished: %s %s", self.reached_eof, self.read_gaps)
-        if self.reached_eof and len(self.read_gaps) == 0:
+        if self.reached_eof and len(self.read_gaps) == 0 and self.fh is not None and self.op_start is not None:
             ofs = self.fh.tell()
             dt = time.time() - self.op_start
             rate = (ofs / dt) / 1024.0
@@ -674,7 +717,7 @@ def __handle_burst_read(self, op, _m) -> MAVFTPReturn:  # noqa PRL0911, PGH004,
                 if op.size > 0 and op.size < self.burst_size:
                     # a burst complete with non-zero size and less than burst packet size
                     # means EOF
-                    if not self.reached_eof and self.ftp_settings.debug > 0:
+                    if not self.reached_eof and self.ftp_settings.debug > 0 and self.op_start is not None:
                         logging.info(
                             "FTP: EOF at %u with %u gaps t=%.2f",
                             self.fh.tell(),
@@ -700,7 +743,7 @@ def __handle_burst_read(self, op, _m) -> MAVFTPReturn:  # noqa PRL0911, PGH004,
                     if self.ftp_settings.debug > 0:
                         logging.error("FTP: burst lost EOF %u %u", self.fh.tell(), op.offset)
                     return MAVFTPReturn("BurstReadFile", ERR_Fail)
-                if not self.reached_eof and self.ftp_settings.debug > 0:
+                if not self.reached_eof and self.ftp_settings.debug > 0 and self.op_start is not None:
                     logging.info(
                         "FTP: EOF at %u with %u gaps t=%.2f", self.fh.tell(), len(self.read_gaps), time.time() - self.op_start
                     )
@@ -809,7 +852,7 @@ def __put_finished(self, flen) -> None:
         if self.put_callback is not None:
             self.put_callback(flen)
             self.put_callback = None
-        else:
+        elif self.op_start is not None:
             dt = time.time() - self.op_start
             rate = (flen / dt) / 1024.0
             logging.info("Put %u bytes to %s file in %.2fs %.1fkByte/s", flen, self.filename, dt, rate)
@@ -829,7 +872,7 @@ def __handle_create_file_reply(self, op, _m) -> MAVFTPReturn:
 
     def __send_more_writes(self) -> None:
         """Send some more writes."""
-        if len(self.write_list) == 0:
+        if len(self.write_list) == 0 or self.write_list is None:
             # all done
             self.__put_finished(self.write_file_size)
             self.__terminate_session()
@@ -962,7 +1005,8 @@ def __handle_crc_reply(self, op, _m) -> MAVFTPReturn:
         if op.opcode == OP_Ack and op.size == 4:
             (crc,) = struct.unpack("<I", op.payload)
             now = time.time()
-            logging.info("crc: %s 0x%08x in %.1fs", self.filename, crc, now - self.op_start)
+            if self.op_start is not None:
+                logging.info("crc: %s 0x%08x in %.1fs", self.filename, crc, now - self.op_start)
         return self.__decode_ftp_ack_and_nack(op)
 
     def cmd_cancel(self) -> MAVFTPReturn:
@@ -972,7 +1016,7 @@ def cmd_cancel(self) -> MAVFTPReturn:
 
     def cmd_status(self) -> MAVFTPReturn:
         """Show status."""
-        if self.fh is None:
+        if self.fh is None or self.op_start is None:
             logging.info("No transfer in progress")
         else:
             ofs = self.fh.tell()