FEATURE: Add MAVFTP support for uploading lua scripts and other files

.vscode/launch.json

@@ -1,11 +1,36 @@
 {
     "configurations": [
+        {
+            "name": "Python Debugger: Current File without Arguments",
+            "type": "debugpy",
+            "request": "launch",
+            "program": "${file}",
+            "console": "integratedTerminal"
+        },
+        {
+            "name": "Python Debugger: Current File with Arguments",
+            "type": "debugpy",
+            "request": "launch",
+            "program": "${file}",
+            "console": "integratedTerminal",
+            "args": "rmdir lkvn"
+        },
+        {
+            "name": "Python Debugger: Current File with Debug Arguments",
+            "type": "debugpy",
+            "request": "launch",
+            "program": "${file}",
+            "console": "integratedTerminal",
+            "args": "--loglevel DEBUG --debug 2 rmdir lkvn"
+        },
         {
             "name": "Python Debugger: ardupilot_methodic_configurator.py",
             "type": "debugpy",
             "request": "launch",
             "program": "MethodicConfigurator/ardupilot_methodic_configurator.py",
-            "args": ["--allow-editing-template-files"],
+            "args": [
+                "--allow-editing-template-files"
+            ],
         }
     ]
 }

MethodicConfigurator/backend_filesystem.py

@@ -32,6 +32,8 @@
 
 from zipfile import ZipFile
 
+from requests import get as requests_get
+
 from MethodicConfigurator.annotate_params import PARAM_DEFINITION_XML_FILE, Par
 from MethodicConfigurator.annotate_params import get_xml_url
 from MethodicConfigurator.annotate_params import get_xml_dir
@@ -156,7 +158,7 @@ def rename_parameter_files(self):
             for new_filename in self.configuration_steps:
                 if 'old_filenames' in self.configuration_steps[new_filename]:
                     for old_filename in self.configuration_steps[new_filename]['old_filenames']:
-                        if self.intermediate_parameter_file_exists(old_filename) and old_filename != new_filename:
+                        if self.vehicle_configuration_file_exists(old_filename) and old_filename != new_filename:
                             new_filename_path = os_path.join(self.vehicle_dir, new_filename)
                             old_filename_path = os_path.join(self.vehicle_dir, old_filename)
                             os_rename(old_filename_path, new_filename_path)
@@ -272,9 +274,9 @@ def export_to_param(self, params: Dict[str, 'Par'], filename_out: str, annotate_
                                            "missionplanner",
                                            self.param_default_dict)
 
-    def intermediate_parameter_file_exists(self, filename: str) -> bool:
+    def vehicle_configuration_file_exists(self, filename: str) -> bool:
         """
-        Check if an intermediate parameter file exists in the vehicle directory.
+        Check if a vehicle configuration file exists in the vehicle directory.
 
         Parameters:
         - filename (str): The name of the file to check.
@@ -556,6 +558,42 @@ def write_param_default_values_to_file(self, param_default_values: Dict[str, 'Pa
         if self.write_param_default_values(param_default_values):
             Par.export_to_param(Par.format_params(self.param_default_dict), os_path.join(self.vehicle_dir, filename))
 
+    def get_download_url_and_local_filename(self, selected_file: str) -> Tuple[str, str]:
+        if selected_file in self.configuration_steps:
+            if 'download_file' in self.configuration_steps[selected_file] and \
+               self.configuration_steps[selected_file]['download_file']:
+                src = self.configuration_steps[selected_file]['download_file'].get('source_url', '')
+                dst = self.configuration_steps[selected_file]['download_file'].get('dest_local', '')
+                if self.vehicle_dir and src and dst:
+                    return src, os_path.join(self.vehicle_dir, dst)
+        return '', ''
+
+    def get_upload_local_and_remote_filenames(self, selected_file: str) -> Tuple[str, str]:
+        if selected_file in self.configuration_steps:
+            if 'upload_file' in self.configuration_steps[selected_file] and \
+               self.configuration_steps[selected_file]['upload_file']:
+                src = self.configuration_steps[selected_file]['upload_file'].get('source_local', '')
+                dst = self.configuration_steps[selected_file]['upload_file'].get('dest_on_fc', '')
+                if self.vehicle_dir and src and dst:
+                    return os_path.join(self.vehicle_dir, src), dst
+        return '', ''
+
+    @staticmethod
+    def download_file_from_url(url: str, local_filename: str, timeout: int=5) -> bool:
+        if not url or not local_filename:
+            logging_error("URL or local filename not provided.")
+            return False
+        logging_info("Downloading %s from %s", local_filename, url)
+        response = requests_get(url, timeout=timeout)
+
+        if response.status_code == 200:
+            with open(local_filename, "wb") as file:
+                file.write(response.content)
+            return True
+
+        logging_error("Failed to download the file")
+        return False
+
     @staticmethod
     def add_argparse_arguments(parser):
         parser.add_argument('-t', '--vehicle-type',

MethodicConfigurator/backend_flightcontroller.py

@@ -13,7 +13,6 @@
 from logging import warning as logging_warning
 from logging import error as logging_error
 
-from sys import exit as sys_exit
 from time import sleep as time_sleep
 from time import time as time_time
 from os import path as os_path
@@ -32,7 +31,6 @@
 
 from MethodicConfigurator.backend_flightcontroller_info import BackendFlightcontrollerInfo
 from MethodicConfigurator.backend_mavftp import MAVFTP
-from MethodicConfigurator.param_ftp import ftp_param_decode
 
 from MethodicConfigurator.argparse_check_range import CheckRange
 
@@ -45,23 +43,6 @@
 except Exception: # pylint: disable=broad-exception-caught
     pass
 
-preferred_ports = [
-    '*FTDI*',
-    "*Arduino_Mega_2560*",
-    "*3D*",
-    "*USB_to_UART*",
-    '*Ardu*',
-    '*PX4*',
-    '*Hex_*',
-    '*Holybro_*',
-    '*mRo*',
-    '*FMU*',
-    '*Swift-Flyer*',
-    '*Serial*',
-    '*CubePilot*',
-    '*Qiotek*',
-]
-
 
 class FakeSerialForUnitTests():
     """
