sift_py.ingestion.channel

View Source

  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
183    def __init__(self, name: str, key: int):
184        self.name = name
185        self.key = key
186
187    def as_pb(self, klass: Type[ChannelEnumTypePb]) -> ChannelEnumTypePb:
188        return klass(name=self.name, key=self.key)
189
190    @classmethod
191    def from_pb(cls, message: ChannelEnumTypePb) -> Self:
192        return cls(name=message.name, key=message.key)
193
194
195class ChannelDataTypeStrRep(Enum):
196    DOUBLE = "double"
197    STRING = "string"
198    ENUM = "enum"
199    BIT_FIELD = "bit_field"
200    BOOL = "bool"
201    FLOAT = "float"
202    INT_32 = "int32"
203    INT_64 = "int64"
204    UINT_32 = "uint32"
205    UINT_64 = "uint64"
206
207    @staticmethod
208    def from_api_format(val: str) -> Optional["ChannelDataTypeStrRep"]:
209        try:
210            return {
211                "CHANNEL_DATA_TYPE_DOUBLE": ChannelDataTypeStrRep.DOUBLE,
212                "CHANNEL_DATA_TYPE_STRING": ChannelDataTypeStrRep.STRING,
213                "CHANNEL_DATA_TYPE_ENUM": ChannelDataTypeStrRep.ENUM,
214                "CHANNEL_DATA_TYPE_BIT_FIELD": ChannelDataTypeStrRep.BIT_FIELD,
215                "CHANNEL_DATA_TYPE_BOOL": ChannelDataTypeStrRep.BOOL,
216                "CHANNEL_DATA_TYPE_FLOAT": ChannelDataTypeStrRep.FLOAT,
217                "CHANNEL_DATA_TYPE_INT_32": ChannelDataTypeStrRep.INT_32,
218                "CHANNEL_DATA_TYPE_INT_64": ChannelDataTypeStrRep.INT_64,
219                "CHANNEL_DATA_TYPE_UINT_32": ChannelDataTypeStrRep.UINT_32,
220                "CHANNEL_DATA_TYPE_UINT_64": ChannelDataTypeStrRep.UINT_64,
221            }[val]
222        except KeyError:
223            return None
224
225
226class ChannelDataType(Enum):
227    """
228    Utility enum class to simplify working with channel data-types generated from protobuf
229    """
230
231    DOUBLE = channel_pb.CHANNEL_DATA_TYPE_DOUBLE
232    STRING = channel_pb.CHANNEL_DATA_TYPE_STRING
233    ENUM = channel_pb.CHANNEL_DATA_TYPE_ENUM
234    BIT_FIELD = channel_pb.CHANNEL_DATA_TYPE_BIT_FIELD
235    BOOL = channel_pb.CHANNEL_DATA_TYPE_BOOL
236    FLOAT = channel_pb.CHANNEL_DATA_TYPE_FLOAT
237    INT_32 = channel_pb.CHANNEL_DATA_TYPE_INT_32
238    INT_64 = channel_pb.CHANNEL_DATA_TYPE_INT_64
239    UINT_32 = channel_pb.CHANNEL_DATA_TYPE_UINT_32
240    UINT_64 = channel_pb.CHANNEL_DATA_TYPE_UINT_64
241
242    @classmethod
243    def from_pb(cls, val: channel_pb.ChannelDataType.ValueType) -> "ChannelDataType":
244        if val == cls.DOUBLE.value:
245            return cls.DOUBLE
246        elif val == cls.STRING.value:
247            return cls.STRING
248        elif val == cls.ENUM.value:
249            return cls.ENUM
250        elif val == cls.BIT_FIELD.value:
251            return cls.BIT_FIELD
252        elif val == cls.BOOL.value:
253            return cls.BOOL
254        elif val == cls.FLOAT.value:
255            return cls.FLOAT
256        elif val == cls.INT_32.value:
257            return cls.INT_32
258        elif val == cls.INT_64.value:
259            return cls.INT_64
260        elif val == cls.UINT_32.value:
261            return cls.UINT_32
262        elif val == cls.UINT_64.value:
263            return cls.UINT_64
264        else:
265            raise ValueError(f"Unknown channel data type '{val}'.")
266
267    @classmethod
268    def from_str(cls, raw: str) -> Optional["ChannelDataType"]:
269        if raw.startswith("CHANNEL_DATA_TYPE_"):
270            val = ChannelDataTypeStrRep.from_api_format(raw)
271            if val is None:
272                return None
273        else:
274            try:
275                val = ChannelDataTypeStrRep(raw)
276            except ValueError:
277                return None
278
279        if val == ChannelDataTypeStrRep.DOUBLE:
280            return cls.DOUBLE
281        elif val == ChannelDataTypeStrRep.STRING:
282            return cls.STRING
283        elif val == ChannelDataTypeStrRep.ENUM:
284            return cls.ENUM
285        elif val == ChannelDataTypeStrRep.BIT_FIELD:
286            return cls.BIT_FIELD
287        elif val == ChannelDataTypeStrRep.BOOL:
288            return cls.BOOL
289        elif val == ChannelDataTypeStrRep.FLOAT:
290            return cls.FLOAT
291        elif val == ChannelDataTypeStrRep.INT_32:
292            return cls.INT_32
293        elif val == ChannelDataTypeStrRep.INT_64:
294            return cls.INT_64
295        elif val == ChannelDataTypeStrRep.UINT_32:
296            return cls.UINT_32
297        elif val == ChannelDataTypeStrRep.UINT_64:
298            return cls.UINT_64
299        else:
300            raise Exception("Unreachable")
301
302    def as_human_str(self, api_format: bool = False) -> str:
303        if self == ChannelDataType.DOUBLE:
304            return "CHANNEL_DATA_TYPE_DOUBLE" if api_format else ChannelDataTypeStrRep.DOUBLE.value
305        elif self == ChannelDataType.STRING:
306            return "CHANNEL_DATA_TYPE_STRING" if api_format else ChannelDataTypeStrRep.STRING.value
307        elif self == ChannelDataType.ENUM:
308            return "CHANNEL_DATA_TYPE_ENUM" if api_format else ChannelDataTypeStrRep.ENUM.value
309        elif self == ChannelDataType.BIT_FIELD:
310            return (
311                "CHANNEL_DATA_TYPE_BIT_FIELD"
312                if api_format
313                else ChannelDataTypeStrRep.BIT_FIELD.value
314            )
315        elif self == ChannelDataType.BOOL:
316            return "CHANNEL_DATA_TYPE_BOOL" if api_format else ChannelDataTypeStrRep.BOOL.value
317        elif self == ChannelDataType.FLOAT:
318            return "CHANNEL_DATA_TYPE_FLOAT" if api_format else ChannelDataTypeStrRep.FLOAT.value
319        elif self == ChannelDataType.INT_32:
320            return "CHANNEL_DATA_TYPE_INT_32" if api_format else ChannelDataTypeStrRep.INT_32.value
321        elif self == ChannelDataType.INT_64:
322            return "CHANNEL_DATA_TYPE_INT_64" if api_format else ChannelDataTypeStrRep.INT_64.value
323        elif self == ChannelDataType.UINT_32:
324            return (
325                "CHANNEL_DATA_TYPE_UINT_32" if api_format else ChannelDataTypeStrRep.UINT_32.value
326            )
327        elif self == ChannelDataType.UINT_64:
328            return (
329                "CHANNEL_DATA_TYPE_UINT_64" if api_format else ChannelDataTypeStrRep.UINT_64.value
330            )
331        else:
332            raise Exception("Unreachable.")
333
334
335class _AbstractChannel(TypedDict):
336    channel_name: str
337    component: Optional[str]  # Deprecated
338
339
340def channel_fqn(
341    channel: Union[
342        ChannelConfig,
343        ChannelConfigPb,
344        ChannelValue,
345        ChannelPb,
346        _AbstractChannel,
347    ],
348) -> str:
349    """
350    Computes the fully qualified channel name.
351
352    NOTE: Component field of Channel is deprecated and should not be used. Function is left for code compatibility.
353
354    The fully-qualified channel name of a channel called 'voltage' is simply `voltage'. The
355    fully qualified name of a channel called 'temperature' of component 'motor' is a `motor.temperature'.
356    """
357
358    if isinstance(channel, ChannelConfig):
359        if channel.component:
360            _component_deprecation_warning()
361        return _channel_fqn(channel.name, channel.component)
362    elif isinstance(channel, ChannelConfigPb):
363        return channel.name
364    elif isinstance(channel, ChannelPb):
365        return channel.name
366    else:
367        component = channel.get("component")
368        if component:
369            _component_deprecation_warning()
370        channel_name = channel["channel_name"]
371        return _channel_fqn(name=channel_name, component=component)
372
373
374def string_value(val: str) -> IngestWithConfigDataChannelValue:
375    return IngestWithConfigDataChannelValue(string=val)
376
377
378def double_value(val: float) -> IngestWithConfigDataChannelValue:
379    return IngestWithConfigDataChannelValue(double=val)
380
381
382def float_value(val: float) -> IngestWithConfigDataChannelValue:
383    return IngestWithConfigDataChannelValue(float=val)
384
385
386def bool_value(val: bool) -> IngestWithConfigDataChannelValue:
387    return IngestWithConfigDataChannelValue(bool=val)
388
389
390def int32_value(val: int) -> IngestWithConfigDataChannelValue:
391    return IngestWithConfigDataChannelValue(int32=val)
392
393
394def uint32_value(val: int) -> IngestWithConfigDataChannelValue:
395    return IngestWithConfigDataChannelValue(uint32=val)
396
397
398def int64_value(val: int) -> IngestWithConfigDataChannelValue:
399    return IngestWithConfigDataChannelValue(int64=val)
400
401
402def uint64_value(val: int) -> IngestWithConfigDataChannelValue:
403    return IngestWithConfigDataChannelValue(uint64=val)
404
405
406def bit_field_value(val: bytes) -> IngestWithConfigDataChannelValue:
407    return IngestWithConfigDataChannelValue(bit_field=val)
408
409
410def enum_value(val: int) -> IngestWithConfigDataChannelValue:
411    return IngestWithConfigDataChannelValue(enum=val)
412
413
414def empty_value() -> IngestWithConfigDataChannelValue:
415    return IngestWithConfigDataChannelValue(empty=Empty())
416
417
418def is_data_type(val: IngestWithConfigDataChannelValue, target_type: ChannelDataType) -> bool:
419    if target_type == ChannelDataType.DOUBLE:
420        return val.HasField("double")
421    elif target_type == ChannelDataType.STRING:
422        return val.HasField("string")
423    elif target_type == ChannelDataType.ENUM:
424        return val.HasField("enum")
425    elif target_type == ChannelDataType.BIT_FIELD:
426        return val.HasField("bit_field")
427    elif target_type == ChannelDataType.BOOL:
428        return val.HasField("bool")
429    elif target_type == ChannelDataType.FLOAT:
430        return val.HasField("float")
431    elif target_type == ChannelDataType.INT_32:
432        return val.HasField("int32")
433    elif target_type == ChannelDataType.INT_64:
434        return val.HasField("int64")
435    elif target_type == ChannelDataType.UINT_32:
436        return val.HasField("uint32")
437    elif target_type == ChannelDataType.UINT_64:
438        return val.HasField("uint64")

class ChannelValue(builtins.dict): View Source

25class ChannelValue(TypedDict):
26    """
27    Represents a fully qualified data point for a channel
28    """
29
30    channel_name: str
31    component: NotRequired[str]  # Deprecated
32    value: IngestWithConfigDataChannelValue

Represents a fully qualified data point for a channel

channel_name: str

component: typing_extensions.NotRequired[str]

value: sift.ingest.v1.ingest_pb2.IngestWithConfigDataChannelValue

Inherited Members

builtins.dict: get; setdefault; pop; popitem; keys; items; values; update; fromkeys; clear; copy

class ChannelConfig(abc.ABC, typing.Generic[~T]): View Source

 35class ChannelConfig(AsProtobuf):
 36    """
 37    A description for a channel
 38    """
 39
 40    name: str
 41    data_type: ChannelDataType
 42    description: Optional[str]
 43    unit: Optional[str]
 44    component: Optional[str]  # Deprecated
 45    bit_field_elements: List[ChannelBitFieldElement]
 46    enum_types: List[ChannelEnumType]
 47    identifier: str
 48
 49    def __init__(
 50        self,
 51        name: str,
 52        data_type: ChannelDataType,
 53        description: Optional[str] = None,
 54        unit: Optional[str] = None,
 55        component: Optional[str] = None,  # Deprecated
 56        bit_field_elements: List[ChannelBitFieldElement] = [],
 57        enum_types: List[ChannelEnumType] = [],
 58    ):
 59        self.name = name
 60        self.data_type = data_type
 61        self.description = description
 62        self.unit = unit
 63
 64        self.component = None  # Field kept for backwards compatibility
 65        if component:
 66            _component_deprecation_warning()
 67            self.name = _channel_fqn(name=self.name, component=component)
 68            self.component = None
 69
 70        self.bit_field_elements = bit_field_elements
 71        self.enum_types = enum_types
 72        self.identifier = self.fqn()
 73
 74    def value_from(
 75        self, value: Optional[Union[int, float, bool, str]]
 76    ) -> Optional[IngestWithConfigDataChannelValue]:
 77        """
 78        Like `try_value_from` except will return `None` there is a failure to produce a channel value due to a type mismatch.
 79        """
 80        try:
 81            return self.try_value_from(value)
 82        except ValueError:
 83            return None
 84
 85    def try_value_from(
 86        self, value: Optional[Union[int, float, bool, str]]
 87    ) -> IngestWithConfigDataChannelValue:
 88        """
 89        Generate a channel value for this particular channel configuration. This will raise an exception
 90        if there is a type match, namely, if `value` isn't consistent with the channel's data-type. For a version
 91        of this function that does not raise an exception and simply ignores type mistmatches, see `value_from`. If `value`
 92        is `None` then an empty value will be generated.
 93        """
 94        if value is None:
 95            return empty_value()
 96
 97        if isinstance(value, bool) and self.data_type == ChannelDataType.BOOL:
 98            return bool_value(value)
 99        elif isinstance(value, int) or isinstance(value, float):
