feature: demonstrate settings api

synapse/cli/__main__.py

@@ -17,6 +17,7 @@
     rpc,
     streaming,
     taps,
+    settings,
 )
 from synapse.utils.discover import find_device_by_name
 
@@ -77,6 +78,7 @@ def main():
     taps.add_commands(subparsers)
     deploy.add_commands(subparsers)
     build.add_commands(subparsers)
+    settings.add_commands(subparsers)
     args = parser.parse_args()
 
     # If we need to setup the device URI, do that now

synapse/cli/settings.py

@@ -0,0 +1,73 @@
+import synapse as syn
+from synapse.client import settings
+from rich.console import Console
+from rich.table import Table
+
+
+def add_commands(subparsers):
+    parser = subparsers.add_parser(
+        "settings", help="Manage the persistent device settings"
+    )
+
+    settings_subparsers = parser.add_subparsers(title="Settings")
+
+    get_parser = settings_subparsers.add_parser(
+        "get", help="Get the current settings for a device"
+    )
+    get_parser.set_defaults(func=get_settings)
+
+    set_parser = settings_subparsers.add_parser(
+        "set", help="Set a setting key to a value"
+    )
+    set_parser.add_argument("key", help="The key to set")
+    set_parser.add_argument("value", help="The value to set")
+    set_parser.set_defaults(func=set_setting)
+
+
+def get_settings(args):
+    console = Console()
+
+    try:
+        with console.status("Getting settings", spinner="bouncingBall"):
+            device = syn.Device(args.uri, args.verbose)
+            settings_dict = settings.get_all_settings(device)
+
+        if not settings_dict:
+            console.print(
+                "[yellow]No settings have been configured (all are at default values)[/yellow]"
+            )
+            console.print("\n[dim]Available settings:[/dim]")
+            available = settings.get_available_settings()
+            for name, type_name in available.items():
+                console.print(f"  [cyan]{name}[/cyan] ({type_name})")
+            return
+
+        # Create and populate the settings table
+        settings_table = Table(title="Current Settings", show_lines=True)
+        settings_table.add_column("Setting", style="cyan")
+        settings_table.add_column("Value", style="green")
+
+        for key, value in settings_dict.items():
+            settings_table.add_row(key, str(value))
+
+        console.print(settings_table)
+
+    except Exception as e:
+        console.print(f"[bold red]{e}[/bold red]")
+
+
+def set_setting(args):
+    console = Console()
+
+    try:
+        with console.status("Setting settings", spinner="bouncingBall"):
+            device = syn.Device(args.uri, args.verbose)
+            updated_value = settings.set_setting(device, args.key, args.value)
+
+        console.print(
+            f"[bold green]Setting updated successfully: {args.key} = {args.value}[/bold green]"
+        )
+        console.print(f"[dim]Confirmed value: {updated_value}[/dim]")
+
+    except Exception as e:
+        console.print(f"[bold red]{e}[/bold red]")

synapse/client/__init__.py

@@ -1,3 +1,5 @@
+# linter doesn't know about the imports, so ignore this
+# ruff: noqa: F401
 from synapse.client.node import Node
 from synapse.client.config import Config
 from synapse.client.device import Device

synapse/client/device.py

@@ -14,6 +14,10 @@
 from synapse.api.query_pb2 import StreamQueryRequest, StreamQueryResponse
 from synapse.api.status_pb2 import StatusCode, Status
 from synapse.api.synapse_pb2_grpc import SynapseDeviceStub
+from synapse.api.device_pb2 import (
+    UpdateDeviceSettingsRequest,
+    UpdateDeviceSettingsResponse,
+)
 from synapse.client.config import Config
 from synapse.utils.log import log_level_to_pb
 
