Move Scalar conversion logic into ni-apis-python. (#22)

* Move Scalar conversion logic into ni-apis-python.

* Check the result of WhichOneOf for None.

Signed-off-by: Michael Johansen <[email protected]>

---------

Signed-off-by: Michael Johansen <[email protected]>
Co-authored-by: Michael Johansen <[email protected]>

Author: mjohanse-emr

packages/ni.protobuf.types/src/ni/protobuf/types/scalar_conversion.py

@@ -0,0 +1,82 @@
+"""Methods to convert to and from scalar protobuf messages."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Union
+
+from nitypes.scalar import Scalar
+from nitypes.waveform import ExtendedPropertyDictionary, ExtendedPropertyValue
+from typing_extensions import TypeAlias
+
+import ni.protobuf.types.scalar_pb2 as scalar_pb2
+from ni.protobuf.types.attribute_value_pb2 import AttributeValue
+
+_AnyScalarType: TypeAlias = Union[bool, int, float, str]
+_SCALAR_TYPE_TO_PB_ATTR_MAP = {
+    bool: "bool_value",
+    int: "int32_value",
+    float: "double_value",
+    str: "string_value",
+}
+
+
+def scalar_to_protobuf(value: Scalar[_AnyScalarType], /) -> scalar_pb2.Scalar:
+    """Convert a Scalar python object to a protobuf scalar_pb2.Scalar."""
+    attributes = _extended_properties_to_attributes(value.extended_properties)
+    message = scalar_pb2.Scalar(attributes=attributes)
+
+    # Convert the scalar value
+    value_attr = _SCALAR_TYPE_TO_PB_ATTR_MAP.get(type(value.value), None)
+    if not value_attr:
+        raise TypeError(f"Unexpected type for python_value.value: {type(value.value)}")
+    setattr(message, value_attr, value.value)
+
+    return message
+
+
+def scalar_from_protobuf(message: scalar_pb2.Scalar, /) -> Scalar[_AnyScalarType]:
+    """Convert the protobuf scalar_pb2.Scalar to a Python Scalar."""
+    # Convert the scalar value.
+    pb_type = message.WhichOneof("value")
+    if pb_type is None:
+        raise ValueError("Could not determine the data type of 'value'.")
+
+    if pb_type not in _SCALAR_TYPE_TO_PB_ATTR_MAP.values():
+        raise ValueError(f"Unexpected value for protobuf_value.WhichOneOf: {pb_type}")
+    value = getattr(message, pb_type)
+
+    # Create with blank units. Units from the protobuf message will be populated
+    # when attributes are converted to an ExtendedPropertyDictionary.
+    scalar = Scalar(value, "")
+
+    # Transfer attributes to extended_properties
+    for key, value in message.attributes.items():
+        attr_type = value.WhichOneof("attribute")
+        if attr_type is None:
+            raise ValueError("Could not determine the data type of 'attribute'.")
+        scalar.extended_properties[key] = getattr(value, attr_type)
+
+    return scalar
+
+
+def _extended_properties_to_attributes(
+    extended_properties: ExtendedPropertyDictionary,
+) -> Mapping[str, AttributeValue]:
+    return {key: _value_to_attribute(value) for key, value in extended_properties.items()}
+
+
+def _value_to_attribute(value: ExtendedPropertyValue) -> AttributeValue:
+    attr_value = AttributeValue()
+    if isinstance(value, bool):
+        attr_value.bool_value = value
+    elif isinstance(value, int):
+        attr_value.integer_value = value
+    elif isinstance(value, float):
+        attr_value.double_value = value
+    elif isinstance(value, str):
+        attr_value.string_value = value
+    else:
+        raise TypeError(f"Unexpected type for extended property value {type(value)}")
+
+    return attr_value

packages/ni.protobuf.types/src/ni/protobuf/types/waveform_conversion.py

@@ -78,7 +78,9 @@ def float64_analog_waveform_from_protobuf(
     extended_properties = {}
     for key, value in message.attributes.items():
         attr_type = value.WhichOneof("attribute")
-        extended_properties[key] = getattr(value, str(attr_type))
+        if attr_type is None:
+            raise ValueError("Could not determine the datatype of 'attribute'.")
+        extended_properties[key] = getattr(value, attr_type)
 
     data_array = np.array(message.y_data)
     return AnalogWaveform(

packages/ni.protobuf.types/tests/unit/test_ni_protobuf_types.py

@@ -1,7 +1,9 @@
 """Tests for the ni.protobuf.types package."""
 
 from ni.protobuf.types.array_pb2 import Double2DArray, String2DArray
+from ni.protobuf.types.attribute_value_pb2 import AttributeValue
 from ni.protobuf.types.precision_timestamp_pb2 import PrecisionTimestamp
+from ni.protobuf.types.scalar_pb2 import Scalar
 from ni.protobuf.types.waveform_pb2 import (
     DoubleAnalogWaveform,
     DoubleComplexWaveform,
@@ -18,6 +20,10 @@
     "attr1": WaveformAttributeValue(integer_value=1),
     "attr2": WaveformAttributeValue(string_value="two"),
 }
+EXPECTED_SCALAR_ATTRIBUTES = {
+    "attr1": AttributeValue(integer_value=1),
+    "attr2": AttributeValue(string_value="two"),
+}
 
 
 def test___valid_inputs___createdouble2darray___message_created() -> None:
@@ -114,3 +120,31 @@ def test___valid_inputs___create_doublexydata___message_created() -> None:
 
     assert list(test_wfm.x_data) == [1.0, 2.0, 3.0]
     assert list(test_wfm.y_data) == [4.0, 5.0, 6.0]
+
+
+def test___valid_inputs___create_double_scalar___message_created() -> None:
+    test_scalar = Scalar(attributes=EXPECTED_SCALAR_ATTRIBUTES, double_value=1.0)
+
+    assert test_scalar.double_value == 1.0
+    assert test_scalar.attributes == EXPECTED_SCALAR_ATTRIBUTES
+
+
+def test___valid_inputs___create_int_scalar___message_created() -> None:
+    test_scalar = Scalar(attributes=EXPECTED_SCALAR_ATTRIBUTES, int32_value=1)
+
+    assert test_scalar.int32_value == 1
+    assert test_scalar.attributes == EXPECTED_SCALAR_ATTRIBUTES
+
+
+def test___valid_inputs___create_bool_scalar___message_created() -> None:
+    test_scalar = Scalar(attributes=EXPECTED_SCALAR_ATTRIBUTES, bool_value=True)
+
+    assert test_scalar.bool_value is True
+    assert test_scalar.attributes == EXPECTED_SCALAR_ATTRIBUTES
+
+
+def test___valid_inputs___create_string_scalar___message_created() -> None:
+    test_scalar = Scalar(attributes=EXPECTED_SCALAR_ATTRIBUTES, string_value="one")
+
+    assert test_scalar.string_value == "one"
+    assert test_scalar.attributes == EXPECTED_SCALAR_ATTRIBUTES

packages/ni.protobuf.types/tests/unit/test_scalar_conversion.py

@@ -0,0 +1,161 @@
+import pytest
+from nitypes.scalar import Scalar
+
+import ni.protobuf.types.scalar_pb2 as scalar_pb2
+from ni.protobuf.types.attribute_value_pb2 import AttributeValue
+from ni.protobuf.types.scalar_conversion import scalar_from_protobuf, scalar_to_protobuf
+
+
+# ========================================================
+# Scalar: Protobuf to Python
+# ========================================================
+def test___bool_scalar_protobuf___convert___valid_bool_scalar() -> None:
+    attributes = {"NI_UnitDescription": AttributeValue(string_value="Volts")}
+    protobuf_value = scalar_pb2.Scalar(attributes=attributes)
+    protobuf_value.bool_value = True
+
+    python_value = scalar_from_protobuf(protobuf_value)
+
+    assert isinstance(python_value.value, bool)
+    assert python_value.value is True
+    assert python_value.units == "Volts"
+
+
+def test___int32_scalar_protobuf___convert___valid_int_scalar() -> None:
+    attributes = {"NI_UnitDescription": AttributeValue(string_value="Volts")}
+    protobuf_value = scalar_pb2.Scalar(attributes=attributes)
+    protobuf_value.int32_value = 10
+
+    python_value = scalar_from_protobuf(protobuf_value)
+
+    assert isinstance(python_value.value, int)
+    assert python_value.value == 10
+    assert python_value.units == "Volts"
+
+
+def test___double_scalar_protobuf___convert___valid_float_scalar() -> None:
+    attributes = {"NI_UnitDescription": AttributeValue(string_value="Volts")}
+    protobuf_value = scalar_pb2.Scalar(attributes=attributes)
+    protobuf_value.double_value = 20.0
+
+    python_value = scalar_from_protobuf(protobuf_value)
+
+    assert isinstance(python_value.value, float)
+    assert python_value.value == 20.0
+    assert python_value.units == "Volts"
+
+
+def test___string_scalar_protobuf___convert___valid_str_scalar() -> None:
+    attributes = {"NI_UnitDescription": AttributeValue(string_value="Volts")}
+    protobuf_value = scalar_pb2.Scalar(attributes=attributes)
+    protobuf_value.string_value = "value"
+
+    python_value = scalar_from_protobuf(protobuf_value)
+
+    assert isinstance(python_value.value, str)
+    assert python_value.value == "value"
+    assert python_value.units == "Volts"
+
+
+def test___scalar_protobuf_value_unset___convert___throws_value_error() -> None:
+    attributes = {"NI_UnitDescription": AttributeValue(string_value="Volts")}
+    protobuf_value = scalar_pb2.Scalar(attributes=attributes)
+
+    with pytest.raises(ValueError) as exc:
+        _ = scalar_from_protobuf(protobuf_value)
+
+    assert exc.value.args[0].startswith("Could not determine the data type of 'value'.")
+
+
+def test___scalar_protobuf_units_unset___convert___python_units_blank() -> None:
+    protobuf_value = scalar_pb2.Scalar()
+    protobuf_value.bool_value = True
+
+    python_value = scalar_from_protobuf(protobuf_value)
+
+    assert isinstance(python_value.value, bool)
+    assert python_value.value is True
+    assert python_value.units == ""
+
+
+def test___non_units_attributes___to_python___attributes_converted() -> None:
+    attributes = {
+        "NI_ChannelName": AttributeValue(string_value="Dev1/ai0"),
+        "NI_UnitDescription": AttributeValue(string_value="Volts"),
+    }
+    protobuf_value = scalar_pb2.Scalar(attributes=attributes)
+    protobuf_value.string_value = "value"
+
+    python_value = scalar_from_protobuf(protobuf_value)
+    channel_name = python_value.extended_properties["NI_ChannelName"]
+
+    assert isinstance(python_value.value, str)
+    assert python_value.value == "value"
+    assert python_value.units == "Volts"
+    assert isinstance(channel_name, str)
+    assert channel_name == "Dev1/ai0"
+
+
+# ========================================================
+# Scalar: Python to Protobuf
+# ========================================================
+def test___bool_scalar___convert___valid_bool_scalar_protobuf() -> None:
+    python_value = Scalar(True, "Volts")
+
+    protobuf_value = scalar_to_protobuf(python_value)
+
+    assert protobuf_value.WhichOneof("value") == "bool_value"
+    assert protobuf_value.bool_value is True
+    assert protobuf_value.attributes["NI_UnitDescription"].string_value == "Volts"
+
+
+def test___int_scalar___convert___valid_int32_scalar_protobuf() -> None:
+    python_value = Scalar(10, "Volts")
+
+    protobuf_value = scalar_to_protobuf(python_value)
+
+    assert protobuf_value.WhichOneof("value") == "int32_value"
+    assert protobuf_value.int32_value == 10
+    assert protobuf_value.attributes["NI_UnitDescription"].string_value == "Volts"
+
+
+def test___float_scalar___convert___valid_double_scalar_protobuf() -> None:
+    python_value = Scalar(20.0, "Volts")
+
+    protobuf_value = scalar_to_protobuf(python_value)
+
+    assert protobuf_value.WhichOneof("value") == "double_value"
+    assert protobuf_value.double_value == 20.0
+    assert protobuf_value.attributes["NI_UnitDescription"].string_value == "Volts"
+
+
+def test___str_scalar___convert___valid_string_scalar_protobuf() -> None:
+    python_value = Scalar("value", "Volts")
+
+    protobuf_value = scalar_to_protobuf(python_value)
+
+    assert protobuf_value.WhichOneof("value") == "string_value"
+    assert protobuf_value.string_value == "value"
+    assert protobuf_value.attributes["NI_UnitDescription"].string_value == "Volts"
+
+
+def test___scalar_units_unset___convert___protobuf_units_blank() -> None:
+    python_value = Scalar(10)
+
+    protobuf_value = scalar_to_protobuf(python_value)
+
+    assert protobuf_value.WhichOneof("value") == "int32_value"
+    assert protobuf_value.int32_value == 10
+    assert protobuf_value.attributes["NI_UnitDescription"].string_value == ""
+
+
+def test___non_units_attributes___to_protobuf___attributes_converted() -> None:
+    python_value = Scalar("value", "Volts")
+    python_value.extended_properties["NI_ChannelName"] = "Dev1/ai0"
+
+    protobuf_value = scalar_to_protobuf(python_value)
+
+    assert protobuf_value.WhichOneof("value") == "string_value"
+    assert protobuf_value.string_value == "value"
+    assert protobuf_value.attributes["NI_UnitDescription"].string_value == "Volts"
+    assert protobuf_value.attributes["NI_ChannelName"].string_value == "Dev1/ai0"