core:services:bridget:Migrate settings to pydantic

core/services/bridget/bridget.py

@@ -1,5 +1,4 @@
 import logging
-from pathlib import Path
 from typing import Dict, List
 
 import requests
@@ -8,11 +7,10 @@
 from commonwealth.settings.manager import Manager
 from pydantic import BaseModel, conint
 from serial.tools.list_ports_linux import SysFS
+from config import SERVICE_NAME
 
 from settings import BridgeSettingsSpecV2, SettingsV2
 
-USERDATA = Path("/usr/blueos/userdata/")
-
 
 class BridgeFrontendSpec(BaseModel):
     """Basic interface for 'bridges' links."""
@@ -47,9 +45,8 @@ class Bridget:
 
     def __init__(self) -> None:
         self._bridges: Dict[BridgeFrontendSpec, Bridge] = {}
-        # We use userdata because our regular settings folder is under /root, which regular users
-        # don't have access to.
-        self._settings_manager = Manager("bridget", SettingsV2, USERDATA / "settings" / "bridget")
+        # Following the standard BlueOS pattern: /root/.config/blueos/<service-name>/
+        self._settings_manager = Manager(SERVICE_NAME, SettingsV2)
         self._settings_manager.load()
         for bridge_settings_spec in self._settings_manager.settings.specsv2:
             try:

core/services/bridget/config.py

@@ -0,0 +1,3 @@
+# This file is used to define general configurations for the app
+
+SERVICE_NAME = "bridget"

core/services/bridget/main.py

@@ -13,8 +13,7 @@
 from loguru import logger
 
 from bridget import BridgeFrontendSpec, Bridget
-
-SERVICE_NAME = "bridget"
+from config import SERVICE_NAME
 
 logging.basicConfig(handlers=[InterceptHandler()], level=0)
 init_logger(SERVICE_NAME)

core/services/bridget/settings.py

@@ -1,14 +1,18 @@
-from typing import Any, Dict
+import json
+import pathlib
+from typing import Any, Dict, List
 
-import pykson  # type: ignore
-from commonwealth.settings import settings
+from commonwealth.settings.settings import PydanticSettings
+from pydantic import BaseModel
 
+OLD_SETTINGS_DIR = "/usr/blueos/userdata/settings/bridget/bridget"
 
-class BridgeSettingsSpecV1(pykson.JsonObject):
-    serial_path = pykson.StringField()
-    baudrate = pykson.IntegerField()
-    ip = pykson.StringField()
-    udp_port = pykson.IntegerField()
+
+class BridgeSettingsSpecV1(BaseModel):
+    serial_path: str
+    baudrate: int
+    ip: str
+    udp_port: int
 
     @staticmethod
     def from_spec(spec: "BridgeFrontendSpec") -> "BridgeSettingsSpecV1":  # type: ignore
@@ -24,32 +28,12 @@ def __eq__(self, other: object) -> Any:
             return self.serial_path == other.serial_path
         return False
 
-
-class SettingsV1(settings.BaseSettings):
-    VERSION = 1
-    specs = pykson.ObjectListField(BridgeSettingsSpecV1)
-
-    def __init__(self, *args: str, **kwargs: int) -> None:
-        super().__init__(*args, **kwargs)
-
-        self.VERSION = SettingsV1.VERSION
-
-    def migrate(self, data: Dict[str, Any]) -> None:
-        if data["VERSION"] == SettingsV1.VERSION:
-            return
-
-        if data["VERSION"] < SettingsV1.VERSION:
-            super().migrate(data)
-
-        data["VERSION"] = SettingsV1.VERSION
-
-
-class BridgeSettingsSpecV2(pykson.JsonObject):
-    udp_target_port = pykson.IntegerField()
-    udp_listen_port = pykson.IntegerField()
-    serial_path = pykson.StringField()
-    baudrate = pykson.IntegerField()
-    ip = pykson.StringField()
+class BridgeSettingsSpecV2(BaseModel):
+    udp_target_port: int
+    udp_listen_port: int
+    serial_path: str
+    baudrate: int
+    ip: str
 
     @staticmethod
     def from_spec(spec: "BridgeFrontendSpec") -> "BridgeSettingsSpecV2":  # type: ignore