100            if self.data_type == ChannelDataType.INT_32:
101                return int32_value(int(value))
102            elif self.data_type == ChannelDataType.INT_64:
103                return int64_value(int(value))
104            elif self.data_type == ChannelDataType.UINT_32:
105                return uint32_value(int(value))
106            elif self.data_type == ChannelDataType.UINT_64:
107                return uint64_value(int(value))
108            elif self.data_type == ChannelDataType.FLOAT:
109                return float_value(float(value))
110            elif self.data_type == ChannelDataType.DOUBLE:
111                return double_value(float(value))
112            elif self.data_type == ChannelDataType.ENUM:
113                return enum_value(int(value))
114        elif isinstance(value, str) and self.data_type == ChannelDataType.STRING:
115            return string_value(value)
116
117        raise ValueError(f"Failed to cast value of type {type(value)} to {self.data_type}")
118
119    def as_pb(self, klass: Type[ChannelConfigPb]) -> ChannelConfigPb:
120        return klass(
121            name=self.name,
122            unit=self.unit or "",
123            description=self.description or "",
124            data_type=self.data_type.value,
125            enum_types=[etype.as_pb(ChannelEnumTypePb) for etype in self.enum_types],
126            bit_field_elements=[
127                el.as_pb(ChannelBitFieldElementPb) for el in self.bit_field_elements
128            ],
129        )
130
131    @classmethod
132    def from_pb(cls, message: ChannelConfigPb) -> Self:
133        return cls(
134            name=message.name,
135            data_type=ChannelDataType.from_pb(message.data_type),
136            description=message.description,
137            unit=message.unit,
138            bit_field_elements=[
139                ChannelBitFieldElement.from_pb(el) for el in message.bit_field_elements
140            ],
141            enum_types=[ChannelEnumType.from_pb(etype) for etype in message.enum_types],
142        )
143
144    def fqn(self) -> str:
145        """
146        NOTE: Component field of Channel has been deprecated. Function kept for backwards compatibility.
147
148        The fully-qualified channel name of a channel called 'voltage' is simply `voltage`. The
149        fully qualified name of a channel called 'temperature' of component 'motor' is a `motor.temperature'.
150        """
151        return channel_fqn(self)

A description for a channel

ChannelConfig( name: str, data_type: ChannelDataType, description: Union[str, NoneType] = None, unit: Union[str, NoneType] = None, component: Union[str, NoneType] = None, bit_field_elements: List[ChannelBitFieldElement] = [], enum_types: List[ChannelEnumType] = []) View Source

49    def __init__(
50        self,
51        name: str,
52        data_type: ChannelDataType,
53        description: Optional[str] = None,
54        unit: Optional[str] = None,
55        component: Optional[str] = None,  # Deprecated
56        bit_field_elements: List[ChannelBitFieldElement] = [],
57        enum_types: List[ChannelEnumType] = [],
58    ):
59        self.name = name
60        self.data_type = data_type
61        self.description = description
62        self.unit = unit
63
64        self.component = None  # Field kept for backwards compatibility
65        if component:
66            _component_deprecation_warning()
67            self.name = _channel_fqn(name=self.name, component=component)
68            self.component = None
69
70        self.bit_field_elements = bit_field_elements
71        self.enum_types = enum_types
72        self.identifier = self.fqn()

data_type: ChannelDataType

description: Union[str, NoneType]

unit: Union[str, NoneType]

component: Union[str, NoneType]

bit_field_elements: List[ChannelBitFieldElement]

enum_types: List[ChannelEnumType]

identifier: str

def value_from( self, value: Union[int, float, bool, str, NoneType]) -> Union[sift.ingest.v1.ingest_pb2.IngestWithConfigDataChannelValue, NoneType]: View Source

74    def value_from(
75        self, value: Optional[Union[int, float, bool, str]]
76    ) -> Optional[IngestWithConfigDataChannelValue]:
77        """
78        Like `try_value_from` except will return `None` there is a failure to produce a channel value due to a type mismatch.
79        """
80        try:
81            return self.try_value_from(value)
82        except ValueError:
83            return None

Like try_value_from except will return None there is a failure to produce a channel value due to a type mismatch.

def try_value_from( self, value: Union[int, float, bool, str, NoneType]) -> sift.ingest.v1.ingest_pb2.IngestWithConfigDataChannelValue: View Source

 85    def try_value_from(
 86        self, value: Optional[Union[int, float, bool, str]]
 87    ) -> IngestWithConfigDataChannelValue:
 88        """
 89        Generate a channel value for this particular channel configuration. This will raise an exception
 90        if there is a type match, namely, if `value` isn't consistent with the channel's data-type. For a version
 91        of this function that does not raise an exception and simply ignores type mistmatches, see `value_from`. If `value`
 92        is `None` then an empty value will be generated.
 93        """
 94        if value is None:
 95            return empty_value()
 96
 97        if isinstance(value, bool) and self.data_type == ChannelDataType.BOOL:
 98            return bool_value(value)
 99        elif isinstance(value, int) or isinstance(value, float):
100            if self.data_type == ChannelDataType.INT_32:
101                return int32_value(int(value))
102            elif self.data_type == ChannelDataType.INT_64:
103                return int64_value(int(value))
104            elif self.data_type == ChannelDataType.UINT_32:
105                return uint32_value(int(value))
106            elif self.data_type == ChannelDataType.UINT_64:
107                return uint64_value(int(value))
108            elif self.data_type == ChannelDataType.FLOAT:
109                return float_value(float(value))
110            elif self.data_type == ChannelDataType.DOUBLE:
111                return double_value(float(value))
112            elif self.data_type == ChannelDataType.ENUM:
113                return enum_value(int(value))
114        elif isinstance(value, str) and self.data_type == ChannelDataType.STRING:
115            return string_value(value)
116
117        raise ValueError(f"Failed to cast value of type {type(value)} to {self.data_type}")

Generate a channel value for this particular channel configuration. This will raise an exception if there is a type match, namely, if value isn't consistent with the channel's data-type. For a version of this function that does not raise an exception and simply ignores type mistmatches, see value_from. If value is None then an empty value will be generated.

def as_pb( self, klass: Type[sift.ingestion_configs.v2.ingestion_configs_pb2.ChannelConfig]) -> sift.ingestion_configs.v2.ingestion_configs_pb2.ChannelConfig: View Source

119    def as_pb(self, klass: Type[ChannelConfigPb]) -> ChannelConfigPb:
120        return klass(
121            name=self.name,
122            unit=self.unit or "",
123            description=self.description or "",
124            data_type=self.data_type.value,
125            enum_types=[etype.as_pb(ChannelEnumTypePb) for etype in self.enum_types],
126            bit_field_elements=[
127                el.as_pb(ChannelBitFieldElementPb) for el in self.bit_field_elements
128            ],
129        )

Performs the conversion into a sub-type of ProtobufMessage.

@classmethod

def from_pb( cls, message: sift.ingestion_configs.v2.ingestion_configs_pb2.ChannelConfig) -> typing_extensions.Self: View Source

131    @classmethod
132    def from_pb(cls, message: ChannelConfigPb) -> Self:
133        return cls(
134            name=message.name,
135            data_type=ChannelDataType.from_pb(message.data_type),
136            description=message.description,
137            unit=message.unit,
138            bit_field_elements=[
139                ChannelBitFieldElement.from_pb(el) for el in message.bit_field_elements
140            ],
141            enum_types=[ChannelEnumType.from_pb(etype) for etype in message.enum_types],
142        )

Converts a protobuf object to the type of the sub-class class.

def fqn(self) -> str: View Source

144    def fqn(self) -> str:
145        """
146        NOTE: Component field of Channel has been deprecated. Function kept for backwards compatibility.
147
148        The fully-qualified channel name of a channel called 'voltage' is simply `voltage`. The
149        fully qualified name of a channel called 'temperature' of component 'motor' is a `motor.temperature'.
150        """
151        return channel_fqn(self)

NOTE: Component field of Channel has been deprecated. Function kept for backwards compatibility.

The fully-qualified channel name of a channel called 'voltage' is simply voltage. The fully qualified name of a channel called 'temperature' of component 'motor' is a `motor.temperature'.

class ChannelBitFieldElement(abc.ABC, typing.Generic[~T]): View Source

154class ChannelBitFieldElement(AsProtobuf):
155    name: str
156    index: int
157    bit_count: int
158
159    def __init__(self, name: str, index: int, bit_count: int):
160        self.name = name
161        self.index = index
162        self.bit_count = bit_count
163
164    def as_pb(self, klass: Type[ChannelBitFieldElementPb]) -> ChannelBitFieldElementPb:
165        return klass(
166            name=self.name,
167            index=self.index,
168            bit_count=self.bit_count,
169        )
170
171    @classmethod
172    def from_pb(cls, message: ChannelBitFieldElementPb) -> Self:
173        return cls(
174            name=message.name,
175            index=message.index,
176            bit_count=message.bit_count,
177        )

Abstract base class used to create create sub-types that can be treated as an object that can be converted into an instance of ProtobufMessage.

If there are multiple possible protobuf targets then as_pb may be overloaded.

ChannelBitFieldElement(name: str, index: int, bit_count: int) View Source

159    def __init__(self, name: str, index: int, bit_count: int):
160        self.name = name
161        self.index = index
162        self.bit_count = bit_count

index: int

bit_count: int

def as_pb( self, klass: Type[sift.common.type.v1.channel_bit_field_element_pb2.ChannelBitFieldElement]) -> sift.common.type.v1.channel_bit_field_element_pb2.ChannelBitFieldElement: View Source

164    def as_pb(self, klass: Type[ChannelBitFieldElementPb]) -> ChannelBitFieldElementPb:
165        return klass(
166            name=self.name,
167            index=self.index,
168            bit_count=self.bit_count,
169        )

Performs the conversion into a sub-type of ProtobufMessage.

@classmethod

def from_pb( cls, message: sift.common.type.v1.channel_bit_field_element_pb2.ChannelBitFieldElement) -> typing_extensions.Self: View Source

171    @classmethod
172    def from_pb(cls, message: ChannelBitFieldElementPb) -> Self:
173        return cls(
174            name=message.name,
175            index=message.index,
176            bit_count=message.bit_count,
177        )

Converts a protobuf object to the type of the sub-class class.

class ChannelEnumType(abc.ABC, typing.Generic[~T]): View Source

180class ChannelEnumType(AsProtobuf):
181    name: str
182    key: int
183
184    def __init__(self, name: str, key: int):
185        self.name = name
186        self.key = key
187
188    def as_pb(self, klass: Type[ChannelEnumTypePb]) -> ChannelEnumTypePb:
189        return klass(name=self.name, key=self.key)
190
191    @classmethod
192    def from_pb(cls, message: ChannelEnumTypePb) -> Self:
193        return cls(name=message.name, key=message.key)

Abstract base class used to create create sub-types that can be treated as an object that can be converted into an instance of ProtobufMessage.

If there are multiple possible protobuf targets then as_pb may be overloaded.

ChannelEnumType(name: str, key: int) View Source

184    def __init__(self, name: str, key: int):
185        self.name = name
186        self.key = key

key: int

def as_pb( self, klass: Type[sift.common.type.v1.channel_enum_type_pb2.ChannelEnumType]) -> sift.common.type.v1.channel_enum_type_pb2.ChannelEnumType: View Source

188    def as_pb(self, klass: Type[ChannelEnumTypePb]) -> ChannelEnumTypePb:
189        return klass(name=self.name, key=self.key)

Performs the conversion into a sub-type of ProtobufMessage.

@classmethod

def from_pb( cls, message: sift.common.type.v1.channel_enum_type_pb2.ChannelEnumType) -> typing_extensions.Self: View Source

191    @classmethod
192    def from_pb(cls, message: ChannelEnumTypePb) -> Self:
193        return cls(name=message.name, key=message.key)

Converts a protobuf object to the type of the sub-class class.

class ChannelDataTypeStrRep(enum.Enum): View Source

196class ChannelDataTypeStrRep(Enum):
197    DOUBLE = "double"
198    STRING = "string"
199    ENUM = "enum"
200    BIT_FIELD = "bit_field"
201    BOOL = "bool"
202    FLOAT = "float"
203    INT_32 = "int32"
204    INT_64 = "int64"
205    UINT_32 = "uint32"
206    UINT_64 = "uint64"
207
208    @staticmethod
209    def from_api_format(val: str) -> Optional["ChannelDataTypeStrRep"]:
210        try:
211            return {
212                "CHANNEL_DATA_TYPE_DOUBLE": ChannelDataTypeStrRep.DOUBLE,
213                "CHANNEL_DATA_TYPE_STRING": ChannelDataTypeStrRep.STRING,
214                "CHANNEL_DATA_TYPE_ENUM": ChannelDataTypeStrRep.ENUM,
215                "CHANNEL_DATA_TYPE_BIT_FIELD": ChannelDataTypeStrRep.BIT_FIELD,
216                "CHANNEL_DATA_TYPE_BOOL": ChannelDataTypeStrRep.BOOL,
217                "CHANNEL_DATA_TYPE_FLOAT": ChannelDataTypeStrRep.FLOAT,
218                "CHANNEL_DATA_TYPE_INT_32": ChannelDataTypeStrRep.INT_32,
219                "CHANNEL_DATA_TYPE_INT_64": ChannelDataTypeStrRep.INT_64,
220                "CHANNEL_DATA_TYPE_UINT_32": ChannelDataTypeStrRep.UINT_32,
221                "CHANNEL_DATA_TYPE_UINT_64": ChannelDataTypeStrRep.UINT_64,
222            }[val]
223        except KeyError:
224            return None

An enumeration.

DOUBLE = <ChannelDataTypeStrRep.DOUBLE: 'double'>

STRING = <ChannelDataTypeStrRep.STRING: 'string'>

ENUM = <ChannelDataTypeStrRep.ENUM: 'enum'>

BIT_FIELD = <ChannelDataTypeStrRep.BIT_FIELD: 'bit_field'>

BOOL = <ChannelDataTypeStrRep.BOOL: 'bool'>

FLOAT = <ChannelDataTypeStrRep.FLOAT: 'float'>

INT_32 = <ChannelDataTypeStrRep.INT_32: 'int32'>

INT_64 = <ChannelDataTypeStrRep.INT_64: 'int64'>

UINT_32 = <ChannelDataTypeStrRep.UINT_32: 'uint32'>

UINT_64 = <ChannelDataTypeStrRep.UINT_64: 'uint64'>

@staticmethod

def from_api_format( val: str) -> Union[ChannelDataTypeStrRep, NoneType]: View Source

208    @staticmethod
209    def from_api_format(val: str) -> Optional["ChannelDataTypeStrRep"]:
210        try:
211            return {
212                "CHANNEL_DATA_TYPE_DOUBLE": ChannelDataTypeStrRep.DOUBLE,
213                "CHANNEL_DATA_TYPE_STRING": ChannelDataTypeStrRep.STRING,
214                "CHANNEL_DATA_TYPE_ENUM": ChannelDataTypeStrRep.ENUM,
215                "CHANNEL_DATA_TYPE_BIT_FIELD": ChannelDataTypeStrRep.BIT_FIELD,
216                "CHANNEL_DATA_TYPE_BOOL": ChannelDataTypeStrRep.BOOL,
217                "CHANNEL_DATA_TYPE_FLOAT": ChannelDataTypeStrRep.FLOAT,
218                "CHANNEL_DATA_TYPE_INT_32": ChannelDataTypeStrRep.INT_32,
219                "CHANNEL_DATA_TYPE_INT_64": ChannelDataTypeStrRep.INT_64,
220                "CHANNEL_DATA_TYPE_UINT_32": ChannelDataTypeStrRep.UINT_32,
221                "CHANNEL_DATA_TYPE_UINT_64": ChannelDataTypeStrRep.UINT_64,
222            }[val]
223        except KeyError:
224            return None

Inherited Members

enum.Enum: name; value

class ChannelDataType(enum.Enum): View Source

227class ChannelDataType(Enum):
228    """
229    Utility enum class to simplify working with channel data-types generated from protobuf
230    """
231
232    DOUBLE = channel_pb.CHANNEL_DATA_TYPE_DOUBLE
233    STRING = channel_pb.CHANNEL_DATA_TYPE_STRING
234    ENUM = channel_pb.CHANNEL_DATA_TYPE_ENUM
235    BIT_FIELD = channel_pb.CHANNEL_DATA_TYPE_BIT_FIELD
236    BOOL = channel_pb.CHANNEL_DATA_TYPE_BOOL
237    FLOAT = channel_pb.CHANNEL_DATA_TYPE_FLOAT
238    INT_32 = channel_pb.CHANNEL_DATA_TYPE_INT_32
239    INT_64 = channel_pb.CHANNEL_DATA_TYPE_INT_64
240    UINT_32 = channel_pb.CHANNEL_DATA_TYPE_UINT_32
241    UINT_64 = channel_pb.CHANNEL_DATA_TYPE_UINT_64
242
243    @classmethod
244    def from_pb(cls, val: channel_pb.ChannelDataType.ValueType) -> "ChannelDataType":
245        if val == cls.DOUBLE.value:
246            return cls.DOUBLE
247        elif val == cls.STRING.value:
248            return cls.STRING
249        elif val == cls.ENUM.value:
250            return cls.ENUM
251        elif val == cls.BIT_FIELD.value:
252            return cls.BIT_FIELD
253        elif val == cls.BOOL.value:
254            return cls.BOOL
255        elif val == cls.FLOAT.value:
256            return cls.FLOAT
257        elif val == cls.INT_32.value:
258            return cls.INT_32
259        elif val == cls.INT_64.value:
260            return cls.INT_64
261        elif val == cls.UINT_32.value:
262            return cls.UINT_32
263        elif val == cls.UINT_64.value:
264            return cls.UINT_64
265        else:
266            raise ValueError(f"Unknown channel data type '{val}'.")
267
268    @classmethod
269    def from_str(cls, raw: str) -> Optional["ChannelDataType"]:
270        if raw.startswith("CHANNEL_DATA_TYPE_"):
271            val = ChannelDataTypeStrRep.from_api_format(raw)
272            if val is None:
273                return None
274        else:
275            try:
276                val = ChannelDataTypeStrRep(raw)
277            except ValueError:
278                return None
279
280        if val == ChannelDataTypeStrRep.DOUBLE:
281            return cls.DOUBLE
282        elif val == ChannelDataTypeStrRep.STRING:
283            return cls.STRING
284        elif val == ChannelDataTypeStrRep.ENUM:
285            return cls.ENUM
286        elif val == ChannelDataTypeStrRep.BIT_FIELD:
287            return cls.BIT_FIELD
288        elif val == ChannelDataTypeStrRep.BOOL:
289            return cls.BOOL
290        elif val == ChannelDataTypeStrRep.FLOAT:
291            return cls.FLOAT
292        elif val == ChannelDataTypeStrRep.INT_32:
293            return cls.INT_32
294        elif val == ChannelDataTypeStrRep.INT_64:
295            return cls.INT_64
296        elif val == ChannelDataTypeStrRep.UINT_32:
297            return cls.UINT_32
298        elif val == ChannelDataTypeStrRep.UINT_64:
299            return cls.UINT_64
300        else:
301            raise Exception("Unreachable")
302
303    def as_human_str(self, api_format: bool = False) -> str:
304        if self == ChannelDataType.DOUBLE:
305            return "CHANNEL_DATA_TYPE_DOUBLE" if api_format else ChannelDataTypeStrRep.DOUBLE.value
306        elif self == ChannelDataType.STRING:
307            return "CHANNEL_DATA_TYPE_STRING" if api_format else ChannelDataTypeStrRep.STRING.value
308        elif self == ChannelDataType.ENUM:
309            return "CHANNEL_DATA_TYPE_ENUM" if api_format else ChannelDataTypeStrRep.ENUM.value
310        elif self == ChannelDataType.BIT_FIELD:
311            return (
312                "CHANNEL_DATA_TYPE_BIT_FIELD"
313                if api_format
314                else ChannelDataTypeStrRep.BIT_FIELD.value
315            )
316        elif self == ChannelDataType.BOOL:
317            return "CHANNEL_DATA_TYPE_BOOL" if api_format else ChannelDataTypeStrRep.BOOL.value
318        elif self == ChannelDataType.FLOAT:
319            return "CHANNEL_DATA_TYPE_FLOAT" if api_format else ChannelDataTypeStrRep.FLOAT.value
320        elif self == ChannelDataType.INT_32:
321            return "CHANNEL_DATA_TYPE_INT_32" if api_format else ChannelDataTypeStrRep.INT_32.value
322        elif self == ChannelDataType.INT_64:
323            return "CHANNEL_DATA_TYPE_INT_64" if api_format else ChannelDataTypeStrRep.INT_64.value
324        elif self == ChannelDataType.UINT_32:
325            return (
326                "CHANNEL_DATA_TYPE_UINT_32" if api_format else ChannelDataTypeStrRep.UINT_32.value
327            )
328        elif self == ChannelDataType.UINT_64:
329            return (
330                "CHANNEL_DATA_TYPE_UINT_64" if api_format else ChannelDataTypeStrRep.UINT_64.value
331            )
332        else:
333            raise Exception("Unreachable.")

Utility enum class to simplify working with channel data-types generated from protobuf

DOUBLE = <ChannelDataType.DOUBLE: 1>

STRING = <ChannelDataType.STRING: 2>

ENUM = <ChannelDataType.ENUM: 3>

BIT_FIELD = <ChannelDataType.BIT_FIELD: 4>

BOOL = <ChannelDataType.BOOL: 5>

FLOAT = <ChannelDataType.FLOAT: 6>

INT_32 = <ChannelDataType.INT_32: 7>

INT_64 = <ChannelDataType.INT_64: 9>

UINT_32 = <ChannelDataType.UINT_32: 8>

UINT_64 = <ChannelDataType.UINT_64: 10>

@classmethod

def from_pb(cls, val: int) -> ChannelDataType: View Source

243    @classmethod
244    def from_pb(cls, val: channel_pb.ChannelDataType.ValueType) -> "ChannelDataType":
245        if val == cls.DOUBLE.value:
246            return cls.DOUBLE
247        elif val == cls.STRING.value:
248            return cls.STRING
249        elif val == cls.ENUM.value:
250            return cls.ENUM
251        elif val == cls.BIT_FIELD.value:
252            return cls.BIT_FIELD
253        elif val == cls.BOOL.value:
254            return cls.BOOL
255        elif val == cls.FLOAT.value:
256            return cls.FLOAT
257        elif val == cls.INT_32.value:
258            return cls.INT_32
259        elif val == cls.INT_64.value:
260            return cls.INT_64
261        elif val == cls.UINT_32.value:
262            return cls.UINT_32
263        elif val == cls.UINT_64.value:
264            return cls.UINT_64
265        else:
266            raise ValueError(f"Unknown channel data type '{val}'.")

@classmethod

def from_str( cls, raw: str) -> Union[ChannelDataType, NoneType]: View Source

268    @classmethod
269    def from_str(cls, raw: str) -> Optional["ChannelDataType"]:
270        if raw.startswith("CHANNEL_DATA_TYPE_"):
271            val = ChannelDataTypeStrRep.from_api_format(raw)
272            if val is None:
273                return None
274        else:
275            try:
276                val = ChannelDataTypeStrRep(raw)
277            except ValueError:
278                return None
279
280        if val == ChannelDataTypeStrRep.DOUBLE:
281            return cls.DOUBLE
282        elif val == ChannelDataTypeStrRep.STRING:
283            return cls.STRING
284        elif val == ChannelDataTypeStrRep.ENUM:
285            return cls.ENUM
286        elif val == ChannelDataTypeStrRep.BIT_FIELD:
287            return cls.BIT_FIELD
288        elif val == ChannelDataTypeStrRep.BOOL:
289            return cls.BOOL
290        elif val == ChannelDataTypeStrRep.FLOAT:
291            return cls.FLOAT
292        elif val == ChannelDataTypeStrRep.INT_32:
293            return cls.INT_32
294        elif val == ChannelDataTypeStrRep.INT_64:
295            return cls.INT_64
296        elif val == ChannelDataTypeStrRep.UINT_32:
297            return cls.UINT_32
298        elif val == ChannelDataTypeStrRep.UINT_64:
299            return cls.UINT_64
300        else:
301            raise Exception("Unreachable")

def as_human_str(self, api_format: bool = False) -> str: View Source

303    def as_human_str(self, api_format: bool = False) -> str:
304        if self == ChannelDataType.DOUBLE:
305            return "CHANNEL_DATA_TYPE_DOUBLE" if api_format else ChannelDataTypeStrRep.DOUBLE.value
306        elif self == ChannelDataType.STRING:
307            return "CHANNEL_DATA_TYPE_STRING" if api_format else ChannelDataTypeStrRep.STRING.value
308        elif self == ChannelDataType.ENUM:
309            return "CHANNEL_DATA_TYPE_ENUM" if api_format else ChannelDataTypeStrRep.ENUM.value
310        elif self == ChannelDataType.BIT_FIELD:
311            return (
312                "CHANNEL_DATA_TYPE_BIT_FIELD"
313                if api_format
314                else ChannelDataTypeStrRep.BIT_FIELD.value
315            )
316        elif self == ChannelDataType.BOOL:
317            return "CHANNEL_DATA_TYPE_BOOL" if api_format else ChannelDataTypeStrRep.BOOL.value
318        elif self == ChannelDataType.FLOAT:
319            return "CHANNEL_DATA_TYPE_FLOAT" if api_format else ChannelDataTypeStrRep.FLOAT.value
320        elif self == ChannelDataType.INT_32:
321            return "CHANNEL_DATA_TYPE_INT_32" if api_format else ChannelDataTypeStrRep.INT_32.value
322        elif self == ChannelDataType.INT_64:
323            return "CHANNEL_DATA_TYPE_INT_64" if api_format else ChannelDataTypeStrRep.INT_64.value
324        elif self == ChannelDataType.UINT_32:
325            return (
326                "CHANNEL_DATA_TYPE_UINT_32" if api_format else ChannelDataTypeStrRep.UINT_32.value
327            )
328        elif self == ChannelDataType.UINT_64:
329            return (
330                "CHANNEL_DATA_TYPE_UINT_64" if api_format else ChannelDataTypeStrRep.UINT_64.value
331            )
332        else:
333            raise Exception("Unreachable.")

Inherited Members

enum.Enum: name; value

def channel_fqn( channel: Union[ChannelConfig, sift.ingestion_configs.v2.ingestion_configs_pb2.ChannelConfig, ChannelValue, sift.channels.v3.channels_pb2.Channel, sift_py.ingestion.channel._AbstractChannel]) -> str: View Source

341def channel_fqn(
342    channel: Union[
343        ChannelConfig,
344        ChannelConfigPb,
345        ChannelValue,
346        ChannelPb,
347        _AbstractChannel,
348    ],
349) -> str:
350    """
351    Computes the fully qualified channel name.
352
353    NOTE: Component field of Channel is deprecated and should not be used. Function is left for code compatibility.
354
355    The fully-qualified channel name of a channel called 'voltage' is simply `voltage'. The
356    fully qualified name of a channel called 'temperature' of component 'motor' is a `motor.temperature'.
357    """
358
359    if isinstance(channel, ChannelConfig):
360        if channel.component:
361            _component_deprecation_warning()
362        return _channel_fqn(channel.name, channel.component)
363    elif isinstance(channel, ChannelConfigPb):
364        return channel.name
365    elif isinstance(channel, ChannelPb):
366        return channel.name
367    else:
368        component = channel.get("component")
369        if component:
370            _component_deprecation_warning()
371        channel_name = channel["channel_name"]
372        return _channel_fqn(name=channel_name, component=component)

Computes the fully qualified channel name.

NOTE: Component field of Channel is deprecated and should not be used. Function is left for code compatibility.

The fully-qualified channel name of a channel called 'voltage' is simply voltage'. The fully qualified name of a channel called 'temperature' of component 'motor' is amotor.temperature'.

def string_value(val: str) -> sift.ingest.v1.ingest_pb2.IngestWithConfigDataChannelValue: View Source

375def string_value(val: str) -> IngestWithConfigDataChannelValue:
376    return IngestWithConfigDataChannelValue(string=val)

def double_value(val: float) -> sift.ingest.v1.ingest_pb2.IngestWithConfigDataChannelValue: View Source

379def double_value(val: float) -> IngestWithConfigDataChannelValue:
380    return IngestWithConfigDataChannelValue(double=val)

def float_value(val: float) -> sift.ingest.v1.ingest_pb2.IngestWithConfigDataChannelValue: View Source

383def float_value(val: float) -> IngestWithConfigDataChannelValue:
384    return IngestWithConfigDataChannelValue(float=val)

def bool_value(val: bool) -> sift.ingest.v1.ingest_pb2.IngestWithConfigDataChannelValue: View Source

387def bool_value(val: bool) -> IngestWithConfigDataChannelValue:
388    return IngestWithConfigDataChannelValue(bool=val)

def int32_value(val: int) -> sift.ingest.v1.ingest_pb2.IngestWithConfigDataChannelValue: View Source

391def int32_value(val: int) -> IngestWithConfigDataChannelValue:
392    return IngestWithConfigDataChannelValue(int32=val)

def uint32_value(val: int) -> sift.ingest.v1.ingest_pb2.IngestWithConfigDataChannelValue: View Source

395def uint32_value(val: int) -> IngestWithConfigDataChannelValue:
396    return IngestWithConfigDataChannelValue(uint32=val)

def int64_value(val: int) -> sift.ingest.v1.ingest_pb2.IngestWithConfigDataChannelValue: View Source

399def int64_value(val: int) -> IngestWithConfigDataChannelValue:
400    return IngestWithConfigDataChannelValue(int64=val)

def uint64_value(val: int) -> sift.ingest.v1.ingest_pb2.IngestWithConfigDataChannelValue: View Source

403def uint64_value(val: int) -> IngestWithConfigDataChannelValue:
404    return IngestWithConfigDataChannelValue(uint64=val)

def bit_field_value(val: bytes) -> sift.ingest.v1.ingest_pb2.IngestWithConfigDataChannelValue: View Source

407def bit_field_value(val: bytes) -> IngestWithConfigDataChannelValue:
408    return IngestWithConfigDataChannelValue(bit_field=val)

def enum_value(val: int) -> sift.ingest.v1.ingest_pb2.IngestWithConfigDataChannelValue: View Source

411def enum_value(val: int) -> IngestWithConfigDataChannelValue:
412    return IngestWithConfigDataChannelValue(enum=val)

def empty_value() -> sift.ingest.v1.ingest_pb2.IngestWithConfigDataChannelValue: View Source

415def empty_value() -> IngestWithConfigDataChannelValue:
416    return IngestWithConfigDataChannelValue(empty=Empty())

def is_data_type( val: sift.ingest.v1.ingest_pb2.IngestWithConfigDataChannelValue, target_type: ChannelDataType) -> bool: View Source

419def is_data_type(val: IngestWithConfigDataChannelValue, target_type: ChannelDataType) -> bool:
420    if target_type == ChannelDataType.DOUBLE:
421        return val.HasField("double")
422    elif target_type == ChannelDataType.STRING:
423        return val.HasField("string")
424    elif target_type == ChannelDataType.ENUM:
425        return val.HasField("enum")
426    elif target_type == ChannelDataType.BIT_FIELD:
427        return val.HasField("bit_field")
428    elif target_type == ChannelDataType.BOOL:
429        return val.HasField("bool")
430    elif target_type == ChannelDataType.FLOAT:
431        return val.HasField("float")
432    elif target_type == ChannelDataType.INT_32:
433        return val.HasField("int32")
434    elif target_type == ChannelDataType.INT_64:
435        return val.HasField("int64")
436    elif target_type == ChannelDataType.UINT_32:
437        return val.HasField("uint32")
438    elif target_type == ChannelDataType.UINT_64:
439        return val.HasField("uint64")