WIP

core/services/ardupilot_manager/config.py

@@ -1,9 +1,9 @@
 # This file is used to define general configurations for the app
-import appdirs
-
 from collections import namedtuple
 from pathlib import Path
 
+import appdirs
+
 SERVICE_NAME = "autopilot-manager"
 
 
@@ -15,6 +15,16 @@
 BLUE_OS_FILES_PATH = Path.joinpath(Path.home(), "blueos-files")
 DEFAULTS_PATH = Path.joinpath(BLUE_OS_FILES_PATH, "ardupilot-manager/default")
 
+@staticmethod
+    def create_app_folders() -> None:
+        """Create the necessary folders for proper app function."""
+        for folder in Settings.app_folders:
+            try:
+                Path.mkdir(folder, parents=True, exist_ok=True)
+            except FileExistsError:
+                logger.warning(f"Found file {folder} where a folder should be. Removing file and creating folder.")
+                Path.unlink(folder)
+                Path.mkdir(folder)
 
 StaticFile = namedtuple("StaticFile", "parent filename url")
 DEFAULT_RESOURCES = [

core/services/ardupilot_manager/settings.py

@@ -1,103 +1,19 @@
-import json
-from copy import deepcopy
-from pathlib import Path
-from typing import Any, Dict, Optional, Union, cast
+from typing import Any, Dict, Optional
 
-from loguru import logger
+from commonwealth.settings.bases.pydantic_base import PydanticSettings
 
-from typedefs import SITLFrame
 
+class SettingsV1(PydanticSettings):
+    sitl_frame: Optional[str] = None
+    prefered_router: Optional[str] = None
 
-class Settings:
-    sitl_frame = SITLFrame.UNDEFINED
-    preferred_router: Optional[str] = None
-
-    def __init__(self) -> None:
-        self.root: Dict[str, Union[int, Dict[str, Any]]] = {"version": 0, "content": {}}
-
-    @staticmethod
-    def create_app_folders() -> None:
-        """Create the necessary folders for proper app function."""
-        for folder in Settings.app_folders:
-            try:
-                Path.mkdir(folder, parents=True, exist_ok=True)
-            except FileExistsError:
-                logger.warning(f"Found file {folder} where a folder should be. Removing file and creating folder.")
-                Path.unlink(folder)
-                Path.mkdir(folder)
-
-    def create_settings_file(self) -> None:
-        """Create settings file."""
-        try:
-            if not Path.is_file(self.settings_file):
-                with open(self.settings_file, "w+", encoding="utf-8") as file:
-                    logger.info(f"Creating settings file: {self.settings_file}")
-                    json.dump(self.root, file, sort_keys=True, indent=4)
-
-        except OSError as error:
-            logger.error(
-                f"Could not create settings files: {error}\n No settings will be loaded or saved during this session."
-            )
-
-    @property
-    def content(self) -> Dict[str, Any]:
-        return cast(Dict[str, Any], self.root["content"])
-
-    @property
-    def version(self) -> int:
-        return cast(int, self.root["version"])
-
-    def settings_exist(self) -> bool:
-        """Check if settings file exist
-
-        Returns:
-            bool: True if it exist
-        """
-        return Path.is_file(self.settings_file)
-
-    def load(self) -> bool:
-        """Load settings from file
-
-        Returns:
-            bool: False if failed
-        """
-        if not self.settings_exist():
-            logger.error(f"User settings does not exist on {self.settings_file}.")
-            return False
-
-        data = None
-        try:
-            with open(self.settings_file, encoding="utf-8") as file:
-                data = json.load(file)
-                if data["version"] != self.root["version"]:
-                    logger.error("User settings does not match with our supported version.")
-                    return False
-
-                self.root = data
-        except Exception as error:
-            logger.error(f"Failed to fetch data from file ({self.settings_file}): {error}")
-            logger.debug(data)
-
-        return True
-
-    def save(self, content: Dict[str, Any]) -> None:
-        """Save content to file
-
-        Args:
-            content (list): Configuration list
-        """
-        # We don't want to write in disk if there is nothing different to write
-        if self.root["content"] == content:
-            logger.info("No new data. Not updating settings file.")
+    def migrate(self, data: Dict[str, Any]) -> None:
+        if data["VERSION"] == SettingsV1.STATIC_VERSION:
             return
 
-        self.root["content"] = deepcopy(content)
-
-        try:
-            Path.mkdir(self.settings_path, exist_ok=True)
+        if data["VERSION"] < SettingsV1.STATIC_VERSION:
+            super().migrate(data)
 
-            with open(self.settings_file, "w+", encoding="utf-8") as file:
-                logger.info(f"Updating settings file: {self.settings_file}")
-                json.dump(self.root, file, sort_keys=True, indent=4)
-        except Exception as error:
-            logger.warning(f"Could not save settings to disk: {error}")
+        data["VERSION"] = SettingsV1.STATIC_VERSION
+        data["animal"] = self.animal
+        data["first_variable"] = self.first_variable

core/services/ardupilot_manager/setup.py

@@ -5,14 +5,12 @@
 import sys
 import time
 import urllib.request
-
 from warnings import warn
 
 import setuptools
 
 from config import DEFAULT_RESOURCES
 
-
 ssl._create_default_https_context = ssl._create_unverified_context