@@ -185,6 +189,16 @@ def stream_query(
             self.logger.error(f"Error during StreamQuery: {str(e)}")
             yield StreamQueryResponse(code=StatusCode.kQueryFailed)
 
+    def update_device_settings(
+        self, request: UpdateDeviceSettingsRequest
+    ) -> Optional[UpdateDeviceSettingsResponse]:
+        try:
+            return self.rpc.UpdateDeviceSettings(request)
+
+        except Exception as e:
+            self.logger.error(f"Error during update settings: {str(e)}")
+            return None
+
     def _handle_status_response(self, status):
         if status.code != StatusCode.kOk:
             self.logger.error("Error %d: %s", status.code, status.message)

synapse/client/settings.py

@@ -0,0 +1,223 @@
+from __future__ import annotations
+from typing import Dict, Any, TYPE_CHECKING
+from google.protobuf.descriptor import FieldDescriptor
+
+from synapse.api.device_pb2 import DeviceSettings, UpdateDeviceSettingsRequest
+from synapse.api.query_pb2 import QueryRequest
+
+if TYPE_CHECKING:
+    from synapse.client.device import Device
+
+
+def get_all_settings(device: "Device") -> Dict[str, Any]:
+    """
+    Get all non-default settings from device as a dictionary.
+
+    Args:
+        device: Device instance to fetch settings from
+
+    Returns:
+        Dictionary of setting names to values
+
+    Raises:
+        RuntimeError: If failed to fetch settings from device
+    """
+    request = QueryRequest(
+        query_type=QueryRequest.QueryType.kGetSettings, get_settings_query={}
+    )
+    response = device.query(request)
+
+    if not response or response.status.code != 0:
+        error_msg = response.status.message if response else "Unknown error"
+        raise RuntimeError(f"Failed to get settings from device: {error_msg}")
+
+    settings_proto = response.get_settings_response.settings
+    settings_dict = {}
+
+    for field in settings_proto.DESCRIPTOR.fields:
+        field_name = field.name
+        field_value = getattr(settings_proto, field_name)
+
+        # Check if field has non-default value
+        if _has_non_default_value(settings_proto, field, field_value):
+            settings_dict[field_name] = field_value
+
+    return settings_dict
+
+
+def get_setting(device: "Device", key: str) -> Any:
+    """
+    Get a specific setting value from device.
+
+    Args:
+        device: Device instance
+        key: Setting name
+
+    Returns:
+        Setting value
+
+    Raises:
+        RuntimeError: If failed to fetch settings
+        KeyError: If setting doesn't exist
+    """
+    request = QueryRequest(
+        query_type=QueryRequest.QueryType.kGetSettings, get_settings_query={}
+    )
+    response = device.query(request)
+
+    if not response or response.status.code != 0:
+        error_msg = response.status.message if response else "Unknown error"
+        raise RuntimeError(f"Failed to get settings from device: {error_msg}")
+
+    settings_proto = response.get_settings_response.settings
+
+    if not _has_field(settings_proto, key):
+        available_fields = [field.name for field in settings_proto.DESCRIPTOR.fields]
+        raise KeyError(
+            f"Setting '{key}' not found. Available settings: {available_fields}"
+        )
+
+    return getattr(settings_proto, key)
+
+
+def set_setting(device: "Device", key: str, value: Any) -> Any:
+    """
+    Set a specific setting value on device.
+
+    Args:
+        device: Device instance
+        key: Setting name
+        value: Setting value
+
+    Returns:
+        The actual value that was set (after any device processing)
+
+    Raises:
+        RuntimeError: If failed to update settings
+        KeyError: If setting doesn't exist
+        ValueError: If value is invalid for the setting type
+    """
+    # Create a new settings proto with just this field set
+    settings_proto = DeviceSettings()
+
+    field_descriptor = _get_field_descriptor(settings_proto, key)
+    if not field_descriptor:
+        available_fields = [field.name for field in settings_proto.DESCRIPTOR.fields]
+        raise KeyError(
+            f"Setting '{key}' not found. Available settings: {available_fields}"
+        )
+
+    # Convert and validate value based on field type
+    converted_value = _convert_and_validate_value(field_descriptor, value)
+    setattr(settings_proto, key, converted_value)
+
+    # Send to device
+    request = UpdateDeviceSettingsRequest(settings=settings_proto)
+    response = device.update_device_settings(request)
+
+    if not response or response.status.code != 0:
+        error_msg = response.status.message if response else "Unknown error"
+        raise RuntimeError(f"Failed to update settings on device: {error_msg}")
+
+    # Return the actual value that was set
+    return getattr(response.updated_settings, key)
+
+
+def get_available_settings() -> Dict[str, str]:
+    """
+    Get all available setting names and their types.
+
+    Returns:
+        Dictionary mapping setting names to their protobuf type names
+    """
+    settings_proto = DeviceSettings()
+    return {
+        field.name: _get_field_type_name(field)
+        for field in settings_proto.DESCRIPTOR.fields
+    }
+
+
+# Helper functions
+def _has_field(settings_proto: DeviceSettings, field_name: str) -> bool:
+    """Check if a field exists in the protobuf."""
+    return any(field.name == field_name for field in settings_proto.DESCRIPTOR.fields)
+
+
+def _get_field_descriptor(
+    settings_proto: DeviceSettings, field_name: str
+) -> FieldDescriptor:
+    """Get field descriptor by name."""
+    for field in settings_proto.DESCRIPTOR.fields:
+        if field.name == field_name:
+            return field
+    return None
+
+
+def _has_non_default_value(
+    settings_proto: DeviceSettings, field: FieldDescriptor, value: Any
+) -> bool:
+    """Check if a field has a non-default value."""
+    # For message fields, use HasField to check presence
+    if field.type == field.TYPE_MESSAGE:
+        return settings_proto.HasField(field.name)
+
+    # For scalar fields, check if value is not default
+    if field.type == field.TYPE_STRING:
+        return value != ""
+    elif field.type in [
+        field.TYPE_INT32,
+        field.TYPE_UINT32,
+        field.TYPE_INT64,
+        field.TYPE_UINT64,
+    ]:
+        return value != 0
+    elif field.type in [field.TYPE_FLOAT, field.TYPE_DOUBLE]:
+        return value != 0.0
+    elif field.type == field.TYPE_BOOL:
+        return value is True
+    else:
+        # For other types, always display
+        return True
+
+
+def _convert_and_validate_value(field: FieldDescriptor, value: Any) -> Any:
+    """Convert and validate a value for a specific field type."""
+    try:
+        if field.type == field.TYPE_STRING:
+            return str(value)
+        elif field.type in [field.TYPE_INT32, field.TYPE_UINT32]:
+            return int(value)
+        elif field.type in [field.TYPE_INT64, field.TYPE_UINT64]:
+            return int(value)
+        elif field.type in [field.TYPE_FLOAT, field.TYPE_DOUBLE]:
+            return float(value)
+        elif field.type == field.TYPE_BOOL:
+            if isinstance(value, bool):
+                return value
+            elif isinstance(value, str):
+                return value.lower() in ("true", "1", "yes", "on")
+            else:
+                return bool(value)
+        else:
+            # For other types, try direct assignment
+            return value
+    except (ValueError, TypeError) as e:
+        raise ValueError(
+            f"Invalid value '{value}' for field '{field.name}' of type {_get_field_type_name(field)}: {e}"
+        )
+
+
+def _get_field_type_name(field: FieldDescriptor) -> str:
+    """Get human-readable field type name."""
+    type_names = {
+        field.TYPE_STRING: "string",
+        field.TYPE_INT32: "int32",
+        field.TYPE_UINT32: "uint32",
+        field.TYPE_INT64: "int64",
+        field.TYPE_UINT64: "uint64",
+        field.TYPE_FLOAT: "float",
+        field.TYPE_DOUBLE: "double",
+        field.TYPE_BOOL: "bool",
+        field.TYPE_MESSAGE: "message",
+    }
+    return type_names.get(field.type, f"unknown({field.type})")