Add support for builtin types for get_value and set_value

CONTRIBUTING.md

@@ -43,6 +43,9 @@ poetry run nps lint
 poetry run mypy
 poetry run bandit -c pyproject.toml -r src/nipanel
 
+# Apply safe fixes
+poetry run nps fix
+
 # Run the tests
 poetry run pytest -v
 

examples/placeholder.py

@@ -1 +1,58 @@
 """Placeholder example for the package."""
+
+import enum
+
+import nipanel
+
+
+class MyIntFlags(enum.IntFlag):
+    """Example of an IntFlag enum."""
+
+    VALUE1 = 1
+    VALUE2 = 2
+    VALUE4 = 4
+
+
+class MyIntEnum(enum.IntEnum):
+    """Example of an IntEnum enum."""
+
+    VALUE10 = 10
+    VALUE20 = 20
+    VALUE30 = 30
+
+
+class MyStrEnum(str, enum.Enum):
+    """Example of a mixin string enum."""
+
+    VALUE1 = "value1"
+    VALUE2 = "value2"
+    VALUE3 = "value3"
+
+
+if __name__ == "__main__":
+    my_panel = nipanel.StreamlitPanel(
+        panel_id="placeholder",
+        streamlit_script_uri=__file__,
+    )
+
+    my_types = {
+        "my_str": "im justa smol str",
+        "my_int": 42,
+        "my_float": 13.12,
+        "my_bool": True,
+        "my_bytes": b"robotext",
+        "my_intflags": MyIntFlags.VALUE1 | MyIntFlags.VALUE4,
+        "my_intenum": MyIntEnum.VALUE20,
+        "my_strenum": MyStrEnum.VALUE3,
+    }
+
+    print("Setting values")
+    for name, value in my_types.items():
+        print(f"{name:>15}   {value}")
+        my_panel.set_value(name, value)
+
+    print()
+    print("Getting values")
+    for name in my_types.keys():
+        the_value = my_panel.get_value(name)
+        print(f"{name:>15}   {the_value}")

pyproject.toml

@@ -67,6 +67,7 @@ strict = true
 module = [
   # https://github.com/ni/hightime/issues/4 - Add type annotations
   "hightime.*",
+  "grpc.framework.foundation.*",
 ]
 ignore_missing_imports = true
 

src/nipanel/_converters.py

@@ -0,0 +1,230 @@
+"""Functions to convert between different data formats."""
+
+from __future__ import annotations
+
+import logging
+from abc import ABC, abstractmethod
+from typing import Any, Generic, Type, TypeVar
+
+from google.protobuf import any_pb2, wrappers_pb2
+from google.protobuf.message import Message
+
+_TPythonType_co = TypeVar("_TPythonType_co", covariant=True)
+_TProtobufType_co = TypeVar("_TProtobufType_co", bound=Message, covariant=True)
+
+_logger = logging.getLogger(__name__)
+
+
+class Converter(Generic[_TPythonType_co, _TProtobufType_co], ABC):
+    """A class that defines how to convert between Python objects and protobuf Any messages."""
+
+    @property
+    @abstractmethod
+    def python_type(self) -> Type[_TPythonType_co]:
+        """The Python type that this converter handles."""
+
+    @property
+    @abstractmethod
+    def protobuf_message(self) -> Type[_TProtobufType_co]:
+        """The type-specific protobuf message for the Python type."""
+
+    @property
+    def protobuf_typename(self) -> str:
+        """The protobuf name for the type."""
+        return self.protobuf_message.DESCRIPTOR.full_name  # type: ignore[no-any-return]
+
+    @abstractmethod
+    def to_protobuf(self, python_value: Any) -> any_pb2.Any:
+        """Convert the Python object to its type-specific message and pack it as any_pb2.Any."""
+
+    @abstractmethod
+    def to_python(self, protobuf_value: any_pb2.Any) -> _TPythonType_co:
+        """Convert the protobuf message to its matching Python type."""
+
+
+class BoolConverter(Converter[bool, wrappers_pb2.BoolValue]):
+    """A converter for boolean types."""
+
+    @property
+    def python_type(self) -> Type[bool]:
+        """The Python type that this converter handles."""
+        return bool
+
+    @property
+    def protobuf_message(self) -> Type[wrappers_pb2.BoolValue]:
+        """The type-specific protobuf message for the Python type."""
+        return wrappers_pb2.BoolValue
+
+    def to_protobuf(self, python_value: bool) -> any_pb2.Any:
+        """Convert a bool to a protobuf Any."""
+        wrapped_value = self.protobuf_message(value=python_value)
+        as_any = any_pb2.Any()
+        as_any.Pack(wrapped_value)
+        return as_any
+
+    def to_python(self, protobuf_value: any_pb2.Any) -> bool:
+        """Convert the protobuf message to a Python bool."""
+        protobuf_message = self.protobuf_message()
+        did_unpack = protobuf_value.Unpack(protobuf_message)
+        if not did_unpack:
+            raise ValueError(f"Failed to unpack Any with type '{protobuf_value.TypeName()}'")
+        return protobuf_message.value
+
+
+class BytesConverter(Converter[bytes, wrappers_pb2.BytesValue]):
+    """A converter for byte string types."""
+
+    @property
+    def python_type(self) -> Type[bytes]:
+        """The Python type that this converter handles."""
+        return bytes
+
+    @property
+    def protobuf_message(self) -> Type[wrappers_pb2.BytesValue]:
+        """The type-specific protobuf message for the Python type."""
+        return wrappers_pb2.BytesValue
+
+    def to_protobuf(self, python_value: bytes) -> any_pb2.Any:
+        """Convert bytes to a protobuf Any."""
+        wrapped_value = self.protobuf_message(value=python_value)
+        as_any = any_pb2.Any()
+        as_any.Pack(wrapped_value)
+        return as_any
+
+    def to_python(self, protobuf_value: any_pb2.Any) -> bytes:
+        """Convert the protobuf message to Python bytes."""
+        protobuf_message = self.protobuf_message()
+        did_unpack = protobuf_value.Unpack(protobuf_message)
+        if not did_unpack:
+            raise ValueError(f"Failed to unpack Any with type '{protobuf_value.TypeName()}'")
+        return protobuf_message.value
+
+
+class FloatConverter(Converter[float, wrappers_pb2.DoubleValue]):
+    """A converter for floating point types."""
+
+    @property
+    def python_type(self) -> Type[float]:
+        """The Python type that this converter handles."""
+        return float
+
+    @property
+    def protobuf_message(self) -> Type[wrappers_pb2.DoubleValue]:
+        """The type-specific protobuf message for the Python type."""
+        return wrappers_pb2.DoubleValue
+
+    def to_protobuf(self, python_value: float) -> any_pb2.Any:
+        """Convert a float to a protobuf Any."""
+        wrapped_value = self.protobuf_message(value=python_value)
+        as_any = any_pb2.Any()
+        as_any.Pack(wrapped_value)
+        return as_any
+
+    def to_python(self, protobuf_value: any_pb2.Any) -> float:
+        """Convert the protobuf message to a Python float."""
+        protobuf_message = self.protobuf_message()
+        did_unpack = protobuf_value.Unpack(protobuf_message)
+        if not did_unpack:
+            raise ValueError(f"Failed to unpack Any with type '{protobuf_value.TypeName()}'")
+        return protobuf_message.value
+
+
+class IntConverter(Converter[int, wrappers_pb2.Int64Value]):
+    """A converter for integer types."""
+
+    @property
+    def python_type(self) -> Type[int]:
+        """The Python type that this converter handles."""
+        return int
+
+    @property
+    def protobuf_message(self) -> Type[wrappers_pb2.Int64Value]:
+        """The type-specific protobuf message for the Python type."""
+        return wrappers_pb2.Int64Value
+
+    def to_protobuf(self, python_value: int) -> any_pb2.Any:
+        """Convert an int to a protobuf Any."""
+        wrapped_value = self.protobuf_message(value=python_value)
+        as_any = any_pb2.Any()
+        as_any.Pack(wrapped_value)
+        return as_any
+
+    def to_python(self, protobuf_value: any_pb2.Any) -> int:
+        """Convert the protobuf message to a Python int."""
+        protobuf_message = self.protobuf_message()
+        did_unpack = protobuf_value.Unpack(protobuf_message)
+        if not did_unpack:
+            raise ValueError(f"Failed to unpack Any with type '{protobuf_value.TypeName()}'")
+        return protobuf_message.value
+
+
+class StrConverter(Converter[str, wrappers_pb2.StringValue]):
+    """A converter for text string types."""
+
+    @property
+    def python_type(self) -> Type[str]:
+        """The Python type that this converter handles."""
+        return str
+
+    @property
+    def protobuf_message(self) -> Type[wrappers_pb2.StringValue]:
+        """The type-specific protobuf message for the Python type."""
+        return wrappers_pb2.StringValue
+
+    def to_protobuf(self, python_value: str) -> any_pb2.Any:
+        """Convert a str to a protobuf Any."""
+        wrapped_value = self.protobuf_message(value=python_value)
+        as_any = any_pb2.Any()
+        as_any.Pack(wrapped_value)
+        return as_any
+
+    def to_python(self, protobuf_value: any_pb2.Any) -> str:
+        """Convert the protobuf message to a Python string."""
+        protobuf_message = self.protobuf_message()
+        did_unpack = protobuf_value.Unpack(protobuf_message)
+        if not did_unpack:
+            raise ValueError(f"Failed to unpack Any with type '{protobuf_value.TypeName()}'")
+        return protobuf_message.value
+
+
+# FFV -- consider adding a RegisterConverter mechanism
+_CONVERTIBLE_TYPES = [
+    BoolConverter(),
+    BytesConverter(),
+    FloatConverter(),
+    IntConverter(),
+    StrConverter(),
+]
+
+_CONVERTER_FOR_PYTHON_TYPE = {entry.python_type: entry for entry in _CONVERTIBLE_TYPES}
+_CONVERTER_FOR_GRPC_TYPE = {entry.protobuf_typename: entry for entry in _CONVERTIBLE_TYPES}
+_SUPPORTED_PYTHON_TYPES = _CONVERTER_FOR_PYTHON_TYPE.keys()
+
+
+def to_any(python_value: Any) -> any_pb2.Any:
+    """Convert a Python object to a protobuf Any."""
+    underlying_parents = type(python_value).mro()  # This covers enum.IntEnum and similar
+
+    best_matching_type = next(
+        (parent for parent in underlying_parents if parent in _SUPPORTED_PYTHON_TYPES), None
+    )
+    if not best_matching_type:
+        raise TypeError(
+            f"Unsupported type: {type(python_value)} with parents {underlying_parents}. Supported types are: {_SUPPORTED_PYTHON_TYPES}"
+        )
+    _logger.debug(f"Best matching type for '{repr(python_value)}' resolved to {best_matching_type}")
+
+    converter = _CONVERTER_FOR_PYTHON_TYPE[best_matching_type]
+    return converter.to_protobuf(python_value)
+
+
+def from_any(protobuf_any: any_pb2.Any) -> Any:
+    """Convert a protobuf Any to a Python object."""
+    if not isinstance(protobuf_any, any_pb2.Any):
+        raise ValueError(f"Unexpected type: {type(protobuf_any)}")
+
+    underlying_typename = protobuf_any.TypeName()
+    _logger.debug(f"Unpacking type '{underlying_typename}'")
+
+    converter = _CONVERTER_FOR_GRPC_TYPE[underlying_typename]
+    return converter.to_python(protobuf_any)

src/nipanel/_panel.py

@@ -63,8 +63,7 @@ def get_value(self, value_id: str) -> object:
         Returns:
             The value
         """
-        # TODO: AB#3095681 - get the Any from _client.get_value and convert it to the correct type
-        return "placeholder value"
+        return self._panel_client.get_value(self._panel_id, value_id)
 
     def set_value(self, value_id: str, value: object) -> None:
         """Set the value for a control on the panel.
@@ -73,5 +72,4 @@ def set_value(self, value_id: str, value: object) -> None:
             value_id: The id of the value
             value: The value
         """
-        # TODO: AB#3095681 - Convert the value to an Any and pass it to _client.set_value
-        pass
+        self._panel_client.set_value(self._panel_id, value_id, value)

src/nipanel/_panel_client.py

@@ -7,12 +7,21 @@
 from typing import Callable, TypeVar
 
 import grpc
-from ni.pythonpanel.v1.python_panel_service_pb2 import OpenPanelRequest
+from ni.pythonpanel.v1.python_panel_service_pb2 import (
+    GetValueRequest,
+    OpenPanelRequest,
+    SetValueRequest,
+)
 from ni.pythonpanel.v1.python_panel_service_pb2_grpc import PythonPanelServiceStub
 from ni_measurement_plugin_sdk_service.discovery import DiscoveryClient
 from ni_measurement_plugin_sdk_service.grpc.channelpool import GrpcChannelPool
 from typing_extensions import ParamSpec
 
+from nipanel._converters import (
+    from_any,
+    to_any,
+)
+
 _P = ParamSpec("_P")
 _T = TypeVar("_T")
 
@@ -53,6 +62,33 @@ def open_panel(self, panel_id: str, panel_uri: str) -> None:
         open_panel_request = OpenPanelRequest(panel_id=panel_id, panel_uri=panel_uri)
         self._invoke_with_retry(self._get_stub().OpenPanel, open_panel_request)
 
+    def set_value(self, panel_id: str, value_id: str, value: object) -> None:
+        """Set the value for the control with value_id.
+
+        Args:
+            panel_id: The ID of the panel.
+            value_id: The ID of the control.
+            value: The value to set.
+        """
+        new_any = to_any(value)
+        set_value_request = SetValueRequest(panel_id=panel_id, value_id=value_id, value=new_any)
+        self._invoke_with_retry(self._get_stub().SetValue, set_value_request)
+
+    def get_value(self, panel_id: str, value_id: str) -> object:
+        """Get the value for the control with value_id.
+
+        Args:
+            panel_id: The ID of the panel.
+            value_id: The ID of the control.
+
+        Returns:
+            The value.
+        """
+        get_value_request = GetValueRequest(panel_id=panel_id, value_id=value_id)
+        response = self._invoke_with_retry(self._get_stub().GetValue, get_value_request)
+        the_value = from_any(response.value)
+        return the_value
+
     def _get_stub(self) -> PythonPanelServiceStub:
         if self._stub is None:
             if self._grpc_channel is not None: