refactor(backend filesystem): Refactor to access to a delegated filesystem wrapper pattern.

Improve code organization by encapsulating JSON schema operations
within a dedicated FilesystemJSONWithSchema class.

- Replace direct vehicle_components attribute with vehicle_components_fs.data for data access
- Delegate JSON operations (load, save, validate) to the new filesystem wrapper
- Update all references throughout the codebase to use the new pattern

Author: amilcarlucas

ardupilot_methodic_configurator/backend_filesystem.py

@@ -115,8 +115,8 @@ def re_init(self, vehicle_dir: str, vehicle_type: str, blank_component_data: boo
 
         if blank_component_data:
             self.wipe_component_info()
-            if self.vehicle_components and "Components" in self.vehicle_components:
-                self.save_vehicle_components_json_data(self.vehicle_components, self.vehicle_dir)
+            if self.vehicle_components_fs.data and "Components" in self.vehicle_components_fs.data:
+                self.save_vehicle_components_json_data(self.vehicle_components_fs.data, self.vehicle_dir)
 
         if not self.fw_version:
             self.fw_version = self.get_fc_fw_version_from_vehicle_components_json()
@@ -159,7 +159,7 @@ def vehicle_configuration_files_exist(self, vehicle_dir: str) -> bool:
             if platform_system() == "Windows":
                 vehicle_configuration_files = [f.lower() for f in vehicle_configuration_files]
             pattern = re_compile(r"^\d{2}_.*\.param$")
-            if self.vehicle_components_json_filename in vehicle_configuration_files and any(
+            if self.vehicle_components_fs.json_filename in vehicle_configuration_files and any(
                 pattern.match(f) for f in vehicle_configuration_files
             ):
                 return True
@@ -524,7 +524,7 @@ def zip_files(self, files_to_zip: list[tuple[bool, str]]) -> None:
                 "00_default.param",
                 "apm.pdef.xml",
                 self.configuration_steps_filename,
-                self.vehicle_components_json_filename,
+                self.vehicle_components_fs.json_filename,
                 "vehicle.jpg",
                 "last_uploaded_filename.txt",
                 "tempcal_gyro.png",
@@ -700,8 +700,12 @@ def backup_fc_parameters_to_file(
 
     def get_eval_variables(self) -> dict[str, dict[str, Any]]:
         variables = {}
-        if hasattr(self, "vehicle_components") and self.vehicle_components and "Components" in self.vehicle_components:
-            variables["vehicle_components"] = self.vehicle_components["Components"]
+        if (
+            hasattr(self, "vehicle_components_fs")
+            and self.vehicle_components_fs.data
+            and "Components" in self.vehicle_components_fs.data
+        ):
+            variables["vehicle_components"] = self.vehicle_components_fs.data["Components"]
         if hasattr(self, "doc_dict") and self.doc_dict:
             variables["doc_dict"] = self.doc_dict
         return variables

ardupilot_methodic_configurator/backend_filesystem_vehicle_components.py

@@ -23,9 +23,8 @@
 from re import match as re_match
 from typing import Any, Union
 
-from jsonschema import ValidationError, validate, validators
-
 from ardupilot_methodic_configurator import _
+from ardupilot_methodic_configurator.backend_filesystem_json_with_schema import FilesystemJSONWithSchema
 from ardupilot_methodic_configurator.backend_filesystem_program_settings import ProgramSettings
 from ardupilot_methodic_configurator.data_model_vehicle_components_json_schema import VehicleComponentsJsonSchema
 from ardupilot_methodic_configurator.middleware_template_overview import TemplateOverview
@@ -35,10 +34,9 @@ class VehicleComponents:
     """Load and save vehicle components configurations from a JSON file."""
 
     def __init__(self, save_component_to_system_templates: bool = False) -> None:
-        self.vehicle_components_json_filename = "vehicle_components.json"
-        self.vehicle_components_schema_filename = "vehicle_components_schema.json"
-        self.vehicle_components: Union[None, dict[str, Any]] = None
-        self.schema: Union[None, dict[Any, Any]] = None
+        self.vehicle_components_fs: FilesystemJSONWithSchema = FilesystemJSONWithSchema(
+            "vehicle_components.json", "vehicle_components_schema.json"
+        )
         self.save_component_to_system_templates = save_component_to_system_templates
 
     def load_schema(self) -> dict:
@@ -47,35 +45,7 @@ def load_schema(self) -> dict:
 
         :return: The schema as a dictionary
         """
-        if self.schema is not None:
-            return self.schema
-
-        # Determine the location of the schema file
-        schema_path = os_path.join(os_path.dirname(__file__), self.vehicle_components_schema_filename)
-
-        try:
-            with open(schema_path, encoding="utf-8") as file:
-                loaded_schema: dict[Any, Any] = json_load(file)
-
-                # Validate the schema itself against the JSON Schema meta-schema
-                try:
-                    # Get the Draft7Validator class which has the META_SCHEMA property
-                    validator_class = validators.Draft7Validator
-                    meta_schema = validator_class.META_SCHEMA
-
-                    # Validate the loaded schema against the meta-schema
-                    validate(instance=loaded_schema, schema=meta_schema)
-                    logging_debug(_("Schema file '%s' is valid."), schema_path)
-                except Exception as e:  # pylint: disable=broad-exception-caught
-                    logging_error(_("Schema file '%s' is not a valid JSON Schema: %s"), schema_path, str(e))
-
-                self.schema = loaded_schema
-            return self.schema
-        except FileNotFoundError:
-            logging_error(_("Schema file '%s' not found."), schema_path)
-        except JSONDecodeError:
-            logging_error(_("Error decoding JSON schema from file '%s'."), schema_path)
-        return {}
+        return self.vehicle_components_fs.load_schema()
 
     def load_component_templates(self) -> dict[str, list[dict]]:
         """
@@ -285,113 +255,42 @@ def validate_vehicle_components(self, data: dict) -> tuple[bool, str]:
         :param data: The vehicle components data to validate
         :return: A tuple of (is_valid, error_message)
         """
-        schema = self.load_schema()
-        if not schema:
-            return False, _("Could not load validation schema")
-
-        try:
-            validate(instance=data, schema=schema)
-            return True, ""
-        except ValidationError as e:
-            return False, f"{_('Validation error')}: {e.message}"
+        return self.vehicle_components_fs.validate_json_against_schema(data)
 
     def load_vehicle_components_json_data(self, vehicle_dir: str) -> dict[Any, Any]:
-        data: dict[Any, Any] = {}
-        filepath = os_path.join(vehicle_dir, self.vehicle_components_json_filename)
-        try:
-            with open(filepath, encoding="utf-8") as file:
-                data = json_load(file)
+        return self.vehicle_components_fs.load_json_data(vehicle_dir)
 
-            # Validate the loaded data against the schema
-            is_valid, error_message = self.validate_vehicle_components(data)
-            if not is_valid:
-                logging_error(_("Invalid vehicle components file '%s': %s"), filepath, error_message)
-                # We still return the data even if invalid for debugging purposes
-        except FileNotFoundError:
-            # Normal users do not need this information
-            logging_debug(_("File '%s' not found in %s."), self.vehicle_components_json_filename, vehicle_dir)
-        except JSONDecodeError:
-            logging_error(_("Error decoding JSON data from file '%s'."), filepath)
-        self.vehicle_components = data
-        return data
-
-    def save_vehicle_components_json_data(self, data: dict, vehicle_dir: str) -> tuple[bool, str]:  # noqa: PLR0911 # pylint: disable=too-many-return-statements
+    def save_vehicle_components_json_data(self, data: dict, vehicle_dir: str) -> tuple[bool, str]:
         """
         Save the vehicle components data to a JSON file.
 
         :param data: The vehicle components data to save
         :param vehicle_dir: The directory to save the file in
         :return: A tuple of (error_occurred, error_message)
         """
-        # Validate before saving
-        # commented out until https://github.com/ArduPilot/MethodicConfigurator/pull/237 gets merged
-        # is_valid, error_message = self.validate_vehicle_components(data)
-        # if not is_valid:
-        #     msg = _("Cannot save invalid vehicle components data: {}").format(error_message)
-        #     logging_error(msg)
-        #     return True, msg
-
-        filepath = os_path.join(vehicle_dir, self.vehicle_components_json_filename)
-        try:
-            with open(filepath, "w", encoding="utf-8", newline="\n") as file:
-                json_dump(data, file, indent=4)
-        except FileNotFoundError:
-            msg = _("Directory '{}' not found").format(vehicle_dir)
-            logging_error(msg)
-            return True, msg
-        except PermissionError:
-            msg = _("Permission denied when writing to file '{}'").format(filepath)
-            logging_error(msg)
-            return True, msg
-        except IsADirectoryError:
-            msg = _("Path '{}' is a directory, not a file").format(filepath)
-            logging_error(msg)
-            return True, msg
-        except OSError as e:
-            msg = _("OS error when writing to file '{}': {}").format(filepath, str(e))
-            logging_error(msg)
-            return True, msg
-        except TypeError as e:
-            msg = _("Type error when serializing data to JSON: {}").format(str(e))
-            logging_error(msg)
-            return True, msg
-        except ValueError as e:
-            msg = _("Value error when serializing data to JSON: {}").format(str(e))
-            logging_error(msg)
-            return True, msg
-        except Exception as e:  # pylint: disable=broad-exception-caught
-            # Still have a fallback for truly unexpected errors
-            msg = _("Unexpected error saving data to file '{}': {}").format(filepath, str(e))
-            logging_error(msg)
-            return True, msg
-
-        return False, ""
+        return self.vehicle_components_fs.save_json_data(data, vehicle_dir)
 
     def get_fc_fw_type_from_vehicle_components_json(self) -> str:
-        if self.vehicle_components and "Components" in self.vehicle_components:
-            components = self.vehicle_components["Components"]
-        else:
-            components = None
+        data = self.vehicle_components_fs.data
+        components = data["Components"] if data and "Components" in data else None
         if components:
             fw_type: str = components.get("Flight Controller", {}).get("Firmware", {}).get("Type", "")
             if fw_type in self.supported_vehicles():
                 return fw_type
-            error_msg = _("Firmware type {fw_type} in {self.vehicle_components_json_filename} is not supported")
-            logging_error(error_msg.format(**locals()))
+            error_msg = _("Firmware type {fw_type} in {filename} is not supported")
+            logging_error(error_msg.format(fw_type=fw_type, filename=self.vehicle_components_fs.json_filename))
         return ""
 
     def get_fc_fw_version_from_vehicle_components_json(self) -> str:
-        if self.vehicle_components and "Components" in self.vehicle_components:
-            components = self.vehicle_components["Components"]
-        else:
-            components = None
+        data = self.vehicle_components_fs.data
+        components = data["Components"] if data and "Components" in data else None
         if components:
             version_str: str = components.get("Flight Controller", {}).get("Firmware", {}).get("Version", "")
             version_str = version_str.lstrip().split(" ")[0] if version_str else ""
             if re_match(r"^\d+\.\d+\.\d+$", version_str):
                 return version_str
-            error_msg = _("FW version string {version_str} on {self.vehicle_components_json_filename} is invalid")
-            logging_error(error_msg.format(**locals()))
+            error_msg = _("FW version string {version_str} on {filename} is invalid")
+            logging_error(error_msg.format(version_str=version_str, filename=self.vehicle_components_fs.json_filename))
         return ""
 
     @staticmethod
@@ -410,7 +309,7 @@ def get_vehicle_components_overviews() -> dict[str, TemplateOverview]:
         :return: A dictionary mapping subdirectory paths to TemplateOverview instances.
         """
         vehicle_components_dict = {}
-        file_to_find = VehicleComponents().vehicle_components_json_filename
+        file_to_find = VehicleComponents().vehicle_components_fs.json_filename
         template_default_dir = ProgramSettings.get_templates_base_dir()
         for root, _dirs, files in os_walk(template_default_dir):
             if file_to_find in files:
@@ -437,8 +336,9 @@ def wipe_component_info(self) -> None:
         Preserves the complete structure of the dictionary including all branches and leaves,
         but sets leaf values to empty values based on their type.
         """
-        if self.vehicle_components is not None:
-            self._recursively_clear_dict(self.vehicle_components)
+        data = self.vehicle_components_fs.data
+        if data is not None:
+            self._recursively_clear_dict(data)
 
     def _recursively_clear_dict(self, data: Union[dict, list, float, bool, str]) -> None:
         """

ardupilot_methodic_configurator/frontend_tkinter_directory_selection.py

@@ -200,7 +200,7 @@ def on_select_directory(self) -> bool:
             self.local_filesystem.vehicle_dir = self.directory
 
             if not self.local_filesystem.vehicle_configuration_files_exist(self.directory):
-                _filename = self.local_filesystem.vehicle_components_json_filename
+                _filename = self.local_filesystem.vehicle_components_fs.json_filename
                 error_msg = _("Selected directory must contain files matching \\d\\d_*\\.param and a {_filename} file")
                 messagebox.showerror(_("Invalid Vehicle Directory Selected"), error_msg.format(**locals()))
                 return False

tests/test_backend_filesystem.py

@@ -383,7 +383,7 @@ def test_get_eval_variables_with_none(self) -> None:
         lfs = LocalFilesystem(
             "vehicle_dir", "vehicle_type", None, allow_editing_template_files=False, save_component_to_system_templates=False
         )
-        lfs.vehicle_components = None
+        lfs.vehicle_components_fs.data = None
         lfs.doc_dict = None
         result = lfs.get_eval_variables()
         assert not result
@@ -483,7 +483,7 @@ def test_get_eval_variables(self) -> None:
         assert not result
 
         # Test with components and doc_dict
-        lfs.vehicle_components = {"Components": {"test": "value"}}
+        lfs.vehicle_components_fs.data = {"Components": {"test": "value"}}
         lfs.doc_dict = {"param": "doc"}
         result = lfs.get_eval_variables()
         assert "vehicle_components" in result