diff --git a/core/libs/commonwealth/pyproject.toml b/core/libs/commonwealth/pyproject.toml index 25aff818f3..db4de41910 100644 --- a/core/libs/commonwealth/pyproject.toml +++ b/core/libs/commonwealth/pyproject.toml @@ -12,14 +12,10 @@ dependencies = [ "packaging>=20.4,<=25.0", "psutil>=5.7.2,<=7.1.3", "pydantic>=1.10.12,<=2.12.5", - "pykson==1.0.2", "sentry-sdk==2.29.1", "starlette>=0.27.0", ] -[tool.uv.sources] -pykson = { url = "https://github.com/patrickelectric/pykson/archive/refs/tags/1.0.2.tar.gz" } - [build-system] requires = ["hatchling"] build-backend = "hatchling.build" diff --git a/core/libs/commonwealth/src/commonwealth/settings/manager.py b/core/libs/commonwealth/src/commonwealth/settings/manager.py index e2e260778c..6f2ef0e595 100644 --- a/core/libs/commonwealth/src/commonwealth/settings/manager.py +++ b/core/libs/commonwealth/src/commonwealth/settings/manager.py @@ -1,4 +1,3 @@ from commonwealth.settings.managers.pydantic_manager import PydanticManager -from commonwealth.settings.managers.pykson_manager import PyksonManager as Manager -__all__ = ["Manager", "PydanticManager"] +__all__ = ["PydanticManager"] diff --git a/core/libs/commonwealth/src/commonwealth/settings/settings.py b/core/libs/commonwealth/src/commonwealth/settings/settings.py index 00ce4ed748..b7770b8a4b 100644 --- a/core/libs/commonwealth/src/commonwealth/settings/settings.py +++ b/core/libs/commonwealth/src/commonwealth/settings/settings.py @@ -1,5 +1,4 @@ from commonwealth.settings.bases.pydantic_base import PydanticSettings -from commonwealth.settings.bases.pykson_base import PyksonSettings as BaseSettings from commonwealth.settings.exceptions import ( BadAttributes, BadSettingsFile, @@ -8,7 +7,6 @@ ) __all__ = [ - "BaseSettings", "BadAttributes", "BadSettingsFile", "MigrationFail", diff --git a/core/libs/commonwealth/src/commonwealth/settings/tests/test_manager.py b/core/libs/commonwealth/src/commonwealth/settings/tests/test_manager.py deleted file mode 100644 index 3a46c2f5c2..0000000000 --- a/core/libs/commonwealth/src/commonwealth/settings/tests/test_manager.py +++ /dev/null @@ -1,274 +0,0 @@ -import os -import pathlib -import tempfile -from typing import Any, Dict - -import pykson # type: ignore - -from .. import manager, settings - - -class SettingsV1(settings.BaseSettings): - VERSION = 1 - version_1_variable = pykson.IntegerField(default_value=42) - - 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 - data["version_1_variable"] = self.version_1_variable - - -class SettingsV2(SettingsV1): - VERSION = 2 - version_2_variable = pykson.IntegerField(default_value=66) - - def __init__(self, *args: str, **kwargs: int) -> None: - super().__init__(*args, **kwargs) - - self.VERSION = SettingsV2.VERSION - - def migrate(self, data: Dict[str, Any]) -> None: - if data["VERSION"] == SettingsV2.VERSION: - return - - if data["VERSION"] < SettingsV2.VERSION: - SettingsV1().migrate(data) - - data["VERSION"] = SettingsV2.VERSION - data["version_2_variable"] = self.version_2_variable - - -class SettingsV3(SettingsV2): - VERSION = 3 - version_3_variable = pykson.IntegerField(default_value=99) - - def __init__(self, *args: str, **kwargs: int) -> None: - super().__init__(*args, **kwargs) - - self.VERSION = SettingsV3.VERSION - - def migrate(self, data: Dict[str, Any]) -> None: - if data["VERSION"] == SettingsV3.VERSION: - return - - if data["VERSION"] < SettingsV3.VERSION: - SettingsV2().migrate(data) - - data["VERSION"] = SettingsV3.VERSION - data["version_3_variable"] = self.version_3_variable - - -class SettingsV12(SettingsV3): - VERSION = 12 - version_12_variable = pykson.IntegerField(default_value=1992) - - def __init__(self, *args: str, **kwargs: int) -> None: - super().__init__(*args, **kwargs) - - self.VERSION = SettingsV12.VERSION - - def migrate(self, data: Dict[str, Any]) -> None: - if data["VERSION"] == SettingsV12.VERSION: - return - - if data["VERSION"] < SettingsV12.VERSION: - SettingsV3().migrate(data) - - data["VERSION"] = SettingsV12.VERSION - data["version_12_variable"] = self.version_12_variable - - -def test_basic_settings_save_load() -> None: - temporary_folder = tempfile.mkdtemp() - config_path = pathlib.Path(temporary_folder) - - # Test v1 save - settings_manager = manager.Manager("ManagerTest", SettingsV1, config_path) - settings_manager.settings.version_1_variable = 2022 - settings_manager.save() - - assert settings_manager.settings.version_1_variable == 2022 - - # Test v1 load - settings_manager = manager.Manager("ManagerTest", SettingsV1, config_path) - - assert settings_manager.settings.version_1_variable == 2022 - - # Test v2 load/save with migration from v1 - settings_manager = manager.Manager("ManagerTest", SettingsV2, config_path) - - assert settings_manager.settings.version_1_variable == 2022 - assert settings_manager.settings.version_2_variable == 66 - - settings_manager.settings.version_2_variable = 123 - settings_manager.save() - - settings_manager = manager.Manager("ManagerTest", SettingsV2, config_path) - - assert settings_manager.settings.version_1_variable == 2022 - assert settings_manager.settings.version_2_variable == 123 - - # Test v3 load/save with migration from v2 - settings_manager = manager.Manager("ManagerTest", SettingsV3, config_path) - - assert settings_manager.settings.version_1_variable == 2022 - assert settings_manager.settings.version_2_variable == 123 - assert settings_manager.settings.version_3_variable == 99 - - settings_manager.settings.version_3_variable = 222 - settings_manager.save() - - settings_manager = manager.Manager("ManagerTest", SettingsV3, config_path) - - assert settings_manager.settings.version_1_variable == 2022 - assert settings_manager.settings.version_2_variable == 123 - assert settings_manager.settings.version_3_variable == 222 - - # Test v12 load/save with migration from v3 - settings_manager = manager.Manager("ManagerTest", SettingsV12, config_path) - - assert settings_manager.settings.version_1_variable == 2022 - assert settings_manager.settings.version_2_variable == 123 - assert settings_manager.settings.version_3_variable == 222 - assert settings_manager.settings.version_12_variable == 1992 - - settings_manager.settings.version_12_variable = 14 - settings_manager.save() - - settings_manager = manager.Manager("ManagerTest", SettingsV12, config_path) - - assert settings_manager.settings.version_1_variable == 2022 - assert settings_manager.settings.version_2_variable == 123 - assert settings_manager.settings.version_3_variable == 222 - assert settings_manager.settings.version_12_variable == 14 - - assert len(os.listdir(config_path.joinpath("managertest"))) == 4 - - -def test_fallback_settings_save_load() -> None: - temporary_folder = tempfile.mkdtemp() - config_path = pathlib.Path(temporary_folder) - - # Test v12 with downgrade to v3 - settings_manager = manager.Manager("ManagerTest", SettingsV12, config_path) - - assert settings_manager.settings.version_1_variable == SettingsV1().version_1_variable - assert settings_manager.settings.version_2_variable == SettingsV2().version_2_variable - assert settings_manager.settings.version_3_variable == SettingsV3().version_3_variable - assert settings_manager.settings.version_12_variable == SettingsV12().version_12_variable - - settings_manager.settings.version_1_variable = 1 - settings_manager.settings.version_2_variable = 2 - settings_manager.settings.version_3_variable = 3 - settings_manager.settings.version_12_variable = 12 - settings_manager.save() - - settings_manager = manager.Manager("ManagerTest", SettingsV3, config_path) - - assert settings_manager.settings.version_1_variable == SettingsV1().version_1_variable - assert settings_manager.settings.version_2_variable == SettingsV2().version_2_variable - assert settings_manager.settings.version_3_variable == SettingsV3().version_3_variable - - settings_manager.settings.version_1_variable = 10 - settings_manager.settings.version_2_variable = 20 - settings_manager.settings.version_3_variable = 30 - settings_manager.save() - - # Check if v3 loads previous v3 configuration - settings_manager = manager.Manager("ManagerTest", SettingsV3, config_path) - - assert settings_manager.settings.version_1_variable == 10 - assert settings_manager.settings.version_2_variable == 20 - assert settings_manager.settings.version_3_variable == 30 - - # Check if v12 loads previous v12 configuration without v3 migration - settings_manager = manager.Manager("ManagerTest", SettingsV12, config_path) - - assert settings_manager.settings.version_1_variable == 1 - assert settings_manager.settings.version_2_variable == 2 - assert settings_manager.settings.version_3_variable == 3 - assert settings_manager.settings.version_12_variable == 12 - - -def test_invalid_json_fallback_to_defaults_v1() -> None: - temporary_folder = tempfile.mkdtemp() - config_path = pathlib.Path(temporary_folder) - - # Create a settings file with invalid JSON - settings_folder = config_path / "managertest" - settings_folder.mkdir(parents=True, exist_ok=True) - invalid_settings_file = settings_folder / "settings-1.json" - with open(invalid_settings_file, "w", encoding="utf-8") as f: - f.write('{"VERSION": 1, "version_1_variable": 999, "invalid_json": }') - - settings_manager = manager.Manager("ManagerTest", SettingsV1, config_path) - - # Verify that the manager falls back to default values - assert settings_manager.settings.VERSION == 1 - assert settings_manager.settings.version_1_variable == 42 - - # Verify that a v1 settings will be replaced - settings_manager.save() - assert len(os.listdir(settings_folder)) == 1 - - -def test_invalid_json_fallback_to_defaults_v1_from_invalid_v2() -> None: - temporary_folder = tempfile.mkdtemp() - config_path = pathlib.Path(temporary_folder) - settings_folder = config_path / "managertest" - settings_folder.mkdir(parents=True, exist_ok=True) - - # Create a valid JSON from settings V1 - default_settings = SettingsV1() - default_settings.version_1_variable = 369 - default_settings.save(settings_folder / "settings-1.json") - - # Create a invalid JSON from settings V2 - invalid_settings_file = settings_folder / "settings-2.json" - with open(invalid_settings_file, "w", encoding="utf-8") as f: - f.write('{"VERSION": 2, "version_2_variable": 999, "invalid_json": }') - - settings_manager = manager.Manager("ManagerTest", SettingsV2, config_path) - - # Verify that the manager falls back to default values - assert settings_manager.settings.VERSION == 2 - assert settings_manager.settings.version_1_variable == 369 - assert settings_manager.settings.version_2_variable == 66 - - # Verify that a v2 settings will be replaced - settings_manager.save() - assert len(os.listdir(settings_folder)) == 2 - - -def test_invalid_json_fallback_to_defaults_v2_from_invalid_v2_and_v1() -> None: - temporary_folder = tempfile.mkdtemp() - config_path = pathlib.Path(temporary_folder) - settings_folder = config_path / "managertest" - settings_folder.mkdir(parents=True, exist_ok=True) - - # Create a invalid JSON from settings V1 - invalid_settings_file = settings_folder / "settings-2.json" - with open(invalid_settings_file, "w", encoding="utf-8") as f: - f.write('{"VERSION": 1, "version_1_variable": 999, "invalid_json": }') - - # Create a invalid JSON from settings V2 - invalid_settings_file = settings_folder / "settings-2.json" - with open(invalid_settings_file, "w", encoding="utf-8") as f: - f.write('{"VERSION": 2, "version_2_variable": 999, "invalid_json": }') - - settings_manager = manager.Manager("ManagerTest", SettingsV2, config_path) - - # Verify that the manager falls back to default values - assert settings_manager.settings.VERSION == 2 - assert settings_manager.settings.version_1_variable == 42 - assert settings_manager.settings.version_2_variable == 66 diff --git a/core/libs/commonwealth/src/commonwealth/settings/tests/test_settings.py b/core/libs/commonwealth/src/commonwealth/settings/tests/test_settings.py deleted file mode 100644 index 55ae2292f9..0000000000 --- a/core/libs/commonwealth/src/commonwealth/settings/tests/test_settings.py +++ /dev/null @@ -1,243 +0,0 @@ -import pathlib -import tempfile -from typing import Any, Dict, List - -import pykson # type: ignore - -from .. import settings - - -class JsonExample(pykson.JsonObject): - name = pykson.StringField() - - -class Animal(pykson.JsonObject): - name: str = pykson.StringField() - animal_type: str = pykson.StringField() - parts: List[str] = pykson.ListField(item_type=str) - animal_json: List[pykson.JsonObject] = pykson.ObjectListField(item_type=JsonExample) - - def __init__(self, *args: str, **kwargs: Any) -> None: - super().__init__(*args, **kwargs) - - -class SettingsV1(settings.BaseSettings): - VERSION = 1 - animal = pykson.ObjectField( - Animal, - default_value=Animal( - name="bilica", - animal_type="dog", - parts=["finger", "eyes"], - animal_json=[{"name": "Json!"}], - ), - ) - first_variable = pykson.IntegerField(default_value=42) - - def __init__(self, *args: str, **kwargs: Any) -> 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 - data["animal"] = self.animal - data["first_variable"] = self.first_variable - - -class SettingsV1Expanded(SettingsV1): - new_variable = pykson.IntegerField(default_value=1992) - - -class SettingsV2(settings.BaseSettings): - VERSION = 2 - first_variable = pykson.IntegerField(default_value=66) - new_animal = pykson.ObjectField( - Animal, - default_value=Animal( - name="bilica", - animal_type="dog", - ), - ) - - def __init__(self, *args: str, **kwargs: Any) -> None: - super().__init__(*args, **kwargs) - - self.VERSION = SettingsV2.VERSION - - def migrate(self, data: Dict[str, Any]) -> None: - if data["VERSION"] == SettingsV2.VERSION: - return - - if data["VERSION"] < SettingsV2.VERSION: - SettingsV1().migrate(data) - - data["VERSION"] = SettingsV2.VERSION - data["first_variable"] = self.first_variable - - # Update variable name - data["new_animal"] = data["animal"] - data.pop("animal") - - -class SettingsV3(SettingsV2): - VERSION = 3 - - def __init__(self, *args: str, **kwargs: Any) -> None: - super().__init__(*args, **kwargs) - - self.VERSION = SettingsV3.VERSION - - def migrate(self, data: Dict[str, Any]) -> None: - if data["VERSION"] == SettingsV3.VERSION: - return - - if data["VERSION"] < SettingsV3.VERSION: - super().migrate(data) - - data["VERSION"] = SettingsV3.VERSION - - -class SettingsV4(SettingsV3): - VERSION = 4 - - def __init__(self, *args: str, **kwargs: Any) -> None: - super().__init__(*args, **kwargs) - - self.VERSION = SettingsV4.VERSION - - def migrate(self, data: Dict[str, Any]) -> None: - if data["VERSION"] == SettingsV4.VERSION: - return - - if data["VERSION"] < SettingsV4.VERSION: - super().migrate(data) - - data["VERSION"] = SettingsV4.VERSION - - -def test_basic_settings_save_load() -> None: - # Check basic access - settings_v1 = SettingsV1() - assert settings_v1.VERSION == 1 - assert settings_v1.first_variable == 42 - assert settings_v1.animal.name == "bilica" - assert settings_v1.animal.animal_type == "dog" - assert settings_v1.animal.parts == ["finger", "eyes"] - assert settings_v1.animal.animal_json[0].name == "Json!" - - # pylint: disable=consider-using-with - temporary_file = tempfile.NamedTemporaryFile().name - file_path = pathlib.Path(temporary_file) - - # Check basic save and load - settings_v1.first_variable = 66 - settings_v1.save(file_path) - - settings_v1_new = SettingsV1() - settings_v1_new.load(file_path) - assert settings_v1.first_variable == settings_v1_new.first_variable - - # Check for reset - settings_v1_new.reset() - settings_v1.save(file_path) - - settings_v1_new = SettingsV1() - settings_v1_new.load(file_path) - assert settings_v1.first_variable == 66 - - -def test_nested_settings_save_load() -> None: - # Check basic access - settings_v1 = SettingsV1() - assert settings_v1.animal.name == SettingsV1().animal.name - assert settings_v1.animal.animal_type == SettingsV1().animal.animal_type - - # pylint: disable=consider-using-with - temporary_file = tempfile.NamedTemporaryFile() - file_path = pathlib.Path(temporary_file.name) - - # Check basic save and load - settings_v1.first_variable = 66 - settings_v1.animal.name = "pingu" - settings_v1.animal.animal_type = "penguin" - - assert settings_v1.first_variable == 66 - assert settings_v1.animal.name == "pingu" - assert settings_v1.animal.animal_type == "penguin" - - settings_v1.save(file_path) - settings_v1_new = SettingsV1() - settings_v1_new.load(file_path) - - assert settings_v1.first_variable == settings_v1_new.first_variable - assert settings_v1.animal.name == settings_v1_new.animal.name - assert settings_v1.animal.animal_type == settings_v1_new.animal.animal_type - - -def test_simple_migration_settings_save_load() -> None: - settings_v1 = SettingsV1() - - # pylint: disable=consider-using-with - temporary_file = tempfile.NamedTemporaryFile() - file_path = pathlib.Path(temporary_file.name) - - settings_v1.first_variable = 66 - settings_v1.animal.name = "pingu" - settings_v1.animal.animal_type = "penguin" - - settings_v1.save(file_path) - settings_v1_new = SettingsV1() - settings_v1_new.load(file_path) - - # Check if migration works - settings_v2 = SettingsV2() - settings_v2.load(file_path) - settings_v2.save(file_path) - - assert settings_v1.first_variable == settings_v2.first_variable - assert settings_v1.animal.name == settings_v2.new_animal.name - assert settings_v1.animal.animal_type == settings_v2.new_animal.animal_type - - settings_v3 = SettingsV3() - settings_v3.load(file_path) - settings_v3.save(file_path) - - settings_v4 = SettingsV4() - settings_v4.load(file_path) - settings_v4.save(file_path) - - assert settings_v2.first_variable == settings_v4.first_variable - assert settings_v2.new_animal.name == settings_v4.new_animal.name - assert settings_v2.new_animal.animal_type == settings_v4.new_animal.animal_type - assert settings_v2.new_animal.parts[0] == settings_v4.new_animal.parts[0] - assert settings_v2.new_animal.parts[1] == settings_v4.new_animal.parts[1] - assert settings_v2.new_animal.animal_json[0].name == settings_v4.new_animal.animal_json[0].name - - -def test_simple_settings_expanded_save_load() -> None: - settings_v1 = SettingsV1() - - # pylint: disable=consider-using-with - temporary_file = tempfile.NamedTemporaryFile() - file_path = pathlib.Path(temporary_file.name) - - settings_v1.first_variable = 66 - settings_v1.animal.name = "pingu" - settings_v1.animal.animal_type = "penguin" - - # Load expanded settings with older settings structure - settings_v1.save(file_path) - settings_v1_expanded = SettingsV1Expanded() - settings_v1_expanded.load(file_path) - - assert settings_v1.first_variable == settings_v1_expanded.first_variable - assert settings_v1.animal.name == settings_v1_expanded.animal.name - assert settings_v1.animal.animal_type == settings_v1_expanded.animal.animal_type - assert settings_v1_expanded.new_variable == 1992 diff --git a/core/services/ardupilot_manager/firmware/test_FirmwareUpload.py b/core/services/ardupilot_manager/firmware/test_FirmwareUpload.py new file mode 100644 index 0000000000..dfa6d4004a --- /dev/null +++ b/core/services/ardupilot_manager/firmware/test_FirmwareUpload.py @@ -0,0 +1,112 @@ +import asyncio +import pathlib +import subprocess +from unittest.mock import AsyncMock, patch + +import pytest +from exceptions import InvalidUploadTool, UploadToolNotFound +from firmware.FirmwareUpload import FirmwareUploader + + +class TestFirmwareUpload: + def test_init_success(self) -> None: + with patch("shutil.which", return_value="ardupilot_fw_uploader.py"): + uploader = FirmwareUploader() + assert uploader._autopilot_port == pathlib.Path("/dev/autopilot") + assert uploader._baudrate_bootloader == 115200 + assert uploader._baudrate_flightstack == 57600 + assert uploader._binary == pathlib.Path("ardupilot_fw_uploader.py") + + def test_init_binary_not_found(self) -> None: + with patch("shutil.which", return_value=None): + with pytest.raises(UploadToolNotFound, match="Uploader binary not found on system's PATH."): + FirmwareUploader() + + def test_binary_name(self) -> None: + assert FirmwareUploader.binary_name() == "ardupilot_fw_uploader.py" + + def test_binary(self) -> None: + with patch("shutil.which", return_value="ardupilot_fw_uploader.py"), patch.object( + FirmwareUploader, "validate_binary" + ): + uploader = FirmwareUploader() + assert uploader.binary() == pathlib.Path("ardupilot_fw_uploader.py") + + def test_validate_binary_success(self) -> None: + with patch("shutil.which", return_value="ardupilot_fw_uploader.py"): + uploader = FirmwareUploader() + uploader.validate_binary() + + def test_validate_binary_failure(self) -> None: + with patch("shutil.which", return_value="ardupilot_fw_uploader.py"): + uploader = FirmwareUploader() + + with patch("subprocess.check_output", side_effect=subprocess.CalledProcessError(2, "cmd", "error output")): + with pytest.raises(InvalidUploadTool, match="Binary returned 2 on '--help' call: error output"): + uploader.validate_binary() + + def test_set_autopilot_port(self) -> None: + with patch("shutil.which", return_value="ardupilot_fw_uploader.py"): + uploader = FirmwareUploader() + new_port = pathlib.Path("/dev/autopilot2") + uploader.set_autopilot_port(new_port) + assert uploader._autopilot_port == new_port + + def test_set_baudrate_bootloader(self) -> None: + with patch("shutil.which", return_value="ardupilot_fw_uploader.py"): + uploader = FirmwareUploader() + new_baudrate = 57600 + uploader.set_baudrate_bootloader(new_baudrate) + assert uploader._baudrate_bootloader == new_baudrate + + def test_set_baudrate_flightstack(self) -> None: + with patch("shutil.which", return_value="ardupilot_fw_uploader.py"): + uploader = FirmwareUploader() + new_baudrate = 38400 + uploader.set_baudrate_flightstack(new_baudrate) + assert uploader._baudrate_flightstack == new_baudrate + + @pytest.mark.asyncio + async def test_upload_success(self) -> None: + with patch("shutil.which", return_value="ardupilot_fw_uploader.py"), patch.object( + FirmwareUploader, "validate_binary" + ): + uploader = FirmwareUploader() + + firmware_path = pathlib.Path("/tmp/test_firmware.bin") + + mock_process = AsyncMock() + mock_process.returncode = 0 + mock_process.stdout = AsyncMock() + mock_process.stdout.readline = AsyncMock( + side_effect=[b"Upload starting...\n", b"Progress: 50%\n", b"Upload complete!\n", b""] + ) + mock_process.wait = AsyncMock(return_value=0) + + with patch("asyncio.create_subprocess_shell", return_value=mock_process) as mock_create_subprocess, patch( + "asyncio.sleep" + ) as mock_sleep, patch("loguru.logger.info") as mock_info, patch("loguru.logger.debug") as mock_debug: + + await uploader.upload(firmware_path) + + expected_command = ( + f"ardupilot_fw_uploader.py {firmware_path}" + f" --port /dev/autopilot" + f" --baud-bootloader 115200" + f" --baud-flightstack 57600" + ) + + mock_create_subprocess.assert_called_once_with( + expected_command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + shell=True, + ) + + mock_info.assert_any_call("Starting upload of firmware to board.") + mock_info.assert_any_call("Successfully uploaded firmware to board.") + mock_debug.assert_any_call("Upload starting...") + mock_debug.assert_any_call("Progress: 50%") + mock_debug.assert_any_call("Upload complete!") + + mock_sleep.assert_called_with(10) diff --git a/core/services/beacon/main.py b/core/services/beacon/main.py index db3ffbe7be..967197d0a2 100755 --- a/core/services/beacon/main.py +++ b/core/services/beacon/main.py @@ -8,7 +8,7 @@ from typing import Any, Dict, List, Optional import psutil -from commonwealth.settings.manager import Manager +from commonwealth.settings.manager import PydanticManager from commonwealth.utils.apis import PrettyJSONResponse from commonwealth.utils.logs import init_logger from commonwealth.utils.sentry_config import init_sentry_async @@ -78,7 +78,7 @@ class Beacon: def __init__(self) -> None: self.runners: Dict[str, AsyncRunner] = {} try: - self.manager = Manager(SERVICE_NAME, SettingsV4) + self.manager: PydanticManager = PydanticManager(SERVICE_NAME, SettingsV4) except Exception as e: logger.warning(f"failed to load configuration file ({e}), loading defaults") self.load_default_settings() @@ -95,7 +95,7 @@ def load_default_settings(self) -> None: current_folder = pathlib.Path(__file__).parent.resolve() default_settings_file = current_folder / "default-settings.json" logger.debug("loading settings from ", default_settings_file) - self.manager = Manager(SERVICE_NAME, SettingsV4, load=False) + self.manager = PydanticManager(SERVICE_NAME, SettingsV4, load=False) self.manager.settings = self.manager.load_from_file(SettingsV4, default_settings_file) self.manager.save() diff --git a/core/services/beacon/settings.py b/core/services/beacon/settings.py index 286c5fdbe4..ff68e9ef58 100644 --- a/core/services/beacon/settings.py +++ b/core/services/beacon/settings.py @@ -1,37 +1,29 @@ import json import re import socket -from typing import Any, Dict, List +from typing import Any, Dict, List, Literal, Optional import psutil -from commonwealth.settings import settings +from commonwealth.settings.settings import PydanticSettings from loguru import logger -from pykson import ( - IntegerField, - JsonObject, - ListField, - MultipleChoiceStringField, - ObjectField, - ObjectListField, - StringField, -) +from pydantic import BaseModel, Field class InvalidIpAddress(Exception): pass -class DefaultSettings(JsonObject): - domain_names = ListField(item_type=str) - advertise = ListField(item_type=str) - ip = StringField() +class DefaultSettings(BaseModel): + domain_names: List[str] + advertise: List[str] + ip: str -class Interface(JsonObject): - name = StringField() - domain_names = ListField(item_type=str) - advertise = ListField(item_type=str) - ip = StringField() +class Interface(BaseModel): + name: str + domain_names: List[str] + advertise: List[str] + ip: str def get_phys(self) -> List[psutil._common.snicaddr]: """ @@ -83,11 +75,11 @@ def __repr__(self) -> str: return str(self.name) -class ServiceTypes(JsonObject): - name = StringField() - protocol = MultipleChoiceStringField(options=["_udp", "_tcp"], null=False) - port = IntegerField() - properties = StringField() +class ServiceTypes(BaseModel): + name: str + protocol: Literal["_udp", "_tcp"] + port: int + properties: Optional[str] = None def get_properties(self) -> Dict[str, Any]: if self.properties is None: @@ -95,26 +87,20 @@ def get_properties(self) -> Dict[str, Any]: return json.loads(self.properties) # type: ignore -class SettingsV1(settings.BaseSettings): - VERSION = 1 - default = ObjectField(DefaultSettings) - blacklist = ListField(item_type=str) - interfaces = ObjectListField(Interface) - advertisement_types = ObjectListField(ServiceTypes) - - def __init__(self, *args: str, **kwargs: int) -> None: - super().__init__(*args, **kwargs) - - self.VERSION = SettingsV1.VERSION +class SettingsV1(PydanticSettings): + default: DefaultSettings = Field(default_factory=lambda: DefaultSettings(domain_names=[], advertise=[], ip="")) + blacklist: List[str] = Field(default_factory=list) + interfaces: List[Interface] = Field(default_factory=list) + advertisement_types: List[ServiceTypes] = Field(default_factory=list) def migrate(self, data: Dict[str, Any]) -> None: - if data["VERSION"] == SettingsV1.VERSION: + if data["VERSION"] == SettingsV1.STATIC_VERSION: return - if data["VERSION"] < SettingsV1.VERSION: + if data["VERSION"] < SettingsV1.STATIC_VERSION: super().migrate(data) - data["VERSION"] = SettingsV1.VERSION + data["VERSION"] = SettingsV1.STATIC_VERSION def get_interface_or_create_default(self, name: str) -> Interface: """ @@ -133,17 +119,11 @@ def get_interface_or_create_default(self, name: str) -> Interface: class SettingsV2(SettingsV1): - VERSION = 2 - - def __init__(self, *args: str, **kwargs: int) -> None: - super().__init__(*args, **kwargs) - self.VERSION = SettingsV2.VERSION - 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["default"]["domain_names"] = [domain for domain in data["default"]["domain_names"] if domain != "blueos"] @@ -156,47 +136,36 @@ def migrate(self, data: Dict[str, Any]) -> None: except Exception as e: logger.error(f"unable to update SettingsV1 to SettingsV2: {e}") - data["VERSION"] = SettingsV2.VERSION + data["VERSION"] = SettingsV2.STATIC_VERSION class SettingsV3(SettingsV2): - VERSION = 3 - - def __init__(self, *args: str, **kwargs: int) -> None: - super().__init__(*args, **kwargs) - self.VERSION = SettingsV3.VERSION - def migrate(self, data: Dict[str, Any]) -> None: - if data["VERSION"] == SettingsV3.VERSION: + if data["VERSION"] == SettingsV3.STATIC_VERSION: return - if data["VERSION"] < SettingsV3.VERSION: + if data["VERSION"] < SettingsV3.STATIC_VERSION: super().migrate(data) try: if not any(interface["name"] == "uap0" for interface in data["interfaces"]): data["interfaces"].append( - Interface(name="uap0", domain_names=["blueos-hotspot"], advertise=["_http"], ip="ips[0]")._data + Interface(name="uap0", domain_names=["blueos-hotspot"], advertise=["_http"], ip="ips[0]").dict() ) except Exception as e: logger.error(f"unable to update SettingsV2 to SettingsV3: {e}") - data["VERSION"] = SettingsV3.VERSION + data["VERSION"] = SettingsV3.STATIC_VERSION class SettingsV4(SettingsV3): - VERSION = 4 - vehicle_name = StringField() - - def __init__(self, *args: str, **kwargs: int) -> None: - super().__init__(*args, **kwargs) - self.VERSION = SettingsV4.VERSION + vehicle_name: str = "" def migrate(self, data: Dict[str, Any]) -> None: - if data["VERSION"] == SettingsV4.VERSION: + if data["VERSION"] == SettingsV4.STATIC_VERSION: return - if data["VERSION"] < SettingsV4.VERSION: + if data["VERSION"] < SettingsV4.STATIC_VERSION: super().migrate(data) - data["VERSION"] = SettingsV4.VERSION + data["VERSION"] = SettingsV4.STATIC_VERSION diff --git a/core/services/bridget/bridget.py b/core/services/bridget/bridget.py index 2031e3e3d7..d8a953b002 100644 --- a/core/services/bridget/bridget.py +++ b/core/services/bridget/bridget.py @@ -5,7 +5,7 @@ import requests from bridges.bridges import Bridge from bridges.serialhelper import Baudrate -from commonwealth.settings.manager import Manager +from commonwealth.settings.manager import PydanticManager from pydantic import BaseModel, conint from serial.tools.list_ports_linux import SysFS from settings import BridgeSettingsSpecV2, SettingsV2 @@ -48,7 +48,9 @@ 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") + self._settings_manager: PydanticManager = PydanticManager( + "bridget", SettingsV2, USERDATA / "settings" / "bridget" + ) self._settings_manager.load() for bridge_settings_spec in self._settings_manager.settings.specsv2: try: diff --git a/core/services/bridget/settings.py b/core/services/bridget/settings.py index c1eaa45242..b89ad79aae 100644 --- a/core/services/bridget/settings.py +++ b/core/services/bridget/settings.py @@ -1,14 +1,14 @@ -from typing import Any, Dict +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, Field -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 @@ -25,31 +25,25 @@ def __eq__(self, other: object) -> Any: 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 +class SettingsV1(PydanticSettings): + specs: List[BridgeSettingsSpecV1] = Field(default_factory=list) def migrate(self, data: Dict[str, Any]) -> None: - if data["VERSION"] == SettingsV1.VERSION: + if data["VERSION"] == SettingsV1.STATIC_VERSION: return - if data["VERSION"] < SettingsV1.VERSION: + if data["VERSION"] < SettingsV1.STATIC_VERSION: super().migrate(data) - data["VERSION"] = SettingsV1.VERSION + data["VERSION"] = SettingsV1.STATIC_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 @@ -68,22 +62,16 @@ def __eq__(self, other: object) -> Any: class SettingsV2(SettingsV1): - VERSION = 2 - specsv2 = pykson.ObjectListField(BridgeSettingsSpecV2) - - def __init__(self, *args: str, **kwargs: int) -> None: - super().__init__(*args, **kwargs) - - self.VERSION = SettingsV2.VERSION + specsv2: List[BridgeSettingsSpecV2] = Field(default_factory=list) 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" diff --git a/core/services/kraken/extension/extension.py b/core/services/kraken/extension/extension.py index 53257a2649..7f032be4fe 100644 --- a/core/services/kraken/extension/extension.py +++ b/core/services/kraken/extension/extension.py @@ -18,7 +18,7 @@ ) from aiodocker.exceptions import DockerError -from commonwealth.settings.manager import Manager +from commonwealth.settings.manager import PydanticManager from config import DEFAULT_INJECTED_ENV_VARIABLES, SERVICE_NAME from extension.exceptions import ( ExtensionInsufficientStorage, @@ -51,7 +51,7 @@ class Extension: start_attempts: Dict[str, Tuple[int, int]] = {} temp_extension_activity: Dict[str, Tuple[int, int]] = {} - _manager: Manager = Manager(SERVICE_NAME, SettingsV2) + _manager: PydanticManager = PydanticManager(SERVICE_NAME, SettingsV2) _settings = _manager.settings def __init__(self, source: ExtensionSource, digest: Optional[str] = None) -> None: @@ -95,9 +95,7 @@ def reset_start_attempt(cls, key: str) -> None: cls.start_attempts.pop(key, None) @classmethod - def _fetch_settings( - cls, identifier: Optional[str] = None, tag: Optional[str] = None - ) -> List[ExtensionSettings] | ExtensionSettings: + def _fetch_settings(cls, identifier: Optional[str] = None, tag: Optional[str] = None) -> List[ExtensionSettings]: extensions: List[ExtensionSettings] = [ ext for ext in cast(List[ExtensionSettings], cls._settings.extensions) @@ -107,7 +105,7 @@ def _fetch_settings( if identifier is not None and tag is not None: if not extensions: raise ExtensionNotFound(f"Extension {identifier}:{tag} not found") - return extensions[0] + return [extensions[0]] return extensions def _save_settings(self, extension: Optional[ExtensionSettings] = None) -> None: diff --git a/core/services/kraken/extension_logs.py b/core/services/kraken/extension_logs.py index 084e78b82f..2d9e22fb0d 100644 --- a/core/services/kraken/extension_logs.py +++ b/core/services/kraken/extension_logs.py @@ -1,7 +1,7 @@ import asyncio import json import time -from typing import Callable, Dict, List, Optional, Tuple, cast +from typing import Callable, Dict, Optional, Tuple from commonwealth.utils.zenoh_helper import ZenohSession from config import SERVICE_NAME @@ -51,7 +51,7 @@ async def _collect_desired_streams(self) -> Optional[Dict[str, ExtensionSettings return None running_names = {container.name.lstrip("/") for container in running_containers} - extensions = cast(List[ExtensionSettings], Extension._fetch_settings()) + extensions = Extension._fetch_settings() desired: Dict[str, ExtensionSettings] = {} for extension in extensions: diff --git a/core/services/kraken/kraken.py b/core/services/kraken/kraken.py index fbad037ab2..d8f91ad64e 100644 --- a/core/services/kraken/kraken.py +++ b/core/services/kraken/kraken.py @@ -4,7 +4,7 @@ from typing import Any, List import aiohttp -from commonwealth.settings.manager import Manager +from commonwealth.settings.manager import PydanticManager from config import DEFAULT_EXTENSIONS, SERVICE_NAME from extension.exceptions import IncompatibleExtension from extension.extension import Extension @@ -21,7 +21,7 @@ class Kraken: def __init__(self) -> None: - self._manager: Manager = Manager(SERVICE_NAME, SettingsV2) + self._manager: PydanticManager = PydanticManager(SERVICE_NAME, SettingsV2) self._settings = self._manager.settings self.is_running = True self.manifest = ManifestManager.instance() diff --git a/core/services/kraken/manifest/manifest.py b/core/services/kraken/manifest/manifest.py index 3fbb33bc27..0989903bcc 100644 --- a/core/services/kraken/manifest/manifest.py +++ b/core/services/kraken/manifest/manifest.py @@ -6,7 +6,7 @@ import aiohttp import semver from aiocache import cached -from commonwealth.settings.manager import Manager +from commonwealth.settings.manager import PydanticManager from config import DEFAULT_MANIFESTS, SERVICE_NAME from manifest.exceptions import ( ManifestBackendOffline, @@ -33,7 +33,7 @@ class ManifestManager: """ _instance: Optional["ManifestManager"] = None - _manager: Manager = Manager(SERVICE_NAME, SettingsV2) + _manager: PydanticManager = PydanticManager(SERVICE_NAME, SettingsV2) _settings = _manager.settings def __init__(self) -> None: diff --git a/core/services/kraken/settings.py b/core/services/kraken/settings.py index e71fe28a42..163f6707c3 100644 --- a/core/services/kraken/settings.py +++ b/core/services/kraken/settings.py @@ -1,19 +1,19 @@ import json import re -from typing import Any, Dict +from typing import Any, Dict, Sequence -from commonwealth.settings import settings -from pykson import BooleanField, IntegerField, JsonObject, ObjectListField, StringField +from commonwealth.settings.settings import PydanticSettings +from pydantic import BaseModel, Field -class ExtensionSettings(JsonObject): - identifier = StringField() - name = StringField() - docker = StringField() - tag = StringField() - permissions = StringField() - enabled = BooleanField() - user_permissions = StringField() +class ExtensionSettings(BaseModel): + identifier: str + name: str + docker: str + tag: str + permissions: str + enabled: bool + user_permissions: str def settings(self) -> Any: if self.user_permissions: @@ -33,49 +33,37 @@ def container_name(self) -> str: return "extension-" + regex.sub("", f"{self.docker}{self.tag}") -class ManifestSettings(JsonObject): - identifier = StringField() - enabled = BooleanField() - priority = IntegerField() - factory = BooleanField() - name = StringField() - url = StringField() +class ManifestSettings(BaseModel): + identifier: str + enabled: bool + priority: int + factory: bool + name: str + url: str -class SettingsV1(settings.BaseSettings): - VERSION = 1 - extensions = ObjectListField(ExtensionSettings) - - def __init__(self, *args: str, **kwargs: int) -> None: - super().__init__(*args, **kwargs) - - self.VERSION = SettingsV1.VERSION +class SettingsV1(PydanticSettings): + extensions: Sequence[ExtensionSettings] = Field(default_factory=list) def migrate(self, data: Dict[str, Any]) -> None: - if data["VERSION"] == SettingsV1.VERSION: + if data["VERSION"] == SettingsV1.STATIC_VERSION: return - if data["VERSION"] < SettingsV1.VERSION: + if data["VERSION"] < SettingsV1.STATIC_VERSION: super().migrate(data) - data["VERSION"] = SettingsV1.VERSION + data["VERSION"] = SettingsV1.STATIC_VERSION class SettingsV2(SettingsV1): - VERSION = 2 - manifests = ObjectListField(ManifestSettings) - - def __init__(self, *args: str, **kwargs: int) -> None: - super().__init__(*args, **kwargs) - - self.VERSION = SettingsV2.VERSION + manifests: Sequence[ManifestSettings] = Field(default_factory=list) 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["manifests"] = [] diff --git a/core/services/nmea_injector/nmea_injector/TrafficController.py b/core/services/nmea_injector/nmea_injector/TrafficController.py index 1f15a3c210..a199ce2a34 100644 --- a/core/services/nmea_injector/nmea_injector/TrafficController.py +++ b/core/services/nmea_injector/nmea_injector/TrafficController.py @@ -6,7 +6,7 @@ import pynmea2 from commonwealth.mavlink_comm.MavlinkComm import MavlinkMessenger -from commonwealth.settings.manager import Manager +from commonwealth.settings.manager import PydanticManager from loguru import logger from nmea_injector.exceptions import UnsupportedSocketKind from nmea_injector.MavlinkNMEA import MavlinkGpsInput, parse_mavlink_from_sentence @@ -96,7 +96,7 @@ class TrafficController: def __init__(self) -> None: self._socks: Dict[NMEASocket, Union[asyncio.AbstractServer, asyncio.BaseTransport]] = {} - self._settings_manager = Manager("nmea-injector", SettingsV1) + self._settings_manager: PydanticManager = PydanticManager("nmea-injector", SettingsV1) async def load_socks_from_settings(self) -> None: self._settings_manager.load() diff --git a/core/services/nmea_injector/nmea_injector/settings.py b/core/services/nmea_injector/nmea_injector/settings.py index fcb26a43cc..c74c3f7c59 100644 --- a/core/services/nmea_injector/nmea_injector/settings.py +++ b/core/services/nmea_injector/nmea_injector/settings.py @@ -1,13 +1,13 @@ -from typing import Any, Dict +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, Field -class NmeaInjectorSettingsSpecV1(pykson.JsonObject): - kind = pykson.StringField() - port = pykson.IntegerField() - component_id = pykson.IntegerField() +class NmeaInjectorSettingsSpecV1(BaseModel): + kind: str + port: int + component_id: int def __eq__(self, other: object) -> Any: if isinstance(other, NmeaInjectorSettingsSpecV1): @@ -15,20 +15,14 @@ def __eq__(self, other: object) -> Any: return False -class SettingsV1(settings.BaseSettings): - VERSION = 1 - specs = pykson.ObjectListField(NmeaInjectorSettingsSpecV1) - - def __init__(self, *args: str, **kwargs: int) -> None: - super().__init__(*args, **kwargs) - - self.VERSION = SettingsV1.VERSION +class SettingsV1(PydanticSettings): + specs: List[NmeaInjectorSettingsSpecV1] = Field(default_factory=list) def migrate(self, data: Dict[str, Any]) -> None: - if data["VERSION"] == SettingsV1.VERSION: + if data["VERSION"] == SettingsV1.STATIC_VERSION: return - if data["VERSION"] < SettingsV1.VERSION: + if data["VERSION"] < SettingsV1.STATIC_VERSION: super().migrate(data) - data["VERSION"] = SettingsV1.VERSION + data["VERSION"] = SettingsV1.STATIC_VERSION diff --git a/core/services/ping/ping1d_driver.py b/core/services/ping/ping1d_driver.py index d765b1c668..c4cb31e9d1 100644 --- a/core/services/ping/ping1d_driver.py +++ b/core/services/ping/ping1d_driver.py @@ -3,7 +3,7 @@ from bridges.bridges import Bridge from bridges.serialhelper import Baudrate -from commonwealth.settings.manager import Manager +from commonwealth.settings.manager import PydanticManager from loguru import logger from ping1d_mavlink import Ping1DMavlinkDriver from pingdriver import PingDriver @@ -19,7 +19,7 @@ class Ping1DDriver(PingDriver): def __init__(self, ping: PingDeviceDescriptor, port: int) -> None: super().__init__(ping, port) # load settings - self.manager = Manager(SERVICE_NAME, SettingsV1, USERDATA / "settings" / SERVICE_NAME) + self.manager: PydanticManager = PydanticManager(SERVICE_NAME, SettingsV1, USERDATA / "settings" / SERVICE_NAME) # our settings file is a list for each sensor type. # check the list to find our current sensor in it connection_info = self.ping.get_hw_or_eth_info() diff --git a/core/services/ping/settings.py b/core/services/ping/settings.py index 9c715caebc..67550c4f0c 100644 --- a/core/services/ping/settings.py +++ b/core/services/ping/settings.py @@ -1,12 +1,12 @@ -from typing import Any, Dict +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, Field -class Ping1dSettingsSpecV1(pykson.JsonObject): - port = pykson.StringField() - mavlink_enabled = pykson.BooleanField() +class Ping1dSettingsSpecV1(BaseModel): + port: str + mavlink_enabled: bool def __str__(self) -> str: return f"{self.port} - {self.mavlink_enabled}" @@ -16,21 +16,15 @@ def new(port: str, enabled: bool) -> "Ping1dSettingsSpecV1": return Ping1dSettingsSpecV1(port=port, mavlink_enabled=enabled) -class SettingsV1(settings.BaseSettings): - VERSION = 1 - ping1d_specs = pykson.ObjectListField(Ping1dSettingsSpecV1) +class SettingsV1(PydanticSettings): + ping1d_specs: List[Ping1dSettingsSpecV1] = Field(default_factory=list) # no settings for ping360 as of V1 - 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: + if data["VERSION"] == SettingsV1.STATIC_VERSION: return - if data["VERSION"] < SettingsV1.VERSION: + if data["VERSION"] < SettingsV1.STATIC_VERSION: super().migrate(data) - data["VERSION"] = SettingsV1.VERSION + data["VERSION"] = SettingsV1.STATIC_VERSION diff --git a/core/services/wifi/settings.py b/core/services/wifi/settings.py index 8260d0f90b..79dad61418 100644 --- a/core/services/wifi/settings.py +++ b/core/services/wifi/settings.py @@ -1,26 +1,19 @@ from typing import Any, Dict -import pykson # type: ignore -from commonwealth.settings import settings +from commonwealth.settings.settings import PydanticSettings -class SettingsV1(settings.BaseSettings): - VERSION = 1 - hotspot_enabled = pykson.BooleanField() - hotspot_ssid = pykson.StringField() - hotspot_password = pykson.StringField() - smart_hotspot_enabled = pykson.BooleanField() - - def __init__(self, *args: str, **kwargs: int) -> None: - super().__init__(*args, **kwargs) - - self.VERSION = SettingsV1.VERSION +class SettingsV1(PydanticSettings): + hotspot_enabled: bool = False + hotspot_ssid: str = "BlueOS" + hotspot_password: str = "blueosap" + smart_hotspot_enabled: bool = False def migrate(self, data: Dict[str, Any]) -> None: - if data["VERSION"] == SettingsV1.VERSION: + if data["VERSION"] == SettingsV1.STATIC_VERSION: return - if data["VERSION"] < SettingsV1.VERSION: + if data["VERSION"] < SettingsV1.STATIC_VERSION: super().migrate(data) - data["VERSION"] = SettingsV1.VERSION + data["VERSION"] = SettingsV1.STATIC_VERSION diff --git a/core/services/wifi/wifi_handlers/AbstractWifiHandler.py b/core/services/wifi/wifi_handlers/AbstractWifiHandler.py index 930298d575..a0fdb72eac 100644 --- a/core/services/wifi/wifi_handlers/AbstractWifiHandler.py +++ b/core/services/wifi/wifi_handlers/AbstractWifiHandler.py @@ -2,7 +2,7 @@ from argparse import ArgumentParser, Namespace from typing import List, Optional -from commonwealth.settings.manager import Manager +from commonwealth.settings.manager import PydanticManager from settings import SettingsV1 from typedefs import SavedWifiNetwork, ScannedWifiNetwork, WifiCredentials, WifiStatus from wifi_handlers.wpa_supplicant.wpa_supplicant import WPASupplicant @@ -12,7 +12,7 @@ class AbstractWifiManager: wpa = WPASupplicant() def __init__(self) -> None: - self._settings_manager = Manager("wifi-manager", SettingsV1) + self._settings_manager: PydanticManager = PydanticManager("wifi-manager", SettingsV1) self._settings_manager.load() @abc.abstractmethod diff --git a/core/uv.lock b/core/uv.lock index 6c28ffd4d2..7d7be2b3ee 100644 --- a/core/uv.lock +++ b/core/uv.lock @@ -452,7 +452,6 @@ dependencies = [ { name = "packaging" }, { name = "psutil" }, { name = "pydantic" }, - { name = "pykson" }, { name = "sentry-sdk" }, { name = "starlette" }, ] @@ -466,7 +465,6 @@ requires-dist = [ { name = "packaging", specifier = ">=20.4,<=25.0" }, { name = "psutil", specifier = ">=5.7.2,<=7.1.3" }, { name = "pydantic", specifier = ">=1.10.12,<=2.12.5" }, - { name = "pykson", url = "https://github.com/patrickelectric/pykson/archive/refs/tags/1.0.2.tar.gz" }, { name = "sentry-sdk", specifier = "==2.29.1" }, { name = "starlette", specifier = ">=0.27.0" }, ] @@ -670,27 +668,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d9/47/0ec3ec948b7b3a0ba44e62adede4dca8b5985ba6aaee59998bed0916bd17/isort-5.8.0-py3-none-any.whl", hash = "sha256:2bb1680aad211e3c9944dbce1d4ba09a989f04e238296c87fe2139faa26d655d", size = 103227 }, ] -[[package]] -name = "jalali-core" -version = "1.0.0" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/2b/3c/21e32e3444c572174a5d774643eb2aa8ab60ef68b99a4c3585a0a11428b4/jalali_core-1.0.0.tar.gz", hash = "sha256:f4287c70c630323dcf0a3ab26df905ba4d451e230ac1f65b3bb2f77797894a2b", size = 2752 } -wheels = [ - { url = "https://files.pythonhosted.org/packages/f6/20/a4e942f9685df720a106da292e29a53212b27903749cc563b86b612b113e/jalali_core-1.0.0-py3-none-any.whl", hash = "sha256:84e6f5090eadfb35234f24fad084be831d00da3c0b238ee001e8a1fd49bf7924", size = 3616 }, -] - -[[package]] -name = "jdatetime" -version = "5.2.0" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "jalali-core" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/6e/9d/5ed59c36f3cbc68c01fab6442e6efb6d35a484ba4eec4f790264fce39f6c/jdatetime-5.2.0.tar.gz", hash = "sha256:c81d5898717b82b609a3ce2a73f8b8d3230b0c757e5c0de9d6b1acfdc224f551", size = 21663 } -wheels = [ - { url = "https://files.pythonhosted.org/packages/16/39/0dd2676d08468692606645db5ea40091290dc20747ff59636c21c0567d3c/jdatetime-5.2.0-py3-none-any.whl", hash = "sha256:d4aa73543e4e6c0e6122b58743773168edee5efe5c5acf05d1dc8c90524ca71c", size = 12199 }, -] - [[package]] name = "loguru" version = "0.5.3" @@ -1127,26 +1104,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d6/ae/c5b11e64bcafb6f293ac353fff966c3b62b964409fc18a90a262ad90e0ee/pyfakefs-5.2.4-py3-none-any.whl", hash = "sha256:8eb95f1dd1c4b8bdce30448fe169875e3a4451c32d3f9c37799157bd4eb7b789", size = 207449 }, ] -[[package]] -name = "pykson" -version = "1.0.2" -source = { url = "https://github.com/patrickelectric/pykson/archive/refs/tags/1.0.2.tar.gz" } -dependencies = [ - { name = "jdatetime" }, - { name = "python-dateutil" }, - { name = "pytz" }, - { name = "six" }, -] -sdist = { hash = "sha256:dbd3b8a3c6e8e874ccfc365551fff200a34893bee17f590b1e5f775cd83c9e8f" } - -[package.metadata] -requires-dist = [ - { name = "jdatetime", specifier = ">=3.6.2" }, - { name = "python-dateutil", specifier = ">=2.8.0" }, - { name = "pytz", specifier = ">=2019.3" }, - { name = "six", specifier = ">=1.12.0" }, -] - [[package]] name = "pylint" version = "3.0.0" @@ -1306,18 +1263,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ca/31/d4e37e9e550c2b92a9cbc2e4d0b7420a27224968580b5a447f420847c975/pytest_xdist-3.8.0-py3-none-any.whl", hash = "sha256:202ca578cfeb7370784a8c33d6d05bc6e13b4f25b5053c30a152269fd10f0b88", size = 46396 }, ] -[[package]] -name = "python-dateutil" -version = "2.9.0.post0" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "six" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/66/c0/0c8b6ad9f17a802ee498c46e004a0eb49bc148f2fd230864601a86dcf6db/python-dateutil-2.9.0.post0.tar.gz", hash = "sha256:37dd54208da7e1cd875388217d5e00ebd4179249f90fb72437e91a35459a0ad3", size = 342432 } -wheels = [ - { url = "https://files.pythonhosted.org/packages/ec/57/56b9bcc3c9c6a792fcbaf139543cee77261f3651ca9da0c93f5c1221264b/python_dateutil-2.9.0.post0-py2.py3-none-any.whl", hash = "sha256:a8b2bc7bffae282281c8140a97d3aa9c14da0b136dfe83f850eea9a5f7470427", size = 229892 }, -] - [[package]] name = "python-multipart" version = "0.0.5" @@ -1327,15 +1272,6 @@ dependencies = [ ] sdist = { url = "https://files.pythonhosted.org/packages/46/40/a933ac570bf7aad12a298fc53458115cc74053474a72fbb8201d7dc06d3d/python-multipart-0.0.5.tar.gz", hash = "sha256:f7bb5f611fc600d15fa47b3974c8aa16e93724513b49b5f95c81e6624c83fa43", size = 32581 } -[[package]] -name = "pytz" -version = "2025.2" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/f8/bf/abbd3cdfb8fbc7fb3d4d38d320f2441b1e7cbe29be4f23797b4a2b5d8aac/pytz-2025.2.tar.gz", hash = "sha256:360b9e3dbb49a209c21ad61809c7fb453643e048b38924c765813546746e81c3", size = 320884 } -wheels = [ - { url = "https://files.pythonhosted.org/packages/81/c4/34e93fe5f5429d7570ec1fa436f1986fb1f00c3e0f43a589fe2bbcd22c3f/pytz-2025.2-py2.py3-none-any.whl", hash = "sha256:5ddf76296dd8c44c26eb8f4b6f35488f3ccbf6fbbd7adee0b7262d43f0ec2f00", size = 509225 }, -] - [[package]] name = "recorder-extractor" version = "0.1.0"