@@ -387,47 +368,29 @@ def __download_params_via_mavlink(self, progress_callback=None) -> Dict[str, flo
         return parameters
 
     def download_params_via_mavftp(self, progress_callback=None) -> Tuple[Dict[str, float], Dict[str, 'Par']]:
-        # FIXME global variables should be avoided but I found no other practical way get the FTP result pylint: disable=fixme
-        global ftp_should_run  # pylint: disable=global-variable-undefined
-        global pdict  # pylint: disable=global-variable-undefined
-        global defdict  # pylint: disable=global-variable-undefined
-        ftp_should_run = True
-        pdict = {}
-        defdict = {}
 
         mavftp = MAVFTP(self.master,
                         target_system=self.master.target_system,
                         target_component=self.master.target_component)
 
-        def callback(fh):
-            '''called on ftp completion'''
-            global ftp_should_run  # pylint: disable=global-variable-undefined
-            global pdict  # pylint: disable=global-variable-not-assigned
-            global defdict  # pylint: disable=global-variable-not-assigned
-            data = fh.read()
-            logging_info("'MAVFTP get' parameter values and defaults done, got %u bytes", len(data))
-            pdata = ftp_param_decode(data)
-            if pdata is None:
-                logging_error("param decode failed")
-                sys_exit(1)
-
-            for (name, value, _ptype) in pdata.params:
-                pdict[name.decode('utf-8')] = value
-            logging_info("got %u parameter values", len(pdict))
-            if pdata.defaults:
-                for (name, value, _ptype) in pdata.defaults:
-                    defdict[name.decode('utf-8')] = Par(value)
-                logging_info("got %u parameter default values", len(defdict))
-            ftp_should_run = False
-            progress_callback(len(data), len(data))
-
-        mavftp.cmd_get(['@PARAM/param.pck?withdefaults=1'], callback=callback, callback_progress=progress_callback)
-
-        while ftp_should_run:
-            m = self.master.recv_match(type=['FILE_TRANSFER_PROTOCOL'], timeout=0.1)
-            if m is not None:
-                mavftp.mavlink_packet(m)
-            mavftp.idle_task()
+        def get_params_progress_callback(completion: float):
+            if progress_callback is not None and completion is not None:
+                progress_callback(int(completion*100), 100)
+
+        complete_param_filename = 'complete.param'
+        default_param_filename = '00_default.param'
+        mavftp.cmd_getparams([complete_param_filename, default_param_filename], progress_callback=get_params_progress_callback)
+        ret = mavftp.process_ftp_reply('getparams', timeout=10)
+        pdict = {}
+        if ret.error_code == 0:
+            # load the parameters from the file
+            par_dict = Par.load_param_file_into_dict(complete_param_filename)
+            for name, data in par_dict.items():
+                pdict[name] = data.value
+            defdict = Par.load_param_file_into_dict(default_param_filename)
+        else:
+            ret.display_message()
+            defdict = {}
 
         return pdict, defdict
 
@@ -500,8 +463,25 @@ def __list_network_ports():
         """
         return ['tcp:127.0.0.1:5760', 'udp:127.0.0.1:14550']
 
+# pylint: disable=duplicate-code
     def __auto_detect_serial(self):
         serial_list = []
+        preferred_ports = [
+            '*FTDI*',
+            "*Arduino_Mega_2560*",
+            "*3D*",
+            "*USB_to_UART*",
+            '*Ardu*',
+            '*PX4*',
+            '*Hex_*',
+            '*Holybro_*',
+            '*mRo*',
+            '*FMU*',
+            '*Swift-Flyer*',
+            '*Serial*',
+            '*CubePilot*',
+            '*Qiotek*',
+        ]
         for connection in self.__connection_tuples:
             if connection[1] and 'mavlink' in connection[1].lower():
                 serial_list.append(mavutil.SerialPort(device=connection[0], description=connection[1]))
@@ -518,13 +498,32 @@ def __auto_detect_serial(self):
                 serial_list.pop(1)
 
         return serial_list
+# pylint: enable=duplicate-code
 
     def get_connection_tuples(self):
         """
         Get all available connections.
         """
         return self.__connection_tuples
 
+    def upload_file(self, local_filename: str, remote_filename: str, progress_callback=None):
+        """ Upload a file to the flight controller. """
+        if self.master is None:
+            return False
+        mavftp = MAVFTP(self.master,
+                        target_system=self.master.target_system,
+                        target_component=self.master.target_component)
+
+        def put_progress_callback(completion: float):
+            if progress_callback is not None and completion is not None:
+                progress_callback(int(completion*100), 100)
+
+        mavftp.cmd_put([local_filename, remote_filename], progress_callback=put_progress_callback)
+        ret = mavftp.process_ftp_reply('CreateFile', timeout=10)
+        if ret.error_code != 0:
+            ret.display_message()
+        return ret.error_code == 0
+
     @staticmethod
     def add_argparse_arguments(parser):
         parser.add_argument('--device',