Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion poetry.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

3 changes: 1 addition & 2 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,7 @@ protobuf = {version=">=4.21"}
ni-measurement-plugin-sdk = {version=">=2.3"}
typing-extensions = ">=4.13.2"
streamlit = ">=1.24"
# Temporary dependency until ni-apis-python is available.
nitypes = ">=0.1.0dev1"
nitypes = {version=">=0.1.0dev1", allow-prereleases=true}

[tool.poetry.group.dev.dependencies]
types-grpcio = ">=1.0"
Expand Down
45 changes: 22 additions & 23 deletions src/nipanel/converters/protobuf_types.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,23 @@
"""Classes to convert between measurement specific protobuf types and containers."""

from typing import Any, Type
from typing import Type, Union

from ni.protobuf.types import scalar_pb2
from nitypes.scalar import Scalar
from typing_extensions import TypeAlias

from nipanel.converters import Converter

_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",
}

class ScalarConverter(Converter[Scalar[Any], scalar_pb2.ScalarData]):

class ScalarConverter(Converter[Scalar[_AnyScalarType], scalar_pb2.ScalarData]):
"""A converter for Scalar objects."""

@property
Expand All @@ -21,36 +30,26 @@ def protobuf_message(self) -> Type[scalar_pb2.ScalarData]:
"""The type-specific protobuf message for the Python type."""
return scalar_pb2.ScalarData

def to_protobuf_message(self, python_value: Scalar[Any]) -> scalar_pb2.ScalarData:
def to_protobuf_message(self, python_value: Scalar[_AnyScalarType]) -> scalar_pb2.ScalarData:
"""Convert the Python Scalar to a protobuf scalar_pb2.ScalarData."""
message = self.protobuf_message()
message.units = python_value.units
if isinstance(python_value.value, bool):
message.bool_value = python_value.value
elif isinstance(python_value.value, int):
message.int32_value = python_value.value
elif isinstance(python_value.value, float):
message.double_value = python_value.value
elif isinstance(python_value.value, str):
message.string_value = python_value.value
else:

value_attr = SCALAR_TYPE_TO_PB_ATTR_MAP.get(type(python_value.value), None)
if not value_attr:
raise TypeError(f"Unexpected type for python_value.value: {type(python_value.value)}")
setattr(message, value_attr, python_value.value)

return message

def to_python_value(self, protobuf_value: scalar_pb2.ScalarData) -> Scalar[Any]:
def to_python_value(self, protobuf_value: scalar_pb2.ScalarData) -> Scalar[_AnyScalarType]:
"""Convert the protobuf message to a Python Scalar."""
if protobuf_value.units is None:
raise ValueError("protobuf.units cannot be None.")

pb_type = protobuf_value.WhichOneof("value")
if pb_type == "bool_value":
return Scalar(protobuf_value.bool_value, protobuf_value.units)
elif pb_type == "int32_value":
return Scalar(protobuf_value.int32_value, protobuf_value.units)
elif pb_type == "double_value":
return Scalar(protobuf_value.double_value, protobuf_value.units)
elif pb_type == "string_value":
return Scalar(protobuf_value.string_value, protobuf_value.units)
else:
pb_type = str(protobuf_value.WhichOneof("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(protobuf_value, pb_type)
return Scalar(value, protobuf_value.units)
1 change: 0 additions & 1 deletion tests/unit/test_protobuf_type_conversion.py
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,6 @@ def test___scalar_protobuf_value_unset___convert___throws_type_error() -> None:
def test___scalar_protobuf_units_unset___convert___python_units_blank() -> None:
protobuf_value = ScalarData()
protobuf_value.bool_value = True
print(f"Protobuf units: {protobuf_value.units!r}")

converter = ScalarConverter()
python_value = converter.to_python_value(protobuf_value)
Expand Down