Replaced internal encoders with message encoder calling public API

packages/service/ni_measurement_plugin_sdk_service/_internal/grpc_servicer.py

@@ -205,7 +205,7 @@ def GetMetadata(  # noqa: N802 - function name should be lowercase
             measurement_signature.configuration_parameters.append(configuration_parameter)
 
         measurement_signature.configuration_defaults.value = encoder.serialize_default_values(
-            self._configuration_metadata, self._get_service_name() + ".Inputs"
+            self._configuration_metadata, self._service_info.service_class + ".Configurations"
         )
 
         for field_number, output_metadata in self._output_metadata.items():
@@ -238,7 +238,7 @@ def Measure(  # noqa: N802 - function name should be lowercase
         mapping_by_id = decoder.deserialize_parameters(
             self._configuration_metadata,
             request.configuration_parameters.value,
-            self._get_service_name() + ".Inputs",
+            self._service_info.service_class + ".Configurations",
         )
         mapping_by_variable_name = _get_mapping_by_parameter_name(
             mapping_by_id, self._measure_function
@@ -265,17 +265,16 @@ def Measure(  # noqa: N802 - function name should be lowercase
             measurement_service_context.get().mark_complete()
             measurement_service_context.reset(token)
 
-    def _serialize_response(self, outputs: Any) -> v1_measurement_service_pb2.MeasureResponse:
+    def _serialize_response(
+        self,
+        outputs: Any,
+    ) -> v1_measurement_service_pb2.MeasureResponse:
         return v1_measurement_service_pb2.MeasureResponse(
             outputs=_serialize_outputs(
-                self._output_metadata, outputs, self._get_service_name() + ".Outputs"
+                self._output_metadata, outputs, self._service_info.service_class + ".Outputs"
             )
         )
 
-    def _get_service_name(self) -> str:
-        service_name = "".join(char for char in self._service_info.service_class if char.isalpha())
-        return service_name
-
 
 class MeasurementServiceServicerV2(v2_measurement_service_pb2_grpc.MeasurementServiceServicer):
     """Measurement v2 servicer."""
@@ -323,7 +322,7 @@ def GetMetadata(  # noqa: N802 - function name should be lowercase
             measurement_signature.configuration_parameters.append(configuration_parameter)
 
         measurement_signature.configuration_defaults.value = encoder.serialize_default_values(
-            self._configuration_metadata, self._get_service_name() + ".Inputs"
+            self._configuration_metadata, self._service_info.service_class + ".Configurations"
         )
 
         for field_number, output_metadata in self._output_metadata.items():
@@ -358,7 +357,7 @@ def Measure(  # noqa: N802 - function name should be lowercase
         mapping_by_id = decoder.deserialize_parameters(
             self._configuration_metadata,
             request.configuration_parameters.value,
-            self._get_service_name() + ".Inputs",
+            self._service_info.service_class + ".Configurations",
         )
         mapping_by_variable_name = _get_mapping_by_parameter_name(
             mapping_by_id, self._measure_function
@@ -387,10 +386,6 @@ def Measure(  # noqa: N802 - function name should be lowercase
     def _serialize_response(self, outputs: Any) -> v2_measurement_service_pb2.MeasureResponse:
         return v2_measurement_service_pb2.MeasureResponse(
             outputs=_serialize_outputs(
-                self._output_metadata, outputs, self._get_service_name() + ".Outputs"
+                self._output_metadata, outputs, self._service_info.service_class + ".Outputs"
             )
         )
-
-    def _get_service_name(self) -> str:
-        service_name = "".join(char for char in self._service_info.service_class if char.isalpha())
-        return service_name

packages/service/ni_measurement_plugin_sdk_service/_internal/parameter/_get_type.py

@@ -0,0 +1,36 @@
+from typing import Any
+
+from google.protobuf.descriptor_pb2 import FieldDescriptorProto
+from google.protobuf.type_pb2 import Field
+
+_TYPE_DEFULAT_MAPPING = {
+    Field.TYPE_FLOAT: float(),
+    Field.TYPE_DOUBLE: float(),
+    Field.TYPE_INT32: int(),
+    Field.TYPE_INT64: int(),
+    Field.TYPE_UINT32: int(),
+    Field.TYPE_UINT64: int(),
+    Field.TYPE_BOOL: bool(),
+    Field.TYPE_STRING: str(),
+    Field.TYPE_ENUM: int(),
+}
+
+_TYPE_FIELD_MAPPING = {
+    Field.TYPE_FLOAT: FieldDescriptorProto.TYPE_FLOAT,
+    Field.TYPE_DOUBLE: FieldDescriptorProto.TYPE_DOUBLE,
+    Field.TYPE_INT32: FieldDescriptorProto.TYPE_INT32,
+    Field.TYPE_INT64: FieldDescriptorProto.TYPE_INT64,
+    Field.TYPE_UINT32: FieldDescriptorProto.TYPE_UINT32,
+    Field.TYPE_UINT64: FieldDescriptorProto.TYPE_UINT64,
+    Field.TYPE_BOOL: FieldDescriptorProto.TYPE_BOOL,
+    Field.TYPE_STRING: FieldDescriptorProto.TYPE_STRING,
+    Field.TYPE_ENUM: FieldDescriptorProto.TYPE_ENUM,
+    Field.TYPE_MESSAGE: FieldDescriptorProto.TYPE_MESSAGE,
+}
+
+
+def get_type_default(type: Field.Kind.ValueType, repeated: bool) -> Any:
+    """Get the default value for the give type."""
+    if repeated:
+        return list()
+    return _TYPE_DEFULAT_MAPPING.get(type)

packages/service/ni_measurement_plugin_sdk_service/_internal/parameter/decoder.py

@@ -1,12 +1,10 @@
 """Parameter Serializer."""
 
-from json import loads
-from typing import Any, Dict
+from typing import Any, Dict, List, Union
 
 from google.protobuf import descriptor_pool, message_factory
 from google.protobuf.descriptor_pb2 import FieldDescriptorProto
 
-from ni_measurement_plugin_sdk_service._annotations import ENUM_VALUES_KEY
 from ni_measurement_plugin_sdk_service._internal.parameter.metadata import (
     ParameterMetadata,
 )
@@ -30,7 +28,7 @@ def deserialize_parameters(
         service_name (str): Unique service name.
 
     Returns:
-        Dict[int, Any]: Deserialized parameters by ID
+        Dict[int, Any]: Deserialized parameters by ID.
     """
     pool = descriptor_pool.Default()
     message_proto = pool.FindMessageTypeByName(service_name)
@@ -40,14 +38,14 @@ def deserialize_parameters(
     message_instance.ParseFromString(parameter_bytes)
     for i in message_proto.fields_by_number.keys():
         parameter_metadata = parameter_metadata_dict[i]
-        field_name = parameter_metadata.sanitized_display_name()
+        field_name = parameter_metadata.field_name
+        enum_type = _get_enum_type(parameter_metadata)
         value = getattr(message_instance, field_name)
 
-        if (
-            parameter_metadata.type == FieldDescriptorProto.TYPE_ENUM
-            and _get_enum_type(parameter_metadata) is not int
-        ):
-            parameter_values[i] = _deserialize_enum_parameter(parameter_metadata, value)
+        if parameter_metadata.type == FieldDescriptorProto.TYPE_ENUM and enum_type is not int:
+            parameter_values[i] = _deserialize_enum_parameter(
+                parameter_metadata.repeated, value, enum_type
+            )
         elif (
             parameter_metadata.type == FieldDescriptorProto.TYPE_MESSAGE
             and not parameter_metadata.repeated
@@ -59,39 +57,14 @@ def deserialize_parameters(
     return parameter_values
 
 
-def _deserialize_enum_parameter(parameter_metadata: ParameterMetadata, field_value: Any) -> Any:
+def _deserialize_enum_parameter(
+    repeated: bool, field_value: Any, enum_type: Any
+) -> Union[List[Any], Any]:
     """Convert all enums into the user defined enum type.
 
-    Args:
-        parameter_metadata (ParameterMetadata): Metadata of current enum value.
-
-        field_value (Any): Value of current field.
-
-    Returns:
-        Any: Enum type or a list of enum types.
-    """
-    enum_dict = loads(parameter_metadata.annotations[ENUM_VALUES_KEY])
-    enum_type = _get_enum_type(parameter_metadata)
-    if parameter_metadata.repeated:
-        return [_get_enum_field(enum_dict, enum_type, value) for value in field_value]
-    else:
-        return _get_enum_field(enum_dict, enum_type, field_value)
-
-
-def _get_enum_field(enum_dict: Dict[Any, int], enum_type: Any, field_value: int) -> Any:
-    """Get enum type and value from 'field_value'.
-
-    Args:
-        enum_dict (Dict[Any, int]): List enum class of 'field_value'.
-
-        enum_type (Any): 'field_value' enum class name.
-
-        field_value (int): Default value of current field.
-
     Returns:
-        Any: Enum type of 'field_value' from 'enum_dict' with the enum value.
+        Union[List[Any], Any]: Enum type or a list of enum types.
     """
-    for name in enum_dict.keys():
-        enum_value = getattr(enum_type, name)
-        if field_value == enum_value.value:
-            return enum_value
+    if repeated:
+        return [enum_type(value) for value in field_value]
+    return enum_type(field_value)

packages/service/ni_measurement_plugin_sdk_service/_internal/parameter/encoder.py

@@ -6,7 +6,7 @@
 from google.protobuf import descriptor_pool, message_factory
 from google.protobuf.descriptor_pb2 import FieldDescriptorProto
 
-from ni_measurement_plugin_sdk_service._internal.parameter.default_value import (
+from ni_measurement_plugin_sdk_service._internal.parameter._get_type import (
     get_type_default,
 )
 from ni_measurement_plugin_sdk_service._internal.parameter.metadata import (
@@ -37,7 +37,7 @@ def serialize_parameters(
 
     for i, parameter in enumerate(parameter_values, start=1):
         parameter_metadata = parameter_metadata_dict[i]
-        field_name = parameter_metadata.sanitized_display_name()
+        field_name = parameter_metadata.field_name
         parameter = _get_enum_values(param=parameter)
         type_default_value = get_type_default(parameter_metadata.type, parameter_metadata.repeated)
 
@@ -74,14 +74,7 @@ def serialize_default_values(
 
 
 def _get_enum_values(param: Any) -> Any:
-    """Get's value of an enum.
-
-    Args:
-        param (Any): A value/parameter of parameter_values.
-
-    Returns:
-        Any: An enum value or a list of enums or the 'param'.
-    """
+    """Get's value of an enum."""
     if param == []:
         return param
     if isinstance(param, list) and isinstance(param[0], Enum):

packages/service/ni_measurement_plugin_sdk_service/_internal/parameter/metadata.py

@@ -10,7 +10,7 @@
     ENUM_VALUES_KEY,
     TYPE_SPECIALIZATION_KEY,
 )
-from ni_measurement_plugin_sdk_service._internal.parameter.default_value import (
+from ni_measurement_plugin_sdk_service._internal.parameter._get_type import (
     get_type_default,
 )
 from ni_measurement_plugin_sdk_service.measurement.info import TypeSpecialization
@@ -42,14 +42,30 @@ class ParameterMetadata(NamedTuple):
     Required when 'type' is Kind.TypeMessage. Ignored for any other 'type'.
     """
 
-    def sanitized_display_name(self) -> str:
-        """Parameter display name of alpha/numerical characters.
+    field_name: str = ""
+    """display_name in snake_case format."""
 
-        Returns:
-            str: Alpha/numerical characters of 'display_name'.
 
-        """
-        return "".join(char for char in self.display_name if char.isalnum())
+def initialize(
+    display_name: str,
+    type: type_pb2.Field.Kind.ValueType,
+    repeated: bool,
+    default_value: Any,
+    annotations: Dict[str, str],
+    message_type: str = "",
+) -> ParameterMetadata:
+    """Initialize ParameterMetadata with field_name."""
+    underscore_display_name = display_name.replace(" ", "_")
+    if all(char.isalnum() or char == "_" for char in underscore_display_name):
+        field_name = underscore_display_name
+    else:
+        field_name = "".join(
+            char for char in underscore_display_name if char.isalnum() or char == "_"
+        )
+
+    return ParameterMetadata(
+        display_name, type, repeated, default_value, annotations, message_type, field_name
+    )
 
 
 def validate_default_value_type(parameter_metadata: ParameterMetadata) -> None: