Add encoder and decoder for DoubleXYData (#358)

* Add message_type to datatypeinfo and ParameterMetadata. Add DoubleXYData type enum. No parsers for it yet.

* ni-python-styleguide fix

* Modify generate stubs to work when test asset .proto files reference ni types in stubs.

* Add serializer and deserializer for DoubleXYData with tests

* Fix lint and styleguide errors

* Get tests passing

* Fix mypy errors

* Put underscore before some private methods.

* Move helpers to _message.py

* Use ValueError

* Add unsupported encoder for Array messages

* Update to _EncodeVarint and _DecodeVarint copied directly

* Remove unnecessary local

* Add type hints for encoder and decoder

* Get tests passing

* Remove unnecessary method

* Brad's review feedback

* Couple more minor feedback items

* Review comments

* Review feedback, type hints etc

* One more type hint

* Add return type hint to _encode_message

Author: dixonjoel

ni_measurementlink_service/_internal/parameter/_message.py

@@ -0,0 +1,121 @@
+import struct
+from typing import Any, Callable, Dict, Optional, Tuple, Type, TypeVar
+
+from google.protobuf.internal import encoder, wire_format
+from google.protobuf.message import Message
+
+from ni_measurementlink_service._internal.parameter._serializer_types import (
+    Decoder,
+    Key,
+    NewDefault,
+    WriteFunction,
+)
+
+
+def _message_encoder_constructor(
+    field_index: int, is_repeated: bool, is_packed: bool
+) -> Callable[[WriteFunction, Message, bool], int]:
+    """Mimics google.protobuf.internal.MessageEncoder.
+
+    This function was forked in order to call SerializeToString instead of _InternalSerialize.
+
+    _InternalSerialize is only defined for the pure-Python protobuf implementation. Our child
+    messages (like DoubleXYData) are defined in .proto files, so they use whichever protobuf
+    implementation that google.protobuf.internal.api_implementation chooses (usually upb).
+    """
+    tag = encoder.TagBytes(field_index, wire_format.WIRETYPE_LENGTH_DELIMITED)
+    encode_varint = _varint_encoder()
+
+    def _encode_message(write: WriteFunction, value: Message, deterministic: bool) -> int:
+        write(tag)
+        bytes = value.SerializeToString()
+        encode_varint(write, len(bytes), deterministic)
+        return write(bytes)
+
+    return _encode_message
+
+
+def _varint_encoder() -> Callable[[WriteFunction, int, Optional[bool]], int]:
+    """Return an encoder for a basic varint value (does not include tag).
+
+    From google.protobuf.internal.encoder.py _VarintEncoder
+    """
+    local_int2byte = struct.Struct(">B").pack
+
+    def encode_varint(
+        write: WriteFunction, value: int, unused_deterministic: Optional[bool] = None
+    ) -> int:
+        bits = value & 0x7F
+        value >>= 7
+        while value:
+            write(local_int2byte(0x80 | bits))
+            bits = value & 0x7F
+            value >>= 7
+        return write(local_int2byte(bits))
+
+    return encode_varint
+
+
+def _message_decoder_constructor(
+    field_index: int, is_repeated: bool, is_packed: bool, key: Key, new_default: NewDefault
+) -> Decoder:
+    """Mimics google.protobuf.internal.MessageDecoder.
+
+    This function was forked in order to call ParseFromString instead of _InternalParse.
+
+    _InternalParse is only defined for the pure-Python protobuf implementation. Our child messages
+    (like DoubleXYData) are defined in .proto files, so they use whichever protobuf implementation
+    that google.protobuf.internal.api_implementation chooses (usually upb).
+    """
+
+    def _decode_message(
+        buffer: memoryview, pos: int, end: int, message: Message, field_dict: Dict[Key, Any]
+    ) -> int:
+        decode_varint = _varint_decoder(mask=(1 << 64) - 1, result_type=int)
+        value = field_dict.get(key)
+        if value is None:
+            value = field_dict.setdefault(key, new_default(message))
+        # Read length.
+        (size, pos) = decode_varint(buffer, pos)
+        new_pos = pos + size
+        if new_pos > end:
+            raise ValueError("Error decoding a message. Message is truncated.")
+        parsed_bytes = value.ParseFromString(buffer[pos:new_pos])
+        if parsed_bytes != size:
+            raise ValueError("Parsed incorrect number of bytes.")
+        return new_pos
+
+    return _decode_message
+
+
+T = TypeVar("T", bound="int")
+
+
+def _varint_decoder(mask: int, result_type: Type[T]) -> Callable[[memoryview, int], Tuple[T, int]]:
+    """Return an encoder for a basic varint value (does not include tag).
+
+    Decoded values will be bitwise-anded with the given mask before being
+    returned, e.g. to limit them to 32 bits.  The returned decoder does not
+    take the usual "end" parameter -- the caller is expected to do bounds checking
+    after the fact (often the caller can defer such checking until later).  The
+    decoder returns a (value, new_pos) pair.
+
+    From google.protobuf.internal.decoder.py _VarintDecoder
+    """
+
+    def decode_varint(buffer: memoryview, pos: int) -> Tuple[T, int]:
+        result = 0
+        shift = 0
+        while 1:
+            b = buffer[pos]
+            result |= (b & 0x7F) << shift
+            pos += 1
+            if not (b & 0x80):
+                result &= mask
+                result = result_type(result)
+                return (result, pos)
+            shift += 7
+            if shift >= 64:
+                raise ValueError("Too many bytes when decoding varint: {shift}")
+
+    return decode_varint

ni_measurementlink_service/_internal/parameter/_serializer_types.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+import sys
+import typing
+from typing import Any, Callable, Dict
+
+from google.protobuf.descriptor import FieldDescriptor
+from google.protobuf.message import Message
+
+if typing.TYPE_CHECKING:
+    if sys.version_info >= (3, 10):
+        from typing import TypeAlias
+    else:
+        from typing_extensions import TypeAlias
+
+
+Key: TypeAlias = FieldDescriptor
+WriteFunction: TypeAlias = Callable[[bytes], int]
+Encoder: TypeAlias = Callable[[WriteFunction, bytes, bool], int]
+PartialEncoderConstructor: TypeAlias = Callable[[int], Encoder]
+EncoderConstructor: TypeAlias = Callable[[int, bool, bool], Encoder]
+
+Decoder: TypeAlias = Callable[[memoryview, int, int, Message, Dict[Key, Any]], int]
+PartialDecoderConstructor: TypeAlias = Callable[[int, Key], Decoder]
+NewDefault: TypeAlias = Callable[[Message], Message]
+DecoderConstructor: TypeAlias = Callable[[int, bool, bool, Key, NewDefault], Decoder]

ni_measurementlink_service/_internal/parameter/serialization_strategy.py

@@ -1,31 +1,23 @@
 """Serialization Strategy."""
 from __future__ import annotations
 
-import sys
-import typing
-from typing import Any, Callable, Dict, Optional, cast
+from typing import Any, Optional, cast
 
 from google.protobuf import type_pb2
-from google.protobuf.descriptor import FieldDescriptor
 from google.protobuf.internal import decoder, encoder
 from google.protobuf.message import Message
 
-if typing.TYPE_CHECKING:
-    if sys.version_info >= (3, 10):
-        from typing import TypeAlias
-    else:
-        from typing_extensions import TypeAlias
-
-Key: TypeAlias = FieldDescriptor
-WriteFunction: TypeAlias = Callable[[bytes], int]
-Encoder: TypeAlias = Callable[[WriteFunction, bytes, bool], int]
-PartialEncoderConstructor: TypeAlias = Callable[[int], Encoder]
-EncoderConstructor: TypeAlias = Callable[[int, bool, bool], Encoder]
-
-Decoder: TypeAlias = Callable[[memoryview, int, int, Message, Dict[Key, Any]], int]
-PartialDecoderConstructor: TypeAlias = Callable[[int, Key], Decoder]
-NewDefault: TypeAlias = Callable[[Message], Message]
-DecoderConstructor: TypeAlias = Callable[[int, bool, bool, Key, NewDefault], Decoder]
+from ni_measurementlink_service._internal.parameter import _message
+from ni_measurementlink_service._internal.parameter._serializer_types import (
+    Decoder,
+    DecoderConstructor,
+    Encoder,
+    EncoderConstructor,
+    Key,
+    PartialDecoderConstructor,
+    PartialEncoderConstructor,
+)
+from ni_measurementlink_service._internal.stubs.ni.protobuf.types import xydata_pb2
 
 
 def _scalar_encoder(encoder: EncoderConstructor) -> PartialEncoderConstructor:
@@ -63,6 +55,10 @@ def vector_encoder(field_index: int) -> Encoder:
     return vector_encoder
 
 
+def _unsupported_encoder(field_index: int, is_repeated: bool, is_packed: bool) -> Encoder:
+    raise NotImplementedError(f"Unsupported data type for field {field_index}")
+
+
 def _scalar_decoder(decoder: DecoderConstructor) -> PartialDecoderConstructor:
     """Constructs a scalar decoder constructor.
 
@@ -106,6 +102,23 @@ def vector_decoder(field_index: int, key: Key) -> Decoder:
     return vector_decoder
 
 
+def _double_xy_data_decoder(decoder: DecoderConstructor) -> PartialDecoderConstructor:
+    """Constructs a DoubleXYData decoder constructor.
+
+    Takes a field index and a key and returns a Decoder for DoubleXYData.
+    """
+
+    def _new_default(unused_message: Optional[Message] = None) -> Any:
+        return xydata_pb2.DoubleXYData()
+
+    def message_decoder(field_index: int, key: Key) -> Decoder:
+        is_repeated = True
+        is_packed = True
+        return decoder(field_index, is_repeated, is_packed, key, _new_default)
+
+    return message_decoder
+
+
 # Cast works around this issue in typeshed
 # https://github.com/python/typeshed/issues/10695
 FloatEncoder = _scalar_encoder(cast(EncoderConstructor, encoder.FloatEncoder))
@@ -114,13 +127,15 @@ def vector_decoder(field_index: int, key: Key) -> Decoder:
 UIntEncoder = _scalar_encoder(cast(EncoderConstructor, encoder.UInt32Encoder))
 BoolEncoder = _scalar_encoder(encoder.BoolEncoder)
 StringEncoder = _scalar_encoder(encoder.StringEncoder)
+MessageEncoder = _scalar_encoder(cast(EncoderConstructor, _message._message_encoder_constructor))
 
 FloatArrayEncoder = _vector_encoder(cast(EncoderConstructor, encoder.FloatEncoder))
 DoubleArrayEncoder = _vector_encoder(cast(EncoderConstructor, encoder.DoubleEncoder))
 IntArrayEncoder = _vector_encoder(cast(EncoderConstructor, encoder.Int32Encoder))
 UIntArrayEncoder = _vector_encoder(cast(EncoderConstructor, encoder.UInt32Encoder))
 BoolArrayEncoder = _vector_encoder(encoder.BoolEncoder)
 StringArrayEncoder = _vector_encoder(encoder.StringEncoder, is_packed=False)
+UnsupportedMessageArrayEncoder = _vector_encoder(_unsupported_encoder)
 
 # Cast works around this issue in typeshed
 # https://github.com/python/typeshed/issues/10697
@@ -132,6 +147,7 @@ def vector_decoder(field_index: int, key: Key) -> Decoder:
 UInt64Decoder = _scalar_decoder(cast(DecoderConstructor, decoder.UInt64Decoder))
 BoolDecoder = _scalar_decoder(cast(DecoderConstructor, decoder.BoolDecoder))
 StringDecoder = _scalar_decoder(cast(DecoderConstructor, decoder.StringDecoder))
+XYDataDecoder = _double_xy_data_decoder(_message._message_decoder_constructor)
 
 FloatArrayDecoder = _vector_decoder(cast(DecoderConstructor, decoder.FloatDecoder))
 DoubleArrayDecoder = _vector_decoder(cast(DecoderConstructor, decoder.DoubleDecoder))
@@ -155,6 +171,7 @@ def vector_decoder(field_index: int, key: Key) -> Decoder:
     type_pb2.Field.TYPE_BOOL: (BoolEncoder, BoolArrayEncoder),
     type_pb2.Field.TYPE_STRING: (StringEncoder, StringArrayEncoder),
     type_pb2.Field.TYPE_ENUM: (IntEncoder, IntArrayEncoder),
+    type_pb2.Field.TYPE_MESSAGE: (MessageEncoder, UnsupportedMessageArrayEncoder),
 }
 
 _FIELD_TYPE_TO_DECODER_MAPPING = {
@@ -181,6 +198,10 @@ def vector_decoder(field_index: int, key: Key) -> Decoder:
     type_pb2.Field.TYPE_ENUM: int(),
 }
 
+_MESSAGE_TYPE_TO_DECODER = {
+    xydata_pb2.DoubleXYData.DESCRIPTOR.full_name: XYDataDecoder,
+}
+
 
 def get_encoder(type: type_pb2.Field.Kind.ValueType, repeated: bool) -> PartialEncoderConstructor:
     """Get the appropriate partial encoder constructor for the specified type.
@@ -195,17 +216,23 @@ def get_encoder(type: type_pb2.Field.Kind.ValueType, repeated: bool) -> PartialE
     return scalar
 
 
-def get_decoder(type: type_pb2.Field.Kind.ValueType, repeated: bool) -> PartialDecoderConstructor:
-    """Get the appropriate partial decoder constructor for the specified type.
-
-    A scalar or vector constructor is returned based on the 'repeated' parameter.
-    """
-    if type not in _FIELD_TYPE_TO_DECODER_MAPPING:
+def get_decoder(
+    type: type_pb2.Field.Kind.ValueType, repeated: bool, message_type: str = ""
+) -> PartialDecoderConstructor:
+    """Get the appropriate partial decoder constructor for the specified type."""
+    decoder_mapping = _FIELD_TYPE_TO_DECODER_MAPPING.get(type)
+    if decoder_mapping is not None:
+        scalar_decoder, array_decoder = decoder_mapping
+        return array_decoder if repeated else scalar_decoder
+    elif type == type_pb2.Field.Kind.TYPE_MESSAGE:
+        if repeated:
+            raise ValueError(f"Repeated message types are not supported '{message_type}'")
+        decoder = _MESSAGE_TYPE_TO_DECODER.get(message_type)
+        if decoder is None:
+            raise ValueError(f"Unknown message type '{message_type}'")
+        return decoder
+    else:
         raise ValueError(f"Error can not decode type '{type}'")
-    scalar, array = _FIELD_TYPE_TO_DECODER_MAPPING[type]
-    if repeated:
-        return array
-    return scalar
 
 
 def get_type_default(type: type_pb2.Field.Kind.ValueType, repeated: bool) -> Any:

ni_measurementlink_service/_internal/parameter/serializer.py

@@ -158,9 +158,10 @@ def _get_overlapping_parameters(
             raise Exception(
                 f"Error occurred while reading the parameter - given protobuf index '{field_index}' is invalid."
             )
-        type = parameter_metadata_dict[field_index].type
-        is_repeated = parameter_metadata_dict[field_index].repeated
-        decoder = serialization_strategy.get_decoder(type, is_repeated)
+        field_metadata = parameter_metadata_dict[field_index]
+        decoder = serialization_strategy.get_decoder(
+            field_metadata.type, field_metadata.repeated, field_metadata.message_type
+        )
         inner_decoder = decoder(field_index, field_index)
         parameter_bytes_io = BytesIO(parameter_bytes)
         parameter_bytes_memory_view = parameter_bytes_io.getbuffer()