@@ -67,23 +51,66 @@ def __eq__(self, other: object) -> Any:
         return False
 
 
-class SettingsV2(SettingsV1):
-    VERSION = 2
-    specsv2 = pykson.ObjectListField(BridgeSettingsSpecV2)
+def migrate_from_old_settings(version: int, target_settings: "SettingsV1") -> None:
+    """Common method to migrate settings from the old location."""
+    old_settings_file_path = pathlib.Path(OLD_SETTINGS_DIR) / f"settings-{version}.json"
+
+    if not old_settings_file_path.exists():
+        return
+
+    try:
+        with open(old_settings_file_path, "r", encoding="utf-8") as file:
+            old_data = json.load(file)
+
+        if version == 1 and "specs" in old_data:
+            target_settings.specs = [BridgeSettingsSpecV1.parse_obj(spec) for spec in old_data["specs"]]
+        elif version == 2 and "specsv2" in old_data:
+            # For V2, we need to cast to SettingsV2 to access specsv2
+            if hasattr(target_settings, 'specsv2'):
+                target_settings.specsv2 = [BridgeSettingsSpecV2.parse_obj(spec) for spec in old_data["specsv2"]]
+
+    except Exception:
+        # If migration fails, just continue with empty settings
+        pass
+
 
-    def __init__(self, *args: str, **kwargs: int) -> None:
-        super().__init__(*args, **kwargs)
+class SettingsV1(PydanticSettings):
+    specs: List[BridgeSettingsSpecV1] = []
 
-        self.VERSION = SettingsV2.VERSION
+    def migrate(self, data: Dict[str, Any]) -> None:
+        if data["VERSION"] == SettingsV1.STATIC_VERSION:
+            return
+
+        if data["VERSION"] < SettingsV1.STATIC_VERSION:
+            super().migrate(data)
+
+        data["VERSION"] = SettingsV1.STATIC_VERSION
+
+    def on_settings_created(self, file_path: pathlib.Path) -> None:
+        """Handle migration from old settings location when creating new settings."""
+        if self.VERSION not in (SettingsV1.STATIC_VERSION, SettingsV1.STATIC_VERSION + 1):
+            return
+
+        settings_v1_file = file_path.parent / "settings-1.json"
+        if settings_v1_file.exists():
+            # If settings v1 already exist, we don't re initialize them
+            return
+
+        # Use common migration method
+        migrate_from_old_settings(1, self)
+
+
+class SettingsV2(SettingsV1):
+    specsv2: List[BridgeSettingsSpecV2] = []
 
     def migrate(self, data: Dict[str, Any]) -> None:
-        if data["VERSION"] == SettingsV2.VERSION:
+        if data["VERSION"] == SettingsV2.STATIC_VERSION:
             return
 
-        if data["VERSION"] < SettingsV2.VERSION:
+        if data["VERSION"] < SettingsV2.STATIC_VERSION:
             super().migrate(data)
 
-            data["VERSION"] = SettingsV2.VERSION
+            data["VERSION"] = SettingsV2.STATIC_VERSION
             data["specsv2"] = []
             for spec in data["specs"]:
                 server = spec["ip"] == "0.0.0.0"
@@ -96,3 +123,16 @@ def migrate(self, data: Dict[str, Any]) -> None:
                         "udp_listen_port": spec["udp_port"] if server else 0,
                     }
                 )
+
+    def on_settings_created(self, file_path: pathlib.Path) -> None:
+        """Handle migration from old settings location when creating new settings."""
+        if self.VERSION not in (SettingsV2.STATIC_VERSION, SettingsV2.STATIC_VERSION + 1):
+            return
+
+        settings_v2_file = file_path.parent / "settings-2.json"
+        if settings_v2_file.exists():
+            # If settings v2 already exist, we don't re initialize them
+            return
+
+        # Use common migration method
+        migrate_from_old_settings(2, self)