diff --git a/src/ni/datastore/client.py b/src/ni/datastore/client.py index bebab18..7fd1014 100644 --- a/src/ni/datastore/client.py +++ b/src/ni/datastore/client.py @@ -20,9 +20,12 @@ bintime_datetime_to_protobuf, ) from ni.protobuf.types.waveform_conversion import float64_analog_waveform_to_protobuf +from ni.protobuf.types.waveform_pb2 import DoubleAnalogWaveform from nitypes.bintime import DateTime from nitypes.waveform import AnalogWaveform +from ni.datastore.conversion.convert import from_any, to_protobuf_message + TRead = TypeVar("TRead") TWrite = TypeVar("TWrite") @@ -73,14 +76,18 @@ def publish_measurement_data( test_adapter_ids=test_adapter_ids, ) + # Perform the actual conversion. For built-in types, this will return a Message, + # not the actual data value, so we're basically ignoring the output of this method. + protobuf_message = to_protobuf_message(data) if isinstance(data, bool): + # For the built-in type case, datastore just assigns to the scalar.bool_value field. + # We won't use the value of protobuf_message in this case. publish_request.scalar.bool_value = data elif isinstance(data, AnalogWaveform): # Assuming data is of type AnalogWaveform - analog_waveform = cast(AnalogWaveform[np.float64], data) - publish_request.double_analog_waveform.CopyFrom( - float64_analog_waveform_to_protobuf(analog_waveform) - ) + # Now we have to assign to publish_request.analog_waveform. I had to add the cast here + # to satisfy CopyFrom. Maybe there's another way to assign it? + publish_request.double_analog_waveform.CopyFrom(cast(DoubleAnalogWaveform, protobuf_message)) publish_response = self._data_store_client.publish_measurement(publish_request) return publish_response.published_measurement @@ -95,9 +102,11 @@ def read_measurement_data( moniker = moniker_source.moniker self._moniker_client._service_location = moniker.service_location result = self._moniker_client.read_from_moniker(moniker) - if not isinstance(result.value, expected_type): + python_value = from_any(result.value) + if not isinstance(python_value, expected_type): raise TypeError(f"Expected type {expected_type}, got {type(result.value)}") - return result.value + + return python_value def create_step( self, diff --git a/src/ni/datastore/conversion/__init__.py b/src/ni/datastore/conversion/__init__.py new file mode 100644 index 0000000..92b189d --- /dev/null +++ b/src/ni/datastore/conversion/__init__.py @@ -0,0 +1,114 @@ +"""Functions and classes to convert types between Python and protobuf.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from collections.abc import Collection +from typing import Generic, Type, TypeVar + +from google.protobuf import any_pb2 +from google.protobuf.message import Message + +_TItemType = TypeVar("_TItemType") +_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_type(self) -> type: + """The Python type that this converter handles.""" + + @property + def python_typename(self) -> str: + """The Python type name that this converter handles.""" + return f"{self.python_type.__module__}.{self.python_type.__name__}" + + @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.""" + + +class CollectionConverter( + Generic[_TItemType, _TProtobufType], + Converter[Collection[_TItemType], _TProtobufType], + ABC, +): + """A converter between a collection of Python objects and protobuf Any messages.""" + + @property + @abstractmethod + def item_type(self) -> type: + """The Python item type that this converter handles.""" + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return Collection + + @property + def python_typename(self) -> str: + """The Python type name that this converter handles.""" + return "{}[{}]".format( + f"{Collection.__module__}.{Collection.__name__}", + f"{self.item_type.__module__}.{self.item_type.__name__}", + ) + + +class CollectionConverter2D( + Generic[_TItemType, _TProtobufType], + Converter[Collection[Collection[_TItemType]], _TProtobufType], + ABC, +): + """A converter between a 2D collection of Python objects and protobuf Any messages.""" + + @property + @abstractmethod + def item_type(self) -> type: + """The Python item type that this converter handles.""" + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return Collection + + @property + def python_typename(self) -> str: + """The Python type name that this converter handles.""" + return "{}[{}[{}]]".format( + f"{Collection.__module__}.{Collection.__name__}", + f"{Collection.__module__}.{Collection.__name__}", + f"{self.item_type.__module__}.{self.item_type.__name__}", + ) diff --git a/src/ni/datastore/conversion/builtin.py b/src/ni/datastore/conversion/builtin.py new file mode 100644 index 0000000..3927c78 --- /dev/null +++ b/src/ni/datastore/conversion/builtin.py @@ -0,0 +1,166 @@ +"""Classes to convert between builtin Python scalars and containers.""" + +import datetime as dt +from typing import Type + +from google.protobuf import duration_pb2, timestamp_pb2, wrappers_pb2 + +from ni.datastore.conversion import Converter + + +class BoolConverter(Converter[bool, wrappers_pb2.BoolValue]): + """A converter for boolean types.""" + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return bool + + @property + def protobuf_message(self) -> Type[wrappers_pb2.BoolValue]: + """The type-specific protobuf message for the Python type.""" + return wrappers_pb2.BoolValue + + def to_protobuf_message(self, python_value: bool) -> wrappers_pb2.BoolValue: + """Convert the Python bool to a protobuf wrappers_pb2.BoolValue.""" + return self.protobuf_message(value=python_value) + + def to_python_value(self, protobuf_message: wrappers_pb2.BoolValue) -> bool: + """Convert the protobuf message to a Python bool.""" + return protobuf_message.value + + +class BytesConverter(Converter[bytes, wrappers_pb2.BytesValue]): + """A converter for byte string types.""" + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return bytes + + @property + def protobuf_message(self) -> Type[wrappers_pb2.BytesValue]: + """The type-specific protobuf message for the Python type.""" + return wrappers_pb2.BytesValue + + def to_protobuf_message(self, python_value: bytes) -> wrappers_pb2.BytesValue: + """Convert the Python bytes string to a protobuf wrappers_pb2.BytesValue.""" + return self.protobuf_message(value=python_value) + + def to_python_value(self, protobuf_message: wrappers_pb2.BytesValue) -> bytes: + """Convert the protobuf message to a Python bytes string.""" + return protobuf_message.value + + +class FloatConverter(Converter[float, wrappers_pb2.DoubleValue]): + """A converter for floating point types.""" + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return float + + @property + def protobuf_message(self) -> Type[wrappers_pb2.DoubleValue]: + """The type-specific protobuf message for the Python type.""" + return wrappers_pb2.DoubleValue + + def to_protobuf_message(self, python_value: float) -> wrappers_pb2.DoubleValue: + """Convert the Python float to a protobuf wrappers_pb2.DoubleValue.""" + return self.protobuf_message(value=python_value) + + def to_python_value(self, protobuf_message: wrappers_pb2.DoubleValue) -> float: + """Convert the protobuf message to a Python float.""" + return protobuf_message.value + + +class IntConverter(Converter[int, wrappers_pb2.Int64Value]): + """A converter for integer types.""" + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return int + + @property + def protobuf_message(self) -> Type[wrappers_pb2.Int64Value]: + """The type-specific protobuf message for the Python type.""" + return wrappers_pb2.Int64Value + + def to_protobuf_message(self, python_value: int) -> wrappers_pb2.Int64Value: + """Convert the Python int to a protobuf wrappers_pb2.Int64Value.""" + return self.protobuf_message(value=python_value) + + def to_python_value(self, protobuf_message: wrappers_pb2.Int64Value) -> int: + """Convert the protobuf message to a Python int.""" + return protobuf_message.value + + +class StrConverter(Converter[str, wrappers_pb2.StringValue]): + """A converter for text string types.""" + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return str + + @property + def protobuf_message(self) -> Type[wrappers_pb2.StringValue]: + """The type-specific protobuf message for the Python type.""" + return wrappers_pb2.StringValue + + def to_protobuf_message(self, python_value: str) -> wrappers_pb2.StringValue: + """Convert the Python str to a protobuf wrappers_pb2.StringValue.""" + return self.protobuf_message(value=python_value) + + def to_python_value(self, protobuf_message: wrappers_pb2.StringValue) -> str: + """Convert the protobuf message to a Python string.""" + return protobuf_message.value + + +class DTDateTimeConverter(Converter[dt.datetime, timestamp_pb2.Timestamp]): + """A converter for datetime.datetime types.""" + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return dt.datetime + + @property + def protobuf_message(self) -> Type[timestamp_pb2.Timestamp]: + """The type-specific protobuf message for the Python type.""" + return timestamp_pb2.Timestamp + + def to_protobuf_message(self, python_value: dt.datetime) -> timestamp_pb2.Timestamp: + """Convert the Python dt.datetime to a protobuf timestamp_pb2.Timestamp.""" + ts = self.protobuf_message() + ts.FromDatetime(python_value) + return ts + + def to_python_value(self, protobuf_message: timestamp_pb2.Timestamp) -> dt.datetime: + """Convert the protobuf timestamp_pb2.Timestamp to a Python dt.datetime.""" + return protobuf_message.ToDatetime() + + +class DTTimeDeltaConverter(Converter[dt.timedelta, duration_pb2.Duration]): + """A converter for datetime.timedelta types.""" + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return dt.timedelta + + @property + def protobuf_message(self) -> Type[duration_pb2.Duration]: + """The type-specific protobuf message for the Python type.""" + return duration_pb2.Duration + + def to_protobuf_message(self, python_value: dt.timedelta) -> duration_pb2.Duration: + """Convert the Python dt.timedelta to a protobuf duration_pb2.Duration.""" + dur = self.protobuf_message() + dur.FromTimedelta(python_value) + return dur + + def to_python_value(self, protobuf_message: duration_pb2.Duration) -> dt.timedelta: + """Convert the protobuf timestamp_pb2.Timestamp to a Python dt.timedelta.""" + return protobuf_message.ToTimedelta() diff --git a/src/ni/datastore/conversion/convert.py b/src/ni/datastore/conversion/convert.py new file mode 100644 index 0000000..76ca061 --- /dev/null +++ b/src/ni/datastore/conversion/convert.py @@ -0,0 +1,209 @@ +"""Functions to convert between different data formats.""" + +from __future__ import annotations + +import enum +import logging +from collections.abc import Collection +from typing import Any, Iterable + +from google.protobuf import any_pb2 +from google.protobuf.message import Message +from nitypes.vector import Vector +from nitypes.waveform import AnalogWaveform, ComplexWaveform + +from ni.datastore.conversion.convert import Converter +from ni.datastore.conversion.builtin import ( + BoolConverter, + BytesConverter, + DTDateTimeConverter, + DTTimeDeltaConverter, + FloatConverter, + IntConverter, + StrConverter, +) +from ni.datastore.conversion.protobuf_types import ( + BTDateTimeConverter, + BTTimeDeltaConverter, + BoolCollectionConverter, + BytesCollectionConverter, + DigitalWaveformConverter, + Double2DArrayConverter, + DoubleAnalogWaveformConverter, + DoubleComplexWaveformConverter, + DoubleSpectrumConverter, + FloatCollectionConverter, + HTDateTimeConverter, + HTTimeDeltaConverter, + Int16AnalogWaveformConverter, + Int16ComplexWaveformConverter, + IntCollectionConverter, + ScalarConverter, + StrCollectionConverter, + VectorConverter, +) + +_logger = logging.getLogger(__name__) + +# FFV -- consider adding a RegisterConverter mechanism +_CONVERTIBLE_TYPES: list[Converter[Any, Any]] = [ + # Built-in Types + BoolConverter(), + BytesConverter(), + FloatConverter(), + IntConverter(), + StrConverter(), + DTDateTimeConverter(), + DTTimeDeltaConverter(), + # Protobuf Types + BTDateTimeConverter(), + BTTimeDeltaConverter(), + BoolCollectionConverter(), + BytesCollectionConverter(), + DigitalWaveformConverter(), + Double2DArrayConverter(), + DoubleAnalogWaveformConverter(), + DoubleComplexWaveformConverter(), + DoubleSpectrumConverter(), + FloatCollectionConverter(), + HTDateTimeConverter(), + HTTimeDeltaConverter(), + Int16AnalogWaveformConverter(), + Int16ComplexWaveformConverter(), + IntCollectionConverter(), + StrCollectionConverter(), + ScalarConverter(), + VectorConverter(), +] + +_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() + +_SKIPPED_COLLECTIONS = ( + str, # Handled by StrConverter + bytes, # Handled by BytesConverter + dict, # Unsupported data type + enum.Enum, # Handled by IntConverter + Vector, # Handled by VectorConverter +) + + +def to_any(python_value: object) -> any_pb2.Any: + """Convert a Python object to a protobuf Any.""" + best_matching_type = _get_best_matching_type(python_value) + converter = _CONVERTER_FOR_PYTHON_TYPE[best_matching_type] + return converter.to_protobuf_any(python_value) + + +def to_protobuf_message(python_value: object) -> Message: + """Convert a Python object to a protobuf Any.""" + best_matching_type = _get_best_matching_type(python_value) + converter = _CONVERTER_FOR_PYTHON_TYPE[best_matching_type] + return converter.to_protobuf_message(python_value) + + +def _get_best_matching_type(python_value: object) -> str: + underlying_parents = type(python_value).mro() # This covers enum.IntEnum and similar + additional_info_string = _get_additional_type_info_string(python_value) + + container_types = [] + value_is_collection = _is_collection_for_convert(python_value) + # Variable to use when traversing down through collection types. + working_python_value = python_value + while value_is_collection: + # Assume Sized -- Generators not supported, callers must use list(), set(), ... as desired + if not isinstance(working_python_value, Collection): + raise TypeError() + if len(working_python_value) == 0: + underlying_parents = type(None).mro() + value_is_collection = False + else: + # Assume homogenous -- collections of mixed-types not supported + visitor = iter(working_python_value) + + # Store off the first element. If it's a container, we'll need it in the next while + # loop iteration. + working_python_value = next(visitor) + underlying_parents = type(working_python_value).mro() + + # If this element is a collection, we want to continue traversing. Once we find a + # non-collection, underlying_parents will refer to the candidates for the non- + # collection type. + value_is_collection = _is_collection_for_convert(working_python_value) + container_types.append(Collection) + + best_matching_type = None + candidates = _get_candidate_strings(underlying_parents) + for candidate in candidates: + python_typename = _create_python_typename( + candidate, container_types, additional_info_string + ) + 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_types}, {payload_type}) with parents " + f"{underlying_parents}.\n\nSupported types are: {_SUPPORTED_PYTHON_TYPES}" + f"\n\nAdditional type info: {additional_info_string}" + ) + _logger.debug(f"Best matching type for '{repr(python_value)}' resolved to {best_matching_type}") + return best_matching_type + + +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) + + +def is_supported_type(value: object) -> bool: + """Check if a given Python value can be converted to protobuf Any.""" + try: + _get_best_matching_type(value) + return True + except TypeError: + return False + + +def _get_candidate_strings(candidates: Iterable[type]) -> list[str]: + candidate_names = [] + for candidate in candidates: + candidate_names.append(f"{candidate.__module__}.{candidate.__name__}") + return candidate_names + + +def _create_python_typename( + candidate_name: str, container_types: Iterable[type], additional_info: str +) -> str: + name = candidate_name + if additional_info: + name = f"{name}[{additional_info}]" + for container_type in container_types: + name = f"{container_type.__module__}.{container_type.__name__}[{name}]" + return name + + +def _get_additional_type_info_string(python_value: object) -> str: + if isinstance(python_value, AnalogWaveform): + return str(python_value.dtype) + elif isinstance(python_value, ComplexWaveform): + return str(python_value.dtype) + else: + return "" + + +def _is_collection_for_convert(python_value: object) -> bool: + return isinstance(python_value, Collection) and not isinstance( + python_value, _SKIPPED_COLLECTIONS + ) diff --git a/src/ni/datastore/conversion/protobuf_types.py b/src/ni/datastore/conversion/protobuf_types.py new file mode 100644 index 0000000..121ae76 --- /dev/null +++ b/src/ni/datastore/conversion/protobuf_types.py @@ -0,0 +1,560 @@ +"""Classes to convert between measurement specific protobuf types and containers.""" + +from __future__ import annotations + +from collections.abc import Collection +from typing import Any, Type, Union + +import hightime as ht +import nitypes.bintime as bt +import numpy as np +from ni.protobuf.types import ( + array_pb2, + precision_duration_pb2, + precision_duration_conversion, + precision_timestamp_pb2, + precision_timestamp_conversion, + scalar_conversion, + scalar_pb2, + vector_pb2, + vector_conversion, + waveform_conversion, + waveform_pb2, +) +from nitypes.complex import ComplexInt32Base +from nitypes.scalar import Scalar +from nitypes.vector import Vector +from nitypes.waveform import AnalogWaveform, ComplexWaveform, DigitalWaveform, Spectrum +from typing_extensions import TypeAlias + +from ni.datastore.conversion import Converter, CollectionConverter, CollectionConverter2D + +_AnyScalarType: TypeAlias = Union[bool, int, float, str] + + +class BoolCollectionConverter(CollectionConverter[bool, array_pb2.BoolArray]): + """A converter for a Collection of bools.""" + + @property + def item_type(self) -> type: + """The Python type that this converter handles.""" + return bool + + @property + def protobuf_message(self) -> Type[array_pb2.BoolArray]: + """The type-specific protobuf message for the Python type.""" + return array_pb2.BoolArray + + def to_protobuf_message(self, python_value: Collection[bool]) -> array_pb2.BoolArray: + """Convert the collection of bools to array_pb2.BoolArray.""" + return self.protobuf_message(values=python_value) + + def to_python_value(self, protobuf_message: array_pb2.BoolArray) -> Collection[bool]: + """Convert the protobuf message to a Python collection of bools.""" + return list(protobuf_message.values) + + +class BytesCollectionConverter(CollectionConverter[bytes, array_pb2.BytesArray]): + """A converter for a Collection of byte strings.""" + + @property + def item_type(self) -> type: + """The Python type that this converter handles.""" + return bytes + + @property + def protobuf_message(self) -> Type[array_pb2.BytesArray]: + """The type-specific protobuf message for the Python type.""" + return array_pb2.BytesArray + + def to_protobuf_message(self, python_value: Collection[bytes]) -> array_pb2.BytesArray: + """Convert the collection of byte strings to array_pb2.BytesArray.""" + return self.protobuf_message(values=python_value) + + def to_python_value(self, protobuf_message: array_pb2.BytesArray) -> Collection[bytes]: + """Convert the protobuf message to a Python collection of byte strings.""" + return list(protobuf_message.values) + + +class FloatCollectionConverter(CollectionConverter[float, array_pb2.DoubleArray]): + """A converter for a Collection of floats.""" + + @property + def item_type(self) -> type: + """The Python type that this converter handles.""" + return float + + @property + def protobuf_message(self) -> Type[array_pb2.DoubleArray]: + """The type-specific protobuf message for the Python type.""" + return array_pb2.DoubleArray + + def to_protobuf_message(self, python_value: Collection[float]) -> array_pb2.DoubleArray: + """Convert the collection of floats to array_pb2.DoubleArray.""" + return self.protobuf_message(values=python_value) + + def to_python_value(self, protobuf_message: array_pb2.DoubleArray) -> Collection[float]: + """Convert the protobuf message to a Python collection of floats.""" + return list(protobuf_message.values) + + +class IntCollectionConverter(CollectionConverter[int, array_pb2.SInt64Array]): + """A converter for a Collection of integers.""" + + @property + def item_type(self) -> type: + """The Python type that this converter handles.""" + return int + + @property + def protobuf_message(self) -> Type[array_pb2.SInt64Array]: + """The type-specific protobuf message for the Python type.""" + return array_pb2.SInt64Array + + def to_protobuf_message(self, python_value: Collection[int]) -> array_pb2.SInt64Array: + """Convert the collection of integers to array_pb2.SInt64Array.""" + return self.protobuf_message(values=python_value) + + def to_python_value(self, protobuf_message: array_pb2.SInt64Array) -> Collection[int]: + """Convert the protobuf message to a Python collection of integers.""" + return list(protobuf_message.values) + + +class StrCollectionConverter(CollectionConverter[str, array_pb2.StringArray]): + """A converter for a Collection of strings.""" + + @property + def item_type(self) -> type: + """The Python type that this converter handles.""" + return str + + @property + def protobuf_message(self) -> Type[array_pb2.StringArray]: + """The type-specific protobuf message for the Python type.""" + return array_pb2.StringArray + + def to_protobuf_message(self, python_value: Collection[str]) -> array_pb2.StringArray: + """Convert the collection of strings to array_pb2.StringCollection.""" + return self.protobuf_message(values=python_value) + + def to_python_value(self, protobuf_message: array_pb2.StringArray) -> Collection[str]: + """Convert the protobuf message to a Python collection of strings.""" + return list(protobuf_message.values) + + +class Double2DArrayConverter(CollectionConverter2D[float, array_pb2.Double2DArray]): + """A converter between Collection[Collection[float]] and Double2DArray.""" + + @property + def item_type(self) -> type: + """The Python item type that this converter handles.""" + return float + + @property + def protobuf_message(self) -> Type[array_pb2.Double2DArray]: + """The type-specific protobuf message for the Python type.""" + return array_pb2.Double2DArray + + def to_protobuf_message( + self, python_value: Collection[Collection[float]] + ) -> array_pb2.Double2DArray: + """Convert the Python Collection[Collection[float]] to a protobuf Double2DArray.""" + rows = len(python_value) + if rows: + visitor = iter(python_value) + first_subcollection = next(visitor) + columns = len(first_subcollection) + else: + columns = 0 + if not all(len(subcollection) == columns for subcollection in python_value): + raise ValueError("All subcollections must have the same length.") + + # Create a flat list in row major order. + flat_list = [item for subcollection in python_value for item in subcollection] + return array_pb2.Double2DArray(rows=rows, columns=columns, data=flat_list) + + def to_python_value( + self, protobuf_message: array_pb2.Double2DArray + ) -> Collection[Collection[float]]: + """Convert the protobuf Double2DArray to a Python Collection[Collection[float]].""" + if not protobuf_message.data: + return [] + if len(protobuf_message.data) % protobuf_message.columns != 0: + raise ValueError("The length of the data list must be divisible by num columns.") + + # Convert from a flat list in row major order into a list of lists. + list_of_lists = [] + for i in range(0, len(protobuf_message.data), protobuf_message.columns): + row = protobuf_message.data[i : i + protobuf_message.columns] + list_of_lists.append(row) + + return list_of_lists + + +class DoubleAnalogWaveformConverter( + Converter[AnalogWaveform[np.float64], waveform_pb2.DoubleAnalogWaveform] +): + """A converter for AnalogWaveform types with double-precision data.""" + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return AnalogWaveform + + @property + def python_typename(self) -> str: + """The Python type name that this converter handles.""" + base_typename = super().python_typename + return f"{base_typename}[float64]" + + @property + def protobuf_message(self) -> Type[waveform_pb2.DoubleAnalogWaveform]: + """The type-specific protobuf message for the Python type.""" + return waveform_pb2.DoubleAnalogWaveform + + def to_protobuf_message( + self, python_value: AnalogWaveform[np.float64] + ) -> waveform_pb2.DoubleAnalogWaveform: + """Convert the Python AnalogWaveform to a protobuf DoubleAnalogWaveform.""" + return waveform_conversion.float64_analog_waveform_to_protobuf(python_value) + + def to_python_value( + self, protobuf_message: waveform_pb2.DoubleAnalogWaveform + ) -> AnalogWaveform[np.float64]: + """Convert the protobuf DoubleAnalogWaveform to a Python AnalogWaveform.""" + return waveform_conversion.float64_analog_waveform_from_protobuf(protobuf_message) + + +class Int16AnalogWaveformConverter( + Converter[AnalogWaveform[np.int16], waveform_pb2.I16AnalogWaveform] +): + """A converter for AnalogWaveform types with 16-bit integer data.""" + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return AnalogWaveform + + @property + def python_typename(self) -> str: + """The Python type name that this converter handles.""" + base_typename = super().python_typename + return f"{base_typename}[int16]" + + @property + def protobuf_message(self) -> Type[waveform_pb2.I16AnalogWaveform]: + """The type-specific protobuf message for the Python type.""" + return waveform_pb2.I16AnalogWaveform + + def to_protobuf_message( + self, python_value: AnalogWaveform[np.int16] + ) -> waveform_pb2.I16AnalogWaveform: + """Convert the Python AnalogWaveform to a protobuf Int16AnalogWaveformConverter.""" + return waveform_conversion.int16_analog_waveform_to_protobuf(python_value) + + def to_python_value( + self, protobuf_message: waveform_pb2.I16AnalogWaveform + ) -> AnalogWaveform[np.int16]: + """Convert the protobuf Int16AnalogWaveformConverter to a Python AnalogWaveform.""" + return waveform_conversion.int16_analog_waveform_from_protobuf(protobuf_message) + + +class DoubleComplexWaveformConverter( + Converter[ComplexWaveform[np.complex128], waveform_pb2.DoubleComplexWaveform] +): + """A converter for complex waveform types with 64-bit real and imaginary data.""" + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return ComplexWaveform + + @property + def python_typename(self) -> str: + """The Python type name that this converter handles.""" + base_typename = super().python_typename + return f"{base_typename}[complex128]" + + @property + def protobuf_message(self) -> Type[waveform_pb2.DoubleComplexWaveform]: + """The type-specific protobuf message for the Python type.""" + return waveform_pb2.DoubleComplexWaveform + + def to_protobuf_message( + self, python_value: ComplexWaveform[np.complex128] + ) -> waveform_pb2.DoubleComplexWaveform: + """Convert the Python ComplexWaveform to a protobuf DoubleComplexWaveform.""" + return waveform_conversion.float64_complex_waveform_to_protobuf(python_value) + + def to_python_value( + self, protobuf_message: waveform_pb2.DoubleComplexWaveform + ) -> ComplexWaveform[np.complex128]: + """Convert the protobuf DoubleComplexWaveform to a Python ComplexWaveform.""" + return waveform_conversion.float64_complex_waveform_from_protobuf(protobuf_message) + + +class Int16ComplexWaveformConverter( + Converter[ComplexWaveform[ComplexInt32Base], waveform_pb2.I16ComplexWaveform] +): + """A converter for complex waveform types with 16-bit real and imaginary data.""" + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return ComplexWaveform + + @property + def python_typename(self) -> str: + """The Python type name that this converter handles.""" + base_typename = super().python_typename + # Use the string representation of ComplexInt32DType + return f"{base_typename}[[('real', ' Type[waveform_pb2.I16ComplexWaveform]: + """The type-specific protobuf message for the Python type.""" + return waveform_pb2.I16ComplexWaveform + + def to_protobuf_message( + self, python_value: ComplexWaveform[ComplexInt32Base] + ) -> waveform_pb2.I16ComplexWaveform: + """Convert the Python ComplexWaveform to a protobuf I16ComplexWaveform.""" + return waveform_conversion.int16_complex_waveform_to_protobuf(python_value) + + def to_python_value( + self, protobuf_message: waveform_pb2.I16ComplexWaveform + ) -> ComplexWaveform[ComplexInt32Base]: + """Convert the protobuf I16ComplexWaveform to a Python ComplexWaveform.""" + return waveform_conversion.int16_complex_waveform_from_protobuf(protobuf_message) + + +class DigitalWaveformConverter(Converter[DigitalWaveform[Any], waveform_pb2.DigitalWaveform]): + """A converter for digital waveform types.""" + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return DigitalWaveform + + @property + def protobuf_message(self) -> Type[waveform_pb2.DigitalWaveform]: + """The type-specific protobuf message for the Python type.""" + return waveform_pb2.DigitalWaveform + + def to_protobuf_message( + self, python_value: DigitalWaveform[Any] + ) -> waveform_pb2.DigitalWaveform: + """Convert the Python DigitalWaveform to a protobuf DigitalWaveform.""" + return waveform_conversion.digital_waveform_to_protobuf(python_value) + + def to_python_value( + self, protobuf_message: waveform_pb2.DigitalWaveform + ) -> DigitalWaveform[Any]: + """Convert the protobuf DigitalWaveform to a Python DigitalWaveform.""" + return waveform_conversion.digital_waveform_from_protobuf(protobuf_message) + + +class DoubleSpectrumConverter(Converter[Spectrum[np.float64], waveform_pb2.DoubleSpectrum]): + """A converter for spectrums with float64 data.""" + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return Spectrum + + @property + def protobuf_message(self) -> Type[waveform_pb2.DoubleSpectrum]: + """The type-specific protobuf message for the Python type.""" + return waveform_pb2.DoubleSpectrum + + def to_protobuf_message( + self, python_value: Spectrum[np.float64] + ) -> waveform_pb2.DoubleSpectrum: + """Convert the Python Spectrum to a protobuf DoubleSpectrum.""" + return waveform_conversion.float64_spectrum_to_protobuf(python_value) + + def to_python_value( + self, protobuf_message: waveform_pb2.DoubleSpectrum + ) -> Spectrum[np.float64]: + """Convert the protobuf DoubleSpectrum to a Python Spectrum.""" + return waveform_conversion.float64_spectrum_from_protobuf(protobuf_message) + + +class BTDateTimeConverter(Converter[bt.DateTime, precision_timestamp_pb2.PrecisionTimestamp]): + """A converter for bintime.DateTime types. + + .. note:: The nipanel package will always convert PrecisionTimestamp messages to + hightime.datetime objects using HTDateTimeConverter. To use bintime.DateTime + values in a panel, you must pass a bintime.DateTime value for the default_value + parameter of the get_value() method on the panel. + """ + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return bt.DateTime + + @property + def protobuf_message(self) -> Type[precision_timestamp_pb2.PrecisionTimestamp]: + """The type-specific protobuf message for the Python type.""" + return precision_timestamp_pb2.PrecisionTimestamp + + @property + def protobuf_typename(self) -> str: + """The protobuf name for the type.""" + # Override the base class here because there can only be one converter that + # converts PrecisionTimestamp objects. Since there are two converters that convert + # to PrecisionTimestamp, we have to choose one to handle conversion from protobuf. + # For the purposes of nipanel, we'll convert PrecisionTimestamp messages to + # hightime.datetime. See HTDateTimeConverter. + return "PrecisionTimestamp_Placeholder" + + def to_protobuf_message( + self, python_value: bt.DateTime + ) -> precision_timestamp_pb2.PrecisionTimestamp: + """Convert the Python DateTime to a protobuf PrecisionTimestamp.""" + return precision_timestamp_conversion.bintime_datetime_to_protobuf(python_value) + + def to_python_value( + self, protobuf_message: precision_timestamp_pb2.PrecisionTimestamp + ) -> bt.DateTime: + """Convert the protobuf PrecisionTimestamp to a Python DateTime.""" + return precision_timestamp_conversion.bintime_datetime_from_protobuf(protobuf_message) + + +class BTTimeDeltaConverter(Converter[bt.TimeDelta, precision_duration_pb2.PrecisionDuration]): + """A converter for bintime.TimeDelta types. + + .. note:: The nipanel package will always convert PrecisionDuration messages to + hightime.timedelta objects using HTTimeDeltaConverter. To use bintime.TimeDelta + values in a panel, you must pass a bintime.TimeDelta value for the default_value + parameter of the get_value() method on the panel. + """ + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return bt.TimeDelta + + @property + def protobuf_message(self) -> Type[precision_duration_pb2.PrecisionDuration]: + """The type-specific protobuf message for the Python type.""" + return precision_duration_pb2.PrecisionDuration + + @property + def protobuf_typename(self) -> str: + """The protobuf name for the type.""" + # Override the base class here because there can only be one converter that + # converts PrecisionDuration objects. Since there are two converters that convert + # to PrecisionDuration, we have to choose one to handle conversion from protobuf. + # For the purposes of nipanel, we'll convert PrecisionDuration messages to + # hightime.timedelta. See HTTimeDeltaConverter. + return "PrecisionDuration_Placeholder" + + def to_protobuf_message( + self, python_value: bt.TimeDelta + ) -> precision_duration_pb2.PrecisionDuration: + """Convert the Python TimeDelta to a protobuf PrecisionDuration.""" + return precision_duration_conversion.bintime_timedelta_to_protobuf(python_value) + + def to_python_value( + self, protobuf_message: precision_duration_pb2.PrecisionDuration + ) -> bt.TimeDelta: + """Convert the protobuf PrecisionDuration to a Python TimeDelta.""" + return precision_duration_conversion.bintime_timedelta_from_protobuf(protobuf_message) + + +class HTDateTimeConverter(Converter[ht.datetime, precision_timestamp_pb2.PrecisionTimestamp]): + """A converter for hightime.datetime objects.""" + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return ht.datetime + + @property + def protobuf_message(self) -> Type[precision_timestamp_pb2.PrecisionTimestamp]: + """The type-specific protobuf message for the Python type.""" + return precision_timestamp_pb2.PrecisionTimestamp + + def to_protobuf_message( + self, python_value: ht.datetime + ) -> precision_timestamp_pb2.PrecisionTimestamp: + """Convert the Python DateTime to a protobuf PrecisionTimestamp.""" + return precision_timestamp_conversion.hightime_datetime_to_protobuf(python_value) + + def to_python_value( + self, protobuf_message: precision_timestamp_pb2.PrecisionTimestamp + ) -> ht.datetime: + """Convert the protobuf PrecisionTimestamp to a Python DateTime.""" + return precision_timestamp_conversion.hightime_datetime_from_protobuf(protobuf_message) + + +class HTTimeDeltaConverter(Converter[ht.timedelta, precision_duration_pb2.PrecisionDuration]): + """A converter for hightime.timedelta objects.""" + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return ht.timedelta + + @property + def protobuf_message(self) -> Type[precision_duration_pb2.PrecisionDuration]: + """The type-specific protobuf message for the Python type.""" + return precision_duration_pb2.PrecisionDuration + + def to_protobuf_message( + self, python_value: ht.timedelta + ) -> precision_duration_pb2.PrecisionDuration: + """Convert the Python timedelta to a protobuf PrecisionDuration.""" + return precision_duration_conversion.hightime_timedelta_to_protobuf(python_value) + + def to_python_value( + self, protobuf_message: precision_duration_pb2.PrecisionDuration + ) -> ht.timedelta: + """Convert the protobuf PrecisionDuration to a Python timedelta.""" + return precision_duration_conversion.hightime_timedelta_from_protobuf(protobuf_message) + + +class ScalarConverter(Converter[Scalar[_AnyScalarType], scalar_pb2.Scalar]): + """A converter for Scalar objects.""" + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return Scalar + + @property + def protobuf_message(self) -> Type[scalar_pb2.Scalar]: + """The type-specific protobuf message for the Python type.""" + return scalar_pb2.Scalar + + def to_protobuf_message(self, python_value: Scalar[_AnyScalarType]) -> scalar_pb2.Scalar: + """Convert the Python Scalar to a protobuf scalar_pb2.Scalar.""" + return scalar_conversion.scalar_to_protobuf(python_value) + + def to_python_value(self, protobuf_message: scalar_pb2.Scalar) -> Scalar[_AnyScalarType]: + """Convert the protobuf message to a Python Scalar.""" + return scalar_conversion.scalar_from_protobuf(protobuf_message) + + +class VectorConverter(Converter[Vector[_AnyScalarType], vector_pb2.Vector]): + """A converter for Vector objects.""" + + @property + def python_type(self) -> type: + """The Python type that this converter handles.""" + return Vector + + @property + def protobuf_message(self) -> Type[vector_pb2.Vector]: + """The type-specific protobuf message for the Python type.""" + return vector_pb2.Vector + + def to_protobuf_message(self, python_value: Vector[Any]) -> vector_pb2.Vector: + """Convert the Python Vector to a protobuf vector_pb2.Vector.""" + return vector_conversion.vector_to_protobuf(python_value) + + def to_python_value(self, protobuf_message: vector_pb2.Vector) -> Vector[_AnyScalarType]: + """Convert the protobuf message to a Python Vector.""" + return vector_conversion.vector_from_protobuf(protobuf_message)