sift_py.ingestion.channel

  1from __future__ import annotations
  2
  3from enum import Enum
  4from typing import List, Optional, Type, TypedDict, Union
  5
  6import sift.common.type.v1.channel_data_type_pb2 as channel_pb
  7from google.protobuf.empty_pb2 import Empty
  8from sift.channels.v3.channels_pb2 import Channel as ChannelPb
  9from sift.common.type.v1.channel_bit_field_element_pb2 import (
 10    ChannelBitFieldElement as ChannelBitFieldElementPb,
 11)
 12from sift.common.type.v1.channel_enum_type_pb2 import (
 13    ChannelEnumType as ChannelEnumTypePb,
 14)
 15from sift.ingest.v1.ingest_pb2 import IngestWithConfigDataChannelValue
 16from sift.ingestion_configs.v2.ingestion_configs_pb2 import ChannelConfig as ChannelConfigPb
 17from typing_extensions import NotRequired, Self
 18
 19from sift_py._internal.channel import channel_fqn as _channel_fqn
 20from sift_py._internal.convert.protobuf import AsProtobuf
 21from sift_py.error import _component_deprecation_warning
 22
 23
 24class ChannelValue(TypedDict):
 25    """
 26    Represents a fully qualified data point for a channel
 27    """
 28
 29    channel_name: str
 30    component: NotRequired[str]  # Deprecated
 31    value: IngestWithConfigDataChannelValue
 32
 33
 34class ChannelConfig(AsProtobuf):
 35    """
 36    A description for a channel
 37    """
 38
 39    name: str
 40    data_type: ChannelDataType
 41    description: Optional[str]
 42    unit: Optional[str]
 43    component: Optional[str]  # Deprecated
 44    bit_field_elements: List[ChannelBitFieldElement]
 45    enum_types: List[ChannelEnumType]
 46    identifier: str
 47
 48    def __init__(
 49        self,
 50        name: str,
 51        data_type: ChannelDataType,
 52        description: Optional[str] = None,
 53        unit: Optional[str] = None,
 54        component: Optional[str] = None,  # Deprecated
 55        bit_field_elements: List[ChannelBitFieldElement] = [],
 56        enum_types: List[ChannelEnumType] = [],
 57    ):
 58        self.name = name
 59        self.data_type = data_type
 60        self.description = description
 61        self.unit = unit
 62
 63        self.component = None  # Field kept for backwards compatibility
 64        if component:
 65            _component_deprecation_warning()
 66            self.name = _channel_fqn(name=self.name, component=component)
 67            self.component = None
 68
 69        self.bit_field_elements = bit_field_elements
 70        self.enum_types = enum_types
 71        self.identifier = self.fqn()
 72
 73    def value_from(
 74        self, value: Optional[Union[int, float, bool, str]]
 75    ) -> Optional[IngestWithConfigDataChannelValue]:
 76        """
 77        Like `try_value_from` except will return `None` there is a failure to produce a channel value due to a type mismatch.
 78        """
 79        try:
 80            return self.try_value_from(value)
 81        except ValueError:
 82            return None
 83
 84    def try_value_from(
 85        self, value: Optional[Union[int, float, bool, str]]
 86    ) -> IngestWithConfigDataChannelValue:
 87        """
 88        Generate a channel value for this particular channel configuration. This will raise an exception
 89        if there is a type match, namely, if `value` isn't consistent with the channel's data-type. For a version
 90        of this function that does not raise an exception and simply ignores type mistmatches, see `value_from`. If `value`
 91        is `None` then an empty value will be generated.
 92        """
 93        if value is None:
 94            return empty_value()
 95
 96        if isinstance(value, bool) and self.data_type == ChannelDataType.BOOL:
 97            return bool_value(value)
 98        elif isinstance(value, int) or isinstance(value, float):
 99            if self.data_type == ChannelDataType.INT_32:
