Refactor converters as a submodule

Signed-off-by: Joe Friedrichsen <[email protected]>

Author: jfriedri-ni

src/nipanel/_convert.py

@@ -0,0 +1,103 @@
+"""Functions to convert between different data formats."""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Collection
+from typing import Any
+
+from google.protobuf import any_pb2
+
+from nipanel.converters import Converter
+from nipanel.converters.builtin import (
+    BoolConverter,
+    BytesConverter,
+    FloatConverter,
+    IntConverter,
+    StrConverter,
+    BoolCollectionConverter,
+    BytesCollectionConverter,
+    FloatCollectionConverter,
+    IntCollectionConverter,
+    StrCollectionConverter,
+)
+
+_logger = logging.getLogger(__name__)
+
+# FFV -- consider adding a RegisterConverter mechanism
+_CONVERTIBLE_TYPES: list[Converter[Any, Any]] = [
+    # Scalars first
+    BoolConverter(),
+    BytesConverter(),
+    FloatConverter(),
+    IntConverter(),
+    StrConverter(),
+    # Containers next
+    BoolCollectionConverter(),
+    BytesCollectionConverter(),
+    FloatCollectionConverter(),
+    IntCollectionConverter(),
+    StrCollectionConverter(),
+]
+
+_CONVERTIBLE_COLLECTION_TYPES = {
+    frozenset,
+    list,
+    set,
+    tuple,
+}
+
+_CONVERTER_FOR_PYTHON_TYPE = {entry.python_typename: 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: object) -> any_pb2.Any:
+    """Convert a Python object to a protobuf Any."""
+    underlying_parents = type(python_value).mro()  # This covers enum.IntEnum and similar
+
+    container_type = None
+    value_is_collection = _CONVERTIBLE_COLLECTION_TYPES.intersection(underlying_parents)
+    if value_is_collection:
+        # Assume Sized -- Generators not supported, callers must use list(), set(), ... as desired
+        if not isinstance(python_value, Collection):
+            raise TypeError()
+        if len(python_value) == 0:
+            underlying_parents = type(None).mro()
+        else:
+            # Assume homogenous -- collections of mixed-types not supported
+            visitor = iter(python_value)
+            first_value = next(visitor)
+            underlying_parents = type(first_value).mro()
+        container_type = Collection
+
+    best_matching_type = None
+    candidates = [parent.__name__ for parent in underlying_parents]
+    for candidate in candidates:
+        python_typename = f"{container_type.__name__}.{candidate}" if container_type else candidate
+        if python_typename not in _SUPPORTED_PYTHON_TYPES:
+            continue
+        best_matching_type = python_typename
+        break
+
+    if not best_matching_type:
+        payload_type = underlying_parents[0]
+        raise TypeError(
+            f"Unsupported type: ({container_type}, {payload_type}) 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_any(python_value)
+
+
+def from_any(protobuf_any: any_pb2.Any) -> object:
+    """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_client.py

@@ -19,7 +19,7 @@
 from ni_measurement_plugin_sdk_service.grpc.channelpool import GrpcChannelPool
 from typing_extensions import ParamSpec
 
-from nipanel._converters import (
+from nipanel._convert import (
     from_any,
     to_any,
 )

src/nipanel/converters/__init__.py

@@ -0,0 +1,54 @@
+"""Functions and classes to convert types between Python and protobuf."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Generic, Type, TypeVar
+
+from google.protobuf import any_pb2
+from google.protobuf.message import Message
+
+_TPythonType = TypeVar("_TPythonType")
+_TProtobufType = TypeVar("_TProtobufType", bound=Message)
+
+
+class Converter(Generic[_TPythonType, _TProtobufType], ABC):
+    """A class that defines how to convert between Python objects and protobuf Any messages."""
+
+    @property
+    @abstractmethod
+    def python_typename(self) -> str:
+        """The Python type that this converter handles."""
+
+    @property
+    @abstractmethod
+    def protobuf_message(self) -> Type[_TProtobufType]:
+        """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]
+
+    def to_protobuf_any(self, python_value: _TPythonType) -> any_pb2.Any:
+        """Convert the Python object to its type-specific message and pack it as any_pb2.Any."""
+        message = self.to_protobuf_message(python_value)
+        as_any = any_pb2.Any()
+        as_any.Pack(message)
+        return as_any
+
+    @abstractmethod
+    def to_protobuf_message(self, python_value: _TPythonType) -> _TProtobufType:
+        """Convert the Python object to its type-specific message."""
+
+    def to_python(self, protobuf_value: any_pb2.Any) -> _TPythonType:
+        """Convert the protobuf Any message to its matching Python type."""
+        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 self.to_python_value(protobuf_message)
+
+    @abstractmethod
+    def to_python_value(self, protobuf_message: _TProtobufType) -> _TPythonType:
+        """Convert the protobuf wrapper message to its matching Python type."""

src/nipanel/_converters.py renamed to src/nipanel/converters/builtin.py

@@ -1,62 +1,12 @@
-"""Functions to convert between different data formats."""
+"""Classes to convert between builtin Python scalars and containers."""
 
-from __future__ import annotations
-
-import logging
-from abc import ABC, abstractmethod
 from collections.abc import Collection
-from typing import Any, Generic, Type, TypeVar
+from typing import Type
 
-from google.protobuf import any_pb2, wrappers_pb2
-from google.protobuf.message import Message
+from google.protobuf import wrappers_pb2
 from ni.pythonpanel.v1 import python_panel_types_pb2
 
-_TPythonType = TypeVar("_TPythonType")
-_TProtobufType = TypeVar("_TProtobufType", bound=Message)
-
-_logger = logging.getLogger(__name__)
-
-
-class Converter(Generic[_TPythonType, _TProtobufType], ABC):
-    """A class that defines how to convert between Python objects and protobuf Any messages."""
-
-    @property
-    @abstractmethod
-    def python_typename(self) -> str:
-        """The Python type that this converter handles."""
-
-    @property
-    @abstractmethod
-    def protobuf_message(self) -> Type[_TProtobufType]:
-        """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]
-
-    def to_protobuf_any(self, python_value: _TPythonType) -> any_pb2.Any:
-        """Convert the Python object to its type-specific message and pack it as any_pb2.Any."""
-        message = self.to_protobuf_message(python_value)
-        as_any = any_pb2.Any()
-        as_any.Pack(message)
-        return as_any
-
-    @abstractmethod
-    def to_protobuf_message(self, python_value: _TPythonType) -> _TProtobufType:
-        """Convert the Python object to its type-specific message."""
-
-    def to_python(self, protobuf_value: any_pb2.Any) -> _TPythonType:
-        """Convert the protobuf Any message to its matching Python type."""
-        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 self.to_python_value(protobuf_message)
-
-    @abstractmethod
-    def to_python_value(self, protobuf_message: _TProtobufType) -> _TPythonType:
-        """Convert the protobuf wrapper message to its matching Python type."""
+from nipanel.converters import Converter
 
 
 class BoolConverter(Converter[bool, wrappers_pb2.BoolValue]):
