diff --git a/ARCHITECTURE_log_analysis.md b/ARCHITECTURE_log_analysis.md index 2fc43bf40..7f397b486 100644 --- a/ARCHITECTURE_log_analysis.md +++ b/ARCHITECTURE_log_analysis.md @@ -12,13 +12,14 @@ The main components are: 1. **Log Analysis Backend** - Loads and validates ArduPilot `.bin` logs and prepares the data required by the analysis layer. - * [`backend_log_analysis.py`](ardupilot_methodic_configurator/backend_log_analysis.py) + * [`backend_log_analysis.py`](ardupilot_methodic_configurator/log_analysis/backend_log_analysis.py) * [`backend_log_extraction.py`](ardupilot_methodic_configurator/log_analysis/backend_log_extraction.py) 2. **Log Analysis Data Models** - Contains the analysis pipeline, shared context, quality models, analysis models, and result structures. * [`data_model_log_analysis.py`](ardupilot_methodic_configurator/log_analysis/data_model_log_analysis.py) * [`data_model_log_analysis_context.py`](ardupilot_methodic_configurator/log_analysis/data_model_log_analysis_context.py) + * [`data_model_parameter_derivation.py`](ardupilot_methodic_configurator/log_analysis/data_model_parameter_derivation.py) * [`data_model_log_quality.py`](ardupilot_methodic_configurator/log_analysis/data_model_log_quality.py) * [`data_model_log_quality_check.py`](ardupilot_methodic_configurator/log_analysis/data_model_log_quality_check.py) * [`data_model_quality_base.py`](ardupilot_methodic_configurator/log_analysis/data_model_quality_base.py) @@ -48,14 +49,41 @@ The main components are: The log extraction backend reads an ArduPilot `.bin` flight log and creates the internal `LogData` representation. -The extraction layer is responsible for parsing the log and providing the data required by the analysis models. Individual analysis models do not parse the `.bin` file themselves. +The extraction layer is responsible for parsing the log and providing the data required by the analysis models. Individual analysis models do not parse the `.bin` file +themselves. The extraction backend can also report progress through a callback so that the frontend can display parsing progress without depending on the parser implementation. +### Numeric Storage and Scaling + +`LogData` keeps one compact NumPy structured array for each log message type. The stored representation is selected to limit the permanent memory cost of long flight +logs; conversion to analysis units happens only when required. + +ArduPilot DataFlash fixed-point format characters `c`, `C`, `e`, `E`, and `L` are stored as their original integer values. Although pymavlink exposes scaled values +through normal attribute access and `to_dict()`, extraction reads the corresponding `DFMessage._elements` entry for these fields. `_elements` is a pymavlink private API, +but it is maintained by the ArduPilot project and is isolated to the extraction adapter with fixture-based regression coverage. + +`LogData.get_field(..., scaled=True)` applies the fixed-point multiplier with a vectorized `float64` NumPy operation. The temporary scaled array is not cached, so +analyses that do not request a field do not pay its memory cost. + +FMTU multipliers use a width-aware policy: + +* `f` and `d` fields apply their dynamic FMTU multiplier while being ingested, retaining their original floating-point dtype and avoiding repeated scaling for common + telemetry fields. +* Integer fields whose multiplier would require a wider or fractional representation retain their compact stored value and scale lazily. +* Multipliers equal to one leave values unchanged. + +Each `MessageSchema` records `stored_units`, `scaled_units`, `multipliers`, and `multipliers_applied_at_ingest`. These fields make the storage-to-analysis conversion +explicit and prevent a multiplier from being applied twice. + +Analysis code always uses `LogData`'s default scaled representation. The `scaled=False` option is retained for low-level diagnostics and regression tests, not for +production analysis. A result timestamp is converted to microseconds only when populating `LogAnalysis.timestamp_us`; parameter values with different documented units +are converted explicitly before comparison. + ## Log Analysis Backend -[`backend_log_analysis.py`](ardupilot_methodic_configurator/backend_log_analysis.py) acts as the orchestration layer between log extraction, Methodic Configurator context, -and the analysis data models. +[`backend_log_analysis.py`](ardupilot_methodic_configurator/log_analysis/backend_log_analysis.py) acts as the orchestration layer between log extraction, Methodic +Configurator context, and the analysis data models. Its responsibilities are: @@ -98,12 +126,18 @@ The context contains: * Methodic Configurator configuration steps * Vehicle component information * ArduPilot parameter documentation +* A parameter-derivation service The backend constructs this context after extracting the log. The analysis models receive the context instead of independently loading these resources. This keeps data loading outside the analysis models and avoids duplicated filesystem and configuration logic. +Detailed analysis models use the parameter-derivation service to evaluate forced +and derived configuration parameters. The default adapter reuses the shared +configuration-step expression evaluator with only the already-loaded context +data; tests can supply a small replacement service without a vehicle directory. + ## Analysis Pipeline [`data_model_log_analysis.py`](ardupilot_methodic_configurator/log_analysis/data_model_log_analysis.py) contains the main domain-level analysis pipeline. diff --git a/ardupilot_methodic_configurator/frontend_tkinter_log_analysis.py b/ardupilot_methodic_configurator/frontend_tkinter_log_analysis.py index 7347e4bb7..d41d48d4b 100644 --- a/ardupilot_methodic_configurator/frontend_tkinter_log_analysis.py +++ b/ardupilot_methodic_configurator/frontend_tkinter_log_analysis.py @@ -11,29 +11,23 @@ """ import tkinter as tk +from collections.abc import Callable from enum import Enum +from functools import partial from pathlib import Path from tkinter import messagebox, ttk from typing import Any from ardupilot_methodic_configurator import _ from ardupilot_methodic_configurator.backend_internet import webbrowser_open_url +from ardupilot_methodic_configurator.data_model_par_dict import Par from ardupilot_methodic_configurator.frontend_tkinter_autoresize_combobox import AutoResizeCombobox from ardupilot_methodic_configurator.frontend_tkinter_base_window import BaseWindow from ardupilot_methodic_configurator.frontend_tkinter_scroll_frame import ScrollFrame from ardupilot_methodic_configurator.frontend_tkinter_show import show_tooltip from ardupilot_methodic_configurator.frontend_tkinter_tuning_report import TuningReportWindow -from ardupilot_methodic_configurator.log_analysis.data_model_log_analysis import QUALITY_AND_ANALYSIS_MODELS, LogSummary -from ardupilot_methodic_configurator.log_analysis.data_model_log_analysis_result import LogAnalysis, LogAnalysisResult -from ardupilot_methodic_configurator.log_analysis.data_model_log_quality import LogQualityResult - -_SUBSYSTEM_TO_COMPONENT_KEYS: dict[str, tuple[str, ...]] = { - "Battery": ("Battery", "Battery Monitor"), - "ESC telemetry": ("ESC", "Motors"), - "IMU": ("Flight Controller",), - "VIBE": ("Flight Controller",), - "GPS": ("GNSS Receiver",), -} +from ardupilot_methodic_configurator.log_analysis.data_model_log_analysis import LogSummary +from ardupilot_methodic_configurator.log_analysis.data_model_log_analysis_result import LogAnalysis class Severity(Enum): @@ -63,28 +57,6 @@ class Severity(Enum): } -def paired_quality_and_analysis_results(summary: LogSummary) -> list[tuple[LogQualityResult, LogAnalysisResult | None]]: - offset = len(summary.quality_results) - len(QUALITY_AND_ANALYSIS_MODELS) - if offset not in (0, 1): - msg = ( - f"Unexpected quality_results length ({len(summary.quality_results)}) vs " - f"QUALITY_AND_ANALYSIS_MODELS length ({len(QUALITY_AND_ANALYSIS_MODELS)})." - ) - raise AssertionError(msg) - - analysis_iter = iter(summary.analysis_results) - paired: list[tuple[LogQualityResult, LogAnalysisResult | None]] = [] - - for i, (_quality_cls, analysis_cls) in enumerate(QUALITY_AND_ANALYSIS_MODELS): - if analysis_cls is None: - continue - quality_result = summary.quality_results[i + offset] - analysis_result = next(analysis_iter) if quality_result.available else None - paired.append((quality_result, analysis_result)) - - return paired - - def _collect_links(quality_dict: dict[str, Any] | None, analysis_dict: dict[str, Any] | None) -> list[dict[str, Any]]: seen: set[tuple[str | None, str | None]] = set() links: list[dict[str, Any]] = [] @@ -146,20 +118,24 @@ def _format_component(component: dict[str, Any]) -> list[str]: # pylint: disabl class LogAnalysisReportWindow(BaseWindow): # pylint: disable=too-many-instance-attributes """Log analysis window.""" - def __init__( + def __init__( # pylint: disable=too-many-arguments, too-many-positional-arguments self, root_tk: tk.Tk | tk.Toplevel, summary: LogSummary, vehicle_dir: str, report: dict[str, Any] | None = None, + is_fc_connected: bool = False, + upload_callback: Callable[[dict[str, Par]], bool | None] | None = None, ) -> None: super().__init__(root_tk) self.summary = summary self.vehicle_dir = vehicle_dir self.report = report + self.is_fc_connected = is_fc_connected + self.upload_callback = upload_callback self._ai_panel_visible = False - self.pairs = paired_quality_and_analysis_results(summary) + self.pairs = summary.paired_quality_and_analysis_results() self.subsystem_names = [q.name for q, _a in self.pairs] self._report_quality_by_name: dict[str, dict[str, Any]] = {} @@ -279,7 +255,7 @@ def _render_subsystem(self, name: str) -> None: # pylint: disable=too-many-loca self._section_link(_("Guide"), link.get("blog_text") or link["blog_url"], link["blog_url"]) vehicle_components = (self.report or {}).get("vehicle_components") or {} - component_keys = _SUBSYSTEM_TO_COMPONENT_KEYS.get(name, ()) + component_keys = self.summary.component_keys_for_subsystem(quality_result.subsystem_key) hardware_lines: list[tuple[str, list[str]]] = [] for key in component_keys: component = vehicle_components.get(key) @@ -331,13 +307,73 @@ def _bullet_line(self, text: str) -> None: def _outcome_line(self, outcome: LogAnalysis) -> None: timestamp_text = f" ({outcome.timestamp_us / 1e6:.1f}s)" if outcome.timestamp_us is not None else "" + row = ttk.Frame(self.body_frame) + row.pack(anchor=tk.W, padx=(10, 0), pady=3, fill=tk.X) ttk.Label( - self.body_frame, + row, text=f"{outcome.message}{timestamp_text}", font=("TkDefaultFont", 14), wraplength=950, justify=tk.LEFT, - ).pack(anchor=tk.W, padx=(10, 0), pady=3, fill=tk.X) + ).pack(side=tk.LEFT, fill=tk.X, expand=True) + + fixes = self._fix_for_outcome(outcome) + if fixes: + fix_button = ttk.Button(row, text=_("Fix"), command=partial(self._open_review_dialog, fixes)) + fix_button.pack(side=tk.RIGHT, padx=(8, 0)) + + def _fix_for_outcome(self, outcome: LogAnalysis) -> list[tuple[str, float, float, list[str]]]: + if not isinstance(outcome.param_name, str) or not isinstance(outcome.suggested_value, (int, float)): + return [] + current = self.summary.related_parameter_values.get(outcome.param_name) + if current is None or float(outcome.suggested_value) == current: + return [] + return [(outcome.param_name, current, float(outcome.suggested_value), [outcome.message])] + + def _open_review_dialog(self, fixes: list[tuple[str, float, float, list[str]]]) -> None: + dialog = tk.Toplevel(self.root) + dialog.title(_("Review Parameter Changes")) + dialog.geometry(self.calculate_scaled_geometry(520, 140 + 60 * len(fixes))) + self.center_window(dialog, self.root) + dialog.transient(self.root) + dialog.grab_set() + + ttk.Label(dialog, text=_("The following parameter change(s) are proposed:"), font=("TkDefaultFont", 11, "bold")).pack( + anchor=tk.W, padx=14, pady=(14, 6) + ) + rows_frame = ttk.Frame(dialog) + rows_frame.pack(fill=tk.BOTH, expand=True, padx=14, pady=(0, 6)) + for param_name, current, proposed, reasons in fixes: + row = ttk.Frame(rows_frame) + row.pack(fill=tk.X, pady=4) + ttk.Label(row, text=param_name, width=18, font=("TkDefaultFont", 11, "bold")).pack(side=tk.LEFT) + ttk.Label(row, text=str(current), foreground="gray").pack(side=tk.LEFT, padx=(0, 6)) + ttk.Label(row, text="->").pack(side=tk.LEFT, padx=(0, 6)) + value_lbl = ttk.Label(row, text=str(proposed), foreground="darkgreen", font=("TkDefaultFont", 11, "bold")) + value_lbl.pack(side=tk.LEFT) + show_tooltip(value_lbl, "\n".join(f"- {reason}" for reason in reasons)) + + button_row = ttk.Frame(dialog) + button_row.pack(fill=tk.X, padx=14, pady=(6, 14)) + ttk.Button(button_row, text=_("Cancel"), command=dialog.destroy).pack(side=tk.RIGHT, padx=(6, 0)) + upload_button = ttk.Button( + button_row, + text=_("Apply & Upload"), + command=partial(self._apply_param_fixes, fixes, dialog), + ) + upload_button.configure(state="normal" if self.is_fc_connected else "disabled") + upload_button.pack(side=tk.RIGHT) + if not self.is_fc_connected: + show_tooltip(upload_button, _("No flight controller connected, upload not available")) + + def _apply_param_fixes(self, fixes: list[tuple[str, float, float, list[str]]], dialog: tk.Toplevel) -> None: + changes = {param_name: Par(proposed, "") for param_name, _current, proposed, _reasons in fixes} + if self.upload_callback is not None: + upload_result = self.upload_callback(changes) + if upload_result is False: + return + self.summary.related_parameter_values.update({name: par.value for name, par in changes.items()}) + dialog.destroy() def _section_link(self, tag: str, text: str, url: str) -> None: row = ttk.Frame(self.body_frame) diff --git a/ardupilot_methodic_configurator/frontend_tkinter_log_quality.py b/ardupilot_methodic_configurator/frontend_tkinter_log_quality.py index c6f25d2ee..7e6b0364d 100644 --- a/ardupilot_methodic_configurator/frontend_tkinter_log_quality.py +++ b/ardupilot_methodic_configurator/frontend_tkinter_log_quality.py @@ -23,10 +23,7 @@ from ardupilot_methodic_configurator.data_model_par_dict import Par from ardupilot_methodic_configurator.formatting import format_filesize from ardupilot_methodic_configurator.frontend_tkinter_base_window import BaseWindow -from ardupilot_methodic_configurator.frontend_tkinter_log_analysis import ( - LogAnalysisReportWindow, - paired_quality_and_analysis_results, -) +from ardupilot_methodic_configurator.frontend_tkinter_log_analysis import LogAnalysisReportWindow from ardupilot_methodic_configurator.frontend_tkinter_log_hardware_quality import build_hardware_tab from ardupilot_methodic_configurator.frontend_tkinter_scroll_frame import ScrollFrame from ardupilot_methodic_configurator.frontend_tkinter_show import show_tooltip @@ -45,6 +42,11 @@ ) +def _format_parameter_value(value: float) -> str: + """Format a parameter value without hiding fractional changes.""" + return str(int(value)) if value.is_integer() else str(value) + + class LogQualityReportWindow(BaseWindow): # pylint: disable=too-many-instance-attributes """Displays log analysis results as a beginner-friendly, detailed dashboard.""" @@ -55,7 +57,7 @@ def __init__( # pylint: disable=too-many-arguments, too-many-positional-argumen summary: LogSummary, vehicle_dir: str, is_fc_connected: bool = False, - upload_callback: Callable[[dict], bool] | None = None, + upload_callback: Callable[[dict[str, Par]], bool | None] | None = None, navigate_callback: Callable[[str], None] | None = None, report: dict | None = None, ) -> None: @@ -146,7 +148,7 @@ def _build_footer(self) -> None: def _on_continue_to_analysis(self) -> None: pending_names = [ quality_result.name - for quality_result, analysis_result in paired_quality_and_analysis_results(self.summary) + for quality_result, analysis_result in self.summary.paired_quality_and_analysis_results() if analysis_result is None ] if pending_names: @@ -157,7 +159,14 @@ def _on_continue_to_analysis(self) -> None: ), parent=self.root, ) - self._analysis_window = LogAnalysisReportWindow(self.root, self.summary, self.vehicle_dir, report=self.report) + self._analysis_window = LogAnalysisReportWindow( + self.root, + self.summary, + self.vehicle_dir, + is_fc_connected=self.is_fc_connected, + upload_callback=self.upload_callback, + report=self.report, + ) def _open_review_dialog(self, fixes: list[tuple[str, float, float, list[str]]]) -> None: dialog = tk.Toplevel(self.root) @@ -178,9 +187,14 @@ def _open_review_dialog(self, fixes: list[tuple[str, float, float, list[str]]]) row = ttk.Frame(rows_frame) row.pack(fill=tk.X, pady=4) ttk.Label(row, text=param_name, width=18, font=("TkDefaultFont", 11, "bold")).pack(side=tk.LEFT) - ttk.Label(row, text=str(int(current)), foreground="gray").pack(side=tk.LEFT, padx=(0, 6)) + ttk.Label(row, text=_format_parameter_value(current), foreground="gray").pack(side=tk.LEFT, padx=(0, 6)) ttk.Label(row, text="->").pack(side=tk.LEFT, padx=(0, 6)) - value_lbl = ttk.Label(row, text=str(int(proposed)), foreground="darkgreen", font=("TkDefaultFont", 11, "bold")) + value_lbl = ttk.Label( + row, + text=_format_parameter_value(proposed), + foreground="darkgreen", + font=("TkDefaultFont", 11, "bold"), + ) value_lbl.pack(side=tk.LEFT) show_tooltip(value_lbl, "\n".join(f"- {r}" for r in reasons)) @@ -200,9 +214,12 @@ def _open_review_dialog(self, fixes: list[tuple[str, float, float, list[str]]]) def _apply_param_fixes(self, fixes: list[tuple[str, float, float, list[str]]], dialog: tk.Toplevel) -> None: changes = {param_name: Par(proposed, "") for param_name, _current, proposed, _reasons in fixes} - self.summary.related_parameter_values.update({name: par.value for name, par in changes.items()}) if self.upload_callback is not None: - self.upload_callback(changes) + upload_result = self.upload_callback(changes) + if upload_result is False: + return + + self.summary.related_parameter_values.update({name: par.value for name, par in changes.items()}) dialog.destroy() @staticmethod @@ -348,8 +365,8 @@ def _build_quality_tab(self, parent: ttk.Frame) -> None: # pylint: disable=too- for kind, item in needs_attention: if kind == "quality": quality_item = cast("LogQualityResult", item) - absorbed_steps = absorbed_by_step.get(quality_item.related_step, []) - self._quality_result_card(inner, quality_item, absorbed_steps) + quality_absorbed_steps = absorbed_by_step.get(quality_item.related_step, []) + self._quality_result_card(inner, quality_item, quality_absorbed_steps) else: self._step_result_card(inner, item) # type: ignore[arg-type] ttk.Separator(inner, orient=tk.HORIZONTAL).pack(fill=tk.X, padx=14, pady=(14, 14)) @@ -361,8 +378,8 @@ def _build_quality_tab(self, parent: ttk.Frame) -> None: # pylint: disable=too- for kind, item in passed_checks: if kind == "quality": quality_item = cast("LogQualityResult", item) - absorbed_steps = absorbed_by_step.get(quality_item.related_step, []) - self._quality_result_card(inner, quality_item, absorbed_steps) + quality_absorbed_steps = absorbed_by_step.get(quality_item.related_step, []) + self._quality_result_card(inner, quality_item, quality_absorbed_steps) else: self._step_result_card(inner, item) # type: ignore[arg-type] diff --git a/ardupilot_methodic_configurator/frontend_tkinter_parameter_editor.py b/ardupilot_methodic_configurator/frontend_tkinter_parameter_editor.py index 029f1532a..530925f7b 100755 --- a/ardupilot_methodic_configurator/frontend_tkinter_parameter_editor.py +++ b/ardupilot_methodic_configurator/frontend_tkinter_parameter_editor.py @@ -605,9 +605,6 @@ def _create_parameter_area_widgets(self) -> None: if self.parameter_editor.parameter_files() else _("No intermediate parameter files available"), ) - # Create skip button - self.skip_button = ttk.Button(buttons_frame, text=_("Skip parameter file"), command=self.on_skip_click) - # Create skip buttons self.skip_button = ttk.Button(buttons_frame, text=_("Skip >"), command=self.on_skip_click) self.skip_button.configure( diff --git a/ardupilot_methodic_configurator/frontend_tkinter_tuning_report.py b/ardupilot_methodic_configurator/frontend_tkinter_tuning_report.py index b09149e63..b6057d684 100644 --- a/ardupilot_methodic_configurator/frontend_tkinter_tuning_report.py +++ b/ardupilot_methodic_configurator/frontend_tkinter_tuning_report.py @@ -74,8 +74,10 @@ def _build_chart_panel(self, parent: ttk.Frame) -> None: right_scroll.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=(4, 8), pady=8) self.figure = Figure(figsize=(7, 6), dpi=100) - self.canvas = FigureCanvasTkAgg(self.figure, master=right_scroll.view_port) - self.canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True) + self.canvas = FigureCanvasTkAgg( # type: ignore[no-untyped-call] + self.figure, master=right_scroll.view_port + ) + self.canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True) # type: ignore[no-untyped-call] self.canvas.mpl_connect("motion_notify_event", self._on_hover) @@ -90,7 +92,7 @@ def _redraw(self) -> None: if num_plots == 0: self.figure.set_figheight(6) - self.canvas.get_tk_widget().configure(height=int(6 * self.figure.dpi)) + self.canvas.get_tk_widget().configure(height=int(6 * self.figure.dpi)) # type: ignore[no-untyped-call] ax = self.figure.add_subplot(111) ax.text( 0.5, @@ -106,13 +108,13 @@ def _redraw(self) -> None: ax.set_yticks([]) for spine in ax.spines.values(): spine.set_visible(False) - self.canvas.draw() - self.canvas.get_tk_widget().update_idletasks() + self.canvas.draw() # type: ignore[no-untyped-call] + self.canvas.get_tk_widget().update_idletasks() # type: ignore[no-untyped-call] return calc_height = max(6.0, num_plots * 1.5) self.figure.set_figheight(calc_height) - self.canvas.get_tk_widget().configure(height=int(calc_height * self.figure.dpi)) + self.canvas.get_tk_widget().configure(height=int(calc_height * self.figure.dpi)) # type: ignore[no-untyped-call] axes = self.figure.subplots(nrows=num_plots, ncols=1, sharex=True, squeeze=False) flat_axes = [ax[0] for ax in axes] @@ -159,8 +161,8 @@ def _redraw(self) -> None: self.figure.align_ylabels(flat_axes) self.figure.subplots_adjust(hspace=0.1) self.figure.tight_layout() - self.canvas.draw() - self.canvas.get_tk_widget().update_idletasks() + self.canvas.draw() # type: ignore[no-untyped-call] + self.canvas.get_tk_widget().update_idletasks() # type: ignore[no-untyped-call] def _on_hover(self, event: Any) -> None: # noqa: ANN401 """Triggered on mouse movement to display exact point values.""" @@ -183,7 +185,7 @@ def _on_hover(self, event: Any) -> None: # noqa: ANN401 annot.xy = (x_val, y_val) step_name = self.report.steps[int(x_val)] - text = f"Step: {step_name}\nValue: {y_val}" + text = f"{_('Step')}: {step_name}\n{_('Value')}: {y_val}" # Only update and redraw if the tooltip is new or changed if not annot.get_visible() or annot.get_text() != text: @@ -199,7 +201,7 @@ def _on_hover(self, event: Any) -> None: # noqa: ANN401 redraw_needed = True if redraw_needed: - self.canvas.draw_idle() + self.canvas.draw_idle() # type: ignore[no-untyped-call] def run(self) -> None: self.root.mainloop() diff --git a/ardupilot_methodic_configurator/log_analysis/backend_log_extraction.py b/ardupilot_methodic_configurator/log_analysis/backend_log_extraction.py index a0816080a..794c60447 100644 --- a/ardupilot_methodic_configurator/log_analysis/backend_log_extraction.py +++ b/ardupilot_methodic_configurator/log_analysis/backend_log_extraction.py @@ -15,7 +15,8 @@ import contextlib import os -from collections.abc import Callable, Iterator +from collections.abc import Callable, Iterator, Mapping +from dataclasses import dataclass from logging import error as logging_error from typing import Any, Protocol, cast @@ -28,6 +29,19 @@ _NO_ID_ASSIGNED = "-" # ArduPilot's FMTU convention: '-' marks a field with no unit/multiplier ID assigned +_MULTIPLIER_TO_PREFIX = { + 0.0: "", + 1.0: "", + 1.0e-1: "d", + 1.0e-2: "c", + 1.0e-3: "m", + 1.0e-6: "ยต", + 1.0e-9: "n", +} + +_FIXED_POINT_FORMATS = frozenset("cCeEL") +_EAGERLY_SCALED_FLOAT_FORMATS = frozenset("fd") + def open_log(logfile: str) -> mavutil.mavfile: """ @@ -109,6 +123,15 @@ class _SchemaSource(Protocol): # pylint: disable=too-few-public-methods formats: dict[Any, _MavFmt] mult_lookup: dict[str, float] + unit_lookup: dict[str, str] + + +@dataclass(frozen=True) +class _FMTUDefinition: + """The UNIT and MULT identifiers assigned by one FMTU message.""" + + unit_ids: str + mult_ids: str _FORMAT_TO_DTYPE: dict[str, Any] = { @@ -119,6 +142,7 @@ class _SchemaSource(Protocol): # pylint: disable=too-few-public-methods "i": np.int32, "I": np.uint32, "f": np.float32, + "g": np.float16, "d": np.float64, "n": "S4", "N": "S16", @@ -137,7 +161,7 @@ class _SchemaSource(Protocol): # pylint: disable=too-few-public-methods def _schema_numpy_dtype(schema: data_model_log_data.MessageSchema) -> np.dtype[Any]: - """Build a structured numpy dtype that mirrors a message schema.""" + """Build a compact structured NumPy dtype matching the log's stored values.""" if len(schema.fields) != len(schema.format): msg = _("Schema {name} has mismatched field and format counts").format(name=schema.name) raise ValueError(msg) @@ -205,20 +229,22 @@ def _set_log_identity(log_data: data_model_log_data.LogData, identity: tuple[str log_data.firmware_version = (major, minor, patch) -def _record_message_counts_fields_and_identity(mlog: mavutil.mavfile, log_data: data_model_log_data.LogData) -> dict[int, str]: +def _record_message_counts_fields_and_identity( + mlog: mavutil.mavfile, log_data: data_model_log_data.LogData +) -> dict[int, _FMTUDefinition]: """ - First pass: count message occurrences, capture each type's FMTU MultIds string, and find log identity. + First pass: count messages, capture FMTU unit/multiplier IDs, and find log identity. - MultIds maps each field position to a single-character multiplier ID, - resolved later against mlog.mult_lookup. + FMTU IDs map each field position to a UNIT and MULT entry, resolved later + against pymavlink's completed lookup tables. """ - mult_ids_by_type: dict[int, str] = {} + fmtu_definitions: dict[int, _FMTUDefinition] = {} msg_fallback_identity: tuple[str, int, int, int] | None = None for msg in _iter_messages(mlog): msg_type = msg.get_type() log_data.msg_count[msg_type] = log_data.msg_count.get(msg_type, 0) + 1 if msg_type == "FMTU": - mult_ids_by_type[int(msg.FmtType)] = msg.MultIds + fmtu_definitions[int(msg.FmtType)] = _FMTUDefinition(unit_ids=msg.UnitIds, mult_ids=msg.MultIds) elif msg_type == "VER" and log_data.vehicle_type is None: identity = process_ver_identity(msg) if identity is not None: @@ -229,20 +255,15 @@ def _record_message_counts_fields_and_identity(mlog: mavutil.mavfile, log_data: if log_data.vehicle_type is None and msg_fallback_identity is not None: _set_log_identity(log_data, msg_fallback_identity) - return mult_ids_by_type + return fmtu_definitions def _resolve_multipliers(fmt: Any, mult_ids: str | None, mult_lookup: dict[str, float]) -> list[float | None]: # noqa: ANN401 - """ - Fields whose format character is one of pymavlink's built-in fixed-point types. - - ('c', 'C', 'e', 'E', 'L') are already scaled by pymavlink itself - inside DFMessage. - """ + """Return each field's stored-to-scaled fixed-point or FMTU multiplier.""" resolved: list[float | None] = [] for i, fixed_mult in enumerate(fmt.msg_mults): if fixed_mult is not None: - resolved.append(None) + resolved.append(fixed_mult) continue if mult_ids is not None and i < len(mult_ids) and mult_ids[i] != _NO_ID_ASSIGNED and mult_ids[i] in mult_lookup: @@ -253,12 +274,85 @@ def _resolve_multipliers(fmt: Any, mult_ids: str | None, mult_lookup: dict[str, return resolved +def _resolve_multipliers_applied_at_ingest(fmt: Any, multipliers: list[float | None]) -> list[bool]: # noqa: ANN401 + """Mark dynamic FMTU multipliers that can be applied without widening float storage.""" + return [ + fmt.format[index] in _EAGERLY_SCALED_FLOAT_FORMATS and fmt.msg_mults[index] is None and multiplier not in (None, 1) + for index, multiplier in enumerate(multipliers) + ] + + +def _resolve_scaled_units(fmt: Any, unit_ids: str | None, unit_lookup: dict[str, str]) -> list[str]: # noqa: ANN401 + """Return units for values after dynamic FMTU multipliers are applied.""" + fallback_units = list(fmt.units) if fmt.units is not None else [""] * len(fmt.columns) + if unit_ids is None: + return fallback_units + + return [ + unit_lookup.get(unit_ids[index], fallback_units[index]) + if index < len(unit_ids) and unit_ids[index] != _NO_ID_ASSIGNED + else fallback_units[index] + for index in range(len(fmt.columns)) + ] + + +def _resolve_stored_units( + scaled_units: list[str], + multipliers: list[float | None], + multipliers_applied_at_ingest: list[bool], +) -> list[str]: + """Return units corresponding to the values physically stored in NumPy arrays.""" + stored_units: list[str] = [] + for unit, multiplier, applied_at_ingest in zip(scaled_units, multipliers, multipliers_applied_at_ingest, strict=True): + if applied_at_ingest or multiplier is None or multiplier == 1: + stored_units.append(unit) + continue + + prefix = _MULTIPLIER_TO_PREFIX.get(multiplier) + stored_units.append(f"{prefix}{unit}" if prefix is not None else f"{multiplier:.4g} {unit}") + return stored_units + + +def _raw_fixed_point_value(msg: Any, field_index: int, field_name: str) -> Any: # noqa: ANN401 + """Return one unscaled fixed-point value from pymavlink's DataFlash message storage.""" + try: + return msg._elements[field_index] # pylint: disable=protected-access # noqa: SLF001 # pymavlink's raw DataFlash representation + except (AttributeError, IndexError) as error: + message = _("pymavlink did not expose raw fixed-point field {field_name}").format(field_name=field_name) + raise ValueError(message) from error + + def _allocate_message_arrays(log_data: data_model_log_data.LogData) -> None: """Allocate one structured numpy array per message type.""" for message_name, schema in log_data.schemas.items(): log_data._raw_messages[message_name] = np.empty(schema.records, dtype=_schema_numpy_dtype(schema)) # pylint: disable=protected-access # noqa: SLF001 +def _message_values( + msg: Any, # noqa: ANN401 + schema: data_model_log_data.MessageSchema, + payload: dict[str, Any], + field_info: Mapping[str, Any], +) -> list[Any]: + """Convert one decoded message into values compatible with its structured-array schema.""" + values: list[Any] = [] + for field_index, field_name in enumerate(schema.fields): + format_char = schema.format[field_index] + value = ( + _raw_fixed_point_value(msg, field_index, field_name) + if format_char in _FIXED_POINT_FORMATS + else payload[field_name] + ) + if schema.multipliers_applied_at_ingest[field_index]: + multiplier = schema.multipliers[field_index] + if multiplier is not None: + value *= multiplier + if field_info[field_name][0].kind == "S" and isinstance(value, str): + value = value.encode("ascii", "ignore") + values.append(value) + return values + + def _fill_message_arrays( # pylint: disable=too-many-locals mlog: mavutil.mavfile, log_data: data_model_log_data.LogData, @@ -300,15 +394,7 @@ def _fill_message_arrays( # pylint: disable=too-many-locals error_message = _("Structured array for {message_type} is missing field metadata").format(message_type=msg_type) raise ValueError(error_message) - values: list[Any] = [] - for field_name in schema.fields: - value = payload[field_name] - - dtype = field_info[field_name][0] - if dtype.kind == "S" and isinstance(value, str): - value = value.encode("ascii", "ignore") - values.append(value) - array[index] = tuple(values) + array[index] = tuple(_message_values(msg, schema, payload, field_info)) write_positions[msg_type] = index + 1 for message_name, schema in log_data.schemas.items(): @@ -323,30 +409,49 @@ def _fill_message_arrays( # pylint: disable=too-many-locals def extract_schemas( mlog: mavutil.mavfile, log_data: data_model_log_data.LogData, - mult_ids_by_type: dict[int, str], + fmtu_definitions: dict[int, _FMTUDefinition], ) -> None: """ Copy pymavlink's discovered FMT/FMTU schemas into log_data.schemas. - Stored dicts (not pymavlink objects). Raw metadata only units and multipliers are stored. + Stored schemas distinguish units for decoded stored values from units for + values returned after dynamic FMTU scaling. Args: mlog: An open pymavlink connection (fully read). log_data: The LogData instance to populate. - mult_ids_by_type: Per message type, the MultIds string from that type's FMTU message. + fmtu_definitions: Per message type, FMTU UNIT and MULT identifiers. """ schema_source = cast("_SchemaSource", mlog) for fmt in schema_source.formats.values(): + fmtu_definition = fmtu_definitions.get(fmt.type) + multipliers = _resolve_multipliers( + fmt, + fmtu_definition.mult_ids if fmtu_definition is not None else None, + schema_source.mult_lookup, + ) + multipliers_applied_at_ingest = _resolve_multipliers_applied_at_ingest(fmt, multipliers) + scaled_units = _resolve_scaled_units( + fmt, + fmtu_definition.unit_ids if fmtu_definition is not None else None, + schema_source.unit_lookup, + ) log_data.schemas[fmt.name] = data_model_log_data.MessageSchema( name=fmt.name, msg_type=fmt.type, length=fmt.len, format=fmt.format, fields=list(fmt.columns), - units=list(fmt.units) if fmt.units is not None else [], - multipliers=_resolve_multipliers(fmt, mult_ids_by_type.get(fmt.type), schema_source.mult_lookup), + scaled_units=scaled_units, + multipliers=multipliers, + multipliers_applied_at_ingest=multipliers_applied_at_ingest, records=log_data.msg_count.get(fmt.name, 0), + stored_units=_resolve_stored_units( + scaled_units, + multipliers, + multipliers_applied_at_ingest, + ), ) @@ -378,17 +483,17 @@ def compute_flight_duration(log_data: data_model_log_data.LogData) -> float | No if records is None or records.size == 0: continue - timeus = log_data.get_field(message_name, "TimeUS", scaled=False) + time_seconds = log_data.get_field(message_name, "TimeUS") states = log_data.get_field(message_name, state_field) total_time = 0 start_time = None # Many logs have multiple arm/disarm events, calculate them separately and sum up - for time_dur, state in zip(timeus, states, strict=True): + for timestamp_seconds, state in zip(time_seconds, states, strict=True): if state == start_value and start_time is None: - start_time = time_dur + start_time = timestamp_seconds elif state == stop_value and start_time is not None: - total_time += time_dur - start_time + total_time += timestamp_seconds - start_time start_time = None # If there is no disarm message the flight time can't be calculated. @@ -398,7 +503,7 @@ def compute_flight_duration(log_data: data_model_log_data.LogData) -> float | No ) if total_time > 0: - return total_time / 1e6 # 1_000_000 + return float(total_time) except (KeyError, ValueError) as error: logging_error(_("Could not compute flight duration: {error}").format(error=error)) @@ -432,9 +537,9 @@ def extract_log( # first pass: count messages, capture identity, and let pymavlink discover schemas for preallocated arrays mlog = open_log(logfile) try: - mult_ids_by_type = _record_message_counts_fields_and_identity(mlog, log_data) + fmtu_definitions = _record_message_counts_fields_and_identity(mlog, log_data) # extract_schemas should not raise any exception if it does it should fail - extract_schemas(mlog, log_data, mult_ids_by_type) + extract_schemas(mlog, log_data, fmtu_definitions) finally: close_log(mlog) diff --git a/ardupilot_methodic_configurator/log_analysis/data_model_log_analysis.py b/ardupilot_methodic_configurator/log_analysis/data_model_log_analysis.py index 5d9325e71..729e2808f 100644 --- a/ardupilot_methodic_configurator/log_analysis/data_model_log_analysis.py +++ b/ardupilot_methodic_configurator/log_analysis/data_model_log_analysis.py @@ -15,7 +15,7 @@ from ardupilot_methodic_configurator import _ from ardupilot_methodic_configurator.log_analysis.data_model_log_analysis_context import LogAnalysisContext -from ardupilot_methodic_configurator.log_analysis.data_model_log_analysis_result import LogAnalysisResult +from ardupilot_methodic_configurator.log_analysis.data_model_log_analysis_result import LogAnalysis, LogAnalysisResult from ardupilot_methodic_configurator.log_analysis.data_model_log_data import LogData from ardupilot_methodic_configurator.log_analysis.data_model_log_quality import ( LogQualityResult, @@ -31,7 +31,10 @@ validate_configuration_steps_data, ) from ardupilot_methodic_configurator.log_analysis.data_model_quality_arm import ArmLogQualityModel -from ardupilot_methodic_configurator.log_analysis.data_model_quality_base import BaseLogModel +from ardupilot_methodic_configurator.log_analysis.data_model_quality_base import ( + BaseLogAnalysisModel, + BaseLogQualityModel, +) from ardupilot_methodic_configurator.log_analysis.data_model_quality_battery import BatteryLogAnalysis, BatteryLogQualityModel from ardupilot_methodic_configurator.log_analysis.data_model_quality_err import ErrLogQualityModel from ardupilot_methodic_configurator.log_analysis.data_model_quality_esc import EscLogAnalysis, EscLogQualityModel @@ -44,18 +47,31 @@ from ardupilot_methodic_configurator.log_analysis.data_model_vehicle_overview import HardwareReport from ardupilot_methodic_configurator.log_analysis.data_model_vehicle_overview_report import extract_hardware_report -QUALITY_AND_ANALYSIS_MODELS: list[tuple[type[BaseLogModel], type[BaseLogModel] | None]] = [ - (BatteryLogQualityModel, BatteryLogAnalysis), - (GPSLogQualityModel, None), - (EscLogQualityModel, EscLogAnalysis), - (ImuLogQualityModel, ImuLogAnalysis), - (VibeLogQualityModel, VibeLogAnalysis), - (FftLogQualityModel, None), - (ErrLogQualityModel, None), - (PmLogQualityModel, None), - (ArmLogQualityModel, None), - (ModeLogQualityModel, None), -] + +@dataclass(frozen=True) +class LogAnalysisModelSpec: + """Registration for one log subsystem and its optional detailed analysis.""" + + key: str + quality_model: type[BaseLogQualityModel] + analysis_model: type[BaseLogAnalysisModel] | None = None + component_keys: tuple[str, ...] = () + + +LOG_ANALYSIS_SUBSYSTEMS: tuple[LogAnalysisModelSpec, ...] = ( + LogAnalysisModelSpec("battery", BatteryLogQualityModel, BatteryLogAnalysis, ("Battery", "Battery Monitor")), + LogAnalysisModelSpec("gps", GPSLogQualityModel, component_keys=("GNSS Receiver",)), + LogAnalysisModelSpec("esc", EscLogQualityModel, EscLogAnalysis, ("ESC", "Motors")), + LogAnalysisModelSpec("imu", ImuLogQualityModel, ImuLogAnalysis, ("Flight Controller",)), + LogAnalysisModelSpec("vibe", VibeLogQualityModel, VibeLogAnalysis, ("Flight Controller",)), + LogAnalysisModelSpec("fft", FftLogQualityModel), + LogAnalysisModelSpec("err", ErrLogQualityModel), + LogAnalysisModelSpec("pm", PmLogQualityModel), + LogAnalysisModelSpec("arm", ArmLogQualityModel), + LogAnalysisModelSpec("mode", ModeLogQualityModel), +) + +ResolvedModel = tuple[type[BaseLogQualityModel], type[BaseLogAnalysisModel] | None, str] def parse_firmware_version(version: object) -> tuple[int, int, int] | None: @@ -91,6 +107,60 @@ def _pm_validation_as_quality_result(validation: MessageValidation | None) -> Lo ) +def _resolve_models( + quality_and_analysis_models: list[tuple[type[BaseLogQualityModel], type[BaseLogAnalysisModel] | None]] | None, +) -> tuple[list[ResolvedModel], dict[str, tuple[str, ...]]]: + """Resolve the default registry or caller-provided model pairs.""" + if quality_and_analysis_models is None: + return ( + [(spec.quality_model, spec.analysis_model, spec.key) for spec in LOG_ANALYSIS_SUBSYSTEMS], + {spec.key: spec.component_keys for spec in LOG_ANALYSIS_SUBSYSTEMS}, + ) + return ( + [ + (quality_model, analysis_model, f"custom_{index}") + for index, (quality_model, analysis_model) in enumerate(quality_and_analysis_models) + ], + {}, + ) + + +def _add_related_parameter_values( + related_values: dict[str, float], + findings: list[QualityIssue] | list[LogAnalysis], + parameters: dict[str, float], +) -> None: + """Add parameters referenced by quality issues or analysis outcomes.""" + for finding in findings: + if finding.param_name is not None and finding.param_name in parameters: + related_values[finding.param_name] = parameters[finding.param_name] + + +def _run_subsystem_models( + resolved_models: list[ResolvedModel], + log_data: LogData, + context: LogAnalysisContext, + related_parameter_values: dict[str, float], +) -> tuple[list[LogQualityResult], list[LogAnalysisResult], list[str]]: + """Run registered quality and available detailed-analysis models.""" + quality_results: list[LogQualityResult] = [] + analysis_results: list[LogAnalysisResult] = [] + analysis_subsystem_keys: list[str] = [] + for quality_model_cls, analysis_model_cls, subsystem_key in resolved_models: + quality_result = quality_model_cls(log_data, context).check() + quality_result.subsystem_key = subsystem_key + quality_results.append(quality_result) + _add_related_parameter_values(related_parameter_values, quality_result.issues, context.parameters) + if analysis_model_cls is not None: + analysis_subsystem_keys.append(subsystem_key) + if analysis_model_cls is not None and quality_result.available: + analysis_result = analysis_model_cls(log_data, context).analyse() + analysis_result.subsystem_key = subsystem_key + analysis_results.append(analysis_result) + _add_related_parameter_values(related_parameter_values, analysis_result.outcomes, context.parameters) + return quality_results, analysis_results, analysis_subsystem_keys + + def validate_log_matches_vehicle( log_vehicle_type: str, log_firmware_version: tuple[int, int, int], @@ -132,12 +202,35 @@ class LogSummary: # pylint: disable=too-many-instance-attributes step_results: list[StepValidationResult] hardware_report: HardwareReport related_parameter_values: dict[str, float] = field(default_factory=dict) + analysis_subsystem_keys: tuple[str, ...] = () + subsystem_component_keys: dict[str, tuple[str, ...]] = field(default_factory=dict) + + def component_keys_for_subsystem(self, subsystem_key: str | None) -> tuple[str, ...]: + """Return vehicle-component keys declared by one registered subsystem.""" + if subsystem_key is None: + return () + if subsystem_key in self.subsystem_component_keys: + return self.subsystem_component_keys[subsystem_key] + return next((spec.component_keys for spec in LOG_ANALYSIS_SUBSYSTEMS if spec.key == subsystem_key), ()) + + def paired_quality_and_analysis_results( + self, + ) -> list[tuple[LogQualityResult, LogAnalysisResult | None]]: + """Return analysis-enabled subsystem results matched by stable subsystem key.""" + quality_by_key = {result.subsystem_key: result for result in self.quality_results if result.subsystem_key is not None} + analysis_by_key = { + result.subsystem_key: result for result in self.analysis_results if result.subsystem_key is not None + } + registered_keys = self.analysis_subsystem_keys or tuple( + spec.key for spec in LOG_ANALYSIS_SUBSYSTEMS if spec.analysis_model is not None + ) + return [(quality_by_key[key], analysis_by_key.get(key)) for key in registered_keys if key in quality_by_key] def analyze_log( # pylint: disable=too-many-locals log_data: LogData, context: LogAnalysisContext, - quality_and_analysis_models: list[tuple[type[BaseLogModel], type[BaseLogModel] | None]] | None = None, + quality_and_analysis_models: list[tuple[type[BaseLogQualityModel], type[BaseLogAnalysisModel] | None]] | None = None, ) -> LogSummary: """ Run log analysis over already loaded datasource values. @@ -154,39 +247,26 @@ def analyze_log( # pylint: disable=too-many-locals Complete log analysis summary. """ - resolved_models: list[tuple[type[BaseLogModel], type[BaseLogModel] | None]] = ( - QUALITY_AND_ANALYSIS_MODELS if quality_and_analysis_models is None else quality_and_analysis_models - ) - parameters = context.parameters - configuration_steps = context.configuration_steps - apm_doc = context.apm_doc - pm_status = get_pm_status(log_data) pm_validation = check_cpu_performance_message(log_data) + resolved_models, subsystem_component_keys = _resolve_models(quality_and_analysis_models) quality_results: list[LogQualityResult] = [] pm_quality_result = _pm_validation_as_quality_result(pm_validation) if pm_quality_result is not None: quality_results.append(pm_quality_result) - analysis_results: list[LogAnalysisResult] = [] related_parameter_values: dict[str, float] = {} for result in quality_results: - for issue in result.issues: - if issue.param_name is not None and issue.param_name in parameters: - related_parameter_values[issue.param_name] = parameters[issue.param_name] - - for quality_model_cls, analysis_model_cls in resolved_models: - quality_model = quality_model_cls(log_data, context) - quality_result = quality_model.check() - quality_results.append(quality_result) - - if analysis_model_cls is not None and quality_result.available: - analysis_results.append(analysis_model_cls(log_data, context).analyse()) + _add_related_parameter_values(related_parameter_values, result.issues, parameters) + subsystem_quality_results, analysis_results, analysis_subsystem_keys = _run_subsystem_models( + resolved_models, log_data, context, related_parameter_values + ) + quality_results.extend(subsystem_quality_results) - step_results = validate_configuration_steps_data(log_data, configuration_steps) - hardware_report = extract_hardware_report(log_data, parameters, apm_doc) + step_results = validate_configuration_steps_data(log_data, context.configuration_steps) + hardware_report = extract_hardware_report(log_data, parameters, context.apm_doc) return LogSummary( flight_duration_sec=log_data.flight_duration_sec, @@ -201,4 +281,6 @@ def analyze_log( # pylint: disable=too-many-locals hardware_report=hardware_report, analysis_results=analysis_results, related_parameter_values=related_parameter_values, + analysis_subsystem_keys=tuple(analysis_subsystem_keys), + subsystem_component_keys=subsystem_component_keys, ) diff --git a/ardupilot_methodic_configurator/log_analysis/data_model_log_analysis_context.py b/ardupilot_methodic_configurator/log_analysis/data_model_log_analysis_context.py index 99ec23a18..7cf530f59 100644 --- a/ardupilot_methodic_configurator/log_analysis/data_model_log_analysis_context.py +++ b/ardupilot_methodic_configurator/log_analysis/data_model_log_analysis_context.py @@ -11,6 +11,10 @@ from dataclasses import dataclass, field from typing import Any +from ardupilot_methodic_configurator.log_analysis.data_model_parameter_derivation import ( + ConfigurationStepParameterDeriver, + ParameterDeriver, +) from ardupilot_methodic_configurator.log_analysis.utils import APMDoc @@ -22,3 +26,4 @@ class LogAnalysisContext: configuration_steps: dict[str, Any] vehicle_components: dict[str, Any] = field(default_factory=dict) apm_doc: APMDoc | None = None + parameter_deriver: ParameterDeriver = field(default_factory=ConfigurationStepParameterDeriver) diff --git a/ardupilot_methodic_configurator/log_analysis/data_model_log_analysis_result.py b/ardupilot_methodic_configurator/log_analysis/data_model_log_analysis_result.py index 7bdea0fa1..74925441d 100644 --- a/ardupilot_methodic_configurator/log_analysis/data_model_log_analysis_result.py +++ b/ardupilot_methodic_configurator/log_analysis/data_model_log_analysis_result.py @@ -33,3 +33,4 @@ class LogAnalysisResult: name: str | None reason: str related_step: str | None = None + subsystem_key: str | None = None diff --git a/ardupilot_methodic_configurator/log_analysis/data_model_log_data.py b/ardupilot_methodic_configurator/log_analysis/data_model_log_data.py index 81bd4f38f..0c950acb6 100644 --- a/ardupilot_methodic_configurator/log_analysis/data_model_log_data.py +++ b/ardupilot_methodic_configurator/log_analysis/data_model_log_data.py @@ -20,7 +20,7 @@ @dataclass class MessageSchema: # pylint: disable=too-many-instance-attributes - """Message type's FMT schema: fields, units, multipliers, types.""" + """Message type's FMT schema and units for its stored and scaled values.""" name: str msg_type: int @@ -28,8 +28,10 @@ class MessageSchema: # pylint: disable=too-many-instance-attributes format: str fields: list[str] - units: list[str] + stored_units: list[str] + scaled_units: list[str] multipliers: list[float | None] + multipliers_applied_at_ingest: list[bool] records: int = 0 @@ -40,8 +42,12 @@ class LogData: Store parsed log metadata and one structured numpy array per message type. Attributes: - schemas: Raw message definitions extracted from FMT/FMTU/UNIT/MULT, - keyed by message name. + schemas: Message definitions extracted from FMT/FMTU/UNIT/MULT, keyed + by message name. ``stored_units`` describes decoded values in + ``_raw_messages`` and returned with ``scaled=False``; + ``scaled_units`` describes values returned with ``scaled=True``. + ``multipliers_applied_at_ingest`` records fields whose multiplier + has already been applied without widening the stored dtype. _raw_messages: Per message type, a structured numpy array containing all decoded values for that message type. Access via get_message_columns(), get_field() or iter_message_records(). @@ -91,8 +97,8 @@ def get_field(self, message_name: str, field_name: str, scaled: bool = True) -> if not scaled: return values - multiplier, format_char = self._field_multiplier_and_format(message_name, field_name) - return scale_field_values(values, multiplier, format_char) + multiplier = self._field_multiplier(message_name, field_name) + return scale_field_values(values, multiplier) def iter_message_records(self, message_name: str, scaled: bool = True) -> Iterator[dict[str, Any]]: """ @@ -123,60 +129,38 @@ def iter_message_records(self, message_name: str, scaled: bool = True) -> Iterat value = value.decode("ascii", "ignore") if scaled: - multiplier, format_char = self._field_multiplier_and_format(message_name, field_name) + multiplier = self._field_multiplier(message_name, field_name) if multiplier is not None and multiplier != 1 and not isinstance(value, str): if isinstance(value, list): - value = scale_field_values(np.asarray(value), multiplier, format_char).tolist() + value = scale_field_values(np.asarray(value), multiplier).tolist() else: - value = scale_field_values(np.asarray(value), multiplier, format_char)[()] + value = scale_field_values(np.asarray(value), multiplier)[()] record[field_name] = value yield record - def _field_multiplier_and_format(self, message_name: str, field_name: str) -> tuple[float | None, str | None]: + def _field_multiplier(self, message_name: str, field_name: str) -> float | None: schema = self.schemas.get(message_name) if schema is None: - return None, None + return None try: field_index = schema.fields.index(field_name) except ValueError: - return None, None + return None - if field_index >= len(schema.multipliers): - return None, None + if field_index >= len(schema.multipliers) or field_index >= len(schema.multipliers_applied_at_ingest): + return None - format_char = schema.format[field_index] if field_index < len(schema.format) else None - return schema.multipliers[field_index], format_char + if schema.multipliers_applied_at_ingest[field_index]: + return None + return schema.multipliers[field_index] -def promoted_integer_dtype(dtype: np.dtype[Any]) -> np.dtype[Any]: - """Return a wider integer dtype suitable for fixed-point scaled fields.""" - if dtype.kind == "i": - if dtype.itemsize <= 2: - return np.dtype(np.int32) - return np.dtype(np.int64) - if dtype.kind == "u": - if dtype.itemsize <= 2: - return np.dtype(np.uint32) - return np.dtype(np.uint64) - - return dtype - - -def is_integer_multiplier(multiplier: float | None) -> bool: - """Return True when a multiplier can be applied without leaving integer space.""" - return multiplier is not None and float(multiplier).is_integer() - - -def scale_field_values(values: np.ndarray, multiplier: float | None, format_char: str | None = None) -> np.ndarray: - """Apply a field multiplier while preserving integer width for fixed-point fields.""" +def scale_field_values(values: np.ndarray, multiplier: float | None) -> np.ndarray: + """Apply one deferred fixed-point or FMTU multiplier to stored values.""" if multiplier is None or multiplier == 1: return values - if format_char in {"c", "C", "e", "E"} and values.dtype.kind in {"i", "u"} and is_integer_multiplier(multiplier): - promoted_dtype = promoted_integer_dtype(values.dtype) - return values.astype(promoted_dtype, copy=False) * int(multiplier) - return values * multiplier diff --git a/ardupilot_methodic_configurator/log_analysis/data_model_log_quality.py b/ardupilot_methodic_configurator/log_analysis/data_model_log_quality.py index da0bad2ea..e07da4b73 100644 --- a/ardupilot_methodic_configurator/log_analysis/data_model_log_quality.py +++ b/ardupilot_methodic_configurator/log_analysis/data_model_log_quality.py @@ -39,6 +39,7 @@ class LogQualityResult: issues: list[QualityIssue] name: str related_step: str = "" + subsystem_key: str | None = None @dataclass diff --git a/ardupilot_methodic_configurator/log_analysis/data_model_log_quality_check.py b/ardupilot_methodic_configurator/log_analysis/data_model_log_quality_check.py index 2014d2ed4..5f4de7a3c 100644 --- a/ardupilot_methodic_configurator/log_analysis/data_model_log_quality_check.py +++ b/ardupilot_methodic_configurator/log_analysis/data_model_log_quality_check.py @@ -15,6 +15,8 @@ from typing import TYPE_CHECKING, Any +import numpy as np + from ardupilot_methodic_configurator import _ from ardupilot_methodic_configurator.log_analysis.data_model_log_quality import ( MessageValidation, @@ -32,8 +34,6 @@ ) if TYPE_CHECKING: - import numpy as np - from ardupilot_methodic_configurator.log_analysis.data_model_log_data import LogData, MessageSchema _MAX_HEALTHY_AVG_CPU = 80.0 # percent @@ -123,9 +123,12 @@ def get_pm_status(log_data: LogData) -> PMStatus | None: available = set(columns.dtype.names or ()) load = log_data.get_field("PM", "Load") if "Load" in available else None - nlon = log_data.get_field("PM", "NLon", scaled=False) if "NLon" in available else None - max_t = log_data.get_field("PM", "MaxT", scaled=False) if "MaxT" in available else None - mem = log_data.get_field("PM", "Mem", scaled=False) if "Mem" in available else None + nlon = log_data.get_field("PM", "NLon") if "NLon" in available else None + max_t = log_data.get_field("PM", "MaxT") if "MaxT" in available else None + mem = log_data.get_field("PM", "Mem") if "Mem" in available else None + + if any(values is not None and not np.isfinite(values).all() for values in (load, nlon, max_t, mem)): + return PMStatus(0.0, 0.0, 0, 0, 0, None) # compute values into locals first avg_cpu_load = float(load.mean()) if load is not None else 0.0 @@ -153,6 +156,33 @@ def get_pm_status(log_data: LogData) -> PMStatus | None: ) +def _pm_non_finite_issues(log_data: LogData, available: set[str]) -> list[str]: + """Return one issue for each PM field that contains non-finite samples.""" + return [ + _("{field} contains non-finite telemetry values").format(field=field_name) + for field_name in ("Load", "NLon", "MaxT", "Mem", "InE", "ErC", "ErrL") + if field_name in available and not np.isfinite(log_data.get_field("PM", field_name)).all() + ] + + +def _pm_error_signal_issues(log_data: LogData, available: set[str]) -> list[str]: + """Return issues reported by finite PM error counters and masks.""" + issues: list[str] = [] + for field_name, reducer, message in ( + ("InE", np.max, _("Internal firmware errors were detected (InE)")), + ("ErC", np.max, _("Internal error count: {count}")), + ("ErrL", np.max, _("An internal error line was recorded (ErrL)")), + ("NLon", np.sum, _("Detected {count} scheduler long loops")), + ): + if field_name in available: + values = log_data.get_field("PM", field_name) + if np.isfinite(values).all(): + count = int(reducer(values)) + if count > 0: + issues.append(message.format(count=count) if field_name in {"ErC", "NLon"} else message) + return issues + + def check_cpu_performance_message(log_data: LogData) -> MessageValidation: """ Validate the PM (Performance Monitor) message for internal errors and health. @@ -165,34 +195,8 @@ def check_cpu_performance_message(log_data: LogData) -> MessageValidation: return MessageValidation(valid=False, issues=[_("PM message not logged")]) available = set(columns.dtype.names or ()) - issues: list[str] = [] - - # Internal error mask - if "InE" in available: - ine = log_data.get_field("PM", "InE", scaled=False) - if ine.max() > 0: - issues.append(_("Internal firmware errors were detected (InE)")) - - # Internal error count - if "ErC" in available: - erc = log_data.get_field("PM", "ErC", scaled=False) - count = int(erc.max()) - if count > 0: - issues.append(_("Internal error count: {count}").format(count=count)) - - # Internal error line number - if "ErrL" in available: - errl = log_data.get_field("PM", "ErrL", scaled=False) - if errl.max() > 0: - issues.append(_("An internal error line was recorded (ErrL)")) - - # Long loops - if "NLon" in available: - nlon = log_data.get_field("PM", "NLon", scaled=False) - count = int(nlon.sum()) - - if count > 0: - issues.append(_("Detected {count} scheduler long loops").format(count=count)) + issues = _pm_non_finite_issues(log_data, available) + issues.extend(_pm_error_signal_issues(log_data, available)) return MessageValidation(valid=not issues, issues=issues) @@ -217,10 +221,14 @@ def validate_fmt_schema(schema: MessageSchema, columns: np.ndarray | None) -> Me issues.append(_("Missing format string")) if schema.length <= 0: issues.append(_("Invalid message length")) - if schema.units and len(schema.units) != len(schema.fields): - issues.append(_("Unit count mismatch")) + if len(schema.stored_units) != len(schema.fields): + issues.append(_("Stored unit count mismatch")) + if len(schema.scaled_units) != len(schema.fields): + issues.append(_("Scaled unit count mismatch")) if schema.multipliers and len(schema.multipliers) != len(schema.fields): issues.append(_("Multiplier count mismatch")) + if len(schema.multipliers_applied_at_ingest) != len(schema.fields): + issues.append(_("Multiplier ingestion state count mismatch")) if columns is None or columns.size == 0: issues.append(_("{message} has no logging data").format(message=schema.name)) diff --git a/ardupilot_methodic_configurator/log_analysis/data_model_parameter_derivation.py b/ardupilot_methodic_configurator/log_analysis/data_model_parameter_derivation.py new file mode 100644 index 000000000..0d2d207d3 --- /dev/null +++ b/ardupilot_methodic_configurator/log_analysis/data_model_parameter_derivation.py @@ -0,0 +1,97 @@ +""" +Parameter-derivation service used by log-analysis models. + +The service adapts Methodic Configurator's configuration-step evaluator to a +small, in-memory API. Analysis models depend on this service rather than on +the filesystem-oriented ConfigurationSteps class. + +SPDX-FileCopyrightText: 2026 Amilcar do Carmo Lucas + +SPDX-License-Identifier: GPL-3.0-or-later +""" + +import re +from dataclasses import dataclass +from typing import Any, Protocol + +from ardupilot_methodic_configurator.backend_filesystem_configuration_steps import ConfigurationSteps +from ardupilot_methodic_configurator.log_analysis.utils import APMDoc + + +@dataclass(frozen=True) +class ParameterDerivationInputs: + """In-memory context required to evaluate configuration-step parameters.""" + + configuration_steps: dict[str, Any] + parameters: dict[str, float] + vehicle_components: dict[str, Any] + apm_doc: APMDoc | None + + +class ParameterDeriver(Protocol): + """Minimal parameter-derivation API required by detailed log analysis.""" + + def expected_parameter_value( + self, + step_filename: str, + param_name: str, + inputs: ParameterDerivationInputs, + ) -> tuple[float | None, str]: + """Return the expected value and its source, or ``(None, "")`` when unavailable.""" + + def derived_and_forced_parameters_matching(self, pattern: str, inputs: ParameterDerivationInputs) -> dict[str, str]: + """Return matching derived/forced parameter names mapped to their step filenames.""" + + +class ConfigurationStepParameterDeriver: + """In-memory adapter around the shared configuration-step expression evaluator.""" + + def __init__(self) -> None: + # Construction is intentionally side-effect free: configuration data is + # supplied by LogAnalysisContext, not loaded from a vehicle directory. + self._evaluator = ConfigurationSteps(_vehicle_dir="", vehicle_type="") + + def expected_parameter_value( + self, + step_filename: str, + param_name: str, + inputs: ParameterDerivationInputs, + ) -> tuple[float | None, str]: + """Evaluate one forced or derived parameter from supplied in-memory inputs.""" + if step_filename not in inputs.configuration_steps: + return None, "" + + eval_variables: dict[str, Any] = {"vehicle_components": inputs.vehicle_components} + if inputs.apm_doc: + eval_variables["doc_dict"] = inputs.apm_doc + if inputs.parameters: + eval_variables["fc_parameters"] = inputs.parameters + + step_dict = inputs.configuration_steps[step_filename] + forced_error, derived_error = self._evaluator.compute_forced_and_derived_parameters( + step_filename, step_dict, eval_variables, ignore_fc_derived_param_warnings=True + ) + if forced_error or derived_error: + return None, "" + + for source, values_by_step in ( + ("forced", self._evaluator.forced_parameters), + ("derived", self._evaluator.derived_parameters), + ): + if step_filename in values_by_step and param_name in values_by_step[step_filename]: + return values_by_step[step_filename][param_name].value, source + + return None, "" + + def derived_and_forced_parameters_matching(self, pattern: str, inputs: ParameterDerivationInputs) -> dict[str, str]: + """Return all forced and derived parameters matching a regular expression.""" + compiled = re.compile(pattern) + result: dict[str, str] = {} + for step_filename, step_info in inputs.configuration_steps.items(): + for param_name in step_info.get("derived_parameters", {}): + if compiled.match(param_name): + result[param_name] = step_filename + for param_name in step_info.get("forced_parameters", {}): + if compiled.match(param_name): + result[param_name] = step_filename + return result diff --git a/ardupilot_methodic_configurator/log_analysis/data_model_quality_arm.py b/ardupilot_methodic_configurator/log_analysis/data_model_quality_arm.py index e80d9095c..79651514b 100644 --- a/ardupilot_methodic_configurator/log_analysis/data_model_quality_arm.py +++ b/ardupilot_methodic_configurator/log_analysis/data_model_quality_arm.py @@ -11,7 +11,7 @@ from ardupilot_methodic_configurator import _ from ardupilot_methodic_configurator.log_analysis.data_model_log_quality import LogQualityState from ardupilot_methodic_configurator.log_analysis.data_model_quality_base import ( - BaseLogModel, + BaseLogQualityModel, LogQualityResult, QualityIssue, ) @@ -19,7 +19,7 @@ _NAME = "ARM" -class ArmLogQualityModel(BaseLogModel): +class ArmLogQualityModel(BaseLogQualityModel): """ Checks presence of arming/disarming event data. @@ -39,4 +39,4 @@ def check(self) -> LogQualityResult: def check_arm_fields(self) -> list[QualityIssue]: """Check that ArmState/ArmChecks/Forced/Method fields are present and have readable data.""" - return self.check_fields_present("ARM", ("ArmState", "ArmChecks", "Forced", "Method"), scaled=False) + return self.check_fields_present("ARM", ("ArmState", "ArmChecks", "Forced", "Method")) diff --git a/ardupilot_methodic_configurator/log_analysis/data_model_quality_base.py b/ardupilot_methodic_configurator/log_analysis/data_model_quality_base.py index b4fa950ab..10054475c 100644 --- a/ardupilot_methodic_configurator/log_analysis/data_model_quality_base.py +++ b/ardupilot_methodic_configurator/log_analysis/data_model_quality_base.py @@ -10,17 +10,18 @@ SPDX-License-Identifier: GPL-3.0-or-later """ -import re from typing import TYPE_CHECKING, Any +import numpy as np + from ardupilot_methodic_configurator import _ -from ardupilot_methodic_configurator.backend_filesystem_configuration_steps import ConfigurationSteps from ardupilot_methodic_configurator.log_analysis.data_model_log_analysis_context import LogAnalysisContext from ardupilot_methodic_configurator.log_analysis.data_model_log_quality import ( LogQualityResult, LogQualityState, QualityIssue, ) +from ardupilot_methodic_configurator.log_analysis.data_model_parameter_derivation import ParameterDerivationInputs from ardupilot_methodic_configurator.log_analysis.utils import ( find_configuration_step_for_message, find_configuration_step_for_parameter, @@ -29,28 +30,24 @@ ) if TYPE_CHECKING: + from ardupilot_methodic_configurator.log_analysis.data_model_log_analysis_result import LogAnalysisResult from ardupilot_methodic_configurator.log_analysis.data_model_log_data import LogData -class BaseLogModel(ConfigurationSteps): - """Base class for log analysis models.""" +class BaseLogModel: + """Common log-data services shared by quality and analysis models.""" def __init__( self, log_data: "LogData", context: LogAnalysisContext, ) -> None: - ConfigurationSteps.__init__(self, _vehicle_dir="", vehicle_type="") self.configuration_steps = context.configuration_steps self.log_data = log_data self.parameters = context.parameters self.vehicle_components = context.vehicle_components self.apm_doc = context.apm_doc - - def check(self) -> LogQualityResult: - """Run the model-specific quality analysis and return a result.""" - msg = f"{self.__class__.__name__} must implement check()" - raise NotImplementedError(msg) + self.parameter_deriver = context.parameter_deriver def step_for_parameter(self, param_name: str) -> str: try: @@ -58,18 +55,6 @@ def step_for_parameter(self, param_name: str) -> str: except ValueError: return "" - def build_result(self, issues: list[QualityIssue], name: str, related_step: str = "") -> LogQualityResult: - return LogQualityResult( - available=True, - state=LogQualityState.INFO if not issues else LogQualityState.WARNING, - reason=_("{name} data present and good for analysis").format(name=name) - if not issues - else _("{name} data has quality issues").format(name=name), - issues=issues, - name=name, - related_step=related_step, - ) - def resolve_message_step(self, message_name: str, fallback_name: str) -> tuple[str, str]: """ Resolve the configuration step and display name for a message this model checks. @@ -110,6 +95,10 @@ def field_values_or_issue( # pylint: disable=too-many-arguments issues.append(QualityIssue(missing_values_message)) return None, issues + if np.issubdtype(values.dtype, np.number) and not np.isfinite(values).all(): + issues.append(QualityIssue(_("{field} contains non-finite telemetry values").format(field=field_name))) + return None, issues + return values, issues def diagnose_bitmask_absence( @@ -174,6 +163,39 @@ def check_fields_present( issues += field_issues return issues + +class BaseLogQualityModel(BaseLogModel): + """Base class for subsystem quality models.""" + + def check(self) -> LogQualityResult: + """Run the model-specific quality analysis and return a result.""" + msg = f"{self.__class__.__name__} must implement check()" + raise NotImplementedError(msg) + + def build_result(self, issues: list[QualityIssue], name: str, related_step: str = "") -> LogQualityResult: + return LogQualityResult( + available=True, + state=LogQualityState.INFO if not issues else LogQualityState.WARNING, + reason=_("{name} data present and good for analysis").format(name=name) + if not issues + else _("{name} data has quality issues").format(name=name), + issues=issues, + name=name, + related_step=related_step, + ) + + +class BaseLogAnalysisModel(BaseLogModel): + """Base class for detailed analysis models requiring configuration evaluation.""" + + def __init__(self, log_data: "LogData", context: LogAnalysisContext) -> None: + BaseLogModel.__init__(self, log_data, context) + + def analyse(self) -> "LogAnalysisResult": + """Run the model-specific detailed analysis and return its result.""" + msg = f"{self.__class__.__name__} must implement analyse()" + raise NotImplementedError(msg) + def expected_parameter_value(self, step_filename: str, param_name: str) -> tuple[float | None, str]: """ Compute what a derived or forced parameter's value should be at a given step. @@ -181,28 +203,7 @@ def expected_parameter_value(self, step_filename: str, param_name: str) -> tuple Returns: (expected_value, source) where source is "forced" or "derived", or None if this parameter isn't forced or derived at this step. """ - if step_filename not in self.configuration_steps: - return None, "" - - step_dict = self.configuration_steps[step_filename] - eval_variables: dict[str, Any] = {"vehicle_components": self.vehicle_components} - if self.apm_doc: - eval_variables["doc_dict"] = self.apm_doc - if self.parameters: - eval_variables["fc_parameters"] = self.parameters - - forced_error, derived_error = self.compute_forced_and_derived_parameters( - step_filename, step_dict, eval_variables, ignore_fc_derived_param_warnings=True - ) - if forced_error or derived_error: - return None, "" - - for parameter_type in ("forced", "derived"): - destination = self.forced_parameters if parameter_type == "forced" else self.derived_parameters - if step_filename in destination and param_name in destination[step_filename]: - return destination[step_filename][param_name].value, parameter_type - - return None, "" + return self.parameter_deriver.expected_parameter_value(step_filename, param_name, self._parameter_derivation_inputs()) def derived_and_forced_parameters_matching(self, pattern: str) -> dict[str, str]: """ @@ -210,13 +211,13 @@ def derived_and_forced_parameters_matching(self, pattern: str) -> dict[str, str] Returns: param_name: step_filename """ - compiled = re.compile(pattern) - result: dict[str, str] = {} - for step_filename, step_info in self.configuration_steps.items(): - for param_name in step_info.get("derived_parameters", {}): - if compiled.match(param_name): - result[param_name] = step_filename - for param_name in step_info.get("forced_parameters", {}): - if compiled.match(param_name): - result[param_name] = step_filename - return result + return self.parameter_deriver.derived_and_forced_parameters_matching(pattern, self._parameter_derivation_inputs()) + + def _parameter_derivation_inputs(self) -> ParameterDerivationInputs: + """Build the in-memory inputs used by the injected derivation service.""" + return ParameterDerivationInputs( + configuration_steps=self.configuration_steps, + parameters=self.parameters, + vehicle_components=self.vehicle_components, + apm_doc=self.apm_doc, + ) diff --git a/ardupilot_methodic_configurator/log_analysis/data_model_quality_battery.py b/ardupilot_methodic_configurator/log_analysis/data_model_quality_battery.py index 9640e743b..2c160c253 100644 --- a/ardupilot_methodic_configurator/log_analysis/data_model_quality_battery.py +++ b/ardupilot_methodic_configurator/log_analysis/data_model_quality_battery.py @@ -13,14 +13,25 @@ from ardupilot_methodic_configurator.log_analysis.data_model_log_analysis_result import LogAnalysis, LogAnalysisResult from ardupilot_methodic_configurator.log_analysis.data_model_log_quality import LogQualityState from ardupilot_methodic_configurator.log_analysis.data_model_quality_base import ( - BaseLogModel, + BaseLogAnalysisModel, + BaseLogQualityModel, LogQualityResult, QualityIssue, ) from ardupilot_methodic_configurator.log_analysis.utils import find_log_bit_in_apm_file -class BatteryLogQualityModel(BaseLogModel): +def _battery_parameter(instance: int, parameter: str) -> str: + """Return an ArduPilot battery parameter name for a one-based monitor instance.""" + return f"BATT{'' if instance == 1 else instance}_{parameter}" + + +def _configured_battery_instances(parameters: dict[str, float]) -> list[int]: + """Return all configured battery-monitor instances found in logged parameters.""" + return [instance for instance in range(1, 10) if _battery_parameter(instance, "MONITOR") in parameters] + + +class BatteryLogQualityModel(BaseLogQualityModel): """Checks battery telemetry and configuration quality.""" def check(self) -> LogQualityResult: @@ -42,11 +53,23 @@ def _diagnose_absence(self) -> LogQualityResult: "BAT", "Battery Monitor", "Battery", not_logged_hint=_("check the battery monitor physical connection") ) if not bitmask_disabled: - if self.parameters.get("BATT_MONITOR") == 0: + instances = _configured_battery_instances(self.parameters) + enabled_instances = [ + instance for instance in instances if self.parameters[_battery_parameter(instance, "MONITOR")] != 0 + ] + if instances == [1] and not enabled_instances: reason = _("Battery logging enabled but BATT_MONITOR is 0 (monitor disabled)") issues = [ QualityIssue(_("Set BATT_MONITOR to enable the battery monitor"), self.step_for_parameter("BATT_MONITOR")) ] + elif instances and not enabled_instances: + reason = _("Battery logging enabled but all configured battery monitors are disabled") + issues = [ + QualityIssue( + _("Enable a configured battery monitor"), + self.step_for_parameter(_battery_parameter(instances[0], "MONITOR")), + ) + ] else: reason = _("Battery logging enabled but no data, monitor may not be configured properly") issues = [QualityIssue(_("No BAT messages found"), step)] @@ -84,28 +107,36 @@ def check_curr_total(self) -> list[QualityIssue]: _cur_tot, issues = self.field_values_or_issue( "BAT", "CurrTot", - scaled=False, missing_field_message=_("CurrTot field not present in this firmware's BAT schema"), missing_values_message=_("CurrTot missing from BAT records"), ) return issues def check_parameters(self) -> list[QualityIssue]: + """Check failsafe thresholds for every enabled battery monitor instance.""" issues: list[QualityIssue] = [] - monitor = self.parameters.get("BATT_MONITOR") - if monitor is None: - return issues + for instance in _configured_battery_instances(self.parameters): + monitor_param = _battery_parameter(instance, "MONITOR") + if self.parameters[monitor_param] == 0: + continue - if self.parameters.get("BATT_LOW_VOLT") == 0: - issues.append( - QualityIssue(_("Battery low-voltage failsafe threshold disabled"), self.step_for_parameter("BATT_LOW_VOLT")) - ) - if self.parameters.get("BATT_CRT_VOLT") == 0: - issues.append( - QualityIssue( - _("Battery critical-voltage failsafe threshold disabled"), self.step_for_parameter("BATT_CRT_VOLT") + prefix = "" if instance == 1 else f"Battery {instance}: " + low_volt_param = _battery_parameter(instance, "LOW_VOLT") + if self.parameters.get(low_volt_param) == 0: + issues.append( + QualityIssue( + _("{prefix}low-voltage failsafe threshold disabled").format(prefix=prefix), + self.step_for_parameter(low_volt_param), + ) + ) + crt_volt_param = _battery_parameter(instance, "CRT_VOLT") + if self.parameters.get(crt_volt_param) == 0: + issues.append( + QualityIssue( + _("{prefix}critical-voltage failsafe threshold disabled").format(prefix=prefix), + self.step_for_parameter(crt_volt_param), + ) ) - ) return issues @@ -114,7 +145,7 @@ def check_parameters(self) -> list[QualityIssue]: # Hardware and parameter analysis. -class BatteryLogAnalysis(BaseLogModel): +class BatteryLogAnalysis(BaseLogAnalysisModel): """ Battery analysis on the data from the log. @@ -151,39 +182,36 @@ def analyse(self) -> LogAnalysisResult: ) def _last_timestamp_us(self) -> float | None: - """Return the TimeUS value for this message, in microseconds.""" - time_us, _issues = self.field_values_or_issue( + """Return the final canonical timestamp as microseconds for a LogAnalysis result.""" + time_seconds, _issues = self.field_values_or_issue( "BAT", "TimeUS", - scaled=False, missing_field_message=_("TimeUS field not present in this firmware's BAT schema"), missing_values_message=_("TimeUS missing from BAT records"), ) - if time_us is None or len(time_us) == 0: + if time_seconds is None or len(time_seconds) == 0: return None - return float(time_us[-1]) + return float(time_seconds[-1] * 1e6) def check_battery_capacity_retention(self) -> list[LogAnalysis]: """Check what percentage of rated BATT_CAPACITY was consumed during the flight.""" outcomes: list[LogAnalysis] = [] - bat_capacity = self.parameters.get("BATT_CAPACITY") - if bat_capacity is None or bat_capacity <= 0: + bat_capacity_mah = self.parameters.get("BATT_CAPACITY") + if bat_capacity_mah is None or bat_capacity_mah <= 0: return outcomes + bat_capacity_ah = bat_capacity_mah / 1000 curr_tot, _issues = self.field_values_or_issue( "BAT", "CurrTot", - # 3 test logs were tested where scaled values were physically implausible (<1 mAh total consumed) To be tested - # on more logs later until then scaled should be False. - scaled=False, missing_field_message=_("CurrTot field not present in this firmware's BAT schema"), missing_values_message=_("CurrTot missing from BAT records"), ) if curr_tot is None: return outcomes - used_percentage = float(curr_tot.max() / bat_capacity * 100) + used_percentage = float(curr_tot.max() / bat_capacity_ah * 100) timestamp_us = self._last_timestamp_us() if used_percentage > 100: @@ -193,7 +221,7 @@ def check_battery_capacity_retention(self) -> list[LogAnalysis]: "Consumed {used:.1f}% of rated battery capacity ({capacity:.0f} mAh). " "This exceeds 100%, check BATT_CAPACITY is correct for your battery, " "or BATT_AMP_PERVLT." - ).format(used=used_percentage, capacity=bat_capacity), + ).format(used=used_percentage, capacity=bat_capacity_mah), timestamp_us=timestamp_us, value=used_percentage, param_name="BATT_CAPACITY", @@ -203,7 +231,7 @@ def check_battery_capacity_retention(self) -> list[LogAnalysis]: outcomes.append( LogAnalysis( message=_("Used {used:.1f}% of rated battery capacity ({capacity:.0f} mAh)").format( - used=used_percentage, capacity=bat_capacity + used=used_percentage, capacity=bat_capacity_mah ), timestamp_us=timestamp_us, value=used_percentage, @@ -372,14 +400,14 @@ def check_battery_parameter_derivation(self) -> list[LogAnalysis]: "in {step}, based on your vehicle_components specifications." ).format(param=param_name, actual=actual, expected=expected, source=source, step=step_filename) - outcomes.append( - LogAnalysis( - message=message, - timestamp_us=None, - value=actual, - param_name=param_name, - suggested_value=expected, - related_step=step_filename, - ) + outcomes.append( + LogAnalysis( + message=message, + timestamp_us=None, + value=actual, + param_name=param_name, + suggested_value=expected, + related_step=step_filename, ) + ) return outcomes diff --git a/ardupilot_methodic_configurator/log_analysis/data_model_quality_err.py b/ardupilot_methodic_configurator/log_analysis/data_model_quality_err.py index 7fb9309e8..bc25ff350 100644 --- a/ardupilot_methodic_configurator/log_analysis/data_model_quality_err.py +++ b/ardupilot_methodic_configurator/log_analysis/data_model_quality_err.py @@ -11,13 +11,13 @@ from ardupilot_methodic_configurator import _ from ardupilot_methodic_configurator.log_analysis.data_model_log_quality import LogQualityState from ardupilot_methodic_configurator.log_analysis.data_model_quality_base import ( - BaseLogModel, + BaseLogQualityModel, LogQualityResult, QualityIssue, ) -class ErrLogQualityModel(BaseLogModel): +class ErrLogQualityModel(BaseLogQualityModel): """ Checks presence and readability of subsystem error/recovery events. @@ -43,4 +43,4 @@ def check(self) -> LogQualityResult: def check_err_fields(self) -> list[QualityIssue]: """Check that Subsys/ECode fields are present and have readable data.""" - return self.check_fields_present("ERR", ("Subsys", "ECode"), scaled=False) + return self.check_fields_present("ERR", ("Subsys", "ECode")) diff --git a/ardupilot_methodic_configurator/log_analysis/data_model_quality_esc.py b/ardupilot_methodic_configurator/log_analysis/data_model_quality_esc.py index 73a39eba7..c0d2bf435 100644 --- a/ardupilot_methodic_configurator/log_analysis/data_model_quality_esc.py +++ b/ardupilot_methodic_configurator/log_analysis/data_model_quality_esc.py @@ -15,7 +15,8 @@ from ardupilot_methodic_configurator.log_analysis.data_model_log_analysis_result import LogAnalysis, LogAnalysisResult from ardupilot_methodic_configurator.log_analysis.data_model_log_quality import LogQualityState from ardupilot_methodic_configurator.log_analysis.data_model_quality_base import ( - BaseLogModel, + BaseLogAnalysisModel, + BaseLogQualityModel, LogQualityResult, QualityIssue, ) @@ -23,9 +24,12 @@ from ardupilot_methodic_configurator.log_analysis.utils import find_matching_param_values _MOT_SPIN_MIN_REQUIRED_MARGIN = 0.03 +_MIN_ZERO_RPM_OBSERVATION_SECONDS = 1.0 +_MIN_ZERO_RPM_ARMED_COVERAGE = 0.5 +_DSHOT_OUTPUT_RATE_WARN_THRESHOLD = 1000.0 # Amilcar's stated threshold, Hz -class EscLogQualityModel(BaseLogModel): +class EscLogQualityModel(BaseLogQualityModel): """Checks ESC telemetry and configuration quality.""" def check(self) -> LogQualityResult: @@ -104,7 +108,7 @@ def check_error_rate(self) -> list[QualityIssue]: return issues -class EscLogAnalysis(BaseLogModel): +class EscLogAnalysis(BaseLogAnalysisModel): """ ESC analysis on the data from the log. @@ -133,27 +137,27 @@ def analyse(self) -> LogAnalysisResult: ) def _armed_time_windows(self) -> list[tuple[float, float]]: - """Build (arm_time_us, disarm_time_us) pairs from ARM message transitions.""" + """Build canonical-second arm/disarm windows from ARM message transitions.""" columns = self.log_data.get_message_columns("ARM") names = tuple(columns.dtype.names or ()) if columns is not None else () if columns is None or "ArmState" not in names or "TimeUS" not in names: return [] - arm_state = self.log_data.get_field("ARM", "ArmState", scaled=False) - time_us = self.log_data.get_field("ARM", "TimeUS", scaled=False) + arm_state = self.log_data.get_field("ARM", "ArmState") + time_seconds = self.log_data.get_field("ARM", "TimeUS") windows: list[tuple[float, float]] = [] arm_time: float | None = None - for state, ts in zip(arm_state, time_us, strict=True): + for state, timestamp_seconds in zip(arm_state, time_seconds, strict=True): if state and arm_time is None: - arm_time = float(ts) + arm_time = float(timestamp_seconds) elif not state and arm_time is not None: - windows.append((arm_time, float(ts))) + windows.append((arm_time, float(timestamp_seconds))) arm_time = None # Vehicle still armed at end of log (no matching disarm event) - if arm_time is not None and len(time_us) > 0: - windows.append((arm_time, float(time_us[-1]))) + if arm_time is not None and len(time_seconds) > 0: + windows.append((arm_time, float(time_seconds[-1]))) return windows @@ -168,14 +172,14 @@ def check_rpm_while_armed(self) -> list[LogAnalysis]: # pylint: disable=too-man if "Instance" not in names or "RPM" not in names or "TimeUS" not in names: return [] - instance_numbers = self.log_data.get_field("ESC", "Instance", scaled=False) + instance_numbers = self.log_data.get_field("ESC", "Instance") rpm_values = self.log_data.get_field("ESC", "RPM") - time_us = self.log_data.get_field("ESC", "TimeUS", scaled=False) + time_seconds = self.log_data.get_field("ESC", "TimeUS") outcomes: list[LogAnalysis] = [] for instance in sorted(set(instance_numbers.tolist())): mask = instance_numbers == instance - inst_times = time_us[mask] + inst_times = time_seconds[mask] inst_rpm = rpm_values[mask] for arm_time, disarm_time in windows: @@ -183,14 +187,23 @@ def check_rpm_while_armed(self) -> list[LogAnalysis]: # pylint: disable=too-man if not window_mask.any(): continue windowed_rpm = inst_rpm[window_mask] - if windowed_rpm.max() == 0: + windowed_times = inst_times[window_mask] + observed_duration = float(windowed_times[-1] - windowed_times[0]) + armed_duration = float(disarm_time - arm_time) + has_meaningful_coverage = ( + len(windowed_rpm) >= 2 + and observed_duration >= _MIN_ZERO_RPM_OBSERVATION_SECONDS + and armed_duration > 0 + and observed_duration / armed_duration >= _MIN_ZERO_RPM_ARMED_COVERAGE + ) + if windowed_rpm.max() == 0 and has_meaningful_coverage: outcomes.append( LogAnalysis( message=_( "ESC output {instance} reported zero RPM throughout an armed period " "({start:.1f}s to {end:.1f}s), the motor may not have responded." - ).format(instance=int(instance), start=arm_time / 1e6, end=disarm_time / 1e6), - timestamp_us=arm_time, + ).format(instance=int(instance), start=arm_time, end=disarm_time), + timestamp_us=arm_time * 1e6, value=0.0, ) ) @@ -233,24 +246,26 @@ def check_per_instance_errors(self) -> list[LogAnalysis]: if "Instance" not in names or "Err" not in names or "TimeUS" not in names: return [] - instance_numbers = self.log_data.get_field("ESC", "Instance", scaled=False) + instance_numbers = self.log_data.get_field("ESC", "Instance") err_values = self.log_data.get_field("ESC", "Err") - time_us = self.log_data.get_field("ESC", "TimeUS", scaled=False) + time_seconds = self.log_data.get_field("ESC", "TimeUS") outcomes: list[LogAnalysis] = [] for instance in sorted(set(instance_numbers.tolist())): mask = instance_numbers == instance instance_errs = err_values[mask] - instance_times = time_us[mask] + instance_times = time_seconds[mask] peak_err = float(instance_errs.max()) + if peak_err <= 0: + continue peak_idx = instance_errs.argmax() outcomes.append( LogAnalysis( message=_("ESC output {instance} peak error rate: {err:.1f}%").format( instance=int(instance), err=peak_err ), - timestamp_us=float(instance_times[peak_idx]), + timestamp_us=float(instance_times[peak_idx] * 1e6), value=peak_err, ) ) @@ -268,27 +283,34 @@ def check_current_imbalance(self) -> list[LogAnalysis]: # pylint: disable=too-m if "Instance" not in names or "Curr" not in names or "TimeUS" not in names: return [] - instance_numbers = self.log_data.get_field("ESC", "Instance", scaled=False) + instance_numbers = self.log_data.get_field("ESC", "Instance") curr_values = self.log_data.get_field("ESC", "Curr") - time_us = self.log_data.get_field("ESC", "TimeUS", scaled=False) + time_seconds = self.log_data.get_field("ESC", "TimeUS") instances = sorted(set(instance_numbers.tolist())) - if len(instances) < 3: + windows = self._armed_time_windows() + if len(instances) < 3 or not windows: return [] - means = {inst: float(curr_values[instance_numbers == inst].mean()) for inst in instances} + armed_masks = [(time_seconds >= arm_time) & (time_seconds <= disarm_time) for arm_time, disarm_time in windows] + armed_mask = armed_masks[0] + for window_mask in armed_masks[1:]: + armed_mask |= window_mask + + means: dict[float, float] = {} + for instance in instances: + instance_mask = armed_mask & (instance_numbers == instance) + if not instance_mask.any(): + return [] + means[instance] = float(curr_values[instance_mask].mean()) sorted_items = sorted(means.items(), key=lambda kv: kv[1]) values = [v for _, v in sorted_items] total_range = values[-1] - values[0] - if total_range == 0: - return [] - gaps = [values[i + 1] - values[i] for i in range(len(values) - 1)] max_gap = max(gaps) max_gap_idx = gaps.index(max_gap) - - if max_gap <= (total_range - max_gap): + if total_range == 0 or max_gap <= (total_range - max_gap): return [] lower_group = sorted_items[: max_gap_idx + 1] @@ -302,50 +324,58 @@ def check_current_imbalance(self) -> list[LogAnalysis]: # pylint: disable=too-m median_curr = statistics.median(values) outcomes: list[LogAnalysis] = [] for inst, inst_mean in outlier_group: - mask = instance_numbers == inst + mask = armed_mask & (instance_numbers == inst) inst_curr = curr_values[mask] - inst_times = time_us[mask] + inst_times = time_seconds[mask] furthest_idx = abs(inst_curr - median_curr).argmax() outcomes.append( LogAnalysis( message=_( - "ESC output {instance} drew {mean:.2f}A on average," + "ESC output {instance} drew {mean:.2f}A on average, " "current drawn by the other {n} ESC output(s) (median {median:.2f}A). If unexpected " "for this vehicle's configuration, check for a mechanical or wiring issue." ).format(instance=int(inst), mean=inst_mean, n=len(instances) - len(outlier_group), median=median_curr), - timestamp_us=float(inst_times[furthest_idx]), + timestamp_us=float(inst_times[furthest_idx] * 1e6), value=inst_mean, ) ) return outcomes - _DSHOT_OUTPUT_RATE_WARN_THRESHOLD = 1000.0 # Amilcar's stated threshold, Hz - def check_dshot_output_rate(self) -> list[LogAnalysis]: """Check SCHED_LOOP_RATE * SERVO_DSHOT_RATE effective output rate against a minimum threshold.""" loop_rate = self.parameters.get("SCHED_LOOP_RATE") dshot_rate_code = self.parameters.get("SERVO_DSHOT_RATE") - if loop_rate is None or dshot_rate_code is None: + pwm_type = self.parameters.get("MOT_PWM_TYPE") + if loop_rate is None or dshot_rate_code is None or pwm_type is None or self.apm_doc is None: + return [] + + dshot_values = find_matching_param_values(self.apm_doc, "MOT_PWM_TYPE", "DShot") + if str(int(pwm_type)) not in dshot_values: return [] label = enum_value_name(self.apm_doc, "SERVO_DSHOT_RATE", dshot_rate_code) if label is not None and label.lower() == "1khz": - effective_rate = 1000.0 + effective_rate = _DSHOT_OUTPUT_RATE_WARN_THRESHOLD else: multiplier = self._dshot_rate_multiplier(dshot_rate_code) if multiplier is None: return [] # unknown enum value, skip effective_rate = loop_rate * multiplier - if effective_rate <= 1000.0: + if effective_rate <= _DSHOT_OUTPUT_RATE_WARN_THRESHOLD: return [ LogAnalysis( message=_( "Effective DShot output rate is {rate:.0f}Hz (SCHED_LOOP_RATE={loop:.0f}Hz, " "SERVO_DSHOT_RATE={dshot!r}), at or below the {threshold:.0f}Hz level Amilcar flagged as " "problematic. apm.pdef.xml also states SERVO_DSHOT_RATE should never be set below 500Hz." - ).format(rate=effective_rate, loop=loop_rate, dshot=label, threshold=1000.0), + ).format( + rate=effective_rate, + loop=loop_rate, + dshot=label, + threshold=_DSHOT_OUTPUT_RATE_WARN_THRESHOLD, + ), timestamp_us=None, value=effective_rate, param_name="SERVO_DSHOT_RATE", diff --git a/ardupilot_methodic_configurator/log_analysis/data_model_quality_fft.py b/ardupilot_methodic_configurator/log_analysis/data_model_quality_fft.py index 62cf90d42..95f0429a4 100644 --- a/ardupilot_methodic_configurator/log_analysis/data_model_quality_fft.py +++ b/ardupilot_methodic_configurator/log_analysis/data_model_quality_fft.py @@ -11,13 +11,13 @@ from ardupilot_methodic_configurator import _ from ardupilot_methodic_configurator.log_analysis.data_model_log_quality import LogQualityState from ardupilot_methodic_configurator.log_analysis.data_model_quality_base import ( - BaseLogModel, + BaseLogQualityModel, LogQualityResult, QualityIssue, ) -class FftLogQualityModel(BaseLogModel): +class FftLogQualityModel(BaseLogQualityModel): """Checks presence of raw IMU batch logging data (ISBH and ISBD samples).""" def check(self) -> LogQualityResult: diff --git a/ardupilot_methodic_configurator/log_analysis/data_model_quality_gnss.py b/ardupilot_methodic_configurator/log_analysis/data_model_quality_gnss.py index 5be1f7c3b..a6c4d2c61 100644 --- a/ardupilot_methodic_configurator/log_analysis/data_model_quality_gnss.py +++ b/ardupilot_methodic_configurator/log_analysis/data_model_quality_gnss.py @@ -11,13 +11,13 @@ from ardupilot_methodic_configurator import _ from ardupilot_methodic_configurator.log_analysis.data_model_log_quality import LogQualityState from ardupilot_methodic_configurator.log_analysis.data_model_quality_base import ( - BaseLogModel, + BaseLogQualityModel, LogQualityResult, QualityIssue, ) -class GPSLogQualityModel(BaseLogModel): +class GPSLogQualityModel(BaseLogQualityModel): """Checks GPS/GNSS telemetry and configuration quality.""" def check(self) -> LogQualityResult: diff --git a/ardupilot_methodic_configurator/log_analysis/data_model_quality_imu.py b/ardupilot_methodic_configurator/log_analysis/data_model_quality_imu.py index 843dc6061..7a2f2746b 100644 --- a/ardupilot_methodic_configurator/log_analysis/data_model_quality_imu.py +++ b/ardupilot_methodic_configurator/log_analysis/data_model_quality_imu.py @@ -12,7 +12,8 @@ from ardupilot_methodic_configurator.log_analysis.data_model_log_analysis_result import LogAnalysis, LogAnalysisResult from ardupilot_methodic_configurator.log_analysis.data_model_log_quality import LogQualityState from ardupilot_methodic_configurator.log_analysis.data_model_quality_base import ( - BaseLogModel, + BaseLogAnalysisModel, + BaseLogQualityModel, LogQualityResult, QualityIssue, ) @@ -33,7 +34,7 @@ _INVALID_CALTEMP = -300.0 -class ImuLogQualityModel(BaseLogModel): +class ImuLogQualityModel(BaseLogQualityModel): """Checks IMU telemetry quality (error counts, sensor health, raw signal presence).""" def check(self) -> LogQualityResult: @@ -141,7 +142,7 @@ def check_signal_present(self) -> list[QualityIssue]: return issues -class ImuLogAnalysis(BaseLogModel): +class ImuLogAnalysis(BaseLogAnalysisModel): """ IMU analysis on the data from the log. diff --git a/ardupilot_methodic_configurator/log_analysis/data_model_quality_mode.py b/ardupilot_methodic_configurator/log_analysis/data_model_quality_mode.py index 9725c88d4..1c5485e2b 100644 --- a/ardupilot_methodic_configurator/log_analysis/data_model_quality_mode.py +++ b/ardupilot_methodic_configurator/log_analysis/data_model_quality_mode.py @@ -11,13 +11,13 @@ from ardupilot_methodic_configurator import _ from ardupilot_methodic_configurator.log_analysis.data_model_log_quality import LogQualityState from ardupilot_methodic_configurator.log_analysis.data_model_quality_base import ( - BaseLogModel, + BaseLogQualityModel, LogQualityResult, QualityIssue, ) -class ModeLogQualityModel(BaseLogModel): +class ModeLogQualityModel(BaseLogQualityModel): """Checks presence of flight mode change data.""" def check(self) -> LogQualityResult: @@ -48,4 +48,4 @@ def _diagnose_absence(self) -> LogQualityResult: def check_mode_fields(self) -> list[QualityIssue]: """Check that Mode/ModeNum/Rsn fields are present and have readable data.""" - return self.check_fields_present("MODE", ("Mode", "ModeNum", "Rsn"), scaled=False) + return self.check_fields_present("MODE", ("Mode", "ModeNum", "Rsn")) diff --git a/ardupilot_methodic_configurator/log_analysis/data_model_quality_pm.py b/ardupilot_methodic_configurator/log_analysis/data_model_quality_pm.py index d5b9f4769..a853d48ed 100644 --- a/ardupilot_methodic_configurator/log_analysis/data_model_quality_pm.py +++ b/ardupilot_methodic_configurator/log_analysis/data_model_quality_pm.py @@ -11,13 +11,13 @@ from ardupilot_methodic_configurator import _ from ardupilot_methodic_configurator.log_analysis.data_model_log_quality import LogQualityState from ardupilot_methodic_configurator.log_analysis.data_model_quality_base import ( - BaseLogModel, + BaseLogQualityModel, LogQualityResult, QualityIssue, ) -class PmLogQualityModel(BaseLogModel): +class PmLogQualityModel(BaseLogQualityModel): """ Checks presence and readability of system performance (PM) data. @@ -47,5 +47,7 @@ def _diagnose_absence(self) -> LogQualityResult: ) def check_pm_fields(self) -> list[QualityIssue]: - """Check that key PM fields are present and have readable data.""" - return self.check_fields_present("PM", ("Load", "Mem", "NLon", "InE", "ErC"), scaled=False) + """Validate whichever known PM fields are provided by this firmware's schema.""" + optional_fields = ("Load", "Mem", "NLon", "InE", "ErC") + available_fields = tuple(field_name for field_name in optional_fields if self.field_available("PM", field_name)) + return self.check_fields_present("PM", available_fields) diff --git a/ardupilot_methodic_configurator/log_analysis/data_model_quality_vibe.py b/ardupilot_methodic_configurator/log_analysis/data_model_quality_vibe.py index 5ad15d7ad..f66681135 100644 --- a/ardupilot_methodic_configurator/log_analysis/data_model_quality_vibe.py +++ b/ardupilot_methodic_configurator/log_analysis/data_model_quality_vibe.py @@ -14,7 +14,8 @@ from ardupilot_methodic_configurator.log_analysis.data_model_log_analysis_result import LogAnalysis, LogAnalysisResult from ardupilot_methodic_configurator.log_analysis.data_model_log_quality import LogQualityState from ardupilot_methodic_configurator.log_analysis.data_model_quality_base import ( - BaseLogModel, + BaseLogAnalysisModel, + BaseLogQualityModel, LogQualityResult, QualityIssue, ) @@ -25,7 +26,7 @@ _VIBE_AXES = ("VibeX", "VibeY", "VibeZ") -class VibeLogQualityModel(BaseLogModel): +class VibeLogQualityModel(BaseLogQualityModel): """ Checks VIBE data presence and availability for analysis. @@ -80,7 +81,7 @@ def check_clipping(self) -> list[QualityIssue]: return issues -class VibeLogAnalysis(BaseLogModel): +class VibeLogAnalysis(BaseLogAnalysisModel): """ VIBE analysis on the data from the log. @@ -110,23 +111,22 @@ def analyse(self) -> LogAnalysisResult: related_step=step, ) - def _time_us(self) -> Any | None: # noqa: ANN401 - """Return the raw TimeUS array for the VIBE message, or None if unavailable.""" - time_us, _issues = self.field_values_or_issue( + def _time_seconds(self) -> Any | None: # noqa: ANN401 + """Return canonical TimeUS values in seconds, or None if unavailable.""" + time_seconds, _issues = self.field_values_or_issue( "VIBE", "TimeUS", - scaled=False, missing_field_message=_("TimeUS field not present in this firmware's VIBE schema"), missing_values_message=_("TimeUS missing from VIBE records"), ) - return time_us + return time_seconds def check_vibration_levels(self) -> list[LogAnalysis]: """Check peak vibration per axis against the ArduPilot wiki's 30/60 m/s/s guidance.""" outcomes: list[LogAnalysis] = [] - time_us = self._time_us() - if time_us is None: + time_seconds = self._time_seconds() + if time_seconds is None: return outcomes for axis_field in _VIBE_AXES: @@ -141,7 +141,7 @@ def check_vibration_levels(self) -> list[LogAnalysis]: peak_idx = values.argmax() peak_value = float(values[peak_idx]) - peak_timestamp = float(time_us[peak_idx]) + peak_timestamp_us = float(time_seconds[peak_idx] * 1e6) if peak_value >= _VIBE_SEVERE_THRESHOLD: outcomes.append( @@ -150,7 +150,7 @@ def check_vibration_levels(self) -> list[LogAnalysis]: "{axis} peaked at {peak:.1f} m/s/s, above the {severe:.0f} m/s/s level at which " "position/altitude hold problems are nearly always present." ).format(axis=axis_field, peak=peak_value, severe=_VIBE_SEVERE_THRESHOLD), - timestamp_us=peak_timestamp, + timestamp_us=peak_timestamp_us, value=peak_value, ) ) @@ -161,7 +161,7 @@ def check_vibration_levels(self) -> list[LogAnalysis]: "{axis} peaked at {peak:.1f} m/s/s, above the {warn:.0f} m/s/s level that may cause " "position/altitude hold problems." ).format(axis=axis_field, peak=peak_value, warn=_VIBE_WARNING_THRESHOLD), - timestamp_us=peak_timestamp, + timestamp_us=peak_timestamp_us, value=peak_value, ) ) @@ -172,15 +172,14 @@ def check_clipping(self) -> list[LogAnalysis]: """Report total accelerometer clipping events, if any.""" outcomes: list[LogAnalysis] = [] - time_us = self._time_us() + time_seconds = self._time_seconds() clip_values, _issues = self.field_values_or_issue( "VIBE", "Clip", - scaled=False, missing_field_message=_("Clip field not present in this firmware's VIBE schema"), missing_values_message=_("Clip values missing from VIBE records"), ) - if clip_values is None or time_us is None or len(clip_values) == 0: + if clip_values is None or time_seconds is None or len(clip_values) == 0: return outcomes total_clips = float(clip_values.max()) @@ -192,7 +191,7 @@ def check_clipping(self) -> list[LogAnalysis]: "Accelerometer reported {count:.0f} clipping event(s) during the flight, " "meaning the sensor physically saturated at some point." ).format(count=total_clips), - timestamp_us=float(time_us[worst_idx]), + timestamp_us=float(time_seconds[worst_idx] * 1e6), value=total_clips, ) ) diff --git a/ardupilot_methodic_configurator/log_analysis/data_model_tuning_report.py b/ardupilot_methodic_configurator/log_analysis/data_model_tuning_report.py index b08827d79..a1bc9c28f 100644 --- a/ardupilot_methodic_configurator/log_analysis/data_model_tuning_report.py +++ b/ardupilot_methodic_configurator/log_analysis/data_model_tuning_report.py @@ -33,7 +33,16 @@ def load_tuning_report(csv_path: str) -> TuningReport: """Load and forward-fill tuning_report.csv.""" with open(csv_path, encoding="utf-8", newline="") as f: reader = csv.reader(f) - header = next(reader) + try: + header = next(reader) + except StopIteration as exc: + msg = "tuning_report.csv is empty" + raise ValueError(msg) from exc + + if len(header) < 2 or not header[0].strip(): + msg = "tuning_report.csv has no parameter-step columns" + raise ValueError(msg) + # Apply the display name cleanup directly to the headers steps = [step_display_name(step) for step in header[1:]] @@ -46,6 +55,8 @@ def load_tuning_report(csv_path: str) -> TuningReport: # Defensive: some CSV writers drop trailing empty cells, pad to match step count. if len(raw_values) < len(steps): raw_values = raw_values + [""] * (len(steps) - len(raw_values)) + elif len(raw_values) > len(steps): + raw_values = raw_values[: len(steps)] raw_rows.append((param_name, raw_values)) values: dict[str, list[float | None]] = {} diff --git a/tests/test_backend_log_analysis.py b/tests/test_backend_log_analysis.py index db3d0530a..e29f79d50 100755 --- a/tests/test_backend_log_analysis.py +++ b/tests/test_backend_log_analysis.py @@ -31,8 +31,10 @@ def test_analyze_log_file_loads_inputs_and_builds_context(monkeypatch: Any) -> N length=1, format="Nf", fields=["Name", "Value"], - units=[], + stored_units=["", ""], + scaled_units=["", ""], multipliers=[None, None], + multipliers_applied_at_ingest=[False, False], records=1, ), ) diff --git a/tests/test_data_model_battery_analysis.py b/tests/test_data_model_battery_analysis.py index 31e0a18e1..6bf3daf96 100755 --- a/tests/test_data_model_battery_analysis.py +++ b/tests/test_data_model_battery_analysis.py @@ -247,8 +247,8 @@ def test_user_sees_capacity_used_percentage_for_a_normal_flight(self) -> None: WHEN: Capacity retention analysis runs THEN: It reports the correct percentage without flagging an anomaly """ - # Arrange: 500 mAh consumed out of a 1000 mAh pack - log_data = FakeLogData({"BAT": {"CurrTot": [100.0, 500.0], "TimeUS": [0, 1_000_000]}}) + # Arrange: CurrTot is scaled to Ah by LogData: 500 mAh from a 1000 mAh pack + log_data = FakeLogData({"BAT": {"CurrTot": [0.1, 0.5], "TimeUS": [0, 1_000_000]}}) model = BatteryLogAnalysis(log_data, _context({"BATT_CAPACITY": 1000.0})) # Act: run capacity retention analysis @@ -267,8 +267,8 @@ def test_user_is_warned_when_consumed_current_exceeds_rated_capacity(self) -> No WHEN: Capacity retention analysis runs THEN: It flags the anomaly and points at BATT_CAPACITY as the likely cause """ - # Arrange: 1200 mAh consumed against a 1000 mAh rated pack - log_data = FakeLogData({"BAT": {"CurrTot": [1200.0], "TimeUS": [1_000_000]}}) + # Arrange: CurrTot is scaled to Ah by LogData: 1200 mAh against a 1000 mAh rated pack + log_data = FakeLogData({"BAT": {"CurrTot": [1.2], "TimeUS": [1_000_000]}}) model = BatteryLogAnalysis(log_data, _context({"BATT_CAPACITY": 1000.0})) # Act: run capacity retention analysis diff --git a/tests/test_data_model_log_analysis.py b/tests/test_data_model_log_analysis.py index d40731fd7..a1b2e7047 100755 --- a/tests/test_data_model_log_analysis.py +++ b/tests/test_data_model_log_analysis.py @@ -22,11 +22,12 @@ from ardupilot_methodic_configurator.log_analysis.data_model_log_data import LogData from ardupilot_methodic_configurator.log_analysis.data_model_log_quality import LogQualityResult, LogQualityState from ardupilot_methodic_configurator.log_analysis.data_model_quality_base import ( - BaseLogModel, + BaseLogAnalysisModel, + BaseLogQualityModel, ) -class RecordingQualityModel(BaseLogModel): +class RecordingQualityModel(BaseLogQualityModel): """Quality model test double that records constructor inputs.""" seen_log_data: ClassVar[LogData | None] = None @@ -47,7 +48,7 @@ def check(self) -> LogQualityResult: ) -class DummyQualityModel(BaseLogModel): +class DummyQualityModel(BaseLogQualityModel): """Minimal concrete model to exercise base-class context wiring.""" def check(self) -> LogQualityResult: @@ -60,6 +61,26 @@ def check(self) -> LogQualityResult: ) +class DummyAnalysisModel(BaseLogAnalysisModel): + """Minimal detailed model used to exercise parameter-derivation wiring.""" + + +class RecordingParameterDeriver: + """Test double proving detailed models use the context-provided service.""" + + def __init__(self) -> None: + self.expected_call: tuple[str, str] | None = None + self.matching_pattern: str | None = None + + def expected_parameter_value(self, step_filename: str, param_name: str, _inputs: object) -> tuple[float, str]: + self.expected_call = (step_filename, param_name) + return 42.0, "derived" + + def derived_and_forced_parameters_matching(self, pattern: str, _inputs: object) -> dict[str, str]: + self.matching_pattern = pattern + return {"TEST_PARAM": "01_test.param"} + + def test_analyze_log_passes_context_to_quality_models(monkeypatch: Any) -> None: # noqa: ANN401 """ Pass the same context object through to each quality model constructor. @@ -129,6 +150,31 @@ def test_base_quality_model_reads_fields_from_context() -> None: assert model.apm_doc is context.apm_doc +def test_base_analysis_model_delegates_parameter_derivation_to_context_service() -> None: + """Detailed models can be tested with an injected parameter-derivation service.""" + parameter_deriver = RecordingParameterDeriver() + context = LogAnalysisContext( + parameters={"TEST_PARAM": 1.0}, + configuration_steps={"01_test.param": {}}, + parameter_deriver=parameter_deriver, + ) + model = DummyAnalysisModel(LogData(), context) + + assert model.expected_parameter_value("01_test.param", "TEST_PARAM") == (42.0, "derived") + assert model.derived_and_forced_parameters_matching(r"TEST_.*") == {"TEST_PARAM": "01_test.param"} + assert parameter_deriver.expected_call == ("01_test.param", "TEST_PARAM") + assert parameter_deriver.matching_pattern == r"TEST_.*" + + +def test_subsystem_component_metadata_is_declared_by_the_registry() -> None: + """Component guidance belongs to subsystem registration, not translated UI names.""" + component_keys = {spec.key: spec.component_keys for spec in data_model_log_analysis.LOG_ANALYSIS_SUBSYSTEMS} + + assert component_keys["battery"] == ("Battery", "Battery Monitor") + assert component_keys["gps"] == ("GNSS Receiver",) + assert component_keys["esc"] == ("ESC", "Motors") + + def test_base_quality_model_tolerates_ambiguous_configuration_metadata() -> None: """ Keep report generation alive when configuration metadata has duplicate references. diff --git a/tests/test_data_model_log_quality_check.py b/tests/test_data_model_log_quality_check.py index 8f6ddc594..940e7defc 100755 --- a/tests/test_data_model_log_quality_check.py +++ b/tests/test_data_model_log_quality_check.py @@ -13,6 +13,8 @@ from ardupilot_methodic_configurator.log_analysis.data_model_log_data import LogData, MessageSchema from ardupilot_methodic_configurator.log_analysis.data_model_log_quality_check import ( + check_cpu_performance_message, + get_pm_status, validate_configuration_steps_data, validate_fmt_schema, ) @@ -27,8 +29,10 @@ def _make_schema(fields: list[str]) -> MessageSchema: length=4, format="f", fields=fields, - units=[""] * len(fields), + stored_units=[""] * len(fields), + scaled_units=[""] * len(fields), multipliers=[None] * len(fields), + multipliers_applied_at_ingest=[False] * len(fields), ) @@ -88,6 +92,64 @@ def test_matching_records_are_valid(self, schema_with_two_fields: MessageSchema, assert not result.issues +def test_pm_non_finite_values_are_reported_as_invalid() -> None: + """NaN PM telemetry must not be treated as a valid performance report.""" + log_data = LogData() + log_data.add_message_columns( + "PM", + np.array( + [(np.nan, np.nan, np.nan, np.nan, np.nan, np.nan, np.nan)], + dtype=[ + ("Load", "f8"), + ("NLon", "f8"), + ("MaxT", "f8"), + ("Mem", "f8"), + ("InE", "f8"), + ("ErC", "f8"), + ("ErrL", "f8"), + ], + ), + ) + + validation = check_cpu_performance_message(log_data) + status = get_pm_status(log_data) + + assert validation.valid is False + assert len(validation.issues) == 7 + assert all("non-finite telemetry values" in issue for issue in validation.issues) + assert status is not None + assert status.healthy is None + + +def test_pm_error_signals_preserve_report_order() -> None: + """PM error findings retain the established order used by reports and snapshots.""" + log_data = LogData() + log_data.add_message_columns( + "PM", + np.array( + [(25.0, 3, 0, 20_000, 1, 2, 42)], + dtype=[ + ("Load", "f8"), + ("NLon", "i4"), + ("MaxT", "i4"), + ("Mem", "i4"), + ("InE", "i4"), + ("ErC", "i4"), + ("ErrL", "i4"), + ], + ), + ) + + validation = check_cpu_performance_message(log_data) + + assert validation.issues == [ + "Internal firmware errors were detected (InE)", + "Internal error count: 2", + "An internal error line was recorded (ErrL)", + "Detected 3 scheduler long loops", + ] + + class TestValidateConfigurationSteps: """Validate configuration steps using extracted log data.""" diff --git a/tests/test_data_model_quality_models.py b/tests/test_data_model_quality_models.py index 0bcd174f0..ca1e3d5c0 100755 --- a/tests/test_data_model_quality_models.py +++ b/tests/test_data_model_quality_models.py @@ -13,9 +13,10 @@ from ardupilot_methodic_configurator.log_analysis.data_model_log_analysis_context import LogAnalysisContext from ardupilot_methodic_configurator.log_analysis.data_model_log_data import LogData from ardupilot_methodic_configurator.log_analysis.data_model_log_quality import LogQualityState -from ardupilot_methodic_configurator.log_analysis.data_model_quality_battery import BatteryLogQualityModel -from ardupilot_methodic_configurator.log_analysis.data_model_quality_esc import EscLogQualityModel +from ardupilot_methodic_configurator.log_analysis.data_model_quality_battery import BatteryLogAnalysis, BatteryLogQualityModel +from ardupilot_methodic_configurator.log_analysis.data_model_quality_esc import EscLogAnalysis, EscLogQualityModel from ardupilot_methodic_configurator.log_analysis.data_model_quality_gnss import GPSLogQualityModel +from ardupilot_methodic_configurator.log_analysis.data_model_quality_pm import PmLogQualityModel def _context( @@ -109,3 +110,119 @@ def test_battery_model_checks_present_log_data_without_datasource_access() -> No assert result.available is True assert result.state == LogQualityState.WARNING assert any("Voltage is zero" in issue.message for issue in result.issues) + + +def test_battery_model_rejects_non_finite_telemetry_values() -> None: + """NaN telemetry must not be reported as healthy battery data.""" + log_data = LogData() + log_data.add_message_columns( + "BAT", + np.array( + [(np.nan, np.nan, np.nan)], + dtype=[("Volt", "f8"), ("Curr", "f8"), ("CurrTot", "f8")], + ), + ) + + result = BatteryLogQualityModel(log_data, _context({"BATT_MONITOR": 4.0})).check() + + assert result.available is True + assert result.state == LogQualityState.WARNING + assert len(result.issues) == 3 + assert all("non-finite telemetry values" in issue.message for issue in result.issues) + + +def test_battery_model_checks_failsafes_for_each_enabled_monitor() -> None: + """An enabled secondary monitor must not bypass its own voltage failsafes.""" + log_data = LogData() + log_data.add_message_columns( + "BAT", + np.array([(22.0, 1.0, 0.1)], dtype=[("Volt", "f8"), ("Curr", "f8"), ("CurrTot", "f8")]), + ) + + result = BatteryLogQualityModel( + log_data, + _context({"BATT_MONITOR": 4.0, "BATT2_MONITOR": 4.0, "BATT2_LOW_VOLT": 0.0, "BATT2_CRT_VOLT": 0.0}), + ).check() + + assert result.state == LogQualityState.WARNING + assert [issue.message for issue in result.issues] == [ + "Battery 2: low-voltage failsafe threshold disabled", + "Battery 2: critical-voltage failsafe threshold disabled", + ] + + +def test_pm_model_accepts_valid_legacy_schema_without_internal_error_fields() -> None: + """Older PM schemas do not contain the current InE and ErC fields.""" + log_data = LogData() + log_data.add_message_columns( + "PM", + np.array([(20.0, 10_000.0, 0.0)], dtype=[("Load", "f8"), ("Mem", "f8"), ("NLon", "f8")]), + ) + + result = PmLogQualityModel(log_data, _context({})).check() + + assert result.state == LogQualityState.INFO + + +def test_battery_capacity_uses_canonical_amp_hours() -> None: + """A mAh capacity parameter must be converted before comparison with scaled CurrTot.""" + log_data = LogData() + log_data.add_message_columns( + "BAT", + np.array([(0.5, 1.0)], dtype=[("CurrTot", "f8"), ("TimeUS", "f8")]), + ) + + outcomes = BatteryLogAnalysis(log_data, _context({"BATT_CAPACITY": 1000.0})).check_battery_capacity_retention() + + assert len(outcomes) == 1 + assert outcomes[0].value == 50.0 + assert outcomes[0].timestamp_us == 1_000_000.0 + + +def test_esc_analysis_does_not_report_zero_error_rates_as_findings() -> None: + """A healthy ESC log should have no error-rate findings.""" + log_data = LogData() + log_data.add_message_columns( + "ESC", + np.array( + [(0.0, 0.0, 1_000_000.0), (1.0, 0.0, 2_000_000.0)], + dtype=[("Instance", "f8"), ("Err", "f8"), ("TimeUS", "f8")], + ), + ) + + outcomes = EscLogAnalysis(log_data, _context({})).check_per_instance_errors() + + assert outcomes == [] + + +def test_esc_analysis_ignores_a_single_zero_rpm_sample() -> None: + """A sparse zero-RPM sample must not be described as a full armed-period failure.""" + log_data = LogData() + log_data.add_message_columns( + "ARM", + np.array([(1.0, 0.0), (0.0, 2.0)], dtype=[("ArmState", "f8"), ("TimeUS", "f8")]), + ) + log_data.add_message_columns( + "ESC", + np.array([(0.0, 0.0, 1.0)], dtype=[("Instance", "f8"), ("RPM", "f8"), ("TimeUS", "f8")]), + ) + + outcomes = EscLogAnalysis(log_data, _context({})).check_rpm_while_armed() + + assert outcomes == [] + + +def test_esc_analysis_skips_dshot_rate_for_pwm_output() -> None: + """A SERVO_DSHOT_RATE parameter must not trigger findings when PWM outputs are active.""" + model = EscLogAnalysis( + LogData(), + _context( + {"MOT_PWM_TYPE": 0.0, "SCHED_LOOP_RATE": 400.0, "SERVO_DSHOT_RATE": 1.0}, + apm_doc={ + "MOT_PWM_TYPE": {"values": {"0": "Normal", "4": "DShot150"}}, + "SERVO_DSHOT_RATE": {"values": {"1": "loop-rate"}}, + }, + ), + ) + + assert model.check_dshot_output_rate() == [] diff --git a/tests/test_data_model_tuning_report.py b/tests/test_data_model_tuning_report.py new file mode 100755 index 000000000..0626d1745 --- /dev/null +++ b/tests/test_data_model_tuning_report.py @@ -0,0 +1,35 @@ +#!/usr/bin/env python3 + +""" +Behavior-driven tests for parsing tuning_report.csv files. + +This file is part of ArduPilot Methodic Configurator. https://github.com/ArduPilot/MethodicConfigurator + +SPDX-FileCopyrightText: 2024-2026 Amilcar do Carmo Lucas + +SPDX-License-Identifier: GPL-3.0-or-later +""" + +from pathlib import Path + +from ardupilot_methodic_configurator.log_analysis.data_model_tuning_report import load_tuning_report + + +def test_load_tuning_report_truncates_cells_beyond_the_header(tmp_path: Path) -> None: + """ + Users can open a report with a row that contains an unexpected trailing value. + + GIVEN: A tuning report with two step headers and a parameter row with three values + WHEN: The report is loaded for graphing + THEN: The parameter series remains aligned with the two configured steps + """ + # Arrange: A malformed but recoverable report row. + report_path = tmp_path / "tuning_report.csv" + report_path.write_text("Parameter,01_setup.param,02_tune.param\nMOT_SPIN_MIN,0.1,0.12,0.99\n", encoding="utf-8") + + # Act: Parse the report for the tuning graph. + report = load_tuning_report(str(report_path)) + + # Assert: Ignore only the trailing value that lacks a step header. + assert report.steps == ["Setup", "Tune"] + assert report.values["MOT_SPIN_MIN"] == [0.1, 0.12] diff --git a/tests/test_data_model_vehicle_overview_report.py b/tests/test_data_model_vehicle_overview_report.py index 7b24c2b6a..afed42ec2 100755 --- a/tests/test_data_model_vehicle_overview_report.py +++ b/tests/test_data_model_vehicle_overview_report.py @@ -78,8 +78,10 @@ def test_build_baro_info_handles_missing_instance_field_without_crashing() -> No length=1, format="B", fields=["Health"], - units=[], + stored_units=[""], + scaled_units=[""], multipliers=[None], + multipliers_applied_at_ingest=[False], records=1, ) diff --git a/tests/test_frontend_tkinter_log_analysis.py b/tests/test_frontend_tkinter_log_analysis.py index e33336704..2ddec9f9f 100755 --- a/tests/test_frontend_tkinter_log_analysis.py +++ b/tests/test_frontend_tkinter_log_analysis.py @@ -23,14 +23,17 @@ LogAnalysisReportWindow, _collect_links, _format_component, - paired_quality_and_analysis_results, ) +from ardupilot_methodic_configurator.log_analysis.data_model_log_analysis import LogSummary if TYPE_CHECKING: from pathlib import Path from pytest_mock import MockerFixture + from ardupilot_methodic_configurator.log_analysis.data_model_log_analysis_result import LogAnalysisResult + from ardupilot_methodic_configurator.log_analysis.data_model_log_quality import LogQualityResult + MODULE = "ardupilot_methodic_configurator.frontend_tkinter_log_analysis" # pylint: disable=protected-access, redefined-outer-name @@ -50,6 +53,7 @@ def _make_quality_result( reason: str = "Battery data present and good for analysis", issues: list[MagicMock] | None = None, related_step: str = "", + subsystem_key: str | None = "battery", ) -> MagicMock: result = MagicMock() result.name = name @@ -57,6 +61,7 @@ def _make_quality_result( result.reason = reason result.issues = issues or [] result.related_step = related_step + result.subsystem_key = subsystem_key return result @@ -64,10 +69,14 @@ def _make_outcome( *, message: str = "finding", timestamp_us: float | None = None, + param_name: str | None = None, + suggested_value: float | None = None, ) -> MagicMock: outcome = MagicMock() outcome.message = message outcome.timestamp_us = timestamp_us + outcome.param_name = param_name + outcome.suggested_value = suggested_value return outcome @@ -77,34 +86,39 @@ def _make_analysis_result( available: bool = True, outcomes: list[MagicMock] | None = None, reason: str = "Battery analysis complete", + subsystem_key: str | None = "battery", ) -> MagicMock: result = MagicMock() result.name = name result.available = available result.outcomes = outcomes if outcomes is not None else [] result.reason = reason + result.subsystem_key = subsystem_key return result -def _make_summary(quality_results: list[MagicMock], analysis_results: list[MagicMock]) -> MagicMock: - summary = MagicMock() - summary.quality_results = quality_results - summary.analysis_results = analysis_results - return summary - - -class _FakeQualityCls: # pylint: disable=too-few-public-methods - """Stand-in for a real quality model class - identity only matters for pairing.""" - - -class _FakeAnalysisCls: # pylint: disable=too-few-public-methods - """Stand-in for a real analysis model class - identity only matters for pairing.""" +def _make_summary(quality_results: list[MagicMock], analysis_results: list[MagicMock]) -> LogSummary: + return LogSummary( + flight_duration_sec=None, + file_size_bytes=0, + total_messages=0, + message_types=0, + parameter_count=0, + pm_status=None, + pm_validation=None, + quality_results=cast("list[LogQualityResult]", quality_results), + analysis_results=cast("list[LogAnalysisResult]", analysis_results), + step_results=[], + hardware_report=MagicMock(), + analysis_subsystem_keys=tuple(result.subsystem_key for result in analysis_results if result.subsystem_key is not None) + or ("battery",), + ) class TestPairedQualityAndAnalysisResults: - """Cover the offset-detection pairing logic between quality and analysis results.""" + """Cover stable-key pairing between quality and analysis results.""" - def test_pairs_available_quality_with_its_analysis_result(self, mocker: MockerFixture) -> None: + def test_pairs_available_quality_with_its_analysis_result(self) -> None: """ Quality models with an analysis counterpart are paired when data is available. @@ -112,16 +126,15 @@ def test_pairs_available_quality_with_its_analysis_result(self, mocker: MockerFi WHEN: Results are paired THEN: The quality result is paired with the corresponding analysis result """ - mocker.patch(f"{MODULE}.QUALITY_AND_ANALYSIS_MODELS", [(_FakeQualityCls, _FakeAnalysisCls)]) quality = _make_quality_result(name="Battery", available=True) analysis = _make_analysis_result(name="Battery Analysis") summary = _make_summary([quality], [analysis]) - pairs = paired_quality_and_analysis_results(summary) + pairs = summary.paired_quality_and_analysis_results() assert pairs == [(quality, analysis)] - def test_pairs_unavailable_quality_with_none(self, mocker: MockerFixture) -> None: + def test_pairs_unavailable_quality_with_none(self) -> None: """ Quality models whose data was unavailable pair with None instead of an analysis result. @@ -130,15 +143,14 @@ def test_pairs_unavailable_quality_with_none(self, mocker: MockerFixture) -> Non THEN: The analysis side of the pair is None AND: No analysis result is consumed from the iterator """ - mocker.patch(f"{MODULE}.QUALITY_AND_ANALYSIS_MODELS", [(_FakeQualityCls, _FakeAnalysisCls)]) quality = _make_quality_result(name="ESC telemetry", available=False) summary = _make_summary([quality], []) - pairs = paired_quality_and_analysis_results(summary) + pairs = summary.paired_quality_and_analysis_results() assert pairs == [(quality, None)] - def test_skips_quality_models_with_no_analysis_class(self, mocker: MockerFixture) -> None: + def test_skips_quality_models_with_no_analysis_class(self) -> None: """ Subsystems with no analysis model (analysis_cls is None) are excluded entirely. @@ -146,48 +158,48 @@ def test_skips_quality_models_with_no_analysis_class(self, mocker: MockerFixture WHEN: Results are paired THEN: GPS does not appear in the paired output at all """ - mocker.patch(f"{MODULE}.QUALITY_AND_ANALYSIS_MODELS", [(_FakeQualityCls, None)]) - quality = _make_quality_result(name="GPS", available=True) + quality = _make_quality_result(name="GPS", available=True, subsystem_key="gps") summary = _make_summary([quality], []) - pairs = paired_quality_and_analysis_results(summary) + pairs = summary.paired_quality_and_analysis_results() assert not pairs - def test_handles_one_prepended_quality_result_via_offset(self, mocker: MockerFixture) -> None: + def test_ignores_unregistered_prepended_quality_result(self) -> None: """ A single extra quality result (e.g. System Performance) prepended by analyze_log(). - GIVEN: quality_results has one more entry than QUALITY_AND_ANALYSIS_MODELS + GIVEN: the extra result has no subsystem key WHEN: Results are paired - THEN: The registry's Battery entry pairs with the second quality_results entry, not the first + THEN: the keyed Battery entry is paired, independently of list position """ - mocker.patch(f"{MODULE}.QUALITY_AND_ANALYSIS_MODELS", [(_FakeQualityCls, _FakeAnalysisCls)]) - prepended = _make_quality_result(name="System Performance", available=True) + prepended = _make_quality_result(name="System Performance", available=True, subsystem_key=None) battery_quality = _make_quality_result(name="Battery", available=True) analysis = _make_analysis_result(name="Battery Analysis") summary = _make_summary([prepended, battery_quality], [analysis]) - pairs = paired_quality_and_analysis_results(summary) + pairs = summary.paired_quality_and_analysis_results() assert pairs == [(battery_quality, analysis)] - def test_raises_when_offset_is_not_zero_or_one(self, mocker: MockerFixture) -> None: + def test_ignores_results_without_a_registered_subsystem_key(self) -> None: """ - An unexpected length mismatch fails loudly instead of silently misaligning results. + Unknown or legacy result entries do not affect keyed pairing. - GIVEN: quality_results has two more entries than the registry (not 0 or 1 offset) + GIVEN: quality results have no registered subsystem keys WHEN: Results are paired - THEN: An AssertionError is raised describing the mismatch + THEN: no positional pairing is attempted """ - mocker.patch(f"{MODULE}.QUALITY_AND_ANALYSIS_MODELS", [(_FakeQualityCls, _FakeAnalysisCls)]) summary = _make_summary( - [_make_quality_result(), _make_quality_result(), _make_quality_result()], + [ + _make_quality_result(subsystem_key=None), + _make_quality_result(subsystem_key=None), + _make_quality_result(subsystem_key=None), + ], [], ) - with pytest.raises(AssertionError, match="Unexpected quality_results length"): - paired_quality_and_analysis_results(summary) + assert summary.paired_quality_and_analysis_results() == [] class TestCollectLinks: @@ -401,6 +413,7 @@ def bare_window() -> LogAnalysisReportWindow: window = LogAnalysisReportWindow.__new__(LogAnalysisReportWindow) window.root = MagicMock() window.main_frame = MagicMock() + window.summary = MagicMock(related_parameter_values={}) return window @@ -417,7 +430,6 @@ def _build_window( report: dict | None = None, vehicle_dir: str = "/vehicle", ) -> LogAnalysisReportWindow: - mocker.patch(f"{MODULE}.QUALITY_AND_ANALYSIS_MODELS", [(_FakeQualityCls, _FakeAnalysisCls)] * len(quality_results)) mocker.patch.object(LogAnalysisReportWindow, "calculate_scaled_geometry", return_value="1050x800") mocker.patch.object(LogAnalysisReportWindow, "center_window") summary = _make_summary(quality_results, analysis_results) @@ -439,10 +451,10 @@ def test_selector_defaults_to_first_subsystem(self, mocker: MockerFixture, patch WHEN: The window is constructed THEN: The selector's set() is called with the first subsystem's name """ - quality_battery = _make_quality_result(name="Battery") - quality_imu = _make_quality_result(name="IMU") - analysis_battery = _make_analysis_result(name="Battery Analysis") - analysis_imu = _make_analysis_result(name="IMU Analysis") + quality_battery = _make_quality_result(name="Battery", subsystem_key="battery") + quality_imu = _make_quality_result(name="IMU", subsystem_key="imu") + analysis_battery = _make_analysis_result(name="Battery Analysis", subsystem_key="battery") + analysis_imu = _make_analysis_result(name="IMU Analysis", subsystem_key="imu") window = self._build_window(mocker, patched_widgets, [quality_battery, quality_imu], [analysis_battery, analysis_imu]) @@ -477,7 +489,7 @@ def test_button_enabled_when_tuning_report_exists( THEN: The button is configured with state="normal" """ (tmp_path / "tuning_report.csv").write_text("param,00_default.param\n", encoding="utf-8") - window = cast("LogAnalysisReportWindow", LogAnalysisReportWindow.__new__(LogAnalysisReportWindow)) + window = LogAnalysisReportWindow.__new__(LogAnalysisReportWindow) window.main_frame = MagicMock() window.vehicle_dir = str(tmp_path) window.report = None @@ -497,7 +509,7 @@ def test_button_disabled_when_tuning_report_missing( WHEN: The footer is built THEN: The button is configured with state="disabled" """ - window = cast("LogAnalysisReportWindow", LogAnalysisReportWindow.__new__(LogAnalysisReportWindow)) + window = LogAnalysisReportWindow.__new__(LogAnalysisReportWindow) window.main_frame = MagicMock() window.vehicle_dir = str(tmp_path) window.report = None @@ -610,6 +622,38 @@ def test_renders_each_outcome_as_an_outcome_line(self, bare_window: LogAnalysisR mock_outcome_line.assert_any_call(outcomes[0]) mock_outcome_line.assert_any_call(outcomes[1]) + def test_actionable_outcome_exposes_a_parameter_fix(self, bare_window: LogAnalysisReportWindow) -> None: + """Analysis recommendations should use the same parameter-fix workflow as quality issues.""" + outcome = _make_outcome(param_name="MOT_SPIN_MIN", suggested_value=0.15) + bare_window.summary.related_parameter_values = {"MOT_SPIN_MIN": 0.1} + + fixes = bare_window._fix_for_outcome(outcome) + + assert fixes == [("MOT_SPIN_MIN", 0.1, 0.15, ["finding"])] + + @pytest.mark.parametrize("upload_result", [None, True]) + def test_analysis_fix_accepts_non_false_upload_result( + self, bare_window: LogAnalysisReportWindow, upload_result: bool | None + ) -> None: + """ + Treat callbacks without an explicit failure result as successful. + + GIVEN a parameter-fix callback returning None or True, + WHEN the analysis report applies a fix, + THEN the displayed parameter state is updated and the dialog closes. + """ + # Arrange: provide a side-effect callback and a pending parameter fix. + bare_window.upload_callback = MagicMock(return_value=upload_result) + fixes = [("MOT_SPIN_MIN", 0.1, 0.15, ["finding"])] + dialog = MagicMock() + + # Act: apply the proposed change. + bare_window._apply_param_fixes(fixes, dialog) + + # Assert: None is not mistaken for an upload failure. + assert bare_window.summary.related_parameter_values == {"MOT_SPIN_MIN": 0.15} + dialog.destroy.assert_called_once() + def test_renders_no_hardware_section_when_no_component_data(self, bare_window: LogAnalysisReportWindow) -> None: """ Subsystems with no matching vehicle_components entries do not show a Hardware section. diff --git a/tests/test_frontend_tkinter_log_quality.py b/tests/test_frontend_tkinter_log_quality.py new file mode 100755 index 000000000..e46f01901 --- /dev/null +++ b/tests/test_frontend_tkinter_log_quality.py @@ -0,0 +1,66 @@ +#!/usr/bin/env python3 + +""" +Tests for ardupilot_methodic_configurator/frontend_tkinter_log_quality.py. + +This file is part of ArduPilot Methodic Configurator. https://github.com/ArduPilot/MethodicConfigurator + +SPDX-FileCopyrightText: 2024-2026 Amilcar do Carmo Lucas + +SPDX-License-Identifier: GPL-3.0-or-later +""" + +from unittest.mock import MagicMock + +import pytest + +from ardupilot_methodic_configurator.frontend_tkinter_log_quality import LogQualityReportWindow, _format_parameter_value + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + (123.0, "123"), + (0.15, "0.15"), + (1.25, "1.25"), + ], +) +def test_parameter_value_formatting_preserves_fractional_changes(value: float, expected: str) -> None: + """ + Display parameter values without silently changing their meaning. + + GIVEN integral and fractional parameter values, + WHEN they are formatted for the parameter-change review dialog, + THEN integral values omit an unnecessary decimal and fractional values are preserved. + """ + # Arrange: parametrized parameter values represent pending FC changes. + + # Act: format the value shown to the user. + actual = _format_parameter_value(value) + + # Assert: the text faithfully represents the value that will be uploaded. + assert actual == expected + + +@pytest.mark.parametrize("upload_result", [None, True]) +def test_quality_fix_accepts_non_false_upload_result(upload_result: bool | None) -> None: + """ + Treat callbacks without an explicit failure result as successful. + + GIVEN a parameter-fix callback returning None or True, + WHEN the quality report applies a fix, + THEN the displayed parameter state is updated and the dialog closes. + """ + # Arrange: bypass Tk construction and provide a side-effect callback. + window = LogQualityReportWindow.__new__(LogQualityReportWindow) + window.summary = MagicMock(related_parameter_values={}) + window.upload_callback = MagicMock(return_value=upload_result) + dialog = MagicMock() + fixes = [("MOT_SPIN_MIN", 0.1, 0.15, ["finding"])] + + # Act: apply the proposed change. + window._apply_param_fixes(fixes, dialog) + + # Assert: None is not mistaken for an upload failure. + assert window.summary.related_parameter_values == {"MOT_SPIN_MIN": 0.15} + dialog.destroy.assert_called_once() diff --git a/tests/unit_backend_log_extraction.py b/tests/unit_backend_log_extraction.py index bf117a28f..b7560b186 100755 --- a/tests/unit_backend_log_extraction.py +++ b/tests/unit_backend_log_extraction.py @@ -8,6 +8,7 @@ SPDX-License-Identifier: GPL-3.0-or-later """ +from pathlib import Path from types import SimpleNamespace from unittest.mock import MagicMock, patch @@ -18,6 +19,7 @@ from ardupilot_methodic_configurator.log_analysis.backend_log_extraction import ( _allocate_message_arrays, _fill_message_arrays, + _FMTUDefinition, _schema_numpy_dtype, close_log, extract_schemas, @@ -34,6 +36,7 @@ class MessageStub: def __init__(self, msg_type: str, **payload: object) -> None: self._msg_type = msg_type self._payload = payload + self._elements = list(payload.values()) for key, value in payload.items(): setattr(self, key, value) @@ -75,8 +78,10 @@ def populated_log_data_with_float_field() -> LogData: length=4, format="f", fields=["Value"], - units=["V"], + stored_units=["V"], + scaled_units=["V"], multipliers=[0.01], + multipliers_applied_at_ingest=[False], ) log_data.msg_count["PARM"] = 2 log_data.schemas["PARM"].records = 2 @@ -148,18 +153,21 @@ def test_schema_numpy_dtype_maps_numeric_fields(self) -> None: name="TEST", msg_type=1, length=4, - format="bHf", - fields=["a", "b", "c"], - units=["", "", ""], - multipliers=[None, None, None], + format="bHfg", + fields=["a", "b", "c", "d"], + stored_units=["", "", "", ""], + scaled_units=["", "", "", ""], + multipliers=[None, None, None, None], + multipliers_applied_at_ingest=[False, False, False, False], ) dtype = _schema_numpy_dtype(schema) - assert dtype.names == ("a", "b", "c") + assert dtype.names == ("a", "b", "c", "d") assert dtype["a"] == np.dtype(np.int8) assert dtype["b"] == np.dtype(np.uint16) assert dtype["c"] == np.dtype(np.float32) + assert dtype["d"] == np.dtype(np.float16) def test_preallocated_arrays_store_and_scale_message_values(self, populated_log_data_with_float_field: LogData) -> None: """ @@ -198,14 +206,14 @@ def test_get_message_columns_returns_raw_structured_array(self) -> None: assert columns is not None assert columns.dtype.names == ("Value",) - def test_integer_scaling_promotes_dtype(self) -> None: + def test_fixed_point_fields_store_raw_values_and_scale_lazily(self) -> None: """ - Promote fixed-point integers when scaling. + Store compact fixed-point values and scale them on access. - GIVEN fixed-point integer fields, + GIVEN fixed-point fields with raw pymavlink elements, - WHEN scaled values are accessed, - THEN the returned dtype should be widened to avoid overflow. + WHEN values are stored in the preallocated array, + THEN the raw integer storage is retained and scaled values are floats. """ log_data = LogData() log_data.schemas["INTS"] = MessageSchema( @@ -214,8 +222,10 @@ def test_integer_scaling_promotes_dtype(self) -> None: length=8, format="ce", fields=["Small", "Large"], - units=["", ""], - multipliers=[100, 100], + stored_units=["", ""], + scaled_units=["", ""], + multipliers=[0.01, 0.01], + multipliers_applied_at_ingest=[False, False], ) log_data.msg_count["INTS"] = 2 log_data.schemas["INTS"].records = 2 @@ -229,13 +239,19 @@ def test_integer_scaling_promotes_dtype(self) -> None: ) _fill_message_arrays(mock_mlog, log_data) + raw_small = log_data.get_field("INTS", "Small", scaled=False) + raw_large = log_data.get_field("INTS", "Large", scaled=False) small = log_data.get_field("INTS", "Small") large = log_data.get_field("INTS", "Large") - assert small.dtype == np.int32 - assert large.dtype == np.int64 - np.testing.assert_array_equal(small, np.array([100, 300], dtype=np.int32)) - np.testing.assert_array_equal(large, np.array([200, 400], dtype=np.int64)) + assert raw_small.dtype == np.int16 + assert raw_large.dtype == np.int32 + assert small.dtype == np.float64 + assert large.dtype == np.float64 + np.testing.assert_array_equal(raw_small, np.array([1, 3], dtype=np.int16)) + np.testing.assert_array_equal(raw_large, np.array([2, 4], dtype=np.int32)) + np.testing.assert_array_equal(small, np.array([0.01, 0.03])) + np.testing.assert_array_equal(large, np.array([0.02, 0.04])) def test_fill_message_arrays_raises_on_field_mismatch(self) -> None: """ @@ -253,8 +269,10 @@ def test_fill_message_arrays_raises_on_field_mismatch(self) -> None: length=4, format="f", fields=["Value"], - units=["V"], + stored_units=["V"], + scaled_units=["V"], multipliers=[None], + multipliers_applied_at_ingest=[False], ) log_data.msg_count["PARM"] = 1 log_data.schemas["PARM"].records = 1 @@ -285,8 +303,10 @@ def test_second_pass_reports_progress_against_known_message_count(self) -> None: length=4, format="f", fields=["Value"], - units=["V"], + stored_units=["V"], + scaled_units=["V"], multipliers=[None], + multipliers_applied_at_ingest=[False], records=2, ) log_data.msg_count = {"PARM": 2, "MSG": 1} @@ -326,7 +346,7 @@ def test_extract_schemas_populates_message_schema(self, empty_log_data: LogData) units=["V"], msg_mults=[1.0], ) - mock_mlog = SimpleNamespace(formats={"PARM": mock_fmt}, mult_lookup={}) + mock_mlog = SimpleNamespace(formats={"PARM": mock_fmt}, mult_lookup={}, unit_lookup={}) empty_log_data.msg_count["PARM"] = 5 extract_schemas(mock_mlog, empty_log_data, {}) @@ -335,5 +355,57 @@ def test_extract_schemas_populates_message_schema(self, empty_log_data: LogData) assert isinstance(schema, MessageSchema) assert schema.name == "PARM" assert schema.fields == ["Value"] - assert schema.units == ["V"] + assert schema.scaled_units == ["V"] + assert schema.stored_units == ["V"] + assert schema.multipliers_applied_at_ingest == [False] assert schema.records == 5 + + def test_extract_schemas_distinguishes_stored_and_scaled_units(self, empty_log_data: LogData) -> None: # pylint: disable=redefined-outer-name + """FMTU multipliers convert stored values to the unprefixed UNIT unit.""" + mock_fmt = SimpleNamespace( + name="BAT", + type=88, + len=10, + format="f", + columns=["CurrTot"], + units=["mAh"], + msg_mults=[None], + ) + mock_mlog = SimpleNamespace( + formats={"BAT": mock_fmt}, + mult_lookup={"C": 0.001}, + unit_lookup={"a": "Ah"}, + ) + + extract_schemas( + mock_mlog, + empty_log_data, + {88: _FMTUDefinition(unit_ids="a", mult_ids="C")}, + ) + + schema = empty_log_data.schemas["BAT"] + assert schema.stored_units == ["Ah"] + assert schema.scaled_units == ["Ah"] + assert schema.multipliers == [0.001] + assert schema.multipliers_applied_at_ingest == [True] + + def test_fixture_preserves_fixed_and_dynamic_scaling(self) -> None: + """A real log retains fixed-point precision and scales FMTU values once.""" + fixture = Path(__file__).parent / "fixtures" / "backend_log_80k.bin" + + log_data = backend_log_extraction.extract_log(str(fixture)) + + battery_schema = log_data.schemas["BAT"] + current_total_index = battery_schema.fields.index("CurrTot") + current_total = log_data.get_field("BAT", "CurrTot", scaled=False)[0] + scaled_current_total = log_data.get_field("BAT", "CurrTot")[0] + temperature = log_data.get_field("BARO", "Temp", scaled=False)[0] + + assert battery_schema.stored_units[current_total_index] == "Ah" + assert battery_schema.scaled_units[current_total_index] == "Ah" + assert battery_schema.multipliers[current_total_index] == 0.001 + assert battery_schema.multipliers_applied_at_ingest[current_total_index] is True + assert scaled_current_total == pytest.approx(current_total) + assert current_total == pytest.approx(0.0017867862) + assert temperature == 3607 + assert log_data.get_field("BARO", "Temp")[0] == pytest.approx(36.07)