|
8 | 8 | SPDX-License-Identifier: GPL-3.0-or-later |
9 | 9 | """ |
10 | 10 |
|
| 11 | +import os |
11 | 12 | from argparse import ArgumentParser |
12 | 13 | from logging import debug as logging_debug |
13 | 14 | from logging import error as logging_error |
@@ -1088,6 +1089,169 @@ def put_progress_callback(completion: float) -> None: |
1088 | 1089 | ret.display_message() |
1089 | 1090 | return ret.error_code == 0 |
1090 | 1091 |
|
| 1092 | + def download_last_flight_log( |
| 1093 | + self, local_filename: str, progress_callback: Union[None, Callable[[int, int], None]] = None |
| 1094 | + ) -> bool: |
| 1095 | + """Download the last flight log from the flight controller.""" |
| 1096 | + error_msg = "" |
| 1097 | + |
| 1098 | + if self.master is None: |
| 1099 | + error_msg = _("No flight controller connected") |
| 1100 | + elif not self.info.is_mavftp_supported: |
| 1101 | + error_msg = _("MAVFTP is not supported by the flight controller") |
| 1102 | + |
| 1103 | + if error_msg: |
| 1104 | + logging_error(error_msg) |
| 1105 | + return False |
| 1106 | + |
| 1107 | + mavftp = MAVFTP(self.master, target_system=self.master.target_system, target_component=self.master.target_component) |
| 1108 | + |
| 1109 | + def get_progress_callback(completion: float) -> None: |
| 1110 | + if progress_callback is not None and completion is not None: |
| 1111 | + progress_callback(int(completion * 100), 100) |
| 1112 | + |
| 1113 | + try: |
| 1114 | + # Try to get the last log number using different methods |
| 1115 | + remote_filenumber = self._get_last_log_number(mavftp) |
| 1116 | + if remote_filenumber is None: |
| 1117 | + return False |
| 1118 | + |
| 1119 | + # We want the previous log, not the current one (which might be incomplete) |
| 1120 | + remote_filenumber -= 1 |
| 1121 | + if remote_filenumber < 1: |
| 1122 | + logging_error(_("No previous flight log available")) |
| 1123 | + return False |
| 1124 | + |
| 1125 | + return self._download_log_file(mavftp, remote_filenumber, local_filename, get_progress_callback) |
| 1126 | + |
| 1127 | + except Exception as e: |
| 1128 | + logging_error(_("Error during flight log download: %s"), str(e)) |
| 1129 | + return False |
| 1130 | + |
| 1131 | + def _get_last_log_number(self, mavftp: MAVFTP) -> Union[int, None]: |
| 1132 | + """Get the last log number using multiple fallback methods.""" |
| 1133 | + # Method 1: Try to get LASTLOG.TXT |
| 1134 | + log_number = self._get_log_number_from_lastlog_txt(mavftp) |
| 1135 | + if log_number is not None: |
| 1136 | + return log_number |
| 1137 | + |
| 1138 | + # Method 2: Try to list the logs directory and find the highest numbered log |
| 1139 | + log_number = self._get_log_number_from_directory_listing(mavftp) |
| 1140 | + if log_number is not None: |
| 1141 | + return log_number |
| 1142 | + |
| 1143 | + # Method 3: Try common log numbers (scan backwards from a reasonable max) |
| 1144 | + log_number = self._get_log_number_by_scanning(mavftp) |
| 1145 | + if log_number is not None: |
| 1146 | + return log_number |
| 1147 | + |
| 1148 | + logging_error(_("Could not determine the last log number using any method")) |
| 1149 | + return None |
| 1150 | + |
| 1151 | + def _get_log_number_from_lastlog_txt(self, mavftp: MAVFTP) -> Union[int, None]: |
| 1152 | + """Try to get the log number from LASTLOG.TXT file.""" |
| 1153 | + try: |
| 1154 | + temp_lastlog_file = "temp_lastlog.txt" |
| 1155 | + logging_info(_("Trying to get log number from LASTLOG.TXT")) |
| 1156 | + mavftp.cmd_get(["/APM/LOGS/LASTLOG.TXT", temp_lastlog_file]) |
| 1157 | + ret = mavftp.process_ftp_reply("OpenFileRO", timeout=10) |
| 1158 | + if ret.error_code != 0: |
| 1159 | + logging_warning(_("LASTLOG.TXT not available, trying alternative methods")) |
| 1160 | + return None |
| 1161 | + |
| 1162 | + return self._extract_log_number_from_file(temp_lastlog_file) |
| 1163 | + except Exception as e: |
| 1164 | + logging_warning(_("Failed to get log number from LASTLOG.TXT: %s"), str(e)) |
| 1165 | + return None |
| 1166 | + |
| 1167 | + def _get_log_number_from_directory_listing(self, _mavftp: MAVFTP) -> Union[int, None]: |
| 1168 | + """Try to get the log number by listing the logs directory.""" |
| 1169 | + try: |
| 1170 | + logging_info(_("Trying to get log number from directory listing")) |
| 1171 | + # This would require implementing directory listing functionality |
| 1172 | + # For now, we'll skip this method as it's more complex |
| 1173 | + logging_info(_("Directory listing method not implemented yet")) |
| 1174 | + return None |
| 1175 | + except Exception as e: |
| 1176 | + logging_warning(_("Failed to get log number from directory listing: %s"), str(e)) |
| 1177 | + return None |
| 1178 | + |
| 1179 | + def _get_log_number_by_scanning(self, mavftp: MAVFTP) -> Union[int, None]: |
| 1180 | + """Try to find the last log using binary search for efficiency.""" |
| 1181 | + try: |
| 1182 | + logging_info(_("Trying to find log number using binary search")) |
| 1183 | + |
| 1184 | + # Binary search to find the highest log number |
| 1185 | + low = 1 |
| 1186 | + high = 9999 # Reasonable upper bound for log numbers |
| 1187 | + last_found = None |
| 1188 | + |
| 1189 | + while low <= high: |
| 1190 | + mid = (low + high) // 2 |
| 1191 | + remote_filename = f"/APM/LOGS/{mid:08}.BIN" |
| 1192 | + |
| 1193 | + # Test if this log file exists |
| 1194 | + temp_test_file = f"temp_test_{mid}.tmp" |
| 1195 | + mavftp.cmd_get([remote_filename, temp_test_file]) |
| 1196 | + ret = mavftp.process_ftp_reply("OpenFileRO", timeout=5) # Must be > idle_detection_time (3.7s) |
| 1197 | + |
| 1198 | + # Clean up the temp file if it was created |
| 1199 | + if os.path.exists(temp_test_file): |
| 1200 | + os.remove(temp_test_file) |
| 1201 | + |
| 1202 | + if ret.error_code == 0: |
| 1203 | + # File exists, search in upper half |
| 1204 | + last_found = mid |
| 1205 | + low = mid + 1 |
| 1206 | + logging_debug(_("Log %d exists, searching higher"), mid) |
| 1207 | + else: |
| 1208 | + # File doesn't exist, search in lower half |
| 1209 | + high = mid - 1 |
| 1210 | + logging_debug(_("Log %d doesn't exist, searching lower"), mid) |
| 1211 | + |
| 1212 | + if last_found is not None: |
| 1213 | + logging_info(_("Found highest log number using binary search: %d"), last_found) |
| 1214 | + return last_found |
| 1215 | + |
| 1216 | + logging_warning(_("No log files found using binary search")) |
| 1217 | + return None |
| 1218 | + |
| 1219 | + except Exception as e: |
| 1220 | + logging_warning(_("Failed to scan for log numbers using binary search: %s"), str(e)) |
| 1221 | + return None |
| 1222 | + |
| 1223 | + def _download_log_file( |
| 1224 | + self, mavftp: MAVFTP, remote_filenumber: int, local_filename: str, get_progress_callback: Callable |
| 1225 | + ) -> bool: |
| 1226 | + """Download the actual log file from the flight controller.""" |
| 1227 | + remote_filename = f"/APM/LOGS/{remote_filenumber:08}.BIN" |
| 1228 | + logging_info(_("Downloading flight log %s to %s"), remote_filename, local_filename) |
| 1229 | + |
| 1230 | + # Download the actual log file |
| 1231 | + mavftp.cmd_get([remote_filename, local_filename], progress_callback=get_progress_callback) |
| 1232 | + ret = mavftp.process_ftp_reply("OpenFileRO", timeout=0) # No timeout for large log files |
| 1233 | + if ret.error_code != 0: |
| 1234 | + logging_error(_("Failed to download flight log %s"), remote_filename) |
| 1235 | + ret.display_message() |
| 1236 | + return False |
| 1237 | + |
| 1238 | + logging_info(_("Successfully downloaded flight log to %s"), local_filename) |
| 1239 | + return True |
| 1240 | + |
| 1241 | + def _extract_log_number_from_file(self, temp_lastlog_file: str) -> Union[int, None]: |
| 1242 | + """Extract log number from LASTLOG.TXT file and clean up the temporary file.""" |
| 1243 | + try: |
| 1244 | + with open(temp_lastlog_file, encoding="UTF-8") as file: |
| 1245 | + file_contents = file.readline() |
| 1246 | + return int(file_contents.strip()) |
| 1247 | + except (FileNotFoundError, ValueError) as e: |
| 1248 | + logging_error(_("Could not extract last log file number from LASTLOG.TXT: %s"), e) |
| 1249 | + return None |
| 1250 | + finally: |
| 1251 | + # Clean up the temporary file |
| 1252 | + if os.path.exists(temp_lastlog_file): |
| 1253 | + os.remove(temp_lastlog_file) |
| 1254 | + |
1091 | 1255 | @staticmethod |
1092 | 1256 | def add_argparse_arguments(parser: ArgumentParser) -> ArgumentParser: |
1093 | 1257 | parser.add_argument( |
|
0 commit comments