@@ -301,82 +251,3 @@ def to_python_value(
     ) -> Collection[str]:
         """Convert the protobuf message to a Python collection of strings."""
         return list(protobuf_value.values)
-
-
-# FFV -- consider adding a RegisterConverter mechanism
-_CONVERTIBLE_TYPES: list[Converter[Any, Any]] = [
-    # Scalars first
-    BoolConverter(),
-    BytesConverter(),
-    FloatConverter(),
-    IntConverter(),
-    StrConverter(),
-    # Containers next
-    BoolCollectionConverter(),
-    BytesCollectionConverter(),
-    FloatCollectionConverter(),
-    IntCollectionConverter(),
-    StrCollectionConverter(),
-]
-
-_CONVERTIBLE_COLLECTION_TYPES = {
-    frozenset,
-    list,
-    set,
-    tuple,
-}
-
-_CONVERTER_FOR_PYTHON_TYPE = {entry.python_typename: 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: object) -> any_pb2.Any:
-    """Convert a Python object to a protobuf Any."""
-    underlying_parents = type(python_value).mro()  # This covers enum.IntEnum and similar
-
-    container_type = None
-    value_is_collection = _CONVERTIBLE_COLLECTION_TYPES.intersection(underlying_parents)
-    if value_is_collection:
-        # Assume Sized -- Generators not supported, callers must use list(), set(), ... as desired
-        if not isinstance(python_value, Collection):
-            raise TypeError()
-        if len(python_value) == 0:
-            underlying_parents = type(None).mro()
-        else:
-            # Assume homogenous -- collections of mixed-types not supported
-            visitor = iter(python_value)
-            first_value = next(visitor)
-            underlying_parents = type(first_value).mro()
-        container_type = Collection
-
-    best_matching_type = None
-    candidates = [parent.__name__ for parent in underlying_parents]
-    for candidate in candidates:
-        python_typename = f"{container_type.__name__}.{candidate}" if container_type else candidate
-        if python_typename not in _SUPPORTED_PYTHON_TYPES:
-            continue
-        best_matching_type = python_typename
-        break
-
-    if not best_matching_type:
-        payload_type = underlying_parents[0]
-        raise TypeError(
-            f"Unsupported type: ({container_type}, {payload_type}) 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_any(python_value)
-
-
-def from_any(protobuf_any: any_pb2.Any) -> object:
-    """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)