100                return int32_value(int(value))
101            elif self.data_type == ChannelDataType.INT_64:
102                return int64_value(int(value))
103            elif self.data_type == ChannelDataType.UINT_32:
104                return uint32_value(int(value))
105            elif self.data_type == ChannelDataType.UINT_64:
106                return uint64_value(int(value))
107            elif self.data_type == ChannelDataType.FLOAT:
108                return float_value(float(value))
109            elif self.data_type == ChannelDataType.DOUBLE:
110                return double_value(float(value))
111            elif self.data_type == ChannelDataType.ENUM:
112                return enum_value(int(value))
113        elif isinstance(value, str) and self.data_type == ChannelDataType.STRING:
114            return string_value(value)
115
116        raise ValueError(f"Failed to cast value of type {type(value)} to {self.data_type}")
117
118    def as_pb(self, klass: Type[ChannelConfigPb]) -> ChannelConfigPb:
119        return klass(
120            name=self.name,
121            unit=self.unit or "",
122            description=self.description or "",
123            data_type=self.data_type.value,
124            enum_types=[etype.as_pb(ChannelEnumTypePb) for etype in self.enum_types],
125            bit_field_elements=[
126                el.as_pb(ChannelBitFieldElementPb) for el in self.bit_field_elements
127            ],
128        )
129
130    @classmethod
131    def from_pb(cls, message: ChannelConfigPb) -> Self:
132        return cls(
133            name=message.name,
134            data_type=ChannelDataType.from_pb(message.data_type),
135            description=message.description,
136            unit=message.unit,
137            bit_field_elements=[
138                ChannelBitFieldElement.from_pb(el) for el in message.bit_field_elements
139            ],
140            enum_types=[ChannelEnumType.from_pb(etype) for etype in message.enum_types],
141        )
142
143    def fqn(self) -> str:
144        """
145        NOTE: Component field of Channel has been deprecated. Function kept for backwards compatibility.
146
147        The fully-qualified channel name of a channel called 'voltage' is simply `voltage`. The
148        fully qualified name of a channel called 'temperature' of component 'motor' is a `motor.temperature'.
149        """
150        return channel_fqn(self)
151
152
153class ChannelBitFieldElement(AsProtobuf):
154    name: str
155    index: int
156    bit_count: int
157
158    def __init__(self, name: str, index: int, bit_count: int):
159        self.name = name
160        self.index = index
161        self.bit_count = bit_count
162
163    def as_pb(self, klass: Type[ChannelBitFieldElementPb]) -> ChannelBitFieldElementPb:
164        return klass(
165            name=self.name,
166            index=self.index,
167            bit_count=self.bit_count,
168        )
169
170    @classmethod
171    def from_pb(cls, message: ChannelBitFieldElementPb) -> Self:
172        return cls(
173            name=message.name,
174            index=message.index,
175            bit_count=message.bit_count,
176        )
177
178
179class ChannelEnumType(AsProtobuf):
180    name: str
181    key: int
182    is_signed: Optional[bool] = None
183
184    def __init__(self, name: str, key: int, is_signed: Optional[bool] = None):
185        self.name = name
186        self.key = key
187        self.is_signed = is_signed
188
189    def as_pb(self, klass: Type[ChannelEnumTypePb]) -> ChannelEnumTypePb:
190        return klass(
191            name=self.name,
192            key=self.key,
193            is_signed=False if self.is_signed is None else self.is_signed,
194        )
195
196    @classmethod
197    def from_pb(cls, message: ChannelEnumTypePb) -> Self:
198        return cls(name=message.name, key=message.key, is_signed=message.is_signed)
199
200
201class ChannelDataTypeStrRep(Enum):
202    DOUBLE = "double"
203    STRING = "string"
204    ENUM = "enum"
205    BIT_FIELD = "bit_field"
206    BOOL = "bool"
207    FLOAT = "float"
208    INT_32 = "int32"
209    INT_64 = "int64"
210    UINT_32 = "uint32"
211    UINT_64 = "uint64"
212
213    @staticmethod
214    def from_api_format(val: str) -> Optional["ChannelDataTypeStrRep"]:
215        try:
216            return {
217                "CHANNEL_DATA_TYPE_DOUBLE": ChannelDataTypeStrRep.DOUBLE,
218                "CHANNEL_DATA_TYPE_STRING": ChannelDataTypeStrRep.STRING,
219                "CHANNEL_DATA_TYPE_ENUM": ChannelDataTypeStrRep.ENUM,
220                "CHANNEL_DATA_TYPE_BIT_FIELD": ChannelDataTypeStrRep.BIT_FIELD,
221                "CHANNEL_DATA_TYPE_BOOL": ChannelDataTypeStrRep.BOOL,
222                "CHANNEL_DATA_TYPE_FLOAT": ChannelDataTypeStrRep.FLOAT,
223                "CHANNEL_DATA_TYPE_INT_32": ChannelDataTypeStrRep.INT_32,
224                "CHANNEL_DATA_TYPE_INT_64": ChannelDataTypeStrRep.INT_64,
225                "CHANNEL_DATA_TYPE_UINT_32": ChannelDataTypeStrRep.UINT_32,
226                "CHANNEL_DATA_TYPE_UINT_64": ChannelDataTypeStrRep.UINT_64,
227            }[val]
228        except KeyError:
229            return None
230
231
232class ChannelDataType(Enum):
233    """
234    Utility enum class to simplify working with channel data-types generated from protobuf
235    """
236
237    DOUBLE = channel_pb.CHANNEL_DATA_TYPE_DOUBLE
238    STRING = channel_pb.CHANNEL_DATA_TYPE_STRING
239    ENUM = channel_pb.CHANNEL_DATA_TYPE_ENUM
240    BIT_FIELD = channel_pb.CHANNEL_DATA_TYPE_BIT_FIELD
241    BOOL = channel_pb.CHANNEL_DATA_TYPE_BOOL
242    FLOAT = channel_pb.CHANNEL_DATA_TYPE_FLOAT
243    INT_32 = channel_pb.CHANNEL_DATA_TYPE_INT_32
244    INT_64 = channel_pb.CHANNEL_DATA_TYPE_INT_64
245    UINT_32 = channel_pb.CHANNEL_DATA_TYPE_UINT_32
246    UINT_64 = channel_pb.CHANNEL_DATA_TYPE_UINT_64
247
248    @classmethod
249    def from_pb(cls, val: channel_pb.ChannelDataType.ValueType) -> "ChannelDataType":
250        if val == cls.DOUBLE.value:
251            return cls.DOUBLE
252        elif val == cls.STRING.value:
253            return cls.STRING
254        elif val == cls.ENUM.value:
255            return cls.ENUM
256        elif val == cls.BIT_FIELD.value:
257            return cls.BIT_FIELD
258        elif val == cls.BOOL.value:
259            return cls.BOOL
260        elif val == cls.FLOAT.value:
261            return cls.FLOAT
262        elif val == cls.INT_32.value:
263            return cls.INT_32
264        elif val == cls.INT_64.value:
265            return cls.INT_64
266        elif val == cls.UINT_32.value:
267            return cls.UINT_32
268        elif val == cls.UINT_64.value:
269            return cls.UINT_64
270        else:
271            raise ValueError(f"Unknown channel data type '{val}'.")
272
273    @classmethod
274    def from_str(cls, raw: str) -> Optional["ChannelDataType"]:
275        if raw.startswith("CHANNEL_DATA_TYPE_"):
276            val = ChannelDataTypeStrRep.from_api_format(raw)
277            if val is None:
278                return None
279        else:
280            try:
281                val = ChannelDataTypeStrRep(raw)
282            except ValueError:
283                return None
284
285        if val == ChannelDataTypeStrRep.DOUBLE:
286            return cls.DOUBLE
287        elif val == ChannelDataTypeStrRep.STRING:
288            return cls.STRING
289        elif val == ChannelDataTypeStrRep.ENUM:
290            return cls.ENUM
291        elif val == ChannelDataTypeStrRep.BIT_FIELD:
292            return cls.BIT_FIELD
293        elif val == ChannelDataTypeStrRep.BOOL:
294            return cls.BOOL
295        elif val == ChannelDataTypeStrRep.FLOAT:
296            return cls.FLOAT
297        elif val == ChannelDataTypeStrRep.INT_32:
298            return cls.INT_32
299        elif val == ChannelDataTypeStrRep.INT_64:
300            return cls.INT_64
301        elif val == ChannelDataTypeStrRep.UINT_32:
302            return cls.UINT_32
303        elif val == ChannelDataTypeStrRep.UINT_64:
304            return cls.UINT_64
305        else:
306            raise Exception("Unreachable")
307
308    def as_human_str(self, api_format: bool = False) -> str:
309        if self == ChannelDataType.DOUBLE:
310            return "CHANNEL_DATA_TYPE_DOUBLE" if api_format else ChannelDataTypeStrRep.DOUBLE.value
311        elif self == ChannelDataType.STRING:
312            return "CHANNEL_DATA_TYPE_STRING" if api_format else ChannelDataTypeStrRep.STRING.value
313        elif self == ChannelDataType.ENUM:
314            return "CHANNEL_DATA_TYPE_ENUM" if api_format else ChannelDataTypeStrRep.ENUM.value
315        elif self == ChannelDataType.BIT_FIELD:
316            return (
317                "CHANNEL_DATA_TYPE_BIT_FIELD"
318                if api_format
319                else ChannelDataTypeStrRep.BIT_FIELD.value
320            )
321        elif self == ChannelDataType.BOOL:
322            return "CHANNEL_DATA_TYPE_BOOL" if api_format else ChannelDataTypeStrRep.BOOL.value
323        elif self == ChannelDataType.FLOAT:
324            return "CHANNEL_DATA_TYPE_FLOAT" if api_format else ChannelDataTypeStrRep.FLOAT.value
325        elif self == ChannelDataType.INT_32:
326            return "CHANNEL_DATA_TYPE_INT_32" if api_format else ChannelDataTypeStrRep.INT_32.value
327        elif self == ChannelDataType.INT_64:
328            return "CHANNEL_DATA_TYPE_INT_64" if api_format else ChannelDataTypeStrRep.INT_64.value
329        elif self == ChannelDataType.UINT_32:
330            return (
331                "CHANNEL_DATA_TYPE_UINT_32" if api_format else ChannelDataTypeStrRep.UINT_32.value
332            )
333        elif self == ChannelDataType.UINT_64:
334            return (
335                "CHANNEL_DATA_TYPE_UINT_64" if api_format else ChannelDataTypeStrRep.UINT_64.value
336            )
337        else:
338            raise Exception("Unreachable.")
339
340
341class _AbstractChannel(TypedDict):
342    channel_name: str
343    component: Optional[str]  # Deprecated
344
345
346def channel_fqn(
347    channel: Union[
348        ChannelConfig,
349        ChannelConfigPb,
350        ChannelValue,
351        ChannelPb,
352        _AbstractChannel,
353    ],
354) -> str:
355    """
356    Computes the fully qualified channel name.
357
358    NOTE: Component field of Channel is deprecated and should not be used. Function is left for code compatibility.
359
360    The fully-qualified channel name of a channel called 'voltage' is simply `voltage'. The
361    fully qualified name of a channel called 'temperature' of component 'motor' is a `motor.temperature'.
362    """
363
364    if isinstance(channel, ChannelConfig):
365        if channel.component:
366            _component_deprecation_warning()
367        return _channel_fqn(channel.name, channel.component)
368    elif isinstance(channel, ChannelConfigPb):
369        return channel.name
370    elif isinstance(channel, ChannelPb):
371        return channel.name
372    else:
373        component = channel.get("component")
374        if component:
375            _component_deprecation_warning()
376        channel_name = channel["channel_name"]
377        return _channel_fqn(name=channel_name, component=component)
378
379
380def string_value(val: str) -> IngestWithConfigDataChannelValue:
381    return IngestWithConfigDataChannelValue(string=val)
382
383
384def double_value(val: float) -> IngestWithConfigDataChannelValue:
385    return IngestWithConfigDataChannelValue(double=val)
386
387
388def float_value(val: float) -> IngestWithConfigDataChannelValue:
389    return IngestWithConfigDataChannelValue(float=val)
390
391
392def bool_value(val: bool) -> IngestWithConfigDataChannelValue:
393    return IngestWithConfigDataChannelValue(bool=val)
394
395
396def int32_value(val: int) -> IngestWithConfigDataChannelValue:
397    return IngestWithConfigDataChannelValue(int32=val)
398
399
400def uint32_value(val: int) -> IngestWithConfigDataChannelValue:
401    return IngestWithConfigDataChannelValue(uint32=val)
402
403
404def int64_value(val: int) -> IngestWithConfigDataChannelValue:
405    return IngestWithConfigDataChannelValue(int64=val)
406
407
408def uint64_value(val: int) -> IngestWithConfigDataChannelValue:
409    return IngestWithConfigDataChannelValue(uint64=val)
410
411
412def bit_field_value(val: bytes) -> IngestWithConfigDataChannelValue:
413    return IngestWithConfigDataChannelValue(bit_field=val)
414
415
416def enum_value(val: int) -> IngestWithConfigDataChannelValue:
417    return IngestWithConfigDataChannelValue(enum=val)
418
419
420def empty_value() -> IngestWithConfigDataChannelValue:
421    return IngestWithConfigDataChannelValue(empty=Empty())
422
423
424def is_data_type(val: IngestWithConfigDataChannelValue, target_type: ChannelDataType) -> bool:
425    if target_type == ChannelDataType.DOUBLE:
426        return val.HasField("double")
427    elif target_type == ChannelDataType.STRING:
428        return val.HasField("string")
429    elif target_type == ChannelDataType.ENUM:
430        return val.HasField("enum")
431    elif target_type == ChannelDataType.BIT_FIELD:
432        return val.HasField("bit_field")
433    elif target_type == ChannelDataType.BOOL:
434        return val.HasField("bool")
435    elif target_type == ChannelDataType.FLOAT:
436        return val.HasField("float")
437    elif target_type == ChannelDataType.INT_32:
438        return val.HasField("int32")
439    elif target_type == ChannelDataType.INT_64:
440        return val.HasField("int64")
441    elif target_type == ChannelDataType.UINT_32:
442        return val.HasField("uint32")
443    elif target_type == ChannelDataType.UINT_64:
444        return val.HasField("uint64")