diff --git a/ardupilot_methodic_configurator/data_model_parameter_editor.py b/ardupilot_methodic_configurator/data_model_parameter_editor.py index 70d5e9907..4d963c529 100644 --- a/ardupilot_methodic_configurator/data_model_parameter_editor.py +++ b/ardupilot_methodic_configurator/data_model_parameter_editor.py @@ -187,13 +187,21 @@ def __init__( # Track parameters added by user (not in original file) or renamed by the system in the current configuration step self._added_parameters: set[str] = set() + # Rows created by conditional derivation in this editing session, including after export. + self._inline_derived_parameters: set[str] = set() + # Saved automatic rows retired since the last export, rather than explicitly deleted. + self._retired_inline_derived_parameters: set[str] = set() # Track parameters deleted by user (were in original file) or renamed by the system in the current configuration step self._deleted_parameters: set[str] = set() + # Keep explicit derived-row deletions suppressed after export until the step is reloaded. + self._suppressed_derived_parameters: set[str] = set() # Track connection renames applied for the current step: maps original_file_param_name → current renamed name. # Used by refresh_current_step_connection_renames() to undo/redo renames without a full rebuild. self._connection_renames: dict[str, str] = {} + # Saved sources whose row (possibly already renamed) was explicitly deleted during this edit session. + self._suppressed_connection_renames: set[str] = set() self._at_least_one_changed = False @@ -2078,7 +2086,7 @@ def _filter_derived_parameters_for_manual_overrides(self, derived_params: ParDic } ) - def _repopulate_configuration_step_parameters( # pylint: disable=too-many-locals, too-many-branches + def _repopulate_configuration_step_parameters( # noqa: PLR0915 # pylint: disable=too-many-locals, too-many-branches, too-many-statements self, ) -> tuple[list[tuple[str, str]], list[tuple[str, str]]]: """ @@ -2090,8 +2098,12 @@ def _repopulate_configuration_step_parameters( # pylint: disable=too-many-local """ # Reset tracking sets when navigating to new file self._added_parameters.clear() + self._inline_derived_parameters.clear() + self._retired_inline_derived_parameters.clear() self._deleted_parameters.clear() + self._suppressed_derived_parameters.clear() self._connection_renames.clear() + self._suppressed_connection_renames.clear() # Process configuration step and get operations to apply ( @@ -2127,6 +2139,7 @@ def _repopulate_configuration_step_parameters( # pylint: disable=too-many-local param_name, derived_par, self.current_file, self.fc_parameters ) self._added_parameters.add(param_name) + self._inline_derived_parameters.add(param_name) # Apply rename operations to domain model using add/delete tracking for old_name in duplicates_to_remove: @@ -2144,6 +2157,9 @@ def _repopulate_configuration_step_parameters( # pylint: disable=too-many-local # Mark old parameter as deleted self._deleted_parameters.add(old_name) + # The saved source, not the temporary derived destination, owns this row. + self._inline_derived_parameters.discard(new_name) + # Create new parameter with renamed name old_par = original_params[old_name] self.current_step_parameters[new_name] = self._config_step_processor.create_ardupilot_parameter( @@ -2321,12 +2337,23 @@ def delete_parameter_from_current_file(self, param_name: str) -> None: param_name: The name of the parameter to delete """ - # If parameter was in original file, mark as deleted + self._retired_inline_derived_parameters.discard(param_name) + self._suppressed_connection_renames.update( + source for source, target in self._connection_renames.items() if target == param_name + ) if param_name in self._local_filesystem.file_parameters.get(self.current_file, ParDict()): + self._suppressed_connection_renames.add(param_name) + is_derived = param_name in self._inline_derived_parameters or ( + param_name in self.current_step_parameters and self.current_step_parameters[param_name].is_derived + ) + if is_derived: + self._suppressed_derived_parameters.add(param_name) + if param_name in self._local_filesystem.file_parameters.get(self.current_file, ParDict()) or is_derived: self._deleted_parameters.add(param_name) # If it was previously added in this session, remove from added set self._added_parameters.discard(param_name) + self._inline_derived_parameters.discard(param_name) # Remove from runtime state if param_name in self.current_step_parameters: @@ -2348,6 +2375,15 @@ def get_possible_add_param_names(self) -> list[str]: possible_add_param_names.sort() return possible_add_param_names + def _unsuppress_readded_connection_renames(self, param_name: str) -> None: + """Restore tracking for a saved source when its renamed target is re-added.""" + if param_name in self._suppressed_connection_renames: + self._connection_renames.pop(param_name, None) + self._suppressed_connection_renames.discard(param_name) + self._suppressed_connection_renames.difference_update( + source for source, target in self._connection_renames.items() if target == param_name + ) + def add_parameter_to_current_file(self, param_name: str) -> bool: """ Add a parameter to the current file. @@ -2382,6 +2418,9 @@ def add_parameter_to_current_file(self, param_name: str) -> bool: self._added_parameters.add(param_name) # If was previously deleted, remove from deleted set self._deleted_parameters.discard(param_name) + self._retired_inline_derived_parameters.discard(param_name) + self._suppressed_derived_parameters.discard(param_name) + self._unsuppress_readded_connection_renames(param_name) return True @@ -2398,6 +2437,9 @@ def add_parameter_to_current_file(self, param_name: str) -> bool: self._added_parameters.add(param_name) # If was previously deleted, remove from deleted set self._deleted_parameters.discard(param_name) + self._retired_inline_derived_parameters.discard(param_name) + self._suppressed_derived_parameters.discard(param_name) + self._unsuppress_readded_connection_renames(param_name) return True if fc_parameters: @@ -2615,6 +2657,10 @@ def _export_current_file(self, annotate_doc: bool) -> None: self._added_parameters.clear() self._deleted_parameters.clear() + self._retired_inline_derived_parameters.clear() + # The exported names are now the saved-file sources for the next port change. + self._connection_renames.clear() + self._suppressed_connection_renames.clear() # copy parameters new values to their _values_on_file for param in self.current_step_parameters.values(): param.copy_new_value_to_file() @@ -2753,19 +2799,8 @@ def _has_component_unsaved_changes(self) -> bool: """Return True if the in-memory component data differs from what was last saved to disk.""" return self._local_filesystem.vehicle_components_fs.has_unsaved_changes("Components") - def refresh_current_step_computed_parameters(self) -> None: - """ - Recompute derived parameters and connection renames for the current step after component data changed. - - Unlike _repopulate_configuration_step_parameters(), this performs a lightweight - in-place update: only the derived parameter *values* are recalculated and patched - into the existing current_step_parameters dict, and any rename_connection renames - are re-evaluated and applied. The full parameter set is NOT rebuilt from - file_parameters and the delete_parameters logic is NOT re-run, so: - - parameters that were already removed by the delete logic stay removed - - user edits to non-derived parameters are preserved - - the 'show only differences' filter does not cause derived params to disappear - """ + def refresh_current_step_computed_parameters(self) -> None: # noqa: PLR0915 # pylint: disable=too-many-branches,too-many-statements + """Reconcile conditional derived rows and connection renames after a component edit.""" if not self.current_file or self.current_file not in self._local_filesystem.configuration_steps: return step_info = self._local_filesystem.configuration_steps[self.current_file] @@ -2781,16 +2816,58 @@ def refresh_current_step_computed_parameters(self) -> None: # Recompute derived params with the updated vehicle_components dict self._local_filesystem.compute_parameters(self.current_file, step_info, "derived", variables) - # Patch only the derived values in the existing domain model - no rebuild, no deletions - for param_name, derived_par in self._local_filesystem.derived_parameters.get(self.current_file, ParDict()).items(): + # A newly computed row must not occupy a saved connection's rename destination. + if "rename_connection" in step_info: + self._refresh_current_step_connection_renames(step_info, variables) + + if "derived_parameters" not in step_info: + return + + derived_params = self._local_filesystem.derived_parameters.get(self.current_file, ParDict()) + available = set(self.fc_parameters) if self.fc_parameters else None + deleted_by_step = self._local_filesystem.compute_deletions(self.current_file, step_info, variables) + self._restore_applicable_retired_parameters(set(derived_params), deleted_by_step, available) + applicable = { + name + for name in derived_params + if (available is None or name in available) + and name not in deleted_by_step + and name not in self._deleted_parameters + and name not in self._suppressed_derived_parameters + } + + for param_name in self._inline_derived_parameters - applicable: + existing = self.current_step_parameters.get(param_name) + if existing is not None and existing.is_manual_override: + self._inline_derived_parameters.discard(param_name) + continue + if existing is not None and self._is_selected_gnss_transport(param_name, step_info, variables): + continue + self.current_step_parameters.pop(param_name, None) + self._added_parameters.discard(param_name) + if param_name in self._local_filesystem.file_parameters.get(self.current_file, ParDict()): + self._deleted_parameters.add(param_name) + self._retired_inline_derived_parameters.add(param_name) + self._inline_derived_parameters.intersection_update(applicable | set(self.current_step_parameters)) + + for param_name in applicable: + derived_par = derived_params[param_name] if param_name in self.current_step_parameters: existing = self.current_step_parameters[param_name] + if existing.is_manual_override: + continue + if ( + "gnss" in self.current_file.lower() + and step_info.get("rename_connection") == _GNSS_FC_CONNECTION_RENAME_EXPR + and param_name in self._connection_renames.values() + ): + continue if not (existing.is_forced or existing.is_derived): # The parameter was created before its derived value could be evaluated (e.g. the # vehicle_components dict was incomplete at step-load time). Now that evaluation # succeeded, re-create it so it carries the derived flag and the correct value. original_par = self._local_filesystem.file_parameters.get(self.current_file, ParDict()).get(param_name) - if original_par is not None: + if original_par is not None and not existing.is_dirty: self.current_step_parameters[param_name] = self._config_step_processor.create_ardupilot_parameter( param_name, original_par, self.current_file, self.fc_parameters or {} ) @@ -2801,13 +2878,55 @@ def refresh_current_step_computed_parameters(self) -> None: self.current_step_parameters[param_name].set_forced_or_derived_change_reason(derived_par.comment) except (ValueError, TypeError) as e: logging_error(_("Failed to patch derived parameter %s: %s"), param_name, str(e)) + else: + saved_par = self._local_filesystem.file_parameters.get(self.current_file, ParDict()).get(param_name) + self.current_step_parameters[param_name] = self._config_step_processor.create_ardupilot_parameter( + param_name, + saved_par if saved_par is not None else derived_par, + self.current_file, + self.fc_parameters or {}, + ) + if saved_par is None: + self._added_parameters.add(param_name) + self._inline_derived_parameters.add(param_name) - if "rename_connection" in step_info: - self._refresh_current_step_connection_renames(step_info, variables) + def _restore_applicable_retired_parameters( + self, derived_names: set[str], deleted_by_step: set[str], available: set[str] | None + ) -> None: + """Cancel automatic retirement when a saved derived row becomes applicable again.""" + restored = self._retired_inline_derived_parameters & derived_names - deleted_by_step + if available is not None: + restored &= available + restored -= self._suppressed_derived_parameters + self._deleted_parameters.difference_update(restored) + self._retired_inline_derived_parameters.difference_update(restored) + + @staticmethod + def _is_selected_gnss_transport(param_name: str, step_info: dict, variables: dict) -> bool: + """Keep a same-UART GNSS transport when its driver no longer derives it.""" + if step_info.get("rename_connection") != _GNSS_FC_CONNECTION_RENAME_EXPR: + return False + selected = variables.get("vehicle_components", {}).get("GNSS Receiver", {}).get("FC Connection", {}).get("Type") + return selected is not None and param_name == f"{selected}_PROTOCOL" + + def _transfer_inline_derived_ownership(self, old_name: str, new_name: str) -> None: + """Keep same-session derivation ownership when a saved source is renamed.""" + if old_name in self._inline_derived_parameters or old_name in self._retired_inline_derived_parameters: + self._inline_derived_parameters.discard(old_name) + self._retired_inline_derived_parameters.discard(old_name) + self._inline_derived_parameters.add(new_name) + + def _take_current_rename_par(self, current_target: str, original_par: Par | None) -> Par | None: + """Remove a rename target and retain its current value for the next name.""" + current = self.current_step_parameters.pop(current_target, None) + if current is None: + return original_par + self._added_parameters.discard(current_target) + return Par(current.get_new_value(), current.change_reason_for_file) if original_par is not None else None def _apply_gnss_protocol_value(self, target_name: str) -> None: """Set the GNSS MAVLink protocol value on a newly renamed _PROTOCOL parameter, ignoring no-op updates.""" - with contextlib.suppress(ParameterUnchangedError): + with contextlib.suppress(ParameterUnchangedError, ParameterForcedOrDerivedError): self.current_step_parameters[target_name].set_new_value(_GNSS_MAVLINK_PROTOCOL_VALUE, ignore_out_of_range=True) def _refresh_current_step_connection_renames(self, step_info: dict, variables: dict) -> None: # pylint: disable=too-many-branches @@ -2841,9 +2960,18 @@ def _refresh_current_step_connection_renames(self, step_info: dict, variables: d # Process renames that already exist: update if the target changed for orig_name, current_target in list(self._connection_renames.items()): + if orig_name in self._suppressed_connection_renames: + self._connection_renames[orig_name] = new_rename_map.get(orig_name, orig_name) + continue + if ( + current_target in self.current_step_parameters + and self.current_step_parameters[current_target].is_manual_override is True + ): + continue new_target = new_rename_map.get(orig_name) if new_target is None: # Rename no longer applicable — restore original name + self._transfer_inline_derived_ownership(current_target, orig_name) if current_target in self.current_step_parameters: self._added_parameters.discard(current_target) del self.current_step_parameters[current_target] @@ -2858,12 +2986,11 @@ def _refresh_current_step_connection_renames(self, step_info: dict, variables: d if new_target in self.current_step_parameters and new_target not in self._connection_renames.values(): logging_warning(_("Skipping connection rename %s → %s: target already exists"), current_target, new_target) else: - if current_target in self.current_step_parameters: - self._added_parameters.discard(current_target) - del self.current_step_parameters[current_target] - if orig_name in original_params: + self._transfer_inline_derived_ownership(current_target, new_target) + current_par = self._take_current_rename_par(current_target, original_params.get(orig_name)) + if current_par is not None: self.current_step_parameters[new_target] = self._config_step_processor.create_ardupilot_parameter( - new_target, original_params[orig_name], self.current_file, self.fc_parameters + new_target, current_par, self.current_file, self.fc_parameters ) if is_gnss and new_target.endswith("_PROTOCOL"): self._apply_gnss_protocol_value(new_target) @@ -2873,6 +3000,8 @@ def _refresh_current_step_connection_renames(self, step_info: dict, variables: d # Apply renames that are new (not previously tracked) for orig_name, new_target in new_rename_map.items(): + if orig_name in self._suppressed_connection_renames: + continue if orig_name in self._connection_renames: continue # already handled above if orig_name in original_params: @@ -2886,9 +3015,12 @@ def _apply_new_connection_rename( is_gnss: bool, ) -> None: """Apply a single connection rename that was not previously tracked.""" + if orig_name in self.current_step_parameters and self.current_step_parameters[orig_name].is_manual_override is True: + return if new_target in self.current_step_parameters: logging_warning(_("Skipping connection rename %s → %s: target already exists"), orig_name, new_target) return + self._transfer_inline_derived_ownership(orig_name, new_target) # Remove the original (un-renamed) parameter if it is still present if orig_name in self.current_step_parameters: del self.current_step_parameters[orig_name] diff --git a/tests/test_data_model_parameter_editor.py b/tests/test_data_model_parameter_editor.py index 726d1963c..bf4b73fb9 100755 --- a/tests/test_data_model_parameter_editor.py +++ b/tests/test_data_model_parameter_editor.py @@ -3831,6 +3831,7 @@ def test_user_not_prompted_when_adding_then_deleting_same_parameter(self, parame # Assert: No net change, so no unsaved changes assert not parameter_editor._has_unsaved_changes() + assert "PARAM2" not in parameter_editor._suppressed_derived_parameters def test_user_not_prompted_when_deleting_then_adding_back_same_parameter(self, parameter_editor) -> None: """ @@ -5410,6 +5411,7 @@ def test_system_patches_existing_derived_parameters_in_place(self, parameter_edi """ step_file = "11_battery.param" parameter_editor.current_file = step_file + parameter_editor._flight_controller.fc_parameters = {} parameter_editor._local_filesystem.configuration_steps = { step_file: {"derived_parameters": {"MOT_BAT_VOLT_MAX": {"expr": "cells * 4.2", "comment": "Max volts"}}} } @@ -5422,6 +5424,7 @@ def test_system_patches_existing_derived_parameters_in_place(self, parameter_edi # Existing parameter objects in the domain model existing_derived = MagicMock() + existing_derived.is_manual_override = False untouched_param = MagicMock() parameter_editor.current_step_parameters = { "MOT_BAT_VOLT_MAX": existing_derived, @@ -5760,6 +5763,654 @@ def test_system_skips_retarget_when_new_target_already_exists(self, parameter_ed assert parameter_editor._connection_renames.get("ORIG_PARAM") == "OLD_TARGET" +class TestConnectionRenameSaveLifecycle: + """Exercise port changes, saves, and deletion through the editor's public operations.""" + + @staticmethod + def prepare(parameter_editor: ParameterEditor) -> tuple[str, dict[str, str]]: + """Load a serial step with a live component expression and saved values.""" + step_file = "10_telemetry.param" + components = {"port": "SERIAL1", "other": "initial"} + filesystem = parameter_editor._local_filesystem + filesystem.vehicle_components_fs.data = {"Components": components} + filesystem.configuration_steps = {step_file: {"rename_connection": "vehicle_components['port']"}} + filesystem.file_parameters = { + step_file: ParDict( + { + "SERIAL1_PROTOCOL": Par(2.0, "saved protocol"), + "SERIAL1_BAUD": Par(57.0, "saved baud"), + "BRD_SER1_RTSCTS": Par(1.0, "saved flow control"), + "OTHER": Par(4.0, "unrelated"), + } + ) + } + parameter_editor.current_file = step_file + parameter_editor._config_step_processor.variables = {"vehicle_components": components} + parameter_editor.current_step_parameters = { + name: parameter_editor._config_step_processor.create_ardupilot_parameter( + name, par, step_file, parameter_editor.fc_parameters + ) + for name, par in filesystem.file_parameters[step_file].items() + } + return step_file, components + + @staticmethod + def change_port(parameter_editor: ParameterEditor, components: dict[str, str], port: str) -> None: + """Apply a component edit and refresh the active step.""" + parameter_editor.update_vehicle_components({**components, "port": port}) + parameter_editor.refresh_current_step_computed_parameters() + + def test_export_then_unrelated_edit_preserves_saved_serial_values(self, parameter_editor: ParameterEditor) -> None: + """GIVEN a saved rename, WHEN another component field changes, THEN both exports retain serial settings.""" + step_file, components = self.prepare(parameter_editor) + self.change_port(parameter_editor, components, "SERIAL2") + assert parameter_editor.current_step_parameters["SERIAL2_BAUD"].get_new_value() == 57.0 + assert parameter_editor.current_step_parameters["BRD_SER2_RTSCTS"].get_new_value() == 1.0 + + parameter_editor._export_current_file(annotate_doc=False) + parameter_editor.update_vehicle_components({**components, "other": "edited"}) + parameter_editor.refresh_current_step_computed_parameters() + parameter_editor._export_current_file(annotate_doc=False) + + saved = parameter_editor._local_filesystem.file_parameters[step_file] + assert saved["SERIAL2_BAUD"].value == 57.0 + assert saved["BRD_SER2_RTSCTS"].value == 1.0 + assert "SERIAL1_BAUD" not in saved + assert "BRD_SER1_RTSCTS" not in saved + + def test_export_then_another_uart_change_uses_new_saved_baseline(self, parameter_editor: ParameterEditor) -> None: + """GIVEN a saved SERIAL2 step, WHEN changed to SERIAL3, THEN the saved values move with it.""" + step_file, components = self.prepare(parameter_editor) + self.change_port(parameter_editor, components, "SERIAL2") + parameter_editor._export_current_file(annotate_doc=False) + + self.change_port(parameter_editor, components, "SERIAL3") + assert parameter_editor.current_step_parameters["SERIAL3_BAUD"].get_new_value() == 57.0 + assert parameter_editor.current_step_parameters["BRD_SER3_RTSCTS"].get_new_value() == 1.0 + parameter_editor._export_current_file(annotate_doc=False) + assert set(parameter_editor._local_filesystem.file_parameters[step_file]) == { + "SERIAL3_PROTOCOL", + "SERIAL3_BAUD", + "BRD_SER3_RTSCTS", + "OTHER", + } + + def test_deleted_renamed_protocol_survives_port_round_trip(self, parameter_editor: ParameterEditor) -> None: + """GIVEN a deleted renamed row, WHEN the port changes twice, THEN it stays out of the export.""" + step_file, components = self.prepare(parameter_editor) + self.change_port(parameter_editor, components, "SERIAL2") + parameter_editor.delete_parameter_from_current_file("SERIAL2_PROTOCOL") + + self.change_port(parameter_editor, components, "SERIAL3") + assert "SERIAL3_PROTOCOL" not in parameter_editor.current_step_parameters + self.change_port(parameter_editor, components, "SERIAL2") + assert "SERIAL2_PROTOCOL" not in parameter_editor.current_step_parameters + parameter_editor._export_current_file(annotate_doc=False) + assert "SERIAL2_PROTOCOL" not in parameter_editor._local_filesystem.file_parameters[step_file] + assert parameter_editor._local_filesystem.file_parameters[step_file]["SERIAL2_BAUD"].value == 57.0 + + def test_deleting_saved_rename_target_survives_another_port_change(self, parameter_editor: ParameterEditor) -> None: + """GIVEN a saved SERIAL2 protocol, WHEN deleted then changed to SERIAL3, THEN it stays deleted.""" + step_file, components = self.prepare(parameter_editor) + self.change_port(parameter_editor, components, "SERIAL2") + parameter_editor._export_current_file(annotate_doc=False) + parameter_editor.delete_parameter_from_current_file("SERIAL2_PROTOCOL") + + self.change_port(parameter_editor, components, "SERIAL3") + assert "SERIAL3_PROTOCOL" not in parameter_editor.current_step_parameters + parameter_editor._export_current_file(annotate_doc=False) + assert "SERIAL3_PROTOCOL" not in parameter_editor._local_filesystem.file_parameters[step_file] + assert parameter_editor._local_filesystem.file_parameters[step_file]["SERIAL3_BAUD"].value == 57.0 + + def test_reload_clears_suppression_and_refresh_uses_saved_file(self, parameter_editor: ParameterEditor) -> None: + """GIVEN an unsaved deletion, WHEN the step reloads, THEN its saved protocol can rename again.""" + step_file, components = self.prepare(parameter_editor) + self.change_port(parameter_editor, components, "SERIAL2") + parameter_editor.delete_parameter_from_current_file("SERIAL2_PROTOCOL") + self.change_port(parameter_editor, components, "SERIAL3") + + parameter_editor.update_vehicle_components({**components, "port": "SERIAL1"}) + parameter_editor._config_step_processor.process_configuration_step = MagicMock( + return_value=( + { + name: parameter_editor._config_step_processor.create_ardupilot_parameter( + name, par, step_file, parameter_editor.fc_parameters + ) + for name, par in parameter_editor._local_filesystem.file_parameters[step_file].items() + }, + [], + [], + set(), + [], + ParDict(), + set(), + ) + ) + parameter_editor._repopulate_configuration_step_parameters() + assert not parameter_editor._suppressed_connection_renames + self.change_port(parameter_editor, components, "SERIAL2") + assert parameter_editor.current_step_parameters["SERIAL2_PROTOCOL"].get_new_value() == 2.0 + + def test_deleted_protocol_can_be_readded_from_docs_without_fc_value(self, parameter_editor: ParameterEditor) -> None: + """GIVEN a deletion and docs only, WHEN re-added, THEN the protocol stays in the export.""" + step_file, components = self.prepare(parameter_editor) + parameter_editor._flight_controller.fc_parameters = {} + parameter_editor._local_filesystem.doc_dict = {"SERIAL2_PROTOCOL": {}} + parameter_editor._local_filesystem.param_default_dict = ParDict({"SERIAL2_PROTOCOL": Par(2.0)}) + self.change_port(parameter_editor, components, "SERIAL2") + parameter_editor.delete_parameter_from_current_file("SERIAL2_PROTOCOL") + + assert parameter_editor.add_parameter_to_current_file("SERIAL2_PROTOCOL") + self.change_port(parameter_editor, components, "SERIAL3") + self.change_port(parameter_editor, components, "SERIAL2") + parameter_editor._export_current_file(annotate_doc=False) + assert parameter_editor._local_filesystem.file_parameters[step_file]["SERIAL2_PROTOCOL"].value == 2.0 + + def test_dirty_value_survives_when_a_plain_row_becomes_derived(self, parameter_editor: ParameterEditor) -> None: + """GIVEN an unsaved edit, WHEN derivation becomes available, THEN it keeps the user's value.""" + step_file, _components = self.prepare(parameter_editor) + parameter_editor._local_filesystem.configuration_steps[step_file]["derived_parameters"] = {} + parameter_editor._local_filesystem.derived_parameters = {step_file: ParDict({"OTHER": Par(8.0)})} + parameter_editor.current_step_parameters["OTHER"].set_new_value("9") + + parameter_editor.refresh_current_step_computed_parameters() + + assert parameter_editor.current_step_parameters["OTHER"].get_new_value() == 9.0 + + def test_existing_rename_target_keeps_its_user_value(self, parameter_editor: ParameterEditor) -> None: + """GIVEN a user-added target, WHEN the port changes, THEN the rename cannot overwrite it.""" + _step_file, components = self.prepare(parameter_editor) + parameter_editor._flight_controller.fc_parameters = {"SERIAL2_BAUD": 115.0} + assert parameter_editor.add_parameter_to_current_file("SERIAL2_BAUD") + + self.change_port(parameter_editor, components, "SERIAL2") + + assert parameter_editor.current_step_parameters["SERIAL2_BAUD"].get_new_value() == 115.0 + assert parameter_editor.current_step_parameters["SERIAL1_BAUD"].get_new_value() == 57.0 + + self.change_port(parameter_editor, components, "SERIAL3") + assert parameter_editor.current_step_parameters["SERIAL2_BAUD"].get_new_value() == 115.0 + assert parameter_editor.current_step_parameters["SERIAL3_BAUD"].get_new_value() == 57.0 + + @pytest.mark.parametrize("add_source", ["fc", "docs"]) + @pytest.mark.parametrize( + ("suffix", "initial_value", "edited_value"), + [("BAUD", 115.0, 230.0), ("PROTOCOL", 3.0, 4.0)], + ) + def test_readded_connection_parameter_moves_with_edited_value( + self, + parameter_editor: ParameterEditor, + add_source: str, + suffix: str, + initial_value: float, + edited_value: float, + ) -> None: + """GIVEN a deleted rename target, WHEN re-added and moved, THEN its new value follows the port.""" + step_file, components = self.prepare(parameter_editor) + source_name = f"SERIAL2_{suffix}" + target_name = f"SERIAL3_{suffix}" + self.change_port(parameter_editor, components, "SERIAL2") + parameter_editor.delete_parameter_from_current_file(source_name) + parameter_editor.update_vehicle_components({**components, "other": "intervening edit"}) + parameter_editor.refresh_current_step_computed_parameters() + if add_source == "fc": + parameter_editor._flight_controller.fc_parameters = {source_name: initial_value} + else: + parameter_editor._flight_controller.fc_parameters = {} + parameter_editor._local_filesystem.doc_dict = {source_name: {}} + parameter_editor._local_filesystem.param_default_dict = ParDict({source_name: Par(initial_value)}) + + assert parameter_editor.add_parameter_to_current_file(source_name) + result = parameter_editor.update_parameter_value(source_name, str(edited_value)) + assert result.status == ParameterValueUpdateStatus.UPDATED + self.change_port(parameter_editor, components, "SERIAL3") + + assert source_name not in parameter_editor.current_step_parameters + assert parameter_editor.current_step_parameters[target_name].get_new_value() == edited_value + parameter_editor._export_current_file(annotate_doc=False) + saved = parameter_editor._local_filesystem.file_parameters[step_file] + assert source_name not in saved + assert saved[target_name].value == edited_value + assert saved["BRD_SER3_RTSCTS"].value == 1.0 + + @pytest.mark.parametrize("add_source", ["fc", "docs"]) + @pytest.mark.parametrize("suffix", ["BAUD", "PROTOCOL"]) + def test_readded_saved_source_keeps_edit_after_refresh( + self, parameter_editor: ParameterEditor, add_source: str, suffix: str + ) -> None: + """Re-adding the saved source after a deletion must not revive an obsolete rename.""" + step_file, components = self.prepare(parameter_editor) + self.change_port(parameter_editor, components, "SERIAL2") + parameter_editor.delete_parameter_from_current_file(f"SERIAL2_{suffix}") + self.change_port(parameter_editor, components, "SERIAL1") + name = f"SERIAL1_{suffix}" + if add_source == "fc": + parameter_editor._flight_controller.fc_parameters = {name: 3.0} + else: + parameter_editor._flight_controller.fc_parameters = {} + parameter_editor._local_filesystem.doc_dict = {name: {}} + parameter_editor._local_filesystem.param_default_dict = ParDict({name: Par(3.0)}) + assert parameter_editor.add_parameter_to_current_file(name) + assert parameter_editor.update_parameter_value(name, "230").status == ParameterValueUpdateStatus.UPDATED + parameter_editor.update_vehicle_components({**components, "other": "unrelated"}) + parameter_editor.refresh_current_step_computed_parameters() + assert parameter_editor.current_step_parameters[name].get_new_value() == 230.0 + parameter_editor._export_current_file(annotate_doc=False) + assert parameter_editor._local_filesystem.file_parameters[step_file][name].value == 230.0 + + @pytest.mark.parametrize("add_source", ["fc", "docs"]) + @pytest.mark.parametrize(("suffix", "saved_value"), [("BAUD", 57.0), ("PROTOCOL", 2.0)]) + def test_independently_readded_live_rename_source_remains_separate( + self, parameter_editor: ParameterEditor, add_source: str, suffix: str, saved_value: float + ) -> None: + """An independent source row and its still-live renamed row both survive a port move.""" + step_file, components = self.prepare(parameter_editor) + self.change_port(parameter_editor, components, "SERIAL2") + name = f"SERIAL1_{suffix}" + if add_source == "fc": + parameter_editor._flight_controller.fc_parameters = {name: 3.0} + else: + parameter_editor._flight_controller.fc_parameters = {} + parameter_editor._local_filesystem.doc_dict = {name: {}} + parameter_editor._local_filesystem.param_default_dict = ParDict({name: Par(3.0)}) + assert parameter_editor.add_parameter_to_current_file(name) + assert parameter_editor.update_parameter_value(name, "230").status == ParameterValueUpdateStatus.UPDATED + self.change_port(parameter_editor, components, "SERIAL3") + parameter_editor._export_current_file(annotate_doc=False) + saved = parameter_editor._local_filesystem.file_parameters[step_file] + assert saved[name].value == 230.0 + assert saved[f"SERIAL3_{suffix}"].value == saved_value + assert f"SERIAL2_{suffix}" not in saved + + @pytest.mark.parametrize("gnss", [False, True]) + @pytest.mark.parametrize("add_source", ["fc", "docs"]) + def test_old_target_readded_after_return_to_source_stays_independent( + self, parameter_editor: ParameterEditor, gnss: bool, add_source: str + ) -> None: + """A suppressed destination re-added after returning to its source is independent.""" + step_file, components = self.prepare(parameter_editor) + if gnss: + step_file = "10_gnss.param" + filesystem = parameter_editor._local_filesystem + filesystem.file_parameters[step_file] = filesystem.file_parameters.pop("10_telemetry.param") + filesystem.configuration_steps = { + step_file: {"rename_connection": "vehicle_components['GNSS Receiver']['FC Connection']['Type']"} + } + connection = {"Type": "SERIAL1"} + parameter_editor._config_step_processor.variables = { + "vehicle_components": {"GNSS Receiver": {"FC Connection": connection}} + } + parameter_editor.current_file = step_file + connection["Type"] = "SERIAL2" + parameter_editor.refresh_current_step_computed_parameters() + else: + self.change_port(parameter_editor, components, "SERIAL2") + parameter_editor.delete_parameter_from_current_file("SERIAL2_BAUD") + if gnss: + connection["Type"] = "SERIAL1" + parameter_editor.refresh_current_step_computed_parameters() + else: + self.change_port(parameter_editor, components, "SERIAL1") + name = "SERIAL2_BAUD" + if add_source == "fc": + parameter_editor._flight_controller.fc_parameters = {name: 3.0} + else: + parameter_editor._flight_controller.fc_parameters = {} + parameter_editor._local_filesystem.doc_dict = {name: {}} + parameter_editor._local_filesystem.param_default_dict = ParDict({name: Par(3.0)}) + assert parameter_editor.add_parameter_to_current_file(name) + assert parameter_editor.update_parameter_value(name, "230").status == ParameterValueUpdateStatus.UPDATED + if gnss: + parameter_editor._config_step_processor.variables["vehicle_components"]["other"] = "edited" + else: + parameter_editor.update_vehicle_components({**components, "other": "edited"}) + parameter_editor.refresh_current_step_computed_parameters() + parameter_editor._export_current_file(annotate_doc=False) + saved = parameter_editor._local_filesystem.file_parameters[step_file] + assert saved[name].value == 230.0 + assert "SERIAL1_BAUD" not in saved + + @pytest.mark.parametrize("add_source", ["fc", "docs"]) + def test_readded_old_target_after_port_move_remains_independent( + self, parameter_editor: ParameterEditor, add_source: str + ) -> None: + """An old target re-added after the connection moved is an independent row.""" + step_file, components = self.prepare(parameter_editor) + self.change_port(parameter_editor, components, "SERIAL2") + parameter_editor.delete_parameter_from_current_file("SERIAL2_BAUD") + self.change_port(parameter_editor, components, "SERIAL3") + name = "SERIAL2_BAUD" + if add_source == "fc": + parameter_editor._flight_controller.fc_parameters = {name: 115.0} + else: + parameter_editor._flight_controller.fc_parameters = {} + parameter_editor._local_filesystem.doc_dict = {name: {}} + parameter_editor._local_filesystem.param_default_dict = ParDict({name: Par(115.0)}) + assert parameter_editor.add_parameter_to_current_file(name) + assert parameter_editor.update_parameter_value(name, "230").status == ParameterValueUpdateStatus.UPDATED + parameter_editor.update_vehicle_components({**components, "other": "unrelated"}) + parameter_editor.refresh_current_step_computed_parameters() + assert parameter_editor.current_step_parameters[name].get_new_value() == 230.0 + assert "SERIAL3_BAUD" not in parameter_editor.current_step_parameters + parameter_editor._export_current_file(annotate_doc=False) + assert parameter_editor._local_filesystem.file_parameters[step_file][name].value == 230.0 + + @pytest.mark.parametrize(("suffix", "saved_value"), [("BAUD", 57.0), ("PROTOCOL", 2.0)]) + def test_readded_value_round_trip_to_original_port( + self, parameter_editor: ParameterEditor, suffix: str, saved_value: float + ) -> None: + """Document the existing saved-source behavior pending a decision on round-trip semantics.""" + step_file, components = self.prepare(parameter_editor) + self.change_port(parameter_editor, components, "SERIAL2") + name = f"SERIAL2_{suffix}" + parameter_editor.delete_parameter_from_current_file(name) + parameter_editor.refresh_current_step_computed_parameters() + parameter_editor._flight_controller.fc_parameters = {name: 3.0} + assert parameter_editor.add_parameter_to_current_file(name) + parameter_editor.update_parameter_value(name, "4") + self.change_port(parameter_editor, components, "SERIAL3") + assert parameter_editor.current_step_parameters[f"SERIAL3_{suffix}"].get_new_value() == 4.0 + self.change_port(parameter_editor, components, "SERIAL1") + assert parameter_editor.current_step_parameters[f"SERIAL1_{suffix}"].get_new_value() == saved_value + parameter_editor._export_current_file(annotate_doc=False) + saved = parameter_editor._local_filesystem.file_parameters[step_file] + assert saved[f"SERIAL1_{suffix}"].value == saved_value + assert f"SERIAL2_{suffix}" not in saved + assert f"SERIAL3_{suffix}" not in saved + assert not parameter_editor._has_unsaved_changes() + + +class TestInlineDerivationOwnership: + """Keep automatically derived rows identifiable across saves within one editing session.""" + + @staticmethod + def prepare(parameter_editor: ParameterEditor, saved: ParDict | None = None) -> tuple[str, dict[str, str]]: + """Set up a generic conditional serial step with real editor operations.""" + step_file = "10_telemetry.param" + components = {"port": "None"} + filesystem = parameter_editor._local_filesystem + filesystem.vehicle_components_fs.data = {"Components": components} + filesystem.configuration_steps = { + step_file: {"rename_connection": "vehicle_components['port']", "derived_parameters": {}} + } + filesystem.file_parameters = {step_file: saved if saved is not None else ParDict()} + filesystem.derived_parameters = {step_file: ParDict()} + filesystem.compute_deletions.return_value = set() + + def compute(*_args: object) -> None: + port = components["port"] + filesystem.derived_parameters[step_file] = ( + ParDict({f"{port}_PROTOCOL": Par(2.0, "selected protocol")}) if port.startswith("SERIAL") else ParDict() + ) + + filesystem.compute_parameters.side_effect = compute + parameter_editor.current_file = step_file + parameter_editor._flight_controller.fc_parameters = {} + parameter_editor._config_step_processor.variables = {"vehicle_components": components} + parameter_editor.current_step_parameters = { + name: parameter_editor._config_step_processor.create_ardupilot_parameter(name, par, step_file, {}) + for name, par in filesystem.file_parameters[step_file].items() + } + return step_file, components + + @staticmethod + def change_port(parameter_editor: ParameterEditor, components: dict[str, str], port: str) -> None: + """Edit the component and reconcile the current step.""" + parameter_editor.update_vehicle_components({**components, "port": port}) + parameter_editor.refresh_current_step_computed_parameters() + + def test_saved_automatic_protocol_follows_port_and_retires_without_spurious_prompt( + self, parameter_editor: ParameterEditor + ) -> None: + """GIVEN an automatic row, WHEN saved then moved and deselected, THEN only its current name is removed.""" + step_file, components = self.prepare(parameter_editor) + self.change_port(parameter_editor, components, "SERIAL2") + parameter_editor._export_current_file(annotate_doc=False) + assert not parameter_editor._has_unsaved_changes() + parameter_editor.refresh_current_step_computed_parameters() + assert not parameter_editor._has_unsaved_changes() + + self.change_port(parameter_editor, components, "SERIAL3") + assert "SERIAL2_PROTOCOL" not in parameter_editor.current_step_parameters + assert parameter_editor.current_step_parameters["SERIAL3_PROTOCOL"].get_new_value() == 2.0 + parameter_editor._export_current_file(annotate_doc=False) + assert not parameter_editor._has_unsaved_changes() + + self.change_port(parameter_editor, components, "None") + assert "SERIAL3_PROTOCOL" not in parameter_editor.current_step_parameters + assert parameter_editor._has_unsaved_changes() + parameter_editor._export_current_file(annotate_doc=False) + parameter_editor._export_current_file(annotate_doc=False) + assert not parameter_editor._local_filesystem.file_parameters[step_file] + assert not parameter_editor._has_unsaved_changes() + self.change_port(parameter_editor, components, "SERIAL2") + parameter_editor._export_current_file(annotate_doc=False) + assert set(parameter_editor._local_filesystem.file_parameters[step_file]) == {"SERIAL2_PROTOCOL"} + assert not parameter_editor._has_unsaved_changes() + + @pytest.mark.parametrize("save_retirement", [False, True]) + def test_saved_automatic_protocol_returns_after_reselection( + self, parameter_editor: ParameterEditor, save_retirement: bool + ) -> None: + """A saved automatic row returns when the same port is selected again.""" + step_file, components = self.prepare(parameter_editor) + self.change_port(parameter_editor, components, "SERIAL5") + parameter_editor._export_current_file(annotate_doc=False) + self.change_port(parameter_editor, components, "RCIN") + assert "SERIAL5_PROTOCOL" not in parameter_editor.current_step_parameters + if save_retirement: + parameter_editor._export_current_file(annotate_doc=False) + self.change_port(parameter_editor, components, "SERIAL5") + assert parameter_editor.current_step_parameters["SERIAL5_PROTOCOL"].get_new_value() == 2.0 + assert parameter_editor._has_unsaved_changes() is save_retirement + parameter_editor._export_current_file(annotate_doc=False) + assert parameter_editor._local_filesystem.file_parameters[step_file]["SERIAL5_PROTOCOL"].value == 2.0 + assert not parameter_editor._has_unsaved_changes() + + @pytest.mark.parametrize("save_first_retirement", [False, True]) + @pytest.mark.parametrize("save_intermediate_port", [False, True]) + def test_retired_protocol_does_not_return_after_another_port( + self, parameter_editor: ParameterEditor, save_first_retirement: bool, save_intermediate_port: bool + ) -> None: + """Retired automatic rows stay retired when selection passes through another port.""" + step_file, components = self.prepare(parameter_editor) + self.change_port(parameter_editor, components, "SERIAL2") + parameter_editor._export_current_file(annotate_doc=False) + self.change_port(parameter_editor, components, "None") + if save_first_retirement: + parameter_editor._export_current_file(annotate_doc=False) + self.change_port(parameter_editor, components, "SERIAL3") + assert parameter_editor.current_step_parameters["SERIAL3_PROTOCOL"].get_new_value() == 2.0 + if save_intermediate_port: + parameter_editor._export_current_file(annotate_doc=False) + self.change_port(parameter_editor, components, "None") + parameter_editor._export_current_file(annotate_doc=False) + assert not parameter_editor._local_filesystem.file_parameters[step_file] + assert not parameter_editor._has_unsaved_changes() + self.change_port(parameter_editor, components, "SERIAL2") + assert parameter_editor.current_step_parameters["SERIAL2_PROTOCOL"].get_new_value() == 2.0 + parameter_editor._export_current_file(annotate_doc=False) + assert set(parameter_editor._local_filesystem.file_parameters[step_file]) == {"SERIAL2_PROTOCOL"} + + @pytest.mark.parametrize("initial_load", [False, True]) + def test_saved_protocol_with_derived_destination_keeps_user_ownership( + self, parameter_editor: ParameterEditor, initial_load: bool + ) -> None: + """A saved source stays user-owned whether a destination is derived on load or refresh.""" + step_file, components = self.prepare(parameter_editor, ParDict({"SERIAL1_PROTOCOL": Par(7.0)})) + if initial_load: + components["port"] = "SERIAL2" + filesystem = parameter_editor._local_filesystem + filesystem.compute_parameters(step_file, filesystem.configuration_steps[step_file], "derived", {}) + parameter_editor._config_step_processor.process_configuration_step = MagicMock( + return_value=( + dict(parameter_editor.current_step_parameters), + [], + [], + set(), + [("SERIAL1_PROTOCOL", "SERIAL2_PROTOCOL")], + filesystem.derived_parameters[step_file], + set(), + ) + ) + parameter_editor._repopulate_configuration_step_parameters() + else: + self.change_port(parameter_editor, components, "SERIAL2") + self.change_port(parameter_editor, components, "None") + parameter_editor._export_current_file(annotate_doc=False) + saved = parameter_editor._local_filesystem.file_parameters[step_file] + assert set(saved) == {"SERIAL1_PROTOCOL"} + assert saved["SERIAL1_PROTOCOL"].value == 7.0 + + def test_preexisting_saved_protocol_and_serial_settings_remain_user_owned(self, parameter_editor: ParameterEditor) -> None: + """GIVEN saved user settings, WHEN no connection is selected, THEN refresh and export keep them.""" + saved = ParDict( + { + "SERIAL2_PROTOCOL": Par(7.0, "user protocol"), + "SERIAL2_BAUD": Par(57.0, "user baud"), + "BRD_SER2_RTSCTS": Par(1.0, "user flow control"), + } + ) + step_file, components = self.prepare(parameter_editor, saved) + self.change_port(parameter_editor, components, "None") + parameter_editor._export_current_file(annotate_doc=False) + + actual = parameter_editor._local_filesystem.file_parameters[step_file] + assert {name: par.value for name, par in actual.items()} == { + "SERIAL2_PROTOCOL": 7.0, + "SERIAL2_BAUD": 57.0, + "BRD_SER2_RTSCTS": 1.0, + } + + def test_saved_automatic_protocol_retires_but_renamed_baud_and_flow_control_remain( + self, parameter_editor: ParameterEditor + ) -> None: + """GIVEN saved baud and flow control, WHEN ports change then disconnect, THEN those settings remain.""" + saved = ParDict({"SERIAL1_BAUD": Par(57.0), "BRD_SER1_RTSCTS": Par(1.0)}) + step_file, components = self.prepare(parameter_editor, saved) + for port in ("SERIAL2", "SERIAL3"): + self.change_port(parameter_editor, components, port) + parameter_editor._export_current_file(annotate_doc=False) + + self.change_port(parameter_editor, components, "None") + parameter_editor._export_current_file(annotate_doc=False) + final = parameter_editor._local_filesystem.file_parameters[step_file] + assert "SERIAL3_PROTOCOL" not in final + assert final["SERIAL3_BAUD"].value == 57.0 + assert final["BRD_SER3_RTSCTS"].value == 1.0 + + def test_user_addition_and_manual_override_survive_automatic_retirement(self, parameter_editor: ParameterEditor) -> None: + """GIVEN a user addition and override, WHEN the selected port ends, THEN both remain in the export.""" + step_file, components = self.prepare(parameter_editor) + parameter_editor._local_filesystem.doc_dict = {"USER_PARAM": {}} + parameter_editor._local_filesystem.param_default_dict = ParDict({"USER_PARAM": Par(7.0)}) + assert parameter_editor.add_parameter_to_current_file("USER_PARAM") + self.change_port(parameter_editor, components, "SERIAL2") + protocol = parameter_editor.current_step_parameters["SERIAL2_PROTOCOL"] + protocol.set_manual_override(True) + protocol.set_new_value("3") + parameter_editor._export_current_file(annotate_doc=False) + + self.change_port(parameter_editor, components, "None") + parameter_editor._export_current_file(annotate_doc=False) + final = parameter_editor._local_filesystem.file_parameters[step_file] + assert final["SERIAL2_PROTOCOL"].value == 3.0 + assert final["USER_PARAM"].value == 7.0 + + def test_post_export_manual_override_is_not_renamed_away(self, parameter_editor: ParameterEditor) -> None: + """GIVEN a new override on a saved row, WHEN another UART is chosen, THEN the override stays.""" + step_file, components = self.prepare(parameter_editor) + self.change_port(parameter_editor, components, "SERIAL2") + parameter_editor._export_current_file(annotate_doc=False) + original = parameter_editor.current_step_parameters["SERIAL2_PROTOCOL"] + original.set_manual_override(True) + original.set_new_value("3") + + self.change_port(parameter_editor, components, "SERIAL3") + assert parameter_editor.current_step_parameters["SERIAL2_PROTOCOL"] is original + assert original.get_new_value() == 3.0 + assert parameter_editor.current_step_parameters["SERIAL3_PROTOCOL"].get_new_value() == 2.0 + parameter_editor._export_current_file(annotate_doc=False) + assert parameter_editor._local_filesystem.file_parameters[step_file]["SERIAL2_PROTOCOL"].value == 3.0 + + def test_reload_forgets_automatic_ownership_but_keeps_saved_value(self, parameter_editor: ParameterEditor) -> None: + """GIVEN a saved automatic row, WHEN the step reloads, THEN it follows saved-file semantics.""" + step_file, components = self.prepare(parameter_editor) + self.change_port(parameter_editor, components, "SERIAL2") + parameter_editor._export_current_file(annotate_doc=False) + parameter_editor.update_vehicle_components({**components, "port": "None"}) + saved = parameter_editor._local_filesystem.file_parameters[step_file] + parameter_editor._config_step_processor.process_configuration_step = MagicMock( + return_value=( + { + name: parameter_editor._config_step_processor.create_ardupilot_parameter(name, par, step_file, {}) + for name, par in saved.items() + }, + [], + [], + set(), + [], + ParDict(), + set(), + ) + ) + + parameter_editor._repopulate_configuration_step_parameters() + parameter_editor.refresh_current_step_computed_parameters() + assert parameter_editor.current_step_parameters["SERIAL2_PROTOCOL"].get_new_value() == 2.0 + + def test_gnss_same_uart_driver_change_keeps_transport_until_port_is_removed( + self, parameter_editor: ParameterEditor + ) -> None: + """GIVEN a saved GNSS UART, WHEN its driver changes, THEN the port remains until deselected.""" + step_file = "10_gnss.param" + components = {"GNSS Receiver": {"FC Connection": {"Type": "SERIAL3", "Protocol": "uBlox"}}} + filesystem = parameter_editor._local_filesystem + filesystem.vehicle_components_fs.data = {"Components": components} + filesystem.configuration_steps = { + step_file: { + "rename_connection": "vehicle_components['GNSS Receiver']['FC Connection']['Type']", + "derived_parameters": {}, + } + } + filesystem.file_parameters = {step_file: ParDict()} + filesystem.derived_parameters = {step_file: ParDict()} + filesystem.compute_deletions.return_value = set() + + def compute(*_args: object) -> None: + selection = components["GNSS Receiver"]["FC Connection"] + filesystem.derived_parameters[step_file] = ( + ParDict({f"{selection['Type']}_PROTOCOL": Par(5.0)}) + if selection["Type"].startswith("SERIAL") and selection["Protocol"] == "uBlox" + else ParDict() + ) + + filesystem.compute_parameters.side_effect = compute + parameter_editor.current_file = step_file + parameter_editor._flight_controller.fc_parameters = {} + parameter_editor._config_step_processor.variables = {"vehicle_components": components} + parameter_editor.current_step_parameters = {} + + parameter_editor.refresh_current_step_computed_parameters() + parameter_editor._export_current_file(annotate_doc=False) + components["GNSS Receiver"]["FC Connection"]["Protocol"] = "MAVLink" + parameter_editor.refresh_current_step_computed_parameters() + assert parameter_editor.current_step_parameters["SERIAL3_PROTOCOL"].get_new_value() == 5.0 + + components["GNSS Receiver"]["FC Connection"]["Protocol"] = "uBlox" + parameter_editor.refresh_current_step_computed_parameters() + assert parameter_editor.current_step_parameters["SERIAL3_PROTOCOL"].get_new_value() == 5.0 + components["GNSS Receiver"]["FC Connection"]["Type"] = "SERIAL4" + parameter_editor.refresh_current_step_computed_parameters() + parameter_editor._export_current_file(annotate_doc=False) + assert set(filesystem.file_parameters[step_file]) == {"SERIAL4_PROTOCOL"} + components["GNSS Receiver"]["FC Connection"].update({"Type": "None", "Protocol": "None"}) + parameter_editor.refresh_current_step_computed_parameters() + parameter_editor._export_current_file(annotate_doc=False) + assert not filesystem.file_parameters[step_file] + assert not parameter_editor._has_unsaved_changes() + + def test_old_flight_log_ui_workflow_is_retired() -> None: """Dialogs and progress belong to FileBrowserWindow, not the model.""" assert not hasattr(ParameterEditor, "download_last_flight_log_workflow") @@ -7419,6 +8070,192 @@ def test_a_duplicate_absent_from_the_file_is_removed_without_a_deletion_record( class TestDerivedParameterPatching: """Test how derived parameter values are patched into an already loaded step.""" + def test_explicitly_deleted_inline_derived_row_stays_deleted_until_readded( + self, parameter_editor: ParameterEditor + ) -> None: + """GIVEN a derived row, WHEN deleted and refreshed, THEN only a manual re-add restores it.""" + step_file = "10_telemetry.param" + parameter_editor.current_file = step_file + parameter_editor._flight_controller.fc_parameters = {"SERIAL2_PROTOCOL": 0.0, "SERIAL3_PROTOCOL": 0.0} + parameter_editor._local_filesystem.configuration_steps = {step_file: {"derived_parameters": {}}} + parameter_editor._local_filesystem.file_parameters = {step_file: ParDict()} + selection = {"name": "SERIAL2_PROTOCOL"} + + def compute(*_args: object) -> None: + name = selection["name"] + parameter_editor._local_filesystem.derived_parameters = {step_file: ParDict({name: Par(2.0)})} + + parameter_editor._local_filesystem.compute_parameters.side_effect = compute + + parameter_editor.refresh_current_step_computed_parameters() + assert "SERIAL2_PROTOCOL" in parameter_editor.current_step_parameters + + parameter_editor.delete_parameter_from_current_file("SERIAL2_PROTOCOL") + assert "SERIAL2_PROTOCOL" not in parameter_editor.current_step_parameters + assert "SERIAL2_PROTOCOL" in parameter_editor._deleted_parameters + + parameter_editor.refresh_current_step_computed_parameters() + assert "SERIAL2_PROTOCOL" not in parameter_editor.current_step_parameters + + selection["name"] = "SERIAL3_PROTOCOL" + parameter_editor.refresh_current_step_computed_parameters() + selection["name"] = "SERIAL2_PROTOCOL" + parameter_editor.refresh_current_step_computed_parameters() + assert "SERIAL2_PROTOCOL" not in parameter_editor.current_step_parameters + + assert parameter_editor.add_parameter_to_current_file("SERIAL2_PROTOCOL") + assert "SERIAL2_PROTOCOL" not in parameter_editor._deleted_parameters + assert "SERIAL2_PROTOCOL" not in parameter_editor._suppressed_derived_parameters + parameter_editor.refresh_current_step_computed_parameters() + assert "SERIAL2_PROTOCOL" in parameter_editor.current_step_parameters + + parameter_editor.delete_parameter_from_current_file("SERIAL2_PROTOCOL") + assert "SERIAL2_PROTOCOL" in parameter_editor._deleted_parameters + parameter_editor.refresh_current_step_computed_parameters() + assert "SERIAL2_PROTOCOL" not in parameter_editor.current_step_parameters + parameter_editor._export_current_file(annotate_doc=False) + exported = parameter_editor._local_filesystem.export_to_param.call_args.args[0] + assert "SERIAL2_PROTOCOL" not in exported + assert "SERIAL2_PROTOCOL" not in parameter_editor._local_filesystem.file_parameters[step_file] + assert not parameter_editor._has_unsaved_changes() + + parameter_editor.refresh_current_step_computed_parameters() + assert "SERIAL2_PROTOCOL" not in parameter_editor.current_step_parameters + + assert parameter_editor.add_parameter_to_current_file("SERIAL2_PROTOCOL") + assert "SERIAL2_PROTOCOL" not in parameter_editor._suppressed_derived_parameters + parameter_editor.refresh_current_step_computed_parameters() + assert "SERIAL2_PROTOCOL" in parameter_editor.current_step_parameters + parameter_editor.delete_parameter_from_current_file("SERIAL2_PROTOCOL") + + parameter_editor._config_step_processor.process_configuration_step = MagicMock( + return_value=({}, [], [], set(), [], ParDict({"SERIAL2_PROTOCOL": Par(2.0)}), set()) + ) + parameter_editor._repopulate_configuration_step_parameters() + assert "SERIAL2_PROTOCOL" in parameter_editor.current_step_parameters + + def test_inline_selection_adds_and_retires_only_automatic_derived_rows(self, parameter_editor: ParameterEditor) -> None: + """GIVEN changing ports, WHEN refreshed, THEN only the final derived row is exported.""" + step_file = "10_telemetry.param" + parameter_editor.current_file = step_file + parameter_editor._flight_controller.fc_parameters = {} + parameter_editor._local_filesystem.configuration_steps = {step_file: {"derived_parameters": {}}} + parameter_editor._local_filesystem.file_parameters = {step_file: ParDict()} + selection = {"name": "SERIAL2_PROTOCOL"} + + def compute(*_args: object) -> None: + name = selection["name"] + parameter_editor._local_filesystem.derived_parameters = { + step_file: ParDict({name: Par(2.0, "telemetry")}) if name else ParDict() + } + + parameter_editor._local_filesystem.compute_parameters.side_effect = compute + unrelated = ArduPilotParameter("OTHER", Par(1.0)) + unrelated.set_new_value("3") + parameter_editor.current_step_parameters = {"OTHER": unrelated} + + parameter_editor.refresh_current_step_computed_parameters() + assert parameter_editor.current_step_parameters["SERIAL2_PROTOCOL"].get_new_value() == 2.0 + assert "SERIAL2_PROTOCOL" in parameter_editor._added_parameters + + selection["name"] = "SERIAL3_PROTOCOL" + parameter_editor.refresh_current_step_computed_parameters() + parameter_editor.refresh_current_step_computed_parameters() + assert set(parameter_editor.current_step_parameters) == {"OTHER", "SERIAL3_PROTOCOL"} + assert parameter_editor._added_parameters == {"SERIAL3_PROTOCOL"} + assert unrelated.get_new_value() == 3.0 + + selection["name"] = "" + parameter_editor.refresh_current_step_computed_parameters() + assert set(parameter_editor.current_step_parameters) == {"OTHER"} + assert not parameter_editor._added_parameters + + selection["name"] = "SERIAL3_PROTOCOL" + parameter_editor.refresh_current_step_computed_parameters() + parameter_editor._export_current_file(annotate_doc=False) + exported = parameter_editor._local_filesystem.export_to_param.call_args.args[0] + assert set(exported) == {"OTHER", "SERIAL3_PROTOCOL"} + + def test_refresh_preserves_saved_manual_and_user_added_parameters(self, parameter_editor: ParameterEditor) -> None: + """GIVEN saved and user owned rows, WHEN derivation ends, THEN their edits survive.""" + step_file = "10_telemetry.param" + parameter_editor.current_file = step_file + parameter_editor._flight_controller.fc_parameters = {} + parameter_editor._local_filesystem.configuration_steps = {step_file: {"derived_parameters": {}}} + parameter_editor._local_filesystem.file_parameters = {step_file: ParDict({"SAVED": Par(1.0)})} + parameter_editor._local_filesystem.derived_parameters = {step_file: ParDict({"SAVED": Par(2.0)})} + saved = ArduPilotParameter("SAVED", Par(1.0), derived_par=Par(2.0)) + saved.set_manual_override(True) + saved.set_new_value("4") + user_added = ArduPilotParameter("USER", Par(5.0)) + parameter_editor.current_step_parameters = {"SAVED": saved, "USER": user_added} + parameter_editor._added_parameters = {"USER"} + + parameter_editor.refresh_current_step_computed_parameters() + assert saved.get_new_value() == 4.0 + parameter_editor._local_filesystem.derived_parameters = {step_file: ParDict()} + parameter_editor.refresh_current_step_computed_parameters() + assert parameter_editor.current_step_parameters == {"SAVED": saved, "USER": user_added} + assert parameter_editor._added_parameters == {"USER"} + + def test_manual_override_of_inserted_row_survives_retirement(self, parameter_editor: ParameterEditor) -> None: + """GIVEN an overridden derived row, WHEN its condition ends, THEN it stays user owned.""" + step_file = "10_telemetry.param" + parameter_editor.current_file = step_file + parameter_editor._flight_controller.fc_parameters = {} + parameter_editor._local_filesystem.configuration_steps = {step_file: {"derived_parameters": {}}} + parameter_editor._local_filesystem.derived_parameters = {step_file: ParDict({"SERIAL2_PROTOCOL": Par(2.0)})} + + parameter_editor.refresh_current_step_computed_parameters() + overridden = parameter_editor.current_step_parameters["SERIAL2_PROTOCOL"] + overridden.set_manual_override(True) + overridden.set_new_value("3") + parameter_editor._local_filesystem.derived_parameters = {step_file: ParDict()} + + parameter_editor.refresh_current_step_computed_parameters() + + assert parameter_editor.current_step_parameters["SERIAL2_PROTOCOL"] is overridden + assert overridden.get_new_value() == 3.0 + assert "SERIAL2_PROTOCOL" in parameter_editor._added_parameters + + def test_deleted_and_unavailable_derived_rows_do_not_return(self, parameter_editor: ParameterEditor) -> None: + """GIVEN deletion and FC availability rules, WHEN refreshed, THEN no row bypasses them.""" + step_file = "10_telemetry.param" + parameter_editor.current_file = step_file + parameter_editor._flight_controller.fc_parameters = {"SUPPORTED": 0.0} + parameter_editor._local_filesystem.configuration_steps = {step_file: {"derived_parameters": {}}} + parameter_editor._local_filesystem.derived_parameters = { + step_file: ParDict({"SUPPORTED": Par(2.0), "UNSUPPORTED": Par(3.0)}) + } + parameter_editor._deleted_parameters = {"SUPPORTED"} + + parameter_editor.refresh_current_step_computed_parameters() + + assert not parameter_editor.current_step_parameters + assert not parameter_editor._added_parameters + + def test_saved_connection_rename_wins_over_new_derived_destination(self, parameter_editor: ParameterEditor) -> None: + """GIVEN a saved serial parameter, WHEN the port changes, THEN it owns the new name.""" + step_file = "10_telemetry.param" + parameter_editor.current_file = step_file + parameter_editor._flight_controller.fc_parameters = {} + parameter_editor._local_filesystem.configuration_steps = { + step_file: {"derived_parameters": {}, "rename_connection": "new_port"} + } + parameter_editor._local_filesystem.file_parameters = {step_file: ParDict({"SERIAL2_PROTOCOL": Par(5.0, "saved")})} + parameter_editor.current_step_parameters = {"SERIAL2_PROTOCOL": ArduPilotParameter("SERIAL2_PROTOCOL", Par(5.0))} + parameter_editor._local_filesystem.derived_parameters = {step_file: ParDict({"SERIAL3_PROTOCOL": Par(2.0)})} + parameter_editor._config_step_processor.calculate_connection_rename_operations = MagicMock( + return_value=(set(), [("SERIAL2_PROTOCOL", "SERIAL3_PROTOCOL")]) + ) + + parameter_editor.refresh_current_step_computed_parameters() + + assert set(parameter_editor.current_step_parameters) == {"SERIAL3_PROTOCOL"} + assert parameter_editor.current_step_parameters["SERIAL3_PROTOCOL"].get_new_value() == 2.0 + assert parameter_editor._connection_renames == {"SERIAL2_PROTOCOL": "SERIAL3_PROTOCOL"} + assert "SERIAL3_PROTOCOL" not in parameter_editor._inline_derived_parameters + def test_a_parameter_that_became_derived_is_rebuilt_from_the_file_value(self, parameter_editor: ParameterEditor) -> None: """ A parameter that only became derived after a component edit is rebuilt with the derived flag. @@ -7429,6 +8266,7 @@ def test_a_parameter_that_became_derived_is_rebuilt_from_the_file_value(self, pa """ step_file = "11_battery.param" parameter_editor.current_file = step_file + parameter_editor._flight_controller.fc_parameters = {} parameter_editor._local_filesystem.configuration_steps = {step_file: {"derived_parameters": {"MOT_X": {}}}} original_par = Par(4.2, "from file") parameter_editor._local_filesystem.file_parameters = {step_file: ParDict({"MOT_X": original_par})} @@ -7440,6 +8278,8 @@ def test_a_parameter_that_became_derived_is_rebuilt_from_the_file_value(self, pa plain = MagicMock() plain.is_forced = False plain.is_derived = False + plain.is_manual_override = False + plain.is_dirty = False parameter_editor.current_step_parameters = {"MOT_X": plain} rebuilt = MagicMock()