[settings] first draft of validation and loading

Author: MrClock8163

pyproject.toml

@@ -17,6 +17,8 @@ dependencies = [
     "rapidfuzz",
     "jsonschema",
     "jmespath",
+    "toml",
+    "PyYAML",
     "click-extra"
 ]
 classifiers = [
@@ -62,6 +64,8 @@ linting = [
     "types-pyserial",
     "types-jsonschema",
     "types-jmespath",
+    "types-toml",
+    "types-PyYAML",
     "mypy",
     "flake8"
 ]

src/instrumentman/__init__.py

@@ -13,6 +13,7 @@
 from . import inclination
 from . import filetransfer
 from . import jobs
+from . import settings
 
 
 @extra_group("iman", params=None)  # type: ignore[misc]
@@ -65,6 +66,8 @@ def cli_download() -> None:
 
 cli.add_command(morse.cli)
 cli.add_command(terminal.cli)
+cli.add_command(settings.cli_load)
+cli.add_command(settings.cli_save)
 cli_measure.add_command(setmeasurement.cli_measure)
 cli_measure.add_command(setup.cli_measure)
 cli_measure.add_command(inclination.cli_measure)
@@ -75,6 +78,7 @@ def cli_download() -> None:
 cli_merge.add_command(setmeasurement.cli_merge)
 cli_merge.add_command(inclination.cli_merge)
 cli_validate.add_command(setmeasurement.cli_validate)
+cli_validate.add_command(settings.cli_validate)
 cli_import.add_command(setup.cli_import)
 cli_list.add_command(filetransfer.cli_list)
 cli_list.add_command(jobs.cli_list)

src/instrumentman/schema_targets.json

@@ -1,5 +1,5 @@
 {
-    "title": "GeoComPy set measurement targets schema",
+    "title": "Instrumentman set measurement targets schema",
     "description": "Target point records for set measurements",
     "tpye": "object",
     "additionalProperties": false,

src/instrumentman/setmeasurement/schema_session.json

@@ -1,5 +1,5 @@
 {
-    "title": "GeoComPy set measurement session schema",
+    "title": "Instrumentman set measurement session schema",
     "description": "Data recorded during a set measurement session",
     "type": "object",
     "additionalProperties": false,

src/instrumentman/settings/__init__.py

@@ -0,0 +1,104 @@
+from typing import Any
+
+from click_extra import (
+    extra_command,
+    argument,
+    option,
+    file_path,
+    Choice
+)
+from cloup import constraint
+from cloup.constraints import If, IsSet, require_all
+
+from ..utils import (
+    com_port_argument,
+    com_option_group
+)
+
+
+@extra_command(
+    "save",
+    params=None,
+    context_settings={"auto_envvar_prefix": None}
+)  # type: ignore[misc]
+@com_port_argument()
+@argument(
+    "file",
+    help="file to save settings to",
+    type=file_path(readable=False)
+)
+@com_option_group()
+@option(
+    "-f",
+    "--format",
+    help="settings file format",
+    type=Choice(["auto", "json", "yaml", "toml"], case_sensitive=False),
+    default="auto"
+)
+@option(
+    "--save-all",
+    help="save every setting, even if not applicable to the instrument",
+    is_flag=True
+)
+@option(
+    "--add-defaults",
+    help="add defaults for settings that cannot be saved",
+    is_flag=True
+)
+@constraint(
+    If(IsSet("add_defaults"), require_all),
+    ["add_defaults", "save_all"]
+)
+def cli_save(**kwargs: Any) -> None:
+    """Save instrument settings to file."""
+    ...
+
+
+@extra_command(
+    "load",
+    params=None,
+    context_settings={"auto_envvar_prefix": None}
+)  # type: ignore[misc]
+@com_port_argument()
+@argument(
+    "settings",
+    help="file containing instrument settings",
+    type=file_path(exists=True, readable=True)
+)
+@com_option_group()
+@option(
+    "-f",
+    "--format",
+    help="settings file format",
+    type=Choice(["auto", "json", "yaml", "toml"], case_sensitive=False),
+    default="auto"
+)
+def cli_load(**kwargs: Any) -> None:
+    """Load instrument settings from file."""
+    from .load import main
+
+    main(**kwargs)
+
+
+@extra_command(
+    "settings",
+    params=None,
+    context_settings={"auto_envvar_prefix": None}
+)  # type: ignore[misc]
+@argument(
+    "file",
+    help="settings file to validate",
+    type=file_path(exists=True, readable=True)
+)
+@option(
+    "-f",
+    "--format",
+    help="settings file format",
+    type=Choice(["auto", "json", "yaml", "toml"], case_sensitive=False),
+    default="auto"
+)
+def cli_validate(**kwargs: Any) -> None:
+    """Validate instrument settings config."""
+    from .validate import main
+
+    main(**kwargs)

src/instrumentman/settings/io.py

@@ -0,0 +1,60 @@
+from typing import Any, cast
+from pathlib import Path
+
+import json
+import yaml
+import toml
+
+from ..utils import echo_red
+
+
+_SettingsDict = dict[str, Any]
+
+
+def read_settings(
+    file: Path,
+    format: str = "auto"
+) -> _SettingsDict:
+    if format == "auto":
+        format = file.suffix[1:].lower() if file.suffix else ""
+
+    match format:
+        case "json":
+            with file.open("rt", encoding="utf8") as settings:
+                data = cast(_SettingsDict, json.load(settings))
+        case "yaml" | "yml":
+            with file.open("rt", encoding="utf8") as settings:
+                data = cast(_SettingsDict, yaml.load(settings, yaml.Loader))
+        case "toml":
+            # The TOML package doesn't support heterogenous arrays, even tho it
+            # was added to the language spec in 2019. Therefore the standard
+            # lib tomllib/tomli has to be used for parsing.
+            import tomllib as toml
+            with file.open("rb") as settings:
+                data = cast(_SettingsDict, toml.load(settings))
+        case _:
+            echo_red(f"Unknown file format: {format}")
+            exit(1)
+
+    return data
+
+
+def write_settings(
+    data: _SettingsDict,
+    file: Path,
+    format: str = "auto"
+) -> None:
+    if format == "auto":
+        format = file.suffix[1:].lower() if file.suffix else ""
+
+    with file.open("wt", encoding="utf8") as settings:
+        match format:
+            case "json":
+                json.dump(data, settings)
+            case "yaml" | "yml":
+                yaml.dump(data, settings, yaml.Dumper)
+            case "toml":
+                toml.dump(data, settings)
+            case _:
+                echo_red(f"Unknown file format: {format}")
+                exit(1)

src/instrumentman/settings/load.py

@@ -0,0 +1,139 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Callable, TypedDict
+
+from geocompy.communication import open_serial
+from geocompy.geo import GeoCom
+from geocompy.geo.gctypes import GeoComSubsystem, GeoComResponse, GeoComCode
+from geocompy.gsi.dna import GsiOnlineDNA
+from geocompy.gsi.gsitypes import GsiOnlineResponse
+from geocompy.gsi.dna.settings import GsiOnlineDNASettings
+
+from ..utils import echo_red, echo_yellow
+from .io import read_settings
+from .validate import validate_settings
+
+
+class _SubsystemSettingsDict(TypedDict):
+    subsystem: str
+    options: dict[str, Any]
+
+
+class _SettingsDict(TypedDict):
+    protocol: str
+    settings: list[_SubsystemSettingsDict]
+
+
+def set_setting_geocom(
+    system: GeoComSubsystem,
+    setting: str,
+    value: int | float | bool | str | list[int | float | bool | str]
+) -> None:
+    if isinstance(value, bool):
+        name = f"switch_{setting}"
+    else:
+        name = f"set_{setting}"
+
+    method: Callable[
+        ...,
+        GeoComResponse[Any]
+    ] | None = getattr(system, name)
+    if method is None:
+        echo_yellow(f"Could not find '{name}' to set '{setting}'")
+        return
+
+    if isinstance(value, list):
+        response = method(*value)
+    else:
+        response = method(value)
+
+    if response.error != GeoComCode.OK:
+        echo_yellow(f"Could not set '{setting}' ({response.error.name})")
+        return
+
+
+def set_setting_gsidna(
+    system: GsiOnlineDNASettings,
+    setting: str,
+    value: int | float | bool | str
+) -> None:
+    name = f"set_{setting}"
+    method: Callable[
+        ...,
+        GsiOnlineResponse[bool]
+    ] | None = getattr(system, name)
+    if method is None:
+        echo_yellow(f"Could not find '{name}' to set '{setting}'")
+        return
+
+    response = method(value)
+
+    if response.value is None or not response.value:
+        echo_yellow(f"Could not set '{setting}' ({response.response})")
+        return
+
+
+def upload_settings_geocom(
+    protocol: GeoCom,
+    settings: _SettingsDict
+) -> None:
+    for item in settings["settings"]:
+        sysname = item["subsystem"]
+        subsystem: Any = getattr(protocol, sysname)
+        if subsystem is None:
+            echo_red(f"Could not find '{sysname}' subsystem")
+            exit(1)
+
+        for option, value in item["options"].items():
+            if value is None:
+                continue
+
+            set_setting_geocom(subsystem, option, value)
+
+
+def upload_settings_gsidna(
+    protocol: GsiOnlineDNA,
+    settings: _SettingsDict
+) -> None:
+    for item in settings["settings"]:
+        sysname = item["subsystem"]
+        subsystem: Any = getattr(protocol, sysname)
+        if subsystem is None:
+            echo_red(f"Could not find '{sysname}' subsystem")
+            exit(1)
+
+        for option, value in item["options"].items():
+            if value is None:
+                continue
+
+            set_setting_gsidna(subsystem, option, value)
+
+
+def main(
+    port: str,
+    settings: Path,
+    baud: int = 9600,
+    timeout: int = 15,
+    retry: int = 1,
+    sync_after_timeout: bool = False,
+    format: str = "auto"
+) -> None:
+    data = read_settings(settings, format)
+    if not validate_settings(data):
+        echo_red("Settings file does not follow schema")
+        exit(1)
+
+    with open_serial(
+        port,
+        retry=retry,
+        sync_after_timeout=sync_after_timeout,
+        speed=baud,
+        timeout=timeout
+    ) as com:
+        if "geocom" in data:
+            tps = GeoCom(com)
+            upload_settings_geocom(tps, data["geocom"])
+        elif "gsidna" in data:
+            dna = GsiOnlineDNA(com)
+            upload_settings_gsidna(dna, data["gsidna"])