mavftp_example.py

#!/usr/bin/env python3

'''
MAVLink File Transfer Protocol support example

SPDX-FileCopyrightText: 2024 Amilcar Lucas

SPDX-License-Identifier: GPL-3.0-or-later
'''

from argparse import ArgumentParser

from logging import basicConfig as logging_basicConfig
from logging import getLevelName as logging_getLevelName

from logging import debug as logging_debug
from logging import info as logging_info
from logging import error as logging_error

import os
import sys
#import time

import requests


from pymavlink import mavutil
from pymavlink import mavftp

old_mavftp_member_variable_values = {}


# pylint: disable=duplicate-code
def argument_parser():
    """
    Parses command-line arguments for the script.
    """
    parser = ArgumentParser(description='This main is just an example, adapt it to your needs')
    parser.add_argument("--baudrate", type=int, default=115200,
                        help="master port baud rate. Defaults to %(default)s")
    parser.add_argument("--device", type=str, default='',
                        help="serial device. For windows use COMx where x is the port number. "
                                "For Unix use /dev/ttyUSBx where x is the port number. Defaults to autodetection")
    parser.add_argument("--source-system", type=int, default=250,
                        help='MAVLink source system for this GCS. Defaults to %(default)s')
    parser.add_argument("--loglevel", default="INFO",
                        help="log level. Defaults to %(default)s")

    # MAVFTP settings
    parser.add_argument("--debug", type=int, default=0, choices=[0, 1, 2],
                        help="Debug level 0 for none, 2 for max verbosity. Defaults to %(default)s")

    return parser.parse_args()

def auto_detect_serial():
    preferred_ports = [
        '*FTDI*',
        "*3D*",
        "*USB_to_UART*",
        '*Ardu*',
        '*PX4*',
        '*Hex_*',
        '*Holybro_*',
        '*mRo*',
        '*FMU*',
        '*Swift-Flyer*',
        '*Serial*',
        '*CubePilot*',
        '*Qiotek*',
    ]
    serial_list = mavutil.auto_detect_serial(preferred_list=preferred_ports)
    serial_list.sort(key=lambda x: x.device)

    # remove OTG2 ports for dual CDC
    if len(serial_list) == 2 and serial_list[0].device.startswith("/dev/serial/by-id"):
        if serial_list[0].device[:-1] == serial_list[1].device[0:-1]:
            serial_list.pop(1)

    return serial_list


def auto_connect(device):
    comport = None
    if device:
        comport = mavutil.SerialPort(device=device, description=device)
    else:
        autodetect_serial = auto_detect_serial()
        if autodetect_serial:
            # Resolve the soft link if it's a Linux system
            if os.name == 'posix':
                try:
                    dev = autodetect_serial[0].device
                    logging_debug("Auto-detected device %s", dev)
                    # Get the directory part of the soft link
                    softlink_dir = os.path.dirname(dev)
                    # Resolve the soft link and join it with the directory part
                    resolved_path = os.path.abspath(os.path.join(softlink_dir, os.readlink(dev)))
                    autodetect_serial[0].device = resolved_path
                    logging_debug("Resolved soft link %s to %s", dev, resolved_path)
                except OSError:
                    pass # Not a soft link, proceed with the original device path
            comport = autodetect_serial[0]
        else:
            logging_error("No serial ports found. Please connect a flight controller and try again.")
            sys.exit(1)
    return comport


def wait_heartbeat(m):
    '''wait for a heartbeat so we know the target system IDs'''
    logging_info("Waiting for flight controller heartbeat")
    m.wait_heartbeat()
    logging_info("Got heartbeat from system %u, component %u", m.target_system, m.target_system)
# pylint: enable=duplicate-code


def delete_local_file_if_exists(filename):
    if os.path.exists(filename):
        os.remove(filename)


def get_list_dir(mav_ftp, directory):
    ret = mav_ftp.cmd_list([directory])
    ret.display_message()
    debug_class_member_variable_changes(mav_ftp)


def get_file(mav_ftp, remote_filename, local_filename, timeout=5):
    #session = mav_ftp.session # save the session to restore it after the file transfer
    mav_ftp.cmd_get([remote_filename, local_filename])
    ret = mav_ftp.process_ftp_reply('OpenFileRO', timeout=timeout)
    ret.display_message()
    #mav_ftp.session = session # FIXME: this is a huge workaround hack # pylint: disable=fixme
    debug_class_member_variable_changes(mav_ftp)
    #time.sleep(0.2)


def get_last_log(mav_ftp):
    try:
        with open('LASTLOG.TXT', 'r', encoding='UTF-8') as file:
            file_contents = file.readline()
            remote_filenumber = int(file_contents.strip())
    except FileNotFoundError:
        logging_error("File LASTLOG.TXT not found.")
        return
    except ValueError:
        logging_error("Could not extract last log file number from LASTLOG.TXT contants %s", file_contents)
        return
    remote_filenumber = remote_filenumber - 1 # we do not want the very last log
    remote_filename = f'/APM/LOGS/{remote_filenumber:08}.BIN'
    get_file(mav_ftp, remote_filename, 'LASTLOG.BIN', 0)


def download_script(url, local_filename):
    # Download the script from the internet to the PC
    response = requests.get(url, timeout=5)

    if response.status_code == 200:
        with open(local_filename, "wb") as file:
            file.write(response.content)
    else:
        logging_error("Failed to download the file")


def create_directory(mav_ftp, remote_directory):
    ret = mav_ftp.cmd_mkdir([remote_directory])
    ret.display_message()
    debug_class_member_variable_changes(mav_ftp)


def remove_directory(mav_ftp, remote_directory):
    ret = mav_ftp.cmd_rmdir([remote_directory])
    ret.display_message()
    debug_class_member_variable_changes(mav_ftp)


def upload_script(mav_ftp, remote_directory, local_filename, timeout):
    # Upload it from the PC to the flight controller
    mav_ftp.cmd_put([local_filename, remote_directory + '/' + local_filename])
    ret = mav_ftp.process_ftp_reply('CreateFile', timeout=timeout)
    ret.display_message()
    debug_class_member_variable_changes(mav_ftp)


def debug_class_member_variable_changes(instance):
    return
    global old_mavftp_member_variable_values  # pylint: disable=global-statement, unreachable
    new_mavftp_member_variable_values = instance.__dict__
    if old_mavftp_member_variable_values and instance.ftp_settings.debug > 1:  # pylint: disable=too-many-nested-blocks
        logging_info(f"{instance.__class__.__name__} member variable changes:")
        for key, value in new_mavftp_member_variable_values.items():
            if old_mavftp_member_variable_values[key] != value:
                old_value = old_mavftp_member_variable_values[key]
                if old_value and isinstance(value, mavftp.FTP_OP):
                    # Convert both new and old FTP_OP instances to dictionaries for comparison
                    new_op_dict = dict(value.items())
                    old_op_dict = dict(old_value.items()) if isinstance(old_value, mavftp.FTP_OP) else {}
                    for op_key, op_value in new_op_dict.items():
                        old_op_value = old_op_dict.get(op_key)
                        if old_op_value != op_value:
                            logging_info(f"CHANGED {key}.{op_key}: {old_op_value} -> {op_value}")
                else:
                    logging_info(f"CHANGED {key}: {old_mavftp_member_variable_values[key]} -> {value}")
    old_mavftp_member_variable_values = new_mavftp_member_variable_values.copy()

def main():
    '''for testing/example purposes only'''
    args = argument_parser()

    logging_basicConfig(level=logging_getLevelName(args.loglevel), format='%(levelname)s - %(message)s')

    # create a mavlink serial instance
    comport = auto_connect(args.device)
    master = mavutil.mavlink_connection(comport.device, baud=args.baudrate, source_system=args.source_system)

    # wait for the heartbeat msg to find the system ID
    wait_heartbeat(master)

    mav_ftp = mavftp.MAVFTP(master,
                            target_system=master.target_system,
                            target_component=master.target_component)

    mav_ftp.ftp_settings.debug = args.debug

    if args.loglevel == 'DEBUG':
        mav_ftp.ftp_settings.debug = 2

    debug_class_member_variable_changes(mav_ftp)

    get_list_dir(mav_ftp, '/APM/LOGS')

    delete_local_file_if_exists("params.param")
    delete_local_file_if_exists("defaults.param")
    mav_ftp.cmd_getparams(["params.param", "defaults.param"])
    ret = mav_ftp.process_ftp_reply('OpenFileRO', timeout=500)
    ret.display_message()

    get_list_dir(mav_ftp, '/APM/LOGS')

    #delete_local_file_if_exists("LASTLOG.TXT")
    delete_local_file_if_exists("LASTLOG.BIN")

    #get_file(mav_ftp, '/APM/LOGS/LASTLOG.TXT', 'LASTLOG.TXT')

    get_list_dir(mav_ftp, '/APM/LOGS')

    #get_file(mav_ftp, '/APM/LOGS/LASTLOG.TXT', 'LASTLOG2.TXT')

    get_last_log(mav_ftp)

    remove_directory(mav_ftp, "test_dir")
    create_directory(mav_ftp, "test_dir")
    remove_directory(mav_ftp, "test_dir")
    create_directory(mav_ftp, "test_dir2")

    remote_directory = '/APM/Scripts'
    #create_directory(mav_ftp, remote_directory)

    url = "https://discuss.ardupilot.org/uploads/short-url/4pyrl7PcfqiMEaRItUhljuAqLSs.lua"
    local_filename = "copter-magfit-helper.lua"

    if not os.path.exists(local_filename):
        download_script(url, local_filename)

    upload_script(mav_ftp, remote_directory, local_filename, 5)

    url = "https://raw.githubusercontent.com/ArduPilot/ardupilot/Copter-4.5/libraries/AP_Scripting/applets/" \
            "VTOL-quicktune.lua"
    local_filename = "VTOL-quicktune.lua"

    if not os.path.exists(local_filename):
        download_script(url, local_filename)

    upload_script(mav_ftp, remote_directory, local_filename, 5)

    master.close()


if __name__ == "__main__":
    main()