from bt2.field import _ArrayField
from bt2.field import _StaticArrayField
from bt2.field import _DynamicArrayField
+from bt2.field import _BoolFieldConst
+from bt2.field import _BitArrayFieldConst
+from bt2.field import _IntegerFieldConst
+from bt2.field import _UnsignedIntegerFieldConst
+from bt2.field import _SignedIntegerFieldConst
+from bt2.field import _RealFieldConst
+from bt2.field import _EnumerationFieldConst
+from bt2.field import _UnsignedEnumerationFieldConst
+from bt2.field import _SignedEnumerationFieldConst
+from bt2.field import _StringFieldConst
+from bt2.field import _StructureFieldConst
+from bt2.field import _OptionFieldConst
+from bt2.field import _VariantFieldConst
+from bt2.field import _ArrayFieldConst
+from bt2.field import _StaticArrayFieldConst
+from bt2.field import _DynamicArrayFieldConst
from bt2.field_class import IntegerDisplayBase
from bt2.field_class import _BoolFieldClass
from bt2.field_class import _BitArrayFieldClass
from bt2.field_class import _ArrayFieldClass
from bt2.field_class import _StaticArrayFieldClass
from bt2.field_class import _DynamicArrayFieldClass
+from bt2.field_class import _BoolFieldClassConst
+from bt2.field_class import _BitArrayFieldClassConst
+from bt2.field_class import _IntegerFieldClassConst
+from bt2.field_class import _UnsignedIntegerFieldClassConst
+from bt2.field_class import _SignedIntegerFieldClassConst
+from bt2.field_class import _RealFieldClassConst
+from bt2.field_class import _EnumerationFieldClassConst
+from bt2.field_class import _UnsignedEnumerationFieldClassConst
+from bt2.field_class import _SignedEnumerationFieldClassConst
+from bt2.field_class import _StringFieldClassConst
+from bt2.field_class import _StructureFieldClassConst
+from bt2.field_class import _OptionFieldClassConst
+from bt2.field_class import _VariantFieldClassConst
+from bt2.field_class import _VariantFieldClassWithoutSelectorConst
+from bt2.field_class import _VariantFieldClassWithSelectorConst
+from bt2.field_class import _VariantFieldClassWithUnsignedSelectorConst
+from bt2.field_class import _VariantFieldClassWithSignedSelectorConst
+from bt2.field_class import _ArrayFieldClassConst
+from bt2.field_class import _StaticArrayFieldClassConst
+from bt2.field_class import _DynamicArrayFieldClassConst
from bt2.field_path import FieldPathScope
from bt2.field_path import _IndexFieldPathItem
from bt2.field_path import _CurrentArrayElementFieldPathItem
from bt2.message import _MessageIteratorInactivityMessage
from bt2.message import _DiscardedEventsMessage
from bt2.message import _DiscardedPacketsMessage
+from bt2.message import _EventMessageConst
+from bt2.message import _PacketBeginningMessageConst
+from bt2.message import _PacketEndMessageConst
+from bt2.message import _StreamBeginningMessageConst
+from bt2.message import _StreamEndMessageConst
+from bt2.message import _MessageIteratorInactivityMessageConst
+from bt2.message import _DiscardedEventsMessageConst
+from bt2.message import _DiscardedPacketsMessageConst
from bt2.message_iterator import _UserMessageIterator
from bt2.mip import get_greatest_operative_mip_version
from bt2.mip import get_maximal_mip_version
from bt2 import field as bt2_field
-class _Event(object._UniqueObject):
+class _EventConst(object._UniqueObject):
+ _borrow_class_ptr = staticmethod(native_bt.event_borrow_class_const)
+ _borrow_packet_ptr = staticmethod(native_bt.event_borrow_packet_const)
+ _borrow_stream_ptr = staticmethod(native_bt.event_borrow_stream_const)
+ _borrow_common_context_field_ptr = staticmethod(
+ native_bt.event_borrow_common_context_field_const
+ )
+ _borrow_specific_context_field_ptr = staticmethod(
+ native_bt.event_borrow_specific_context_field_const
+ )
+ _borrow_payload_field_ptr = staticmethod(native_bt.event_borrow_payload_field_const)
+ _create_field_from_ptr = staticmethod(bt2_field._create_field_from_const_ptr)
+
+ _event_class_pycls = property(lambda _: bt2_event_class._EventClassConst)
+ _packet_pycls = property(lambda _: bt2_packet._PacketConst)
+ _stream_pycls = property(lambda _: bt2_stream._StreamConst)
+
@property
def cls(self):
- event_class_ptr = native_bt.event_borrow_class(self._ptr)
+ event_class_ptr = self._borrow_class_ptr(self._ptr)
assert event_class_ptr is not None
- return bt2_event_class._EventClass._create_from_ptr_and_get_ref(event_class_ptr)
+ return self._event_class_pycls._create_from_ptr_and_get_ref(event_class_ptr)
@property
def name(self):
@property
def packet(self):
- packet_ptr = native_bt.event_borrow_packet(self._ptr)
+ packet_ptr = self._borrow_packet_ptr(self._ptr)
if packet_ptr is None:
return
- return bt2_packet._Packet._create_from_ptr_and_get_ref(packet_ptr)
+ return self._packet_pycls._create_from_ptr_and_get_ref(packet_ptr)
@property
def stream(self):
- stream_ptr = native_bt.event_borrow_stream(self._ptr)
+ stream_ptr = self._borrow_stream_ptr(self._ptr)
assert stream_ptr is not None
- return bt2_stream._Stream._create_from_ptr_and_get_ref(stream_ptr)
+ return self._stream_pycls._create_from_ptr_and_get_ref(stream_ptr)
@property
def common_context_field(self):
- field_ptr = native_bt.event_borrow_common_context_field(self._ptr)
+ field_ptr = self._borrow_common_context_field_ptr(self._ptr)
if field_ptr is None:
return
- return bt2_field._create_field_from_ptr(
+ return self._create_field_from_ptr(
field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
)
@property
def specific_context_field(self):
- field_ptr = native_bt.event_borrow_specific_context_field(self._ptr)
+ field_ptr = self._borrow_specific_context_field_ptr(self._ptr)
if field_ptr is None:
return
- return bt2_field._create_field_from_ptr(
+ return self._create_field_from_ptr(
field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
)
@property
def payload_field(self):
- field_ptr = native_bt.event_borrow_payload_field(self._ptr)
+ field_ptr = self._borrow_payload_field_ptr(self._ptr)
if field_ptr is None:
return
- return bt2_field._create_field_from_ptr(
+ return self._create_field_from_ptr(
field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
)
return packet_context_field[key]
raise KeyError(key)
+
+
+class _Event(_EventConst):
+ _borrow_class_ptr = staticmethod(native_bt.event_borrow_class)
+ _borrow_packet_ptr = staticmethod(native_bt.event_borrow_packet)
+ _borrow_stream_ptr = staticmethod(native_bt.event_borrow_stream)
+ _borrow_common_context_field_ptr = staticmethod(
+ native_bt.event_borrow_common_context_field
+ )
+ _borrow_specific_context_field_ptr = staticmethod(
+ native_bt.event_borrow_specific_context_field
+ )
+ _borrow_payload_field_ptr = staticmethod(native_bt.event_borrow_payload_field)
+ _create_field_from_ptr = staticmethod(bt2_field._create_field_from_ptr)
+
+ _event_class_pycls = property(lambda _: bt2_event_class._EventClass)
+ _packet_pycls = property(lambda _: bt2_packet._Packet)
+ _stream_pycls = property(lambda _: bt2_stream._Stream)
DEBUG = native_bt.EVENT_CLASS_LOG_LEVEL_DEBUG
-class _EventClass(object._SharedObject):
+class _EventClassConst(object._SharedObject):
_get_ref = staticmethod(native_bt.event_class_get_ref)
_put_ref = staticmethod(native_bt.event_class_put_ref)
+ _borrow_stream_class_ptr = staticmethod(
+ native_bt.event_class_borrow_stream_class_const
+ )
+ _borrow_specific_context_field_class_ptr = staticmethod(
+ native_bt.event_class_borrow_specific_context_field_class_const
+ )
+ _borrow_payload_field_class_ptr = staticmethod(
+ native_bt.event_class_borrow_payload_field_class_const
+ )
+ _borrow_user_attributes_ptr = staticmethod(
+ native_bt.event_class_borrow_user_attributes_const
+ )
+ _create_field_class_from_ptr_and_get_ref = staticmethod(
+ bt2_field_class._create_field_class_from_const_ptr_and_get_ref
+ )
+ _create_value_from_ptr_and_get_ref = staticmethod(
+ bt2_value._create_from_const_ptr_and_get_ref
+ )
+ _stream_class_pycls = property(lambda s: bt2_stream_class._StreamClassConst)
@property
def stream_class(self):
- sc_ptr = native_bt.event_class_borrow_stream_class(self._ptr)
+ sc_ptr = self._borrow_stream_class_ptr(self._ptr)
if sc_ptr is not None:
- return bt2_stream_class._StreamClass._create_from_ptr_and_get_ref(sc_ptr)
+ return self._stream_class_pycls._create_from_ptr_and_get_ref(sc_ptr)
@property
def user_attributes(self):
- ptr = native_bt.event_class_borrow_user_attributes(self._ptr)
+ ptr = self._borrow_user_attributes_ptr(self._ptr)
assert ptr is not None
- return bt2_value._create_from_ptr_and_get_ref(ptr)
-
- def _user_attributes(self, user_attributes):
- value = bt2_value.create_value(user_attributes)
- utils._check_type(value, bt2_value.MapValue)
- native_bt.event_class_set_user_attributes(self._ptr, value._ptr)
-
- _user_attributes = property(fset=_user_attributes)
+ return self._create_value_from_ptr_and_get_ref(ptr)
@property
def name(self):
return native_bt.event_class_get_name(self._ptr)
- def _name(self, name):
- utils._check_str(name)
- return native_bt.event_class_set_name(self._ptr, name)
-
- _name = property(fset=_name)
-
@property
def id(self):
id = native_bt.event_class_get_id(self._ptr)
return _EVENT_CLASS_LOG_LEVEL_TO_OBJ[log_level]
+ @property
+ def emf_uri(self):
+ return native_bt.event_class_get_emf_uri(self._ptr)
+
+ @property
+ def specific_context_field_class(self):
+ fc_ptr = self._borrow_specific_context_field_class_ptr(self._ptr)
+
+ if fc_ptr is None:
+ return
+
+ return self._create_field_class_from_ptr_and_get_ref(fc_ptr)
+
+ @property
+ def payload_field_class(self):
+ fc_ptr = self._borrow_payload_field_class_ptr(self._ptr)
+
+ if fc_ptr is None:
+ return
+
+ return self._create_field_class_from_ptr_and_get_ref(fc_ptr)
+
+
+class _EventClass(_EventClassConst):
+ _borrow_stream_class_ptr = staticmethod(native_bt.event_class_borrow_stream_class)
+ _borrow_specific_context_field_class_ptr = staticmethod(
+ native_bt.event_class_borrow_specific_context_field_class
+ )
+ _borrow_payload_field_class_ptr = staticmethod(
+ native_bt.event_class_borrow_payload_field_class
+ )
+ _borrow_user_attributes_ptr = staticmethod(
+ native_bt.event_class_borrow_user_attributes
+ )
+ _create_field_class_from_ptr_and_get_ref = staticmethod(
+ bt2_field_class._create_field_class_from_ptr_and_get_ref
+ )
+ _create_value_from_ptr_and_get_ref = staticmethod(
+ bt2_value._create_from_ptr_and_get_ref
+ )
+ _stream_class_pycls = property(lambda s: bt2_stream_class._StreamClass)
+
+ def _user_attributes(self, user_attributes):
+ value = bt2_value.create_value(user_attributes)
+ utils._check_type(value, bt2_value.MapValue)
+ native_bt.event_class_set_user_attributes(self._ptr, value._ptr)
+
+ _user_attributes = property(fset=_user_attributes)
+
+ def _name(self, name):
+ utils._check_str(name)
+ return native_bt.event_class_set_name(self._ptr, name)
+
+ _name = property(fset=_name)
+
def _log_level(self, log_level):
log_levels = (
EventClassLogLevel.EMERGENCY,
_log_level = property(fset=_log_level)
- @property
- def emf_uri(self):
- return native_bt.event_class_get_emf_uri(self._ptr)
-
def _emf_uri(self, emf_uri):
utils._check_str(emf_uri)
status = native_bt.event_class_set_emf_uri(self._ptr, emf_uri)
_emf_uri = property(fset=_emf_uri)
- @property
- def specific_context_field_class(self):
- fc_ptr = native_bt.event_class_borrow_specific_context_field_class_const(
- self._ptr
- )
-
- if fc_ptr is None:
- return
-
- return bt2_field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
-
def _specific_context_field_class(self, context_field_class):
if context_field_class is not None:
utils._check_type(context_field_class, bt2_field_class._StructureFieldClass)
_specific_context_field_class = property(fset=_specific_context_field_class)
- @property
- def payload_field_class(self):
- fc_ptr = native_bt.event_class_borrow_payload_field_class_const(self._ptr)
-
- if fc_ptr is None:
- return
-
- return bt2_field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
-
def _payload_field_class(self, payload_field_class):
if payload_field_class is not None:
utils._check_type(payload_field_class, bt2_field_class._StructureFieldClass)
import math
-def _create_field_from_ptr(ptr, owner_ptr, owner_get_ref, owner_put_ref):
+def _create_field_from_ptr_template(
+ object_map, ptr, owner_ptr, owner_get_ref, owner_put_ref
+):
+
field_class_ptr = native_bt.field_borrow_class_const(ptr)
typeid = native_bt.field_class_get_type(field_class_ptr)
- field = _TYPE_ID_TO_OBJ[typeid]._create_from_ptr_and_get_ref(
+ field = object_map[typeid]._create_from_ptr_and_get_ref(
ptr, owner_ptr, owner_get_ref, owner_put_ref
)
return field
+def _create_field_from_ptr(ptr, owner_ptr, owner_get_ref, owner_put_ref):
+ return _create_field_from_ptr_template(
+ _TYPE_ID_TO_OBJ, ptr, owner_ptr, owner_get_ref, owner_put_ref
+ )
+
+
+def _create_field_from_const_ptr(ptr, owner_ptr, owner_get_ref, owner_put_ref):
+ return _create_field_from_ptr_template(
+ _TYPE_ID_TO_CONST_OBJ, ptr, owner_ptr, owner_get_ref, owner_put_ref
+ )
+
+
# Get the "effective" field of `field`. If `field` is a variant, return
# the currently selected field. If `field` is an option, return the
# content field. If `field` is of any other type, return `field`
def _get_leaf_field(field):
- if isinstance(field, _VariantField):
+ if isinstance(field, _VariantFieldConst):
return _get_leaf_field(field.selected_option)
- if isinstance(field, _OptionField):
+ if isinstance(field, _OptionFieldConst):
return _get_leaf_field(field.field)
return field
-class _Field(object._UniqueObject):
+class _FieldConst(object._UniqueObject):
+ _create_field_from_ptr = staticmethod(_create_field_from_const_ptr)
+ _create_field_class_from_ptr_and_get_ref = staticmethod(
+ bt2_field_class._create_field_class_from_const_ptr_and_get_ref
+ )
+ _borrow_class_ptr = staticmethod(native_bt.field_borrow_class_const)
+
def __eq__(self, other):
other = _get_leaf_field(other)
return self._spec_eq(other)
@property
def cls(self):
- field_class_ptr = native_bt.field_borrow_class_const(self._ptr)
+ field_class_ptr = self._borrow_class_ptr(self._ptr)
assert field_class_ptr is not None
- return bt2_field_class._create_field_class_from_ptr_and_get_ref(field_class_ptr)
+ return self._create_field_class_from_ptr_and_get_ref(field_class_ptr)
def _repr(self):
raise NotImplementedError
return self._repr()
-class _BitArrayField(_Field):
- _NAME = 'Bit array'
+class _Field(_FieldConst):
+ _create_field_from_ptr = staticmethod(_create_field_from_ptr)
+ _create_field_class_from_ptr_and_get_ref = staticmethod(
+ bt2_field_class._create_field_class_from_ptr_and_get_ref
+ )
+ _borrow_class_ptr = staticmethod(native_bt.field_borrow_class)
+
+
+class _BitArrayFieldConst(_FieldConst):
+ _NAME = 'Const bit array'
@property
def value_as_integer(self):
return native_bt.field_bit_array_get_value_as_integer(self._ptr)
- @value_as_integer.setter
- def value_as_integer(self, value):
- utils._check_uint64(value)
- native_bt.field_bit_array_set_value_as_integer(self._ptr, value)
-
def _spec_eq(self, other):
if type(other) is not type(self):
return False
return self.cls.length
+class _BitArrayField(_BitArrayFieldConst, _Field):
+ _NAME = 'Bit array'
+
+ def _value_as_integer(self, value):
+ utils._check_uint64(value)
+ native_bt.field_bit_array_set_value_as_integer(self._ptr, value)
+
+ value_as_integer = property(
+ fget=_BitArrayFieldConst.value_as_integer.fget, fset=_value_as_integer
+ )
+
+
@functools.total_ordering
-class _NumericField(_Field):
+class _NumericFieldConst(_FieldConst):
@staticmethod
def _extract_value(other):
- if isinstance(other, _BoolField) or isinstance(other, bool):
+ if isinstance(other, _BoolFieldConst) or isinstance(other, bool):
return bool(other)
if isinstance(other, numbers.Integral):
return self._extract_value(base) ** self._value
-class _IntegralField(_NumericField, numbers.Integral):
+class _NumericField(_NumericFieldConst, _Field):
+ pass
+
+
+class _IntegralFieldConst(_NumericFieldConst, numbers.Integral):
def __lshift__(self, other):
return self._value << self._extract_value(other)
return ~self._value
-class _BoolField(_IntegralField, _Field):
- _NAME = 'Boolean'
+class _IntegralField(_IntegralFieldConst, _NumericField):
+ pass
+
+
+class _BoolFieldConst(_IntegralFieldConst, _FieldConst):
+ _NAME = 'Const boolean'
def __bool__(self):
return self._value
- def _value_to_bool(self, value):
- if isinstance(value, _BoolField):
+ @classmethod
+ def _value_to_bool(cls, value):
+ if isinstance(value, _BoolFieldConst):
value = value._value
if not isinstance(value, bool):
raise TypeError(
- "'{}' object is not a 'bool' or '_BoolField' object".format(
+ "'{}' object is not a 'bool', '_BoolFieldConst', or '_BoolField' object".format(
value.__class__
)
)
def _value(self):
return bool(native_bt.field_bool_get_value(self._ptr))
+
+class _BoolField(_BoolFieldConst, _IntegralField, _Field):
+ _NAME = 'Boolean'
+
def _set_value(self, value):
value = self._value_to_bool(value)
native_bt.field_bool_set_value(self._ptr, value)
value = property(fset=_set_value)
-class _IntegerField(_IntegralField, _Field):
+class _IntegerFieldConst(_IntegralFieldConst, _FieldConst):
pass
-class _UnsignedIntegerField(_IntegerField, _Field):
- _NAME = 'Unsigned integer'
+class _IntegerField(_IntegerFieldConst, _IntegralField, _Field):
+ pass
+
+
+class _UnsignedIntegerFieldConst(_IntegerFieldConst, _FieldConst):
+ _NAME = 'Const unsigned integer'
- def _value_to_int(self, value):
+ @classmethod
+ def _value_to_int(cls, value):
if not isinstance(value, numbers.Integral):
raise TypeError('expecting an integral number object')
def _value(self):
return native_bt.field_integer_unsigned_get_value(self._ptr)
+
+class _UnsignedIntegerField(_UnsignedIntegerFieldConst, _IntegerField, _Field):
+ _NAME = 'Unsigned integer'
+
def _set_value(self, value):
value = self._value_to_int(value)
native_bt.field_integer_unsigned_set_value(self._ptr, value)
value = property(fset=_set_value)
-class _SignedIntegerField(_IntegerField, _Field):
- _NAME = 'Signed integer'
+class _SignedIntegerFieldConst(_IntegerFieldConst, _FieldConst):
+ _NAME = 'Const signed integer'
- def _value_to_int(self, value):
+ @classmethod
+ def _value_to_int(cls, value):
if not isinstance(value, numbers.Integral):
raise TypeError('expecting an integral number object')
def _value(self):
return native_bt.field_integer_signed_get_value(self._ptr)
+
+class _SignedIntegerField(_SignedIntegerFieldConst, _IntegerField, _Field):
+ _NAME = 'Signed integer'
+
def _set_value(self, value):
value = self._value_to_int(value)
native_bt.field_integer_signed_set_value(self._ptr, value)
value = property(fset=_set_value)
-class _RealField(_NumericField, numbers.Real):
- _NAME = 'Real'
+class _RealFieldConst(_NumericFieldConst, numbers.Real):
+ _NAME = 'Const real'
- def _value_to_float(self, value):
+ @classmethod
+ def _value_to_float(cls, value):
if not isinstance(value, numbers.Real):
raise TypeError("expecting a real number object")
def _value(self):
return native_bt.field_real_get_value(self._ptr)
+
+class _RealField(_RealFieldConst, _NumericField):
+ _NAME = 'Real'
+
def _set_value(self, value):
value = self._value_to_float(value)
native_bt.field_real_set_value(self._ptr, value)
value = property(fset=_set_value)
-class _EnumerationField(_IntegerField):
+class _EnumerationFieldConst(_IntegerFieldConst):
def _repr(self):
return '{} ({})'.format(self._value, ', '.join(self.labels))
return labels
-class _UnsignedEnumerationField(_EnumerationField, _UnsignedIntegerField):
- _NAME = 'Unsigned Enumeration'
+class _EnumerationField(_EnumerationFieldConst, _IntegerField):
+ pass
+
+
+class _UnsignedEnumerationFieldConst(
+ _EnumerationFieldConst, _UnsignedIntegerFieldConst
+):
+ _NAME = 'Const unsigned Enumeration'
_get_mapping_labels = staticmethod(
native_bt.field_enumeration_unsigned_get_mapping_labels
)
-class _SignedEnumerationField(_EnumerationField, _SignedIntegerField):
- _NAME = 'Signed Enumeration'
+class _UnsignedEnumerationField(
+ _UnsignedEnumerationFieldConst, _EnumerationField, _UnsignedIntegerField
+):
+ _NAME = 'Unsigned enumeration'
+
+
+class _SignedEnumerationFieldConst(_EnumerationFieldConst, _SignedIntegerFieldConst):
+ _NAME = 'Const signed Enumeration'
_get_mapping_labels = staticmethod(
native_bt.field_enumeration_signed_get_mapping_labels
)
+class _SignedEnumerationField(
+ _SignedEnumerationFieldConst, _EnumerationField, _SignedIntegerField
+):
+ _NAME = 'Signed enumeration'
+
+
@functools.total_ordering
-class _StringField(_Field):
- _NAME = 'String'
+class _StringFieldConst(_FieldConst):
+ _NAME = 'Const string'
- def _value_to_str(self, value):
- if isinstance(value, self.__class__):
+ @classmethod
+ def _value_to_str(cls, value):
+ if isinstance(value, _StringFieldConst):
value = value._value
if not isinstance(value, str):
def _value(self):
return native_bt.field_string_get_value(self._ptr)
- def _set_value(self, value):
- value = self._value_to_str(value)
- native_bt.field_string_set_value(self._ptr, value)
-
- value = property(fset=_set_value)
-
def _spec_eq(self, other):
try:
return self._value == self._value_to_str(other)
def __len__(self):
return native_bt.field_string_get_length(self._ptr)
+
+class _StringField(_StringFieldConst, _Field):
+ _NAME = 'String'
+
+ def _set_value(self, value):
+ value = self._value_to_str(value)
+ native_bt.field_string_set_value(self._ptr, value)
+
+ value = property(fset=_set_value)
+
def __iadd__(self, value):
value = self._value_to_str(value)
status = native_bt.field_string_append(self._ptr, value)
return self
-class _ContainerField(_Field):
+class _ContainerFieldConst(_FieldConst):
def __bool__(self):
return len(self) != 0
+ def _count(self):
+ return len(self.cls)
+
def __len__(self):
count = self._count()
assert count >= 0
def __delitem__(self, index):
raise NotImplementedError
+ def __setitem__(self, index, value):
+ raise TypeError(
+ '\'{}\' object does not support item assignment'.format(self.__class__)
+ )
-class _StructureField(_ContainerField, collections.abc.MutableMapping):
- _NAME = 'Structure'
- def _count(self):
- return len(self.cls)
+class _ContainerField(_ContainerFieldConst, _Field):
+ pass
- def __setitem__(self, key, value):
- # raises if key is somehow invalid
- field = self[key]
- # the field's property does the appropriate conversion or raises
- # the appropriate exception
- field.value = value
+class _StructureFieldConst(_ContainerFieldConst, collections.abc.Mapping):
+ _NAME = 'Const structure'
+ _borrow_member_field_ptr_by_index = staticmethod(
+ native_bt.field_structure_borrow_member_field_by_index_const
+ )
+ _borrow_member_field_ptr_by_name = staticmethod(
+ native_bt.field_structure_borrow_member_field_by_name_const
+ )
+
+ def _count(self):
+ return len(self.cls)
def __iter__(self):
# same name iterator
return True
- def _set_value(self, values):
- try:
- for key, value in values.items():
- self[key].value = value
- except Exception:
- raise
-
- value = property(fset=_set_value)
-
def _repr(self):
items = ['{}: {}'.format(repr(k), repr(v)) for k, v in self.items()]
return '{{{}}}'.format(', '.join(items))
def __getitem__(self, key):
utils._check_str(key)
- field_ptr = native_bt.field_structure_borrow_member_field_by_name(
- self._ptr, key
- )
+ field_ptr = self._borrow_member_field_ptr_by_name(self._ptr, key)
if field_ptr is None:
raise KeyError(key)
- return _create_field_from_ptr(
+ return self._create_field_from_ptr(
field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
)
if index >= len(self):
raise IndexError
-
- field_ptr = native_bt.field_structure_borrow_member_field_by_index(
- self._ptr, index
- )
+ field_ptr = self._borrow_member_field_ptr_by_index(self._ptr, index)
assert field_ptr is not None
- return _create_field_from_ptr(
+ return self._create_field_from_ptr(
field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
)
-class _OptionField(_Field):
- _NAME = 'Option'
+class _StructureField(
+ _StructureFieldConst, _ContainerField, collections.abc.MutableMapping
+):
+ _NAME = 'Structure'
+ _borrow_member_field_ptr_by_index = staticmethod(
+ native_bt.field_structure_borrow_member_field_by_index
+ )
+ _borrow_member_field_ptr_by_name = staticmethod(
+ native_bt.field_structure_borrow_member_field_by_name
+ )
+
+ def __setitem__(self, key, value):
+ # raises if key is somehow invalid
+ field = self[key]
+
+ # the field's property does the appropriate conversion or raises
+ # the appropriate exception
+ field.value = value
+
+ def _set_value(self, values):
+ try:
+ for key, value in values.items():
+ self[key].value = value
+ except Exception:
+ raise
+
+ value = property(fset=_set_value)
+
+
+class _OptionFieldConst(_FieldConst):
+ _NAME = 'Const option'
+ _borrow_field_ptr = staticmethod(native_bt.field_option_borrow_field_const)
@property
def field(self):
- field_ptr = native_bt.field_option_borrow_field_const(self._ptr)
+ field_ptr = self._borrow_field_ptr(self._ptr)
if field_ptr is None:
return
- return _create_field_from_ptr(
+ return self._create_field_from_ptr(
field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
)
def has_field(self):
return self.field is not None
- @has_field.setter
- def has_field(self, value):
- utils._check_bool(value)
- native_bt.field_option_set_has_field(self._ptr, value)
-
def _spec_eq(self, other):
return _get_leaf_field(self) == other
def _repr(self):
return repr(self.field)
+
+class _OptionField(_OptionFieldConst, _Field):
+ _NAME = 'Option'
+ _borrow_field_ptr = staticmethod(native_bt.field_option_borrow_field)
+
+ def _has_field(self, value):
+ utils._check_bool(value)
+ native_bt.field_option_set_has_field(self._ptr, value)
+
+ has_field = property(fget=_OptionFieldConst.has_field.fget, fset=_has_field)
+
def _set_value(self, value):
self.has_field = True
field = self.field
value = property(fset=_set_value)
-class _VariantField(_ContainerField, _Field):
- _NAME = 'Variant'
+class _VariantFieldConst(_ContainerFieldConst, _FieldConst):
+ _NAME = 'Const variant'
+ _borrow_selected_option_field_ptr = staticmethod(
+ native_bt.field_variant_borrow_selected_option_field_const
+ )
def _count(self):
return len(self.cls)
def selected_option_index(self):
return native_bt.field_variant_get_selected_option_field_index(self._ptr)
- @selected_option_index.setter
- def selected_option_index(self, index):
- if index < 0 or index >= len(self):
- raise IndexError('{} field object index is out of range'.format(self._NAME))
-
- native_bt.field_variant_select_option_field_by_index(self._ptr, index)
-
@property
def selected_option(self):
# TODO: Is there a way to check if the variant field has a selected_option,
# so we can raise an exception instead of hitting a pre-condition check?
# If there is something, that check should be added to selected_option_index too.
- field_ptr = native_bt.field_variant_borrow_selected_option_field(self._ptr)
+ field_ptr = self._borrow_selected_option_field_ptr(self._ptr)
- return _create_field_from_ptr(
+ return self._create_field_from_ptr(
field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
)
def _repr(self):
return repr(self.selected_option)
+
+class _VariantField(_VariantFieldConst, _ContainerField, _Field):
+ _NAME = 'Variant'
+ _borrow_selected_option_field_ptr = staticmethod(
+ native_bt.field_variant_borrow_selected_option_field
+ )
+
+ def _selected_option_index(self, index):
+ if index < 0 or index >= len(self):
+ raise IndexError('{} field object index is out of range'.format(self._NAME))
+
+ native_bt.field_variant_select_option_field_by_index(self._ptr, index)
+
+ selected_option_index = property(
+ fget=_VariantFieldConst.selected_option_index.fget, fset=_selected_option_index
+ )
+
def _set_value(self, value):
self.selected_option.value = value
value = property(fset=_set_value)
-class _ArrayField(_ContainerField, _Field, collections.abc.MutableSequence):
+class _ArrayFieldConst(_ContainerFieldConst, _FieldConst, collections.abc.Sequence):
+ _borrow_element_field_ptr_by_index = staticmethod(
+ native_bt.field_array_borrow_element_field_by_index_const
+ )
+
def _get_length(self):
return native_bt.field_array_get_length(self._ptr)
if index < 0 or index >= len(self):
raise IndexError('{} field object index is out of range'.format(self._NAME))
- field_ptr = native_bt.field_array_borrow_element_field_by_index(
- self._ptr, index
- )
+ field_ptr = self._borrow_element_field_ptr_by_index(self._ptr, index)
assert field_ptr
- return _create_field_from_ptr(
+ return self._create_field_from_ptr(
field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
)
- def __setitem__(self, index, value):
- # raises if index is somehow invalid
- field = self[index]
-
- if not isinstance(field, (_NumericField, _StringField)):
- raise TypeError('can only set the value of a number or string field')
-
- # the field's property does the appropriate conversion or raises
- # the appropriate exception
- field.value = value
-
def insert(self, index, value):
raise NotImplementedError
return '[{}]'.format(', '.join([repr(v) for v in self]))
-class _StaticArrayField(_ArrayField, _Field):
- _NAME = 'Static array'
+class _ArrayField(
+ _ArrayFieldConst, _ContainerField, _Field, collections.abc.MutableSequence
+):
+ _borrow_element_field_ptr_by_index = staticmethod(
+ native_bt.field_array_borrow_element_field_by_index
+ )
+
+ def __setitem__(self, index, value):
+ # raises if index is somehow invalid
+ field = self[index]
+
+ if not isinstance(field, (_NumericField, _StringField)):
+ raise TypeError('can only set the value of a number or string field')
+
+ # the field's property does the appropriate conversion or raises
+ # the appropriate exception
+ field.value = value
+
+
+class _StaticArrayFieldConst(_ArrayFieldConst, _FieldConst):
+ _NAME = 'Const static array'
def _count(self):
return native_bt.field_array_get_length(self._ptr)
+
+class _StaticArrayField(_StaticArrayFieldConst, _ArrayField, _Field):
+ _NAME = 'Static array'
+
def _set_value(self, values):
if len(self) != len(values):
raise ValueError('expected length of value and array field to match')
value = property(fset=_set_value)
-class _DynamicArrayField(_ArrayField, _Field):
- _NAME = 'Dynamic array'
+class _DynamicArrayFieldConst(_ArrayFieldConst, _FieldConst):
+ _NAME = 'Const dynamic array'
def _count(self):
return self.length
+
+class _DynamicArrayField(_DynamicArrayFieldConst, _ArrayField, _Field):
+ _NAME = 'Dynamic array'
+
def _set_length(self, length):
utils._check_uint64(length)
status = native_bt.field_array_dynamic_set_length(self._ptr, length)
value = property(fset=_set_value)
+_TYPE_ID_TO_CONST_OBJ = {
+ native_bt.FIELD_CLASS_TYPE_BOOL: _BoolFieldConst,
+ native_bt.FIELD_CLASS_TYPE_BIT_ARRAY: _BitArrayFieldConst,
+ native_bt.FIELD_CLASS_TYPE_UNSIGNED_INTEGER: _UnsignedIntegerFieldConst,
+ native_bt.FIELD_CLASS_TYPE_SIGNED_INTEGER: _SignedIntegerFieldConst,
+ native_bt.FIELD_CLASS_TYPE_REAL: _RealFieldConst,
+ native_bt.FIELD_CLASS_TYPE_UNSIGNED_ENUMERATION: _UnsignedEnumerationFieldConst,
+ native_bt.FIELD_CLASS_TYPE_SIGNED_ENUMERATION: _SignedEnumerationFieldConst,
+ native_bt.FIELD_CLASS_TYPE_STRING: _StringFieldConst,
+ native_bt.FIELD_CLASS_TYPE_STRUCTURE: _StructureFieldConst,
+ native_bt.FIELD_CLASS_TYPE_STATIC_ARRAY: _StaticArrayFieldConst,
+ native_bt.FIELD_CLASS_TYPE_DYNAMIC_ARRAY: _DynamicArrayFieldConst,
+ native_bt.FIELD_CLASS_TYPE_OPTION: _OptionFieldConst,
+ native_bt.FIELD_CLASS_TYPE_VARIANT_WITHOUT_SELECTOR: _VariantFieldConst,
+ native_bt.FIELD_CLASS_TYPE_VARIANT_WITH_UNSIGNED_SELECTOR: _VariantFieldConst,
+ native_bt.FIELD_CLASS_TYPE_VARIANT_WITH_SIGNED_SELECTOR: _VariantFieldConst,
+}
+
_TYPE_ID_TO_OBJ = {
native_bt.FIELD_CLASS_TYPE_BOOL: _BoolField,
native_bt.FIELD_CLASS_TYPE_BIT_ARRAY: _BitArrayField,
import bt2
-def _create_field_class_from_ptr_and_get_ref(ptr):
+def _create_field_class_from_ptr_and_get_ref_template(type_map, ptr):
typeid = native_bt.field_class_get_type(ptr)
- return _FIELD_CLASS_TYPE_TO_OBJ[typeid]._create_from_ptr_and_get_ref(ptr)
+ return type_map[typeid]._create_from_ptr_and_get_ref(ptr)
+
+
+def _create_field_class_from_ptr_and_get_ref(ptr):
+ return _create_field_class_from_ptr_and_get_ref_template(
+ _FIELD_CLASS_TYPE_TO_OBJ, ptr
+ )
+
+
+def _create_field_class_from_const_ptr_and_get_ref(ptr):
+ return _create_field_class_from_ptr_and_get_ref_template(
+ _FIELD_CLASS_TYPE_TO_CONST_OBJ, ptr
+ )
class IntegerDisplayBase:
HEXADECIMAL = native_bt.FIELD_CLASS_INTEGER_PREFERRED_DISPLAY_BASE_HEXADECIMAL
-class _FieldClass(object._SharedObject):
+class _FieldClassConst(object._SharedObject):
_get_ref = staticmethod(native_bt.field_class_get_ref)
_put_ref = staticmethod(native_bt.field_class_put_ref)
+ _borrow_user_attributes_ptr = staticmethod(
+ native_bt.field_class_borrow_user_attributes_const
+ )
+ _create_value_from_ptr_and_get_ref = staticmethod(
+ bt2_value._create_from_const_ptr_and_get_ref
+ )
def _check_create_status(self, ptr):
if ptr is None:
@property
def user_attributes(self):
- ptr = native_bt.field_class_borrow_user_attributes(self._ptr)
+ ptr = self._borrow_user_attributes_ptr(self._ptr)
assert ptr is not None
- return bt2_value._create_from_ptr_and_get_ref(ptr)
+ return self._create_value_from_ptr_and_get_ref(ptr)
+
+
+class _FieldClass(_FieldClassConst):
+ _borrow_user_attributes_ptr = staticmethod(
+ native_bt.field_class_borrow_user_attributes
+ )
+ _create_value_from_ptr_and_get_ref = staticmethod(
+ bt2_value._create_from_ptr_and_get_ref
+ )
def _user_attributes(self, user_attributes):
value = bt2_value.create_value(user_attributes)
_user_attributes = property(fset=_user_attributes)
-class _BoolFieldClass(_FieldClass):
+class _BoolFieldClassConst(_FieldClassConst):
+ _NAME = 'Const Boolean'
+
+
+class _BoolFieldClass(_BoolFieldClassConst, _FieldClass):
_NAME = 'Boolean'
-class _BitArrayFieldClass(_FieldClass):
- _NAME = 'Bit array'
+class _BitArrayFieldClassConst(_FieldClassConst):
+ _NAME = 'Const Bit array'
@property
def length(self):
return length
-class _IntegerFieldClass(_FieldClass):
+class _BitArrayFieldClass(_BitArrayFieldClassConst, _FieldClass):
+ _NAME = 'Bit array'
+
+
+class _IntegerFieldClassConst(_FieldClassConst):
@property
def field_value_range(self):
size = native_bt.field_class_integer_get_field_value_range(self._ptr)
assert size >= 1
return size
+ @property
+ def preferred_display_base(self):
+ base = native_bt.field_class_integer_get_preferred_display_base(self._ptr)
+ assert base >= 0
+ return base
+
+
+class _IntegerFieldClass(_FieldClass, _IntegerFieldClassConst):
def _field_value_range(self, size):
if size < 1 or size > 64:
raise ValueError("Value is outside valid range [1, 64] ({})".format(size))
_field_value_range = property(fset=_field_value_range)
- @property
- def preferred_display_base(self):
- base = native_bt.field_class_integer_get_preferred_display_base(self._ptr)
- assert base >= 0
- return base
-
def _preferred_display_base(self, base):
utils._check_uint64(base)
_preferred_display_base = property(fset=_preferred_display_base)
-class _UnsignedIntegerFieldClass(_IntegerFieldClass):
+class _UnsignedIntegerFieldClassConst(_IntegerFieldClassConst, _FieldClassConst):
+ _NAME = 'Const unsigned integer'
+
+
+class _UnsignedIntegerFieldClass(
+ _UnsignedIntegerFieldClassConst, _IntegerFieldClass, _FieldClass
+):
_NAME = 'Unsigned integer'
-class _SignedIntegerFieldClass(_IntegerFieldClass):
+class _SignedIntegerFieldClassConst(_IntegerFieldClassConst, _FieldClassConst):
+ _NAME = 'Const signed integer'
+
+
+class _SignedIntegerFieldClass(
+ _SignedIntegerFieldClassConst, _IntegerFieldClass, _FieldClass
+):
_NAME = 'Signed integer'
-class _RealFieldClass(_FieldClass):
- _NAME = 'Real'
+class _RealFieldClassConst(_FieldClassConst):
+ _NAME = 'Const real'
@property
def is_single_precision(self):
return native_bt.field_class_real_is_single_precision(self._ptr)
+
+class _RealFieldClass(_FieldClass, _RealFieldClassConst):
+ _NAME = 'Real'
+
def _is_single_precision(self, is_single_precision):
utils._check_bool(is_single_precision)
native_bt.field_class_real_set_is_single_precision(
return self._ranges
-class _UnsignedEnumerationFieldClassMapping(_EnumerationFieldClassMapping):
+class _UnsignedEnumerationFieldClassMappingConst(_EnumerationFieldClassMapping):
_ranges_type = bt2_integer_range_set.UnsignedIntegerRangeSet
_as_enumeration_field_class_mapping_ptr = staticmethod(
native_bt.field_class_enumeration_unsigned_mapping_as_mapping_const
)
-class _SignedEnumerationFieldClassMapping(_EnumerationFieldClassMapping):
+class _SignedEnumerationFieldClassMappingConst(_EnumerationFieldClassMapping):
_ranges_type = bt2_integer_range_set.SignedIntegerRangeSet
_as_enumeration_field_class_mapping_ptr = staticmethod(
native_bt.field_class_enumeration_signed_mapping_as_mapping_const
)
-class _EnumerationFieldClass(_IntegerFieldClass, collections.abc.Mapping):
+class _EnumerationFieldClassConst(_IntegerFieldClassConst, collections.abc.Mapping):
def __len__(self):
count = native_bt.field_class_enumeration_get_mapping_count(self._ptr)
assert count >= 0
return count
- def add_mapping(self, label, ranges):
- utils._check_str(label)
- utils._check_type(ranges, self._range_set_type)
-
- if label in self:
- raise ValueError("duplicate mapping label '{}'".format(label))
-
- status = self._add_mapping(self._ptr, label, ranges._ptr)
- utils._handle_func_status(
- status, 'cannot add mapping to enumeration field class object'
- )
-
def mappings_for_value(self, value):
+ self._check_int_type(value)
+
status, labels = self._get_mapping_labels_for_value(self._ptr, value)
utils._handle_func_status(
status, 'cannot get mapping labels for value {}'.format(value)
def __iter__(self):
for idx in range(len(self)):
- mapping = self._get_mapping_by_index(self._ptr, idx)
- yield mapping.label
+ mapping_ptr = self._borrow_mapping_ptr_by_index(self._ptr, idx)
+ yield self._mapping_pycls(mapping_ptr).label
def __getitem__(self, label):
utils._check_str(label)
- mapping = self._get_mapping_by_label(self._ptr, label)
+ mapping_ptr = self._borrow_mapping_ptr_by_label(self._ptr, label)
- if mapping is None:
+ if mapping_ptr is None:
raise KeyError(label)
- return mapping
+ return self._mapping_pycls(mapping_ptr)
+
+
+class _EnumerationFieldClass(_EnumerationFieldClassConst, _IntegerFieldClass):
+ def add_mapping(self, label, ranges):
+ utils._check_str(label)
+ utils._check_type(ranges, self._range_set_type)
+
+ if label in self:
+ raise ValueError("duplicate mapping label '{}'".format(label))
+
+ status = self._add_mapping(self._ptr, label, ranges._ptr)
+ utils._handle_func_status(
+ status, 'cannot add mapping to enumeration field class object'
+ )
def __iadd__(self, mappings):
for label, ranges in mappings:
return self
-class _UnsignedEnumerationFieldClass(
- _EnumerationFieldClass, _UnsignedIntegerFieldClass
+class _UnsignedEnumerationFieldClassConst(
+ _EnumerationFieldClassConst, _UnsignedIntegerFieldClassConst
):
- _NAME = 'Unsigned enumeration'
+ _NAME = 'Const nsigned enumeration'
_range_set_type = bt2_integer_range_set.UnsignedIntegerRangeSet
- _add_mapping = staticmethod(native_bt.field_class_enumeration_unsigned_add_mapping)
-
- @staticmethod
- def _get_mapping_by_index(enum_ptr, index):
- mapping_ptr = native_bt.field_class_enumeration_unsigned_borrow_mapping_by_index_const(
- enum_ptr, index
- )
- assert mapping_ptr is not None
- return _UnsignedEnumerationFieldClassMapping(mapping_ptr)
+ _borrow_mapping_ptr_by_label = staticmethod(
+ native_bt.field_class_enumeration_unsigned_borrow_mapping_by_label_const
+ )
+ _borrow_mapping_ptr_by_index = staticmethod(
+ native_bt.field_class_enumeration_unsigned_borrow_mapping_by_index_const
+ )
+ _mapping_pycls = property(lambda _: _UnsignedEnumerationFieldClassMappingConst)
+ _get_mapping_labels_for_value = staticmethod(
+ native_bt.field_class_enumeration_unsigned_get_mapping_labels_for_value
+ )
+ _check_int_type = staticmethod(utils._check_uint64)
- @staticmethod
- def _get_mapping_by_label(enum_ptr, label):
- mapping_ptr = native_bt.field_class_enumeration_unsigned_borrow_mapping_by_label_const(
- enum_ptr, label
- )
- if mapping_ptr is None:
- return
+class _UnsignedEnumerationFieldClass(
+ _UnsignedEnumerationFieldClassConst,
+ _EnumerationFieldClass,
+ _UnsignedIntegerFieldClass,
+):
+ _NAME = 'Unsigned enumeration'
+ _add_mapping = staticmethod(native_bt.field_class_enumeration_unsigned_add_mapping)
- return _UnsignedEnumerationFieldClassMapping(mapping_ptr)
- @staticmethod
- def _get_mapping_labels_for_value(enum_ptr, value):
- utils._check_uint64(value)
- return native_bt.field_class_enumeration_unsigned_get_mapping_labels_for_value(
- enum_ptr, value
- )
+class _SignedEnumerationFieldClassConst(
+ _EnumerationFieldClassConst, _SignedIntegerFieldClassConst
+):
+ _NAME = 'Const signed enumeration'
+ _range_set_type = bt2_integer_range_set.SignedIntegerRangeSet
+ _borrow_mapping_ptr_by_label = staticmethod(
+ native_bt.field_class_enumeration_signed_borrow_mapping_by_label_const
+ )
+ _borrow_mapping_ptr_by_index = staticmethod(
+ native_bt.field_class_enumeration_signed_borrow_mapping_by_index_const
+ )
+ _mapping_pycls = property(lambda _: _SignedEnumerationFieldClassMappingConst)
+ _get_mapping_labels_for_value = staticmethod(
+ native_bt.field_class_enumeration_signed_get_mapping_labels_for_value
+ )
+ _check_int_type = staticmethod(utils._check_int64)
-class _SignedEnumerationFieldClass(_EnumerationFieldClass, _SignedIntegerFieldClass):
+class _SignedEnumerationFieldClass(
+ _SignedEnumerationFieldClassConst, _EnumerationFieldClass, _SignedIntegerFieldClass
+):
_NAME = 'Signed enumeration'
- _range_set_type = bt2_integer_range_set.SignedIntegerRangeSet
_add_mapping = staticmethod(native_bt.field_class_enumeration_signed_add_mapping)
- @staticmethod
- def _get_mapping_by_index(enum_ptr, index):
- mapping_ptr = native_bt.field_class_enumeration_signed_borrow_mapping_by_index_const(
- enum_ptr, index
- )
- assert mapping_ptr is not None
- return _SignedEnumerationFieldClassMapping(mapping_ptr)
-
- @staticmethod
- def _get_mapping_by_label(enum_ptr, label):
- mapping_ptr = native_bt.field_class_enumeration_signed_borrow_mapping_by_label_const(
- enum_ptr, label
- )
-
- if mapping_ptr is None:
- return
- return _SignedEnumerationFieldClassMapping(mapping_ptr)
-
- @staticmethod
- def _get_mapping_labels_for_value(enum_ptr, value):
- utils._check_int64(value)
- return native_bt.field_class_enumeration_signed_get_mapping_labels_for_value(
- enum_ptr, value
- )
+class _StringFieldClassConst(_FieldClassConst):
+ _NAME = 'Const string'
-class _StringFieldClass(_FieldClass):
+class _StringFieldClass(_StringFieldClassConst, _FieldClass):
_NAME = 'String'
-class _StructureFieldClassMember:
+class _StructureFieldClassMemberConst:
+ _create_field_class_from_ptr_and_get_ref = staticmethod(
+ _create_field_class_from_const_ptr_and_get_ref
+ )
+ _borrow_field_class_ptr = staticmethod(
+ native_bt.field_class_structure_member_borrow_field_class_const
+ )
+ _borrow_user_attributes_ptr = staticmethod(
+ native_bt.field_class_structure_member_borrow_user_attributes_const
+ )
+ _create_value_from_ptr_and_get_ref = staticmethod(
+ bt2_value._create_from_const_ptr_and_get_ref
+ )
+
def __init__(self, owning_struct_fc, member_ptr):
# this field class owns the member; keeping it here maintains
# the member alive as members are not shared objects
@property
def field_class(self):
- fc_ptr = native_bt.field_class_structure_member_borrow_field_class_const(
- self._ptr
- )
+ fc_ptr = self._borrow_field_class_ptr(self._ptr)
assert fc_ptr is not None
- return _create_field_class_from_ptr_and_get_ref(fc_ptr)
+ return self._create_field_class_from_ptr_and_get_ref(fc_ptr)
@property
def user_attributes(self):
- ptr = native_bt.field_class_structure_member_borrow_user_attributes(self._ptr)
+ ptr = self._borrow_user_attributes_ptr(self._ptr)
assert ptr is not None
- return bt2_value._create_from_ptr_and_get_ref(ptr)
+ return self._create_value_from_ptr_and_get_ref(ptr)
+
+
+class _StructureFieldClassMember(_StructureFieldClassMemberConst):
+ _borrow_field_class_ptr = staticmethod(
+ native_bt.field_class_structure_member_borrow_field_class
+ )
+ _borrow_user_attributes_ptr = staticmethod(
+ native_bt.field_class_structure_member_borrow_user_attributes
+ )
+ _create_field_class_from_ptr_and_get_ref = staticmethod(
+ _create_field_class_from_ptr_and_get_ref
+ )
+ _create_value_from_ptr_and_get_ref = staticmethod(
+ bt2_value._create_from_ptr_and_get_ref
+ )
def _user_attributes(self, user_attributes):
value = bt2_value.create_value(user_attributes)
_user_attributes = property(fset=_user_attributes)
-class _StructureFieldClass(_FieldClass, collections.abc.Mapping):
+class _StructureFieldClassConst(_FieldClassConst, collections.abc.Mapping):
+ _NAME = 'Const structure'
+ _borrow_member_ptr_by_index = staticmethod(
+ native_bt.field_class_structure_borrow_member_by_index_const
+ )
+ _borrow_member_ptr_by_name = staticmethod(
+ native_bt.field_class_structure_borrow_member_by_name_const
+ )
+ _structure_member_field_class_pycls = property(
+ lambda _: _StructureFieldClassMemberConst
+ )
+
+ def __len__(self):
+ count = native_bt.field_class_structure_get_member_count(self._ptr)
+ assert count >= 0
+ return count
+
+ def __getitem__(self, key):
+ if not isinstance(key, str):
+ raise TypeError(
+ "key must be a 'str' object, got '{}'".format(key.__class__.__name__)
+ )
+
+ member_ptr = self._borrow_member_ptr_by_name(self._ptr, key)
+
+ if member_ptr is None:
+ raise KeyError(key)
+
+ return self._structure_member_field_class_pycls(self, member_ptr)
+
+ def __iter__(self):
+ for idx in range(len(self)):
+ member_ptr = self._borrow_member_ptr_by_index(self._ptr, idx)
+ assert member_ptr is not None
+ yield native_bt.field_class_structure_member_get_name(member_ptr)
+
+ def member_at_index(self, index):
+ utils._check_uint64(index)
+
+ if index >= len(self):
+ raise IndexError
+
+ member_ptr = self._borrow_member_ptr_by_index(self._ptr, index)
+ assert member_ptr is not None
+ return self._structure_member_field_class_pycls(self, member_ptr)
+
+
+class _StructureFieldClass(_StructureFieldClassConst, _FieldClass):
_NAME = 'Structure'
+ _borrow_member_by_index = staticmethod(
+ native_bt.field_class_structure_borrow_member_by_index
+ )
+ _borrow_member_ptr_by_name = staticmethod(
+ native_bt.field_class_structure_borrow_member_by_name
+ )
+ _structure_member_field_class_pycls = property(lambda _: _StructureFieldClassMember)
def append_member(self, name, field_class, user_attributes=None):
utils._check_str(name)
if user_attributes is not None:
self[name]._user_attributes = user_attributes_value
- def __len__(self):
- count = native_bt.field_class_structure_get_member_count(self._ptr)
- assert count >= 0
- return count
-
- def _create_member_from_ptr(self, member_ptr):
- return _StructureFieldClassMember(self, member_ptr)
-
- def __getitem__(self, key):
- if not isinstance(key, str):
- raise TypeError(
- "key must be a 'str' object, got '{}'".format(key.__class__.__name__)
- )
-
- member_ptr = native_bt.field_class_structure_borrow_member_by_name_const(
- self._ptr, key
- )
-
- if member_ptr is None:
- raise KeyError(key)
-
- return self._create_member_from_ptr(member_ptr)
-
- def __iter__(self):
- for idx in range(len(self)):
- member_ptr = native_bt.field_class_structure_borrow_member_by_index_const(
- self._ptr, idx
- )
- assert member_ptr is not None
- yield native_bt.field_class_structure_member_get_name(member_ptr)
-
def __iadd__(self, members):
for name, field_class in members:
self.append_member(name, field_class)
return self
- def member_at_index(self, index):
- utils._check_uint64(index)
-
- if index >= len(self):
- raise IndexError
-
- member_ptr = native_bt.field_class_structure_borrow_member_by_index_const(
- self._ptr, index
- )
- assert member_ptr is not None
- return self._create_member_from_ptr(member_ptr)
+class _OptionFieldClassConst(_FieldClassConst):
+ _NAME = 'Const Option'
+ _create_field_class_from_ptr_and_get_ref = staticmethod(
+ _create_field_class_from_const_ptr_and_get_ref
+ )
+ _borrow_field_class_ptr = staticmethod(
+ native_bt.field_class_option_borrow_field_class_const
+ )
+ _borrow_selector_field_path = staticmethod(
+ native_bt.field_class_option_borrow_selector_field_path_const
+ )
-class _OptionFieldClass(_FieldClass):
@property
def field_class(self):
- elem_fc_ptr = native_bt.field_class_option_borrow_field_class_const(self._ptr)
- return _create_field_class_from_ptr_and_get_ref(elem_fc_ptr)
+ elem_fc_ptr = self._borrow_field_class_ptr(self._ptr)
+ return self._create_field_class_from_ptr_and_get_ref(elem_fc_ptr)
@property
def selector_field_path(self):
- ptr = native_bt.field_class_option_borrow_selector_field_path_const(self._ptr)
+ ptr = self._borrow_selector_field_path(self._ptr)
if ptr is None:
return
return bt2_field_path._FieldPath._create_from_ptr_and_get_ref(ptr)
-class _VariantFieldClassOption:
+class _OptionFieldClass(_OptionFieldClassConst, _FieldClass):
+ _NAME = 'Option'
+ _borrow_field_class_ptr = staticmethod(
+ native_bt.field_class_option_borrow_field_class
+ )
+ _create_field_class_from_ptr_and_get_ref = staticmethod(
+ _create_field_class_from_ptr_and_get_ref
+ )
+
+
+class _VariantFieldClassOptionConst:
+ _create_field_class_from_ptr_and_get_ref = staticmethod(
+ _create_field_class_from_const_ptr_and_get_ref
+ )
+ _borrow_field_class_ptr = staticmethod(
+ native_bt.field_class_variant_option_borrow_field_class_const
+ )
+ _borrow_user_attributes_ptr = staticmethod(
+ native_bt.field_class_variant_option_borrow_user_attributes_const
+ )
+ _create_value_from_ptr_and_get_ref = staticmethod(
+ bt2_value._create_from_const_ptr_and_get_ref
+ )
+
def __init__(self, owning_var_fc, option_ptr):
# this field class owns the option; keeping it here maintains
# the option alive as options are not shared objects
@property
def field_class(self):
- fc_ptr = native_bt.field_class_variant_option_borrow_field_class_const(
- self._ptr
- )
+ fc_ptr = self._borrow_field_class_ptr(self._ptr)
assert fc_ptr is not None
- return _create_field_class_from_ptr_and_get_ref(fc_ptr)
+ return self._create_field_class_from_ptr_and_get_ref(fc_ptr)
@property
def user_attributes(self):
- ptr = native_bt.field_class_variant_option_borrow_user_attributes(self._ptr)
+ ptr = self._borrow_user_attributes_ptr(self._ptr)
assert ptr is not None
- return bt2_value._create_from_ptr_and_get_ref(ptr)
+ return self._create_value_from_ptr_and_get_ref(ptr)
+
+
+class _VariantFieldClassOption(_VariantFieldClassOptionConst):
+ _create_field_class_from_ptr_and_get_ref = staticmethod(
+ _create_field_class_from_ptr_and_get_ref
+ )
+ _borrow_field_class_ptr = staticmethod(
+ native_bt.field_class_variant_option_borrow_field_class
+ )
+ _borrow_user_attributes_ptr = staticmethod(
+ native_bt.field_class_variant_option_borrow_user_attributes_const
+ )
+ _create_value_from_ptr_and_get_ref = staticmethod(
+ bt2_value._create_from_ptr_and_get_ref
+ )
def _user_attributes(self, user_attributes):
value = bt2_value.create_value(user_attributes)
_user_attributes = property(fset=_user_attributes)
-class _VariantFieldClass(_FieldClass, collections.abc.Mapping):
- _NAME = 'Variant'
- _borrow_option_by_name_ptr = staticmethod(
+class _VariantFieldClassWithSelectorOptionConst(_VariantFieldClassOptionConst):
+ def __init__(self, owning_var_fc, spec_opt_ptr):
+ self._spec_ptr = spec_opt_ptr
+ super().__init__(owning_var_fc, self._as_option_ptr(spec_opt_ptr))
+
+ @property
+ def ranges(self):
+ range_set_ptr = self._borrow_ranges_ptr(self._spec_ptr)
+ assert range_set_ptr is not None
+ return self._range_set_type._create_from_ptr_and_get_ref(range_set_ptr)
+
+
+class _VariantFieldClassWithSelectorOption(
+ _VariantFieldClassWithSelectorOptionConst, _VariantFieldClassOption
+):
+ pass
+
+
+class _VariantFieldClassWithSignedSelectorOptionConst(
+ _VariantFieldClassWithSelectorOptionConst
+):
+ _as_option_ptr = staticmethod(
+ native_bt.field_class_variant_with_selector_signed_option_as_option_const
+ )
+ _borrow_ranges_ptr = staticmethod(
+ native_bt.field_class_variant_with_selector_signed_option_borrow_ranges_const
+ )
+ _range_set_type = bt2_integer_range_set.SignedIntegerRangeSet
+
+
+class _VariantFieldClassWithSignedSelectorOption(
+ _VariantFieldClassWithSignedSelectorOptionConst,
+ _VariantFieldClassWithSelectorOption,
+):
+ pass
+
+
+class _VariantFieldClassWithUnsignedSelectorOptionConst(
+ _VariantFieldClassWithSelectorOptionConst
+):
+ _as_option_ptr = staticmethod(
+ native_bt.field_class_variant_with_selector_unsigned_option_as_option_const
+ )
+ _borrow_ranges_ptr = staticmethod(
+ native_bt.field_class_variant_with_selector_unsigned_option_borrow_ranges_const
+ )
+ _range_set_type = bt2_integer_range_set.UnsignedIntegerRangeSet
+
+
+class _VariantFieldClassWithUnsignedSelectorOption(
+ _VariantFieldClassWithUnsignedSelectorOptionConst,
+ _VariantFieldClassWithSelectorOption,
+):
+ pass
+
+
+class _VariantFieldClassConst(_FieldClassConst, collections.abc.Mapping):
+ _NAME = 'Const Variant'
+ _borrow_option_ptr_by_name = staticmethod(
native_bt.field_class_variant_borrow_option_by_name_const
)
- _borrow_member_by_index_ptr = staticmethod(
+ _borrow_option_ptr_by_index = staticmethod(
native_bt.field_class_variant_borrow_option_by_index_const
)
+ _variant_option_pycls = _VariantFieldClassOptionConst
@staticmethod
def _as_option_ptr(opt_ptr):
return opt_ptr
def _create_option_from_ptr(self, opt_ptr):
- return _VariantFieldClassOption(self, opt_ptr)
+ return self._variant_option_pycls(self, opt_ptr)
def __len__(self):
count = native_bt.field_class_variant_get_option_count(self._ptr)
"key must be a 'str' object, got '{}'".format(key.__class__.__name__)
)
- opt_ptr = self._borrow_option_by_name_ptr(self._ptr, key)
+ opt_ptr = self._borrow_option_ptr_by_name(self._ptr, key)
if opt_ptr is None:
raise KeyError(key)
def __iter__(self):
for idx in range(len(self)):
- opt_ptr = self._borrow_member_by_index_ptr(self._ptr, idx)
+ opt_ptr = self._borrow_option_ptr_by_index(self._ptr, idx)
assert opt_ptr is not None
base_opt_ptr = self._as_option_ptr(opt_ptr)
yield native_bt.field_class_variant_option_get_name(base_opt_ptr)
if index >= len(self):
raise IndexError
- opt_ptr = self._borrow_member_by_index_ptr(self._ptr, index)
+ opt_ptr = self._borrow_option_ptr_by_index(self._ptr, index)
assert opt_ptr is not None
return self._create_option_from_ptr(opt_ptr)
-class _VariantFieldClassWithoutSelector(_VariantFieldClass):
+class _VariantFieldClass(_VariantFieldClassConst, _FieldClass, collections.abc.Mapping):
+ _NAME = 'Variant'
+ _borrow_option_ptr_by_name = staticmethod(
+ native_bt.field_class_variant_borrow_option_by_name
+ )
+ _borrow_option_ptr_by_index = staticmethod(
+ native_bt.field_class_variant_borrow_option_by_index
+ )
+ _variant_option_pycls = _VariantFieldClassOption
+
+
+class _VariantFieldClassWithoutSelectorConst(_VariantFieldClassConst):
+ _NAME = 'Const Variant (without selector)'
+
+
+class _VariantFieldClassWithoutSelector(
+ _VariantFieldClassWithoutSelectorConst, _VariantFieldClass
+):
_NAME = 'Variant (without selector)'
def append_option(self, name, field_class, user_attributes=None):
return self
-class _VariantFieldClassWithSelectorOption(_VariantFieldClassOption):
- def __init__(self, owning_var_fc, spec_opt_ptr):
- self._spec_ptr = spec_opt_ptr
- super().__init__(owning_var_fc, self._as_option_ptr(spec_opt_ptr))
-
- @property
- def ranges(self):
- range_set_ptr = self._borrow_ranges_ptr(self._spec_ptr)
- assert range_set_ptr is not None
- return self._range_set_type._create_from_ptr_and_get_ref(range_set_ptr)
-
-
-class _VariantFieldClassWithSignedSelectorOption(_VariantFieldClassWithSelectorOption):
- _as_option_ptr = staticmethod(
- native_bt.field_class_variant_with_selector_signed_option_as_option_const
- )
- _borrow_ranges_ptr = staticmethod(
- native_bt.field_class_variant_with_selector_signed_option_borrow_ranges_const
- )
- _range_set_type = bt2_integer_range_set.SignedIntegerRangeSet
-
-
-class _VariantFieldClassWithUnsignedSelectorOption(
- _VariantFieldClassWithSelectorOption
-):
- _as_option_ptr = staticmethod(
- native_bt.field_class_variant_with_selector_unsigned_option_as_option_const
- )
- _borrow_ranges_ptr = staticmethod(
- native_bt.field_class_variant_with_selector_unsigned_option_borrow_ranges_const
- )
- _range_set_type = bt2_integer_range_set.UnsignedIntegerRangeSet
-
-
-class _VariantFieldClassWithSelector(_VariantFieldClass):
- _NAME = 'Variant (with selector)'
-
- def _create_option_from_ptr(self, opt_ptr):
- return self._option_type(self, opt_ptr)
+class _VariantFieldClassWithSelectorConst(_VariantFieldClassConst):
+ _NAME = 'Const Variant (with selector)'
@property
def selector_field_path(self):
return bt2_field_path._FieldPath._create_from_ptr_and_get_ref(ptr)
+
+class _VariantFieldClassWithSelector(
+ _VariantFieldClassWithSelectorConst, _VariantFieldClass
+):
+ _NAME = 'Variant (with selector)'
+
def append_option(self, name, field_class, ranges, user_attributes=None):
utils._check_str(name)
utils._check_type(field_class, _FieldClass)
- utils._check_type(ranges, self._option_type._range_set_type)
+ utils._check_type(ranges, self._variant_option_pycls._range_set_type)
if name in self:
raise ValueError("duplicate option name '{}'".format(name))
return self
-class _VariantFieldClassWithUnsignedSelector(_VariantFieldClassWithSelector):
- _NAME = 'Variant (with unsigned selector)'
- _borrow_option_by_name_ptr = staticmethod(
+class _VariantFieldClassWithUnsignedSelectorConst(_VariantFieldClassWithSelectorConst):
+ _NAME = 'Const Variant (with unsigned selector)'
+ _borrow_option_ptr_by_name = staticmethod(
native_bt.field_class_variant_with_selector_unsigned_borrow_option_by_name_const
)
- _borrow_member_by_index_ptr = staticmethod(
+ _borrow_option_ptr_by_index = staticmethod(
native_bt.field_class_variant_with_selector_unsigned_borrow_option_by_index_const
)
_append_option = staticmethod(
native_bt.field_class_variant_with_selector_unsigned_append_option
)
- _option_type = _VariantFieldClassWithUnsignedSelectorOption
- _as_option_ptr = staticmethod(_option_type._as_option_ptr)
+ _variant_option_pycls = _VariantFieldClassWithUnsignedSelectorOptionConst
+ _as_option_ptr = staticmethod(_variant_option_pycls._as_option_ptr)
-class _VariantFieldClassWithSignedSelector(_VariantFieldClassWithSelector):
- _NAME = 'Variant (with signed selector)'
- _borrow_option_by_name_ptr = staticmethod(
+class _VariantFieldClassWithUnsignedSelector(
+ _VariantFieldClassWithUnsignedSelectorConst, _VariantFieldClassWithSelector
+):
+ _NAME = 'Variant (with unsigned selector)'
+ _variant_option_pycls = _VariantFieldClassWithUnsignedSelectorOption
+ _as_option_ptr = staticmethod(_variant_option_pycls._as_option_ptr)
+
+
+class _VariantFieldClassWithSignedSelectorConst(_VariantFieldClassWithSelectorConst):
+ _NAME = 'Const Variant (with signed selector)'
+ _borrow_option_ptr_by_name = staticmethod(
native_bt.field_class_variant_with_selector_signed_borrow_option_by_name_const
)
- _borrow_member_by_index_ptr = staticmethod(
+ _borrow_option_ptr_by_index = staticmethod(
native_bt.field_class_variant_with_selector_signed_borrow_option_by_index_const
)
_append_option = staticmethod(
native_bt.field_class_variant_with_selector_signed_append_option
)
- _option_type = _VariantFieldClassWithSignedSelectorOption
- _as_option_ptr = staticmethod(_option_type._as_option_ptr)
+ _variant_option_pycls = _VariantFieldClassWithSignedSelectorOptionConst
+ _as_option_ptr = staticmethod(_variant_option_pycls._as_option_ptr)
+
+
+class _VariantFieldClassWithSignedSelector(
+ _VariantFieldClassWithSignedSelectorConst, _VariantFieldClassWithSelector
+):
+ _NAME = 'Variant (with signed selector)'
+ _variant_option_pycls = _VariantFieldClassWithSignedSelectorOption
+ _as_option_ptr = staticmethod(_variant_option_pycls._as_option_ptr)
-class _ArrayFieldClass(_FieldClass):
+class _ArrayFieldClassConst(_FieldClassConst):
+ _create_field_class_from_ptr_and_get_ref = staticmethod(
+ _create_field_class_from_const_ptr_and_get_ref
+ )
+ _borrow_element_field_class = staticmethod(
+ native_bt.field_class_array_borrow_element_field_class_const
+ )
+
@property
def element_field_class(self):
- elem_fc_ptr = native_bt.field_class_array_borrow_element_field_class_const(
- self._ptr
- )
- return _create_field_class_from_ptr_and_get_ref(elem_fc_ptr)
+ elem_fc_ptr = self._borrow_element_field_class(self._ptr)
+ return self._create_field_class_from_ptr_and_get_ref(elem_fc_ptr)
-class _StaticArrayFieldClass(_ArrayFieldClass):
+class _ArrayFieldClass(_ArrayFieldClassConst, _FieldClass):
+ _create_field_class_from_ptr_and_get_ref = staticmethod(
+ _create_field_class_from_ptr_and_get_ref
+ )
+ _borrow_element_field_class = staticmethod(
+ native_bt.field_class_array_borrow_element_field_class
+ )
+
+
+class _StaticArrayFieldClassConst(_ArrayFieldClassConst):
+ _NAME = 'Const static array'
+
@property
def length(self):
return native_bt.field_class_array_static_get_length(self._ptr)
-class _DynamicArrayFieldClass(_ArrayFieldClass):
+class _StaticArrayFieldClass(_StaticArrayFieldClassConst, _ArrayFieldClass):
+ _NAME = 'Static array'
+
+
+class _DynamicArrayFieldClassConst(_ArrayFieldClassConst):
+ _NAME = 'Const dynamic array'
+
@property
def length_field_path(self):
ptr = native_bt.field_class_array_dynamic_borrow_length_field_path_const(
return bt2_field_path._FieldPath._create_from_ptr_and_get_ref(ptr)
+class _DynamicArrayFieldClass(_DynamicArrayFieldClassConst, _ArrayFieldClass):
+ _NAME = 'Dynamic Array'
+
+
+_FIELD_CLASS_TYPE_TO_CONST_OBJ = {
+ native_bt.FIELD_CLASS_TYPE_BOOL: _BoolFieldClassConst,
+ native_bt.FIELD_CLASS_TYPE_BIT_ARRAY: _BitArrayFieldClassConst,
+ native_bt.FIELD_CLASS_TYPE_UNSIGNED_INTEGER: _UnsignedIntegerFieldClassConst,
+ native_bt.FIELD_CLASS_TYPE_SIGNED_INTEGER: _SignedIntegerFieldClassConst,
+ native_bt.FIELD_CLASS_TYPE_REAL: _RealFieldClassConst,
+ native_bt.FIELD_CLASS_TYPE_UNSIGNED_ENUMERATION: _UnsignedEnumerationFieldClassConst,
+ native_bt.FIELD_CLASS_TYPE_SIGNED_ENUMERATION: _SignedEnumerationFieldClassConst,
+ native_bt.FIELD_CLASS_TYPE_STRING: _StringFieldClassConst,
+ native_bt.FIELD_CLASS_TYPE_STRUCTURE: _StructureFieldClassConst,
+ native_bt.FIELD_CLASS_TYPE_STATIC_ARRAY: _StaticArrayFieldClassConst,
+ native_bt.FIELD_CLASS_TYPE_DYNAMIC_ARRAY: _DynamicArrayFieldClassConst,
+ native_bt.FIELD_CLASS_TYPE_OPTION: _OptionFieldClassConst,
+ native_bt.FIELD_CLASS_TYPE_VARIANT_WITHOUT_SELECTOR: _VariantFieldClassWithoutSelectorConst,
+ native_bt.FIELD_CLASS_TYPE_VARIANT_WITH_UNSIGNED_SELECTOR: _VariantFieldClassWithUnsignedSelectorConst,
+ native_bt.FIELD_CLASS_TYPE_VARIANT_WITH_SIGNED_SELECTOR: _VariantFieldClassWithSignedSelectorConst,
+}
+
_FIELD_CLASS_TYPE_TO_OBJ = {
native_bt.FIELD_CLASS_TYPE_BOOL: _BoolFieldClass,
native_bt.FIELD_CLASS_TYPE_BIT_ARRAY: _BitArrayFieldClass,
return _MESSAGE_TYPE_TO_CLS[msg_type]._create_from_ptr(ptr)
-class _Message(object._SharedObject):
+class _MessageConst(object._SharedObject):
_get_ref = staticmethod(native_bt.message_get_ref)
_put_ref = staticmethod(native_bt.message_put_ref)
)
+class _Message(_MessageConst):
+ pass
+
+
class _MessageWithDefaultClockSnapshot:
def _get_default_clock_snapshot(self, borrow_clock_snapshot_ptr):
snapshot_ptr = borrow_clock_snapshot_ptr(self._ptr)
)
-class _EventMessage(_Message, _MessageWithDefaultClockSnapshot):
- _borrow_default_clock_snapshot_ptr = staticmethod(
+class _EventMessageConst(_MessageConst, _MessageWithDefaultClockSnapshot):
+ _borrow_default_clock_snapshot = staticmethod(
native_bt.message_event_borrow_default_clock_snapshot_const
)
+ _borrow_event = staticmethod(native_bt.message_event_borrow_event_const)
+ _event_pycls = property(lambda _: bt2_event._EventConst)
@property
def default_clock_snapshot(self):
self._check_has_default_clock_class(self.event.stream.cls.default_clock_class)
- return self._get_default_clock_snapshot(self._borrow_default_clock_snapshot_ptr)
+ return self._get_default_clock_snapshot(self._borrow_default_clock_snapshot)
@property
def event(self):
- event_ptr = native_bt.message_event_borrow_event(self._ptr)
+ event_ptr = self._borrow_event(self._ptr)
assert event_ptr is not None
- return bt2_event._Event._create_from_ptr_and_get_ref(
+ return self._event_pycls._create_from_ptr_and_get_ref(
event_ptr, self._ptr, self._get_ref, self._put_ref
)
-class _PacketMessage(_Message, _MessageWithDefaultClockSnapshot):
+class _EventMessage(_EventMessageConst, _Message):
+ _borrow_event = staticmethod(native_bt.message_event_borrow_event)
+ _stream_pycls = property(lambda _: bt2_stream._Stream)
+ _event_pycls = property(lambda _: bt2_event._Event)
+
+
+class _PacketMessageConst(_MessageConst, _MessageWithDefaultClockSnapshot):
+ _packet_pycls = bt2_packet._PacketConst
+
@property
def default_clock_snapshot(self):
self._check_has_default_clock_class(self.packet.stream.cls.default_clock_class)
@property
def packet(self):
- packet_ptr = self._borrow_packet_ptr(self._ptr)
+ packet_ptr = self._borrow_packet(self._ptr)
assert packet_ptr is not None
- return bt2_packet._Packet._create_from_ptr_and_get_ref(packet_ptr)
+ return self._packet_pycls._create_from_ptr_and_get_ref(packet_ptr)
-class _PacketBeginningMessage(_PacketMessage):
- _borrow_packet_ptr = staticmethod(native_bt.message_packet_beginning_borrow_packet)
+class _PacketMessage(_PacketMessageConst, _Message):
+ _packet_pycls = bt2_packet._Packet
+
+
+class _PacketBeginningMessageConst(_PacketMessageConst):
+ _borrow_packet = staticmethod(
+ native_bt.message_packet_beginning_borrow_packet_const
+ )
_borrow_default_clock_snapshot_ptr = staticmethod(
native_bt.message_packet_beginning_borrow_default_clock_snapshot_const
)
-class _PacketEndMessage(_PacketMessage):
- _borrow_packet_ptr = staticmethod(native_bt.message_packet_end_borrow_packet)
+class _PacketBeginningMessage(_PacketMessage):
+ _borrow_packet = staticmethod(native_bt.message_packet_beginning_borrow_packet)
+
+
+class _PacketEndMessageConst(_PacketMessageConst):
+ _borrow_packet = staticmethod(native_bt.message_packet_end_borrow_packet_const)
_borrow_default_clock_snapshot_ptr = staticmethod(
native_bt.message_packet_end_borrow_default_clock_snapshot_const
)
-class _StreamMessage(_Message, _MessageWithDefaultClockSnapshot):
+class _PacketEndMessage(_PacketMessage):
+ _borrow_packet = staticmethod(native_bt.message_packet_end_borrow_packet)
+
+
+class _StreamMessageConst(_MessageConst, _MessageWithDefaultClockSnapshot):
+ _stream_pycls = property(lambda _: bt2_stream._StreamConst)
+
@property
def stream(self):
stream_ptr = self._borrow_stream_ptr(self._ptr)
assert stream_ptr
- return bt2_stream._Stream._create_from_ptr_and_get_ref(stream_ptr)
+ return self._stream_pycls._create_from_ptr_and_get_ref(stream_ptr)
@property
def default_clock_snapshot(self):
snapshot_ptr, self._ptr, self._get_ref, self._put_ref
)
+
+class _StreamMessage(_StreamMessageConst, _Message):
def _default_clock_snapshot(self, raw_value):
utils._check_uint64(raw_value)
self._set_default_clock_snapshot(self._ptr, raw_value)
- _default_clock_snapshot = property(fset=_default_clock_snapshot)
+ _default_clock_snapshot = property(
+ fget=_StreamMessageConst.default_clock_snapshot.fget,
+ fset=_default_clock_snapshot,
+ )
+ _stream_pycls = property(lambda _: bt2_stream._Stream)
-class _StreamBeginningMessage(_StreamMessage):
- _borrow_stream_ptr = staticmethod(native_bt.message_stream_beginning_borrow_stream)
+class _StreamBeginningMessageConst(_StreamMessageConst):
+ _borrow_stream_ptr = staticmethod(
+ native_bt.message_stream_beginning_borrow_stream_const
+ )
_borrow_default_clock_snapshot_ptr = staticmethod(
native_bt.message_stream_beginning_borrow_default_clock_snapshot_const
)
+
+
+class _StreamBeginningMessage(_StreamMessage):
+ _borrow_stream_ptr = staticmethod(native_bt.message_stream_beginning_borrow_stream)
_set_default_clock_snapshot = staticmethod(
native_bt.message_stream_beginning_set_default_clock_snapshot
)
-class _StreamEndMessage(_StreamMessage):
- _borrow_stream_ptr = staticmethod(native_bt.message_stream_end_borrow_stream)
+class _StreamEndMessageConst(_StreamMessageConst):
+ _borrow_stream_ptr = staticmethod(native_bt.message_stream_end_borrow_stream_const)
_borrow_default_clock_snapshot_ptr = staticmethod(
native_bt.message_stream_end_borrow_default_clock_snapshot_const
)
+
+
+class _StreamEndMessage(_StreamMessage):
+ _borrow_stream_ptr = staticmethod(native_bt.message_stream_end_borrow_stream)
_set_default_clock_snapshot = staticmethod(
native_bt.message_stream_end_set_default_clock_snapshot
)
-class _MessageIteratorInactivityMessage(_Message, _MessageWithDefaultClockSnapshot):
+class _MessageIteratorInactivityMessageConst(
+ _MessageConst, _MessageWithDefaultClockSnapshot
+):
_borrow_default_clock_snapshot_ptr = staticmethod(
native_bt.message_message_iterator_inactivity_borrow_default_clock_snapshot_const
)
return self._get_default_clock_snapshot(self._borrow_default_clock_snapshot_ptr)
-class _DiscardedMessage(_Message, _MessageWithDefaultClockSnapshot):
+class _MessageIteratorInactivityMessage(
+ _MessageIteratorInactivityMessageConst, _Message
+):
+ pass
+
+
+class _DiscardedMessageConst(_MessageConst, _MessageWithDefaultClockSnapshot):
+ _stream_pycls = property(lambda _: bt2_stream._StreamConst)
+
@property
def stream(self):
stream_ptr = self._borrow_stream_ptr(self._ptr)
assert stream_ptr
- return bt2_stream._Stream._create_from_ptr_and_get_ref(stream_ptr)
+ return self._stream_pycls._create_from_ptr_and_get_ref(stream_ptr)
@property
def count(self):
if avail is native_bt.PROPERTY_AVAILABILITY_AVAILABLE:
return count
- def _set_count(self, count):
- utils._check_uint64(count)
- self._set_count(self._ptr, count)
-
- _count = property(fset=_set_count)
-
def _check_has_default_clock_snapshots(self):
if not self._has_default_clock_snapshots:
raise ValueError(
return self._get_default_clock_snapshot(self._borrow_end_clock_snapshot_ptr)
-class _DiscardedEventsMessage(_DiscardedMessage):
+class _DiscardedMessage(_DiscardedMessageConst, _Message):
+ _stream_pycls = property(lambda _: bt2_stream._Stream)
+
+ def _set_count(self, count):
+ utils._check_uint64(count)
+ self._set_count(self._ptr, count)
+
+ _count = property(fget=_DiscardedMessageConst.count.fget, fset=_set_count)
+
+
+class _DiscardedEventsMessageConst(_DiscardedMessageConst):
_borrow_stream_ptr = staticmethod(
native_bt.message_discarded_events_borrow_stream_const
)
_get_count = staticmethod(native_bt.message_discarded_events_get_count)
- _set_count = staticmethod(native_bt.message_discarded_events_set_count)
_borrow_beginning_clock_snapshot_ptr = staticmethod(
native_bt.message_discarded_events_borrow_beginning_default_clock_snapshot_const
)
return self.stream.cls.discarded_events_have_default_clock_snapshots
-class _DiscardedPacketsMessage(_DiscardedMessage):
+class _DiscardedEventsMessage(_DiscardedMessage):
+ _borrow_stream_ptr = staticmethod(native_bt.message_discarded_events_borrow_stream)
+ _set_count = staticmethod(native_bt.message_discarded_events_set_count)
+
+
+class _DiscardedPacketsMessageConst(_DiscardedMessageConst):
_borrow_stream_ptr = staticmethod(
native_bt.message_discarded_packets_borrow_stream_const
)
_get_count = staticmethod(native_bt.message_discarded_packets_get_count)
- _set_count = staticmethod(native_bt.message_discarded_packets_set_count)
_borrow_beginning_clock_snapshot_ptr = staticmethod(
native_bt.message_discarded_packets_borrow_beginning_default_clock_snapshot_const
)
return self.stream.cls.discarded_packets_have_default_clock_snapshots
+class _DiscardedPacketsMessage(_DiscardedPacketsMessageConst, _DiscardedMessage):
+ _borrow_stream_ptr = staticmethod(native_bt.message_discarded_packets_borrow_stream)
+ _set_count = staticmethod(native_bt.message_discarded_packets_set_count)
+
+
_MESSAGE_TYPE_TO_CLS = {
native_bt.MESSAGE_TYPE_EVENT: _EventMessage,
native_bt.MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY: _MessageIteratorInactivityMessage,
native_bt.MESSAGE_TYPE_DISCARDED_EVENTS: _DiscardedEventsMessage,
native_bt.MESSAGE_TYPE_DISCARDED_PACKETS: _DiscardedPacketsMessage,
}
+
+_MESSAGE_TYPE_TO_CLS = {
+ native_bt.MESSAGE_TYPE_EVENT: _EventMessageConst,
+ native_bt.MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY: _MessageIteratorInactivityMessageConst,
+ native_bt.MESSAGE_TYPE_STREAM_BEGINNING: _StreamBeginningMessageConst,
+ native_bt.MESSAGE_TYPE_STREAM_END: _StreamEndMessageConst,
+ native_bt.MESSAGE_TYPE_PACKET_BEGINNING: _PacketBeginningMessageConst,
+ native_bt.MESSAGE_TYPE_PACKET_END: _PacketEndMessageConst,
+ native_bt.MESSAGE_TYPE_DISCARDED_EVENTS: _DiscardedEventsMessageConst,
+ native_bt.MESSAGE_TYPE_DISCARDED_PACKETS: _DiscardedPacketsMessageConst,
+}
except Exception:
raise
- utils._check_type(msg, bt2_message._Message)
+ utils._check_type(msg, bt2_message._MessageConst)
# The reference we return will be given to the message array.
# However, the `msg` Python object may stay alive, if the user has kept
from bt2 import stream as bt2_stream
-class _Packet(object._SharedObject):
+class _PacketConst(object._SharedObject):
_get_ref = staticmethod(native_bt.packet_get_ref)
_put_ref = staticmethod(native_bt.packet_put_ref)
+ _borrow_stream_ptr = staticmethod(native_bt.packet_borrow_stream_const)
+ _borrow_context_field_ptr = staticmethod(
+ native_bt.packet_borrow_context_field_const
+ )
+ _stream_pycls = property(lambda _: bt2_stream._StreamConst)
+ _create_field_from_ptr = staticmethod(bt2_field._create_field_from_const_ptr)
@property
def stream(self):
- stream_ptr = native_bt.packet_borrow_stream(self._ptr)
+ stream_ptr = self._borrow_stream_ptr(self._ptr)
assert stream_ptr is not None
- return bt2_stream._Stream._create_from_ptr_and_get_ref(stream_ptr)
+ return self._stream_pycls._create_from_ptr_and_get_ref(stream_ptr)
@property
def context_field(self):
- field_ptr = native_bt.packet_borrow_context_field(self._ptr)
+ field_ptr = self._borrow_context_field_ptr(self._ptr)
if field_ptr is None:
return
- return bt2_field._create_field_from_ptr(
+ return self._create_field_from_ptr(
field_ptr, self._ptr, self._get_ref, self._put_ref
)
+
+
+class _Packet(_PacketConst):
+ _borrow_stream_ptr = staticmethod(native_bt.packet_borrow_stream)
+ _borrow_context_field_ptr = staticmethod(native_bt.packet_borrow_context_field)
+ _stream_pycls = property(lambda _: bt2_stream._Stream)
+ _create_field_from_ptr = staticmethod(bt2_field._create_field_from_ptr)
import bt2
-class _Stream(bt2_object._SharedObject):
+class _StreamConst(bt2_object._SharedObject):
_get_ref = staticmethod(native_bt.stream_get_ref)
_put_ref = staticmethod(native_bt.stream_put_ref)
+ _borrow_class_ptr = staticmethod(native_bt.stream_borrow_class_const)
+ _borrow_user_attributes_ptr = staticmethod(
+ native_bt.stream_borrow_user_attributes_const
+ )
+ _create_value_from_ptr_and_get_ref = staticmethod(
+ bt2_value._create_from_const_ptr_and_get_ref
+ )
+ _borrow_trace_ptr = staticmethod(native_bt.stream_borrow_trace_const)
+ _stream_class_pycls = bt2_stream_class._StreamClassConst
+ _trace_pycls = bt2_trace._TraceConst
@property
def cls(self):
- stream_class_ptr = native_bt.stream_borrow_class(self._ptr)
+ stream_class_ptr = self._borrow_class_ptr(self._ptr)
assert stream_class_ptr is not None
- return bt2_stream_class._StreamClass._create_from_ptr_and_get_ref(
- stream_class_ptr
- )
+ return self._stream_class_pycls._create_from_ptr_and_get_ref(stream_class_ptr)
@property
def name(self):
return native_bt.stream_get_name(self._ptr)
- def _name(self, name):
- utils._check_str(name)
- native_bt.stream_set_name(self._ptr, name)
-
- _name = property(fset=_name)
-
@property
def user_attributes(self):
- ptr = native_bt.stream_borrow_user_attributes(self._ptr)
+ ptr = self._borrow_user_attributes_ptr(self._ptr)
assert ptr is not None
- return bt2_value._create_from_ptr_and_get_ref(ptr)
-
- def _user_attributes(self, user_attributes):
- value = bt2_value.create_value(user_attributes)
- utils._check_type(value, bt2_value.MapValue)
- native_bt.stream_set_user_attributes(self._ptr, value._ptr)
-
- _user_attributes = property(fset=_user_attributes)
+ return self._create_value_from_ptr_and_get_ref(ptr)
@property
def id(self):
id = native_bt.stream_get_id(self._ptr)
return id if id >= 0 else None
+ @property
+ def trace(self):
+ trace_ptr = self._borrow_trace_ptr(self._ptr)
+ assert trace_ptr is not None
+ return self._trace_pycls._create_from_ptr_and_get_ref(trace_ptr)
+
+
+class _Stream(_StreamConst):
+ _borrow_class_ptr = staticmethod(native_bt.stream_borrow_class)
+ _borrow_user_attributes_ptr = staticmethod(native_bt.stream_borrow_user_attributes)
+ _create_value_from_ptr_and_get_ref = staticmethod(
+ bt2_value._create_from_ptr_and_get_ref
+ )
+ _borrow_trace_ptr = staticmethod(native_bt.stream_borrow_trace)
+ _stream_class_pycls = bt2_stream_class._StreamClass
+ _trace_pycls = bt2_trace._Trace
+
def create_packet(self):
if not self.cls.supports_packets:
raise ValueError(
return bt2_packet._Packet._create_from_ptr(packet_ptr)
- @property
- def trace(self):
- trace_ptr = native_bt.stream_borrow_trace(self._ptr)
- assert trace_ptr is not None
- return bt2_trace._Trace._create_from_ptr_and_get_ref(trace_ptr)
+ def _user_attributes(self, user_attributes):
+ value = bt2_value.create_value(user_attributes)
+ utils._check_type(value, bt2_value.MapValue)
+ native_bt.stream_set_user_attributes(self._ptr, value._ptr)
+
+ _user_attributes = property(
+ fget=_StreamConst.user_attributes.fget, fset=_user_attributes
+ )
+
+ def _name(self, name):
+ utils._check_str(name)
+ native_bt.stream_set_name(self._ptr, name)
+
+ _name = property(fget=_StreamConst.name.fget, fset=_name)
import collections.abc
-class _StreamClass(object._SharedObject, collections.abc.Mapping):
+class _StreamClassConst(object._SharedObject, collections.abc.Mapping):
_get_ref = staticmethod(native_bt.stream_class_get_ref)
_put_ref = staticmethod(native_bt.stream_class_put_ref)
+ _borrow_event_class_ptr_by_id = staticmethod(
+ native_bt.stream_class_borrow_event_class_by_id_const
+ )
+ _borrow_event_class_ptr_by_index = staticmethod(
+ native_bt.stream_class_borrow_event_class_by_index_const
+ )
+ _borrow_trace_class_ptr = staticmethod(
+ native_bt.stream_class_borrow_trace_class_const
+ )
+ _borrow_packet_context_field_class_ptr = staticmethod(
+ native_bt.stream_class_borrow_packet_context_field_class_const
+ )
+ _borrow_event_common_context_field_class_ptr = staticmethod(
+ native_bt.stream_class_borrow_event_common_context_field_class_const
+ )
+ _borrow_default_clock_class_ptr = staticmethod(
+ native_bt.stream_class_borrow_default_clock_class_const
+ )
+
+ _event_class_cls = property(lambda _: bt2_event_class._EventClassConst)
+ _trace_class_cls = property(lambda _: bt2_trace_class._TraceClassConst)
+ _clock_class_cls = property(lambda _: bt2_clock_class._ClockClassConst)
def __getitem__(self, key):
utils._check_int64(key)
- ec_ptr = native_bt.stream_class_borrow_event_class_by_id(self._ptr, key)
+ ec_ptr = self._borrow_event_class_ptr_by_id(self._ptr, key)
if ec_ptr is None:
raise KeyError(key)
- return bt2_event_class._EventClass._create_from_ptr_and_get_ref(ec_ptr)
+ return self._event_class_cls._create_from_ptr_and_get_ref(ec_ptr)
def __len__(self):
count = native_bt.stream_class_get_event_class_count(self._ptr)
def __iter__(self):
for idx in range(len(self)):
- ec_ptr = native_bt.stream_class_borrow_event_class_by_index_const(
- self._ptr, idx
- )
+ ec_ptr = self._borrow_event_class_ptr_by_index(self._ptr, idx)
assert ec_ptr is not None
id = native_bt.event_class_get_id(ec_ptr)
yield id
+ @property
+ def trace_class(self):
+ tc_ptr = self._borrow_trace_class_ptr(self._ptr)
+
+ if tc_ptr is not None:
+ return self._trace_class_cls._create_from_ptr_and_get_ref(tc_ptr)
+
+ @property
+ def user_attributes(self):
+ ptr = native_bt.stream_class_borrow_user_attributes(self._ptr)
+ assert ptr is not None
+ return bt2_value._create_from_ptr_and_get_ref(ptr)
+
+ @property
+ def name(self):
+ return native_bt.stream_class_get_name(self._ptr)
+
+ @property
+ def assigns_automatic_event_class_id(self):
+ return native_bt.stream_class_assigns_automatic_event_class_id(self._ptr)
+
+ @property
+ def assigns_automatic_stream_id(self):
+ return native_bt.stream_class_assigns_automatic_stream_id(self._ptr)
+
+ @property
+ def supports_packets(self):
+ return native_bt.stream_class_supports_packets(self._ptr)
+
+ @property
+ def packets_have_beginning_default_clock_snapshot(self):
+ return native_bt.stream_class_packets_have_beginning_default_clock_snapshot(
+ self._ptr
+ )
+
+ @property
+ def packets_have_end_default_clock_snapshot(self):
+ return native_bt.stream_class_packets_have_end_default_clock_snapshot(self._ptr)
+
+ @property
+ def supports_discarded_events(self):
+ return native_bt.stream_class_supports_discarded_events(self._ptr)
+
+ @property
+ def discarded_events_have_default_clock_snapshots(self):
+ return native_bt.stream_class_discarded_events_have_default_clock_snapshots(
+ self._ptr
+ )
+
+ @property
+ def supports_discarded_packets(self):
+ return native_bt.stream_class_supports_discarded_packets(self._ptr)
+
+ @property
+ def discarded_packets_have_default_clock_snapshots(self):
+ return native_bt.stream_class_discarded_packets_have_default_clock_snapshots(
+ self._ptr
+ )
+
+ @property
+ def id(self):
+ id = native_bt.stream_class_get_id(self._ptr)
+
+ if id < 0:
+ return
+
+ return id
+
+ @property
+ def packet_context_field_class(self):
+ fc_ptr = self._borrow_packet_context_field_class_ptr(self._ptr)
+
+ if fc_ptr is None:
+ return
+
+ return bt2_field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
+
+ @property
+ def event_common_context_field_class(self):
+ fc_ptr = self._borrow_event_common_context_field_class_ptr(self._ptr)
+
+ if fc_ptr is None:
+ return
+
+ return bt2_field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
+
+ @property
+ def default_clock_class(self):
+ cc_ptr = self._borrow_default_clock_class_ptr(self._ptr)
+ if cc_ptr is None:
+ return
+
+ return self._clock_class_cls._create_from_ptr_and_get_ref(cc_ptr)
+
+
+class _StreamClass(_StreamClassConst):
+ _get_ref = staticmethod(native_bt.stream_class_get_ref)
+ _put_ref = staticmethod(native_bt.stream_class_put_ref)
+ _borrow_event_class_ptr_by_id = staticmethod(
+ native_bt.stream_class_borrow_event_class_by_id
+ )
+ _borrow_event_class_ptr_by_index = staticmethod(
+ native_bt.stream_class_borrow_event_class_by_index
+ )
+ _borrow_trace_class_ptr = staticmethod(native_bt.stream_class_borrow_trace_class)
+ _borrow_packet_context_field_class_ptr = staticmethod(
+ native_bt.stream_class_borrow_packet_context_field_class
+ )
+ _borrow_event_common_context_field_class_ptr = staticmethod(
+ native_bt.stream_class_borrow_event_common_context_field_class
+ )
+ _borrow_default_clock_class_ptr = staticmethod(
+ native_bt.stream_class_borrow_default_clock_class
+ )
+ _event_class_cls = property(lambda s: bt2_event_class._EventClass)
+ _trace_class_cls = property(lambda s: bt2_trace_class._TraceClass)
+ _clock_class_cls = property(lambda s: bt2_clock_class._ClockClass)
+
def create_event_class(
self,
id=None,
return event_class
- @property
- def trace_class(self):
- tc_ptr = native_bt.stream_class_borrow_trace_class_const(self._ptr)
-
- if tc_ptr is not None:
- return bt2_trace_class._TraceClass._create_from_ptr_and_get_ref(tc_ptr)
-
- @property
- def user_attributes(self):
- ptr = native_bt.stream_class_borrow_user_attributes(self._ptr)
- assert ptr is not None
- return bt2_value._create_from_ptr_and_get_ref(ptr)
-
def _user_attributes(self, user_attributes):
value = bt2_value.create_value(user_attributes)
utils._check_type(value, bt2_value.MapValue)
_user_attributes = property(fset=_user_attributes)
- @property
- def name(self):
- return native_bt.stream_class_get_name(self._ptr)
-
def _name(self, name):
utils._check_str(name)
status = native_bt.stream_class_set_name(self._ptr, name)
_name = property(fset=_name)
- @property
- def assigns_automatic_event_class_id(self):
- return native_bt.stream_class_assigns_automatic_event_class_id(self._ptr)
-
def _assigns_automatic_event_class_id(self, auto_id):
utils._check_bool(auto_id)
return native_bt.stream_class_set_assigns_automatic_event_class_id(
_assigns_automatic_event_class_id = property(fset=_assigns_automatic_event_class_id)
- @property
- def assigns_automatic_stream_id(self):
- return native_bt.stream_class_assigns_automatic_stream_id(self._ptr)
-
def _assigns_automatic_stream_id(self, auto_id):
utils._check_bool(auto_id)
return native_bt.stream_class_set_assigns_automatic_stream_id(
_assigns_automatic_stream_id = property(fset=_assigns_automatic_stream_id)
- @property
- def supports_packets(self):
- return native_bt.stream_class_supports_packets(self._ptr)
-
- @property
- def packets_have_beginning_default_clock_snapshot(self):
- return native_bt.stream_class_packets_have_beginning_default_clock_snapshot(
- self._ptr
- )
-
- @property
- def packets_have_end_default_clock_snapshot(self):
- return native_bt.stream_class_packets_have_end_default_clock_snapshot(self._ptr)
-
def _set_supports_packets(self, supports, with_begin_cs=False, with_end_cs=False):
utils._check_bool(supports)
utils._check_bool(with_begin_cs)
self._ptr, supports, with_begin_cs, with_end_cs
)
- @property
- def supports_discarded_events(self):
- return native_bt.stream_class_supports_discarded_events(self._ptr)
-
def _set_supports_discarded_events(self, supports, with_cs=False):
utils._check_bool(supports)
utils._check_bool(with_cs)
self._ptr, supports, with_cs
)
- @property
- def discarded_events_have_default_clock_snapshots(self):
- return native_bt.stream_class_discarded_events_have_default_clock_snapshots(
- self._ptr
- )
-
- @property
- def supports_discarded_packets(self):
- return native_bt.stream_class_supports_discarded_packets(self._ptr)
+ _supports_discarded_events = property(fset=_set_supports_discarded_events)
def _set_supports_discarded_packets(self, supports, with_cs):
utils._check_bool(supports)
self._ptr, supports, with_cs
)
- @property
- def discarded_packets_have_default_clock_snapshots(self):
- return native_bt.stream_class_discarded_packets_have_default_clock_snapshots(
- self._ptr
- )
-
- @property
- def id(self):
- id = native_bt.stream_class_get_id(self._ptr)
-
- if id < 0:
- return
-
- return id
-
- @property
- def packet_context_field_class(self):
- fc_ptr = native_bt.stream_class_borrow_packet_context_field_class_const(
- self._ptr
- )
-
- if fc_ptr is None:
- return
-
- return bt2_field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
+ _supports_discarded_packets = property(fset=_set_supports_discarded_packets)
def _packet_context_field_class(self, packet_context_field_class):
if packet_context_field_class is not None:
_packet_context_field_class = property(fset=_packet_context_field_class)
- @property
- def event_common_context_field_class(self):
- fc_ptr = native_bt.stream_class_borrow_event_common_context_field_class_const(
- self._ptr
- )
-
- if fc_ptr is None:
- return
-
- return bt2_field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
-
def _event_common_context_field_class(self, event_common_context_field_class):
if event_common_context_field_class is not None:
utils._check_type(
_event_common_context_field_class = property(fset=_event_common_context_field_class)
- @property
- def default_clock_class(self):
- cc_ptr = native_bt.stream_class_borrow_default_clock_class(self._ptr)
- if cc_ptr is None:
- return
-
- return bt2_clock_class._ClockClass._create_from_ptr_and_get_ref(cc_ptr)
-
def _default_clock_class(self, clock_class):
utils._check_type(clock_class, bt2_clock_class._ClockClass)
native_bt.stream_class_set_default_clock_class(self._ptr, clock_class._ptr)
import uuid as uuidp
-class _TraceEnvironment(collections.abc.MutableMapping):
+class _TraceEnvironmentConst(collections.abc.Mapping):
+ _create_value_from_ptr_and_get_ref = staticmethod(
+ bt2_value._create_from_const_ptr_and_get_ref
+ )
+
def __init__(self, trace):
self._trace = trace
if value_ptr is None:
raise KeyError(key)
- return bt2_value._create_from_ptr_and_get_ref(value_ptr)
-
- def __setitem__(self, key, value):
- if isinstance(value, str):
- set_env_entry_fn = native_bt.trace_set_environment_entry_string
- elif isinstance(value, int):
- set_env_entry_fn = native_bt.trace_set_environment_entry_integer
- else:
- raise TypeError('expected str or int, got {}'.format(type(value)))
-
- status = set_env_entry_fn(self._trace._ptr, key, value)
- utils._handle_func_status(status, "cannot set trace object's environment entry")
-
- def __delitem__(self, key):
- raise NotImplementedError
+ return self._create_value_from_ptr_and_get_ref(value_ptr)
def __len__(self):
count = native_bt.trace_get_environment_entry_count(self._trace._ptr)
return count
def __iter__(self):
- trace_ptr = self._trace_env._trace._ptr
+ trace_ptr = self._trace._ptr
for idx in range(len(self)):
borrow_entry_fn = native_bt.trace_borrow_environment_entry_by_index_const
yield entry_name
-def _trace_destruction_listener_from_native(user_listener, trace_ptr):
- trace = _Trace._create_from_ptr_and_get_ref(trace_ptr)
- user_listener(trace)
+class _TraceEnvironment(_TraceEnvironmentConst, collections.abc.MutableMapping):
+ _create_value_from_ptr_and_get_ref = staticmethod(
+ bt2_value._create_from_ptr_and_get_ref
+ )
+
+ def __setitem__(self, key, value):
+ if isinstance(value, str):
+ set_env_entry_fn = native_bt.trace_set_environment_entry_string
+ elif isinstance(value, int):
+ set_env_entry_fn = native_bt.trace_set_environment_entry_integer
+ else:
+ raise TypeError('expected str or int, got {}'.format(type(value)))
+ status = set_env_entry_fn(self._trace._ptr, key, value)
+ utils._handle_func_status(status, "cannot set trace object's environment entry")
-class _Trace(object._SharedObject, collections.abc.Mapping):
+ def __delitem__(self, key):
+ raise NotImplementedError
+
+
+class _TraceConst(object._SharedObject, collections.abc.Mapping):
_get_ref = staticmethod(native_bt.trace_get_ref)
_put_ref = staticmethod(native_bt.trace_put_ref)
+ _borrow_stream_ptr_by_id = staticmethod(native_bt.trace_borrow_stream_by_id_const)
+ _borrow_stream_ptr_by_index = staticmethod(
+ native_bt.trace_borrow_stream_by_index_const
+ )
+ _borrow_class_ptr = staticmethod(native_bt.trace_borrow_class_const)
+ _borrow_user_attributes_ptr = staticmethod(
+ native_bt.trace_borrow_user_attributes_const
+ )
+ _create_value_from_ptr_and_get_ref = staticmethod(
+ bt2_value._create_from_const_ptr_and_get_ref
+ )
+ _stream_pycls = property(lambda _: bt2_stream._StreamConst)
+ _trace_class_pycls = property(lambda _: bt2_trace_class._TraceClassConst)
+ _trace_env_pycls = property(lambda _: _TraceEnvironmentConst)
def __len__(self):
count = native_bt.trace_get_stream_count(self._ptr)
def __getitem__(self, id):
utils._check_uint64(id)
- stream_ptr = native_bt.trace_borrow_stream_by_id_const(self._ptr, id)
+ stream_ptr = self._borrow_stream_ptr_by_id(self._ptr, id)
if stream_ptr is None:
raise KeyError(id)
- return bt2_stream._Stream._create_from_ptr_and_get_ref(stream_ptr)
+ return self._stream_pycls._create_from_ptr_and_get_ref(stream_ptr)
def __iter__(self):
for idx in range(len(self)):
- stream_ptr = native_bt.trace_borrow_stream_by_index_const(self._ptr, idx)
+ stream_ptr = self._borrow_stream_ptr_by_index(self._ptr, idx)
assert stream_ptr is not None
id = native_bt.stream_get_id(stream_ptr)
@property
def cls(self):
- trace_class_ptr = native_bt.trace_borrow_class(self._ptr)
+ trace_class_ptr = self._borrow_class_ptr(self._ptr)
assert trace_class_ptr is not None
- return bt2_trace_class._TraceClass._create_from_ptr_and_get_ref(trace_class_ptr)
+ return self._trace_class_pycls._create_from_ptr_and_get_ref(trace_class_ptr)
@property
def user_attributes(self):
- ptr = native_bt.trace_borrow_user_attributes(self._ptr)
+ ptr = self._borrow_user_attributes_ptr(self._ptr)
assert ptr is not None
- return bt2_value._create_from_ptr_and_get_ref(ptr)
-
- def _user_attributes(self, user_attributes):
- value = bt2_value.create_value(user_attributes)
- utils._check_type(value, bt2_value.MapValue)
- native_bt.trace_set_user_attributes(self._ptr, value._ptr)
-
- _user_attributes = property(fset=_user_attributes)
+ return self._create_value_from_ptr_and_get_ref(ptr)
@property
def name(self):
return native_bt.trace_get_name(self._ptr)
+ @property
+ def uuid(self):
+ uuid_bytes = native_bt.trace_get_uuid(self._ptr)
+ if uuid_bytes is None:
+ return
+
+ return uuidp.UUID(bytes=uuid_bytes)
+
+ @property
+ def environment(self):
+ return self._trace_env_pycls(self)
+
+ def add_destruction_listener(self, listener):
+ '''Add a listener to be called when the trace is destroyed.'''
+ if not callable(listener):
+ raise TypeError("'listener' parameter is not callable")
+
+ fn = native_bt.bt2_trace_add_destruction_listener
+ listener_from_native = functools.partial(
+ _trace_destruction_listener_from_native, listener
+ )
+
+ status, listener_id = fn(self._ptr, listener_from_native)
+ utils._handle_func_status(
+ status, 'cannot add destruction listener to trace object'
+ )
+
+ return utils._ListenerHandle(listener_id, self)
+
+
+class _Trace(_TraceConst):
+ _borrow_stream_ptr_by_id = staticmethod(native_bt.trace_borrow_stream_by_id)
+ _borrow_stream_ptr_by_index = staticmethod(native_bt.trace_borrow_stream_by_index)
+ _borrow_class_ptr = staticmethod(native_bt.trace_borrow_class)
+ _borrow_user_attributes_ptr = staticmethod(native_bt.trace_borrow_user_attributes)
+ _create_value_from_ptr_and_get_ref = staticmethod(
+ bt2_value._create_from_ptr_and_get_ref
+ )
+ _stream_pycls = property(lambda _: bt2_stream._Stream)
+ _trace_class_pycls = property(lambda _: bt2_trace_class._TraceClass)
+ _trace_env_pycls = property(lambda _: _TraceEnvironment)
+
def _name(self, name):
utils._check_str(name)
status = native_bt.trace_set_name(self._ptr, name)
_name = property(fset=_name)
- @property
- def uuid(self):
- uuid_bytes = native_bt.trace_get_uuid(self._ptr)
- if uuid_bytes is None:
- return
+ def _user_attributes(self, user_attributes):
+ value = bt2_value.create_value(user_attributes)
+ utils._check_type(value, bt2_value.MapValue)
+ native_bt.trace_set_user_attributes(self._ptr, value._ptr)
- return uuidp.UUID(bytes=uuid_bytes)
+ _user_attributes = property(fset=_user_attributes)
def _uuid(self, uuid):
utils._check_type(uuid, uuidp.UUID)
_uuid = property(fset=_uuid)
- @property
- def environment(self):
- return _TraceEnvironment(self)
-
def create_stream(self, stream_class, id=None, name=None, user_attributes=None):
utils._check_type(stream_class, bt2_stream_class._StreamClass)
return stream
- def add_destruction_listener(self, listener):
- '''Add a listener to be called when the trace is destroyed.'''
- if not callable(listener):
- raise TypeError("'listener' parameter is not callable")
-
- fn = native_bt.bt2_trace_add_destruction_listener
- listener_from_native = functools.partial(
- _trace_destruction_listener_from_native, listener
- )
-
- status, listener_id = fn(self._ptr, listener_from_native)
- utils._handle_func_status(
- status, 'cannot add destruction listener to trace object'
- )
- return utils._ListenerHandle(listener_id, self)
+def _trace_destruction_listener_from_native(user_listener, trace_ptr):
+ trace = _TraceConst._create_from_ptr_and_get_ref(trace_ptr)
+ user_listener(trace)
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
-__all__ = ['_TraceClass']
-
-import bt2
from bt2 import native_bt, utils, object
from bt2 import stream_class as bt2_stream_class
from bt2 import field_class as bt2_field_class
from bt2 import value as bt2_value
import collections.abc
import functools
+import bt2
def _trace_class_destruction_listener_from_native(user_listener, trace_class_ptr):
- trace_class = bt2_trace_class._TraceClass._create_from_ptr_and_get_ref(
- trace_class_ptr
- )
+ trace_class = _TraceClass._create_from_ptr_and_get_ref(trace_class_ptr)
user_listener(trace_class)
-class _TraceClass(object._SharedObject, collections.abc.Mapping):
+class _TraceClassConst(object._SharedObject, collections.abc.Mapping):
_get_ref = staticmethod(native_bt.trace_class_get_ref)
_put_ref = staticmethod(native_bt.trace_class_put_ref)
+ _borrow_stream_class_ptr_by_index = staticmethod(
+ native_bt.trace_class_borrow_stream_class_by_index_const
+ )
+ _borrow_stream_class_ptr_by_id = staticmethod(
+ native_bt.trace_class_borrow_stream_class_by_id_const
+ )
+ _stream_class_pycls = bt2_stream_class._StreamClassConst
+ _create_value_from_ptr_and_get_ref = staticmethod(
+ bt2_value._create_from_const_ptr_and_get_ref
+ )
- # Instantiate a trace of this class.
-
- def __call__(self, name=None, user_attributes=None, uuid=None, environment=None):
- trace_ptr = native_bt.trace_create(self._ptr)
-
- if trace_ptr is None:
- raise bt2._MemoryError('cannot create trace class object')
-
- trace = bt2_trace._Trace._create_from_ptr(trace_ptr)
-
- if name is not None:
- trace._name = name
-
- if user_attributes is not None:
- trace._user_attributes = user_attributes
-
- if uuid is not None:
- trace._uuid = uuid
-
- if environment is not None:
- for key, value in environment.items():
- trace.environment[key] = value
-
- return trace
+ @property
+ def user_attributes(self):
+ ptr = native_bt.trace_class_borrow_user_attributes(self._ptr)
+ assert ptr is not None
+ return self._create_value_from_ptr_and_get_ref(ptr)
# Number of stream classes in this trace class.
def __getitem__(self, key):
utils._check_uint64(key)
- sc_ptr = native_bt.trace_class_borrow_stream_class_by_id_const(self._ptr, key)
+ sc_ptr = self._borrow_stream_class_ptr_by_id(self._ptr, key)
if sc_ptr is None:
raise KeyError(key)
- return bt2_stream_class._StreamClass._create_from_ptr_and_get_ref(sc_ptr)
+ return self._stream_class_pycls._create_from_ptr_and_get_ref(sc_ptr)
def __iter__(self):
for idx in range(len(self)):
- sc_ptr = native_bt.trace_class_borrow_stream_class_by_index_const(
- self._ptr, idx
- )
+ sc_ptr = self._borrow_stream_class_ptr_by_index(self._ptr, idx)
assert sc_ptr is not None
id = native_bt.stream_class_get_id(sc_ptr)
yield id
+ @property
+ def assigns_automatic_stream_class_id(self):
+ return native_bt.trace_class_assigns_automatic_stream_class_id(self._ptr)
+
+ # Add a listener to be called when the trace class is destroyed.
+
+ def add_destruction_listener(self, listener):
+
+ if not callable(listener):
+ raise TypeError("'listener' parameter is not callable")
+
+ fn = native_bt.bt2_trace_class_add_destruction_listener
+ listener_from_native = functools.partial(
+ _trace_class_destruction_listener_from_native, listener
+ )
+
+ status, listener_id = fn(self._ptr, listener_from_native)
+ utils._handle_func_status(
+ status, 'cannot add destruction listener to trace class object'
+ )
+
+ return utils._ListenerHandle(listener_id, self)
+
+
+class _TraceClass(_TraceClassConst):
+ _borrow_stream_class_ptr_by_index = staticmethod(
+ native_bt.trace_class_borrow_stream_class_by_index
+ )
+ _borrow_stream_class_ptr_by_id = staticmethod(
+ native_bt.trace_class_borrow_stream_class_by_id
+ )
+ _stream_class_pycls = bt2_stream_class._StreamClass
+ _create_value_from_ptr_and_get_ref = staticmethod(
+ bt2_value._create_from_ptr_and_get_ref
+ )
+
+ # Instantiate a trace of this class.
+
+ def __call__(self, name=None, user_attributes=None, uuid=None, environment=None):
+ trace_ptr = native_bt.trace_create(self._ptr)
+
+ if trace_ptr is None:
+ raise bt2._MemoryError('cannot create trace class object')
+
+ trace = bt2_trace._Trace._create_from_ptr(trace_ptr)
+
+ if name is not None:
+ trace._name = name
+
+ if user_attributes is not None:
+ trace._user_attributes = user_attributes
+
+ if uuid is not None:
+ trace._uuid = uuid
+
+ if environment is not None:
+ for key, value in environment.items():
+ trace.environment[key] = value
+
+ return trace
+
def create_stream_class(
self,
id=None,
)
return sc
- @property
- def user_attributes(self):
- ptr = native_bt.trace_class_borrow_user_attributes(self._ptr)
- assert ptr is not None
- return bt2_value._create_from_ptr_and_get_ref(ptr)
-
def _user_attributes(self, user_attributes):
value = bt2_value.create_value(user_attributes)
utils._check_type(value, bt2_value.MapValue)
_user_attributes = property(fset=_user_attributes)
- @property
- def assigns_automatic_stream_class_id(self):
- return native_bt.trace_class_assigns_automatic_stream_class_id(self._ptr)
-
def _assigns_automatic_stream_class_id(self, auto_id):
utils._check_bool(auto_id)
return native_bt.trace_class_set_assigns_automatic_stream_class_id(
fc = bt2_field_class._create_field_class_from_ptr_and_get_ref(ptr)
self._set_field_class_user_attrs(fc, user_attributes)
return fc
-
- # Add a listener to be called when the trace class is destroyed.
-
- def add_destruction_listener(self, listener):
-
- if not callable(listener):
- raise TypeError("'listener' parameter is not callable")
-
- fn = native_bt.bt2_trace_class_add_destruction_listener
- listener_from_native = functools.partial(
- _trace_class_destruction_listener_from_native, listener
- )
-
- status, listener_id = fn(self._ptr, listener_from_native)
- utils._handle_func_status(
- status, 'cannot add destruction listener to trace class object'
- )
-
- return utils._ListenerHandle(listener_id, self)
import unittest
import uuid
import bt2
+import utils
from utils import run_in_component_init, TestOutputPortMessageIterator
from bt2 import value as bt2_value
from bt2 import clock_class as bt2_clock_class
self.assertRaisesInComponentInit(TypeError, f)
+ def test_const_user_attributes(self):
+ cc = utils.get_const_event_message().default_clock_snapshot.clock_class
+ self.assertIs(type(cc.user_attributes), bt2_value._MapValueConst)
+
class ClockSnapshotTestCase(unittest.TestCase):
def setUp(self):
import unittest
import bt2
+import utils
from utils import TestOutputPortMessageIterator
+from bt2 import field as bt2_field
+from bt2 import stream as bt2_stream
+from bt2 import event_class as bt2_event_class
+from bt2 import clock_snapshot as bt2_clock_snapshot
+
class EventTestCase(unittest.TestCase):
- def _create_test_event_message(
+ def _create_test_const_event_message(
self,
packet_fields_config=None,
event_fields_config=None,
)
for msg in self._msg_iter:
- if type(msg) is bt2._EventMessage:
+ if type(msg) is bt2._EventMessageConst:
+ self._event_msg = msg
return msg
- def test_attr_event_class(self):
- msg = self._create_test_event_message()
+ def test_const_attr_event_class(self):
+ msg = self._create_test_const_event_message()
self.assertEqual(msg.event.cls.addr, self.event_class.addr)
+ self.assertIs(type(msg.event.cls), bt2_event_class._EventClassConst)
+
+ def test_attr_event_class(self):
+ msg = utils.get_event_message()
+ self.assertIs(type(msg.event.cls), bt2_event_class._EventClass)
- def test_attr_name(self):
- msg = self._create_test_event_message()
+ def test_const_attr_name(self):
+ msg = self._create_test_const_event_message()
self.assertEqual(msg.event.name, self.event_class.name)
- def test_attr_id(self):
- msg = self._create_test_event_message()
+ def test_const_attr_id(self):
+ msg = self._create_test_const_event_message()
self.assertEqual(msg.event.id, self.event_class.id)
- def test_get_common_context_field(self):
+ def test_const_get_common_context_field(self):
def event_fields_config(event):
event.common_context_field['cpu_id'] = 1
event.common_context_field['stuff'] = 13.194
- msg = self._create_test_event_message(
+ msg = self._create_test_const_event_message(
event_fields_config=event_fields_config, with_cc=True
)
self.assertEqual(msg.event.common_context_field['cpu_id'], 1)
self.assertEqual(msg.event.common_context_field['stuff'], 13.194)
+ self.assertIs(
+ type(msg.event.common_context_field), bt2_field._StructureFieldConst
+ )
+
+ def test_attr_common_context_field(self):
+ msg = utils.get_event_message()
+ self.assertIs(type(msg.event.common_context_field), bt2_field._StructureField)
- def test_no_common_context_field(self):
- msg = self._create_test_event_message(with_cc=False)
+ def test_const_no_common_context_field(self):
+ msg = self._create_test_const_event_message(with_cc=False)
self.assertIsNone(msg.event.common_context_field)
- def test_get_specific_context_field(self):
+ def test_const_get_specific_context_field(self):
def event_fields_config(event):
event.specific_context_field['ant'] = -1
event.specific_context_field['msg'] = 'hellooo'
- msg = self._create_test_event_message(
+ msg = self._create_test_const_event_message(
event_fields_config=event_fields_config, with_sc=True
)
self.assertEqual(msg.event.specific_context_field['ant'], -1)
self.assertEqual(msg.event.specific_context_field['msg'], 'hellooo')
+ self.assertIs(
+ type(msg.event.specific_context_field), bt2_field._StructureFieldConst
+ )
+
+ def test_attr_specific_context_field(self):
+ msg = utils.get_event_message()
+ self.assertIs(type(msg.event.specific_context_field), bt2_field._StructureField)
- def test_no_specific_context_field(self):
- msg = self._create_test_event_message(with_sc=False)
+ def test_const_no_specific_context_field(self):
+ msg = self._create_test_const_event_message(with_sc=False)
self.assertIsNone(msg.event.specific_context_field)
- def test_get_event_payload_field(self):
+ def test_const_get_event_payload_field(self):
def event_fields_config(event):
event.payload_field['giraffe'] = 1
event.payload_field['gnu'] = 23
event.payload_field['mosquito'] = 42
- msg = self._create_test_event_message(
+ msg = self._create_test_const_event_message(
event_fields_config=event_fields_config, with_ep=True
)
self.assertEqual(msg.event.payload_field['giraffe'], 1)
self.assertEqual(msg.event.payload_field['gnu'], 23)
self.assertEqual(msg.event.payload_field['mosquito'], 42)
+ self.assertIs(type(msg.event.payload_field), bt2_field._StructureFieldConst)
+
+ def test_attr_payload_field(self):
+ msg = utils.get_event_message()
+ self.assertIs(type(msg.event.payload_field), bt2_field._StructureField)
- def test_no_payload_field(self):
- msg = self._create_test_event_message(with_ep=False)
+ def test_const_no_payload_field(self):
+ msg = self._create_test_const_event_message(with_ep=False)
self.assertIsNone(msg.event.payload_field)
+ def test_const_clock_value(self):
+ msg = self._create_test_const_event_message(with_clockclass=True)
+ self.assertEqual(msg.default_clock_snapshot.value, 789)
+ self.assertIs(
+ type(msg.default_clock_snapshot), bt2_clock_snapshot._ClockSnapshotConst
+ )
+
def test_clock_value(self):
- msg = self._create_test_event_message(with_clockclass=True)
+ msg = utils.get_event_message()
self.assertEqual(msg.default_clock_snapshot.value, 789)
+ self.assertIs(
+ type(msg.default_clock_snapshot), bt2_clock_snapshot._ClockSnapshotConst
+ )
- def test_no_clock_value(self):
- msg = self._create_test_event_message(with_clockclass=False)
+ def test_const_no_clock_value(self):
+ msg = self._create_test_const_event_message(with_clockclass=False)
with self.assertRaisesRegex(
ValueError, 'stream class has no default clock class'
):
msg.default_clock_snapshot
- def test_stream(self):
- msg = self._create_test_event_message()
+ def test_const_stream(self):
+ msg = self._create_test_const_event_message()
self.assertEqual(msg.event.stream.addr, self.stream.addr)
+ self.assertIs(type(msg.event.stream), bt2_stream._StreamConst)
- def test_getitem(self):
+ def test_stream(self):
+ msg = utils.get_event_message()
+ self.assertIs(type(msg.event.stream), bt2_stream._Stream)
+
+ def test_const_getitem(self):
def event_fields_config(event):
event.payload_field['giraffe'] = 1
event.payload_field['gnu'] = 23
packet.context_field['something'] = 154
packet.context_field['something_else'] = 17.2
- msg = self._create_test_event_message(
+ msg = self._create_test_const_event_message(
packet_fields_config=packet_fields_config,
event_fields_config=event_fields_config,
with_cc=True,
# Test event fields
self.assertEqual(ev['giraffe'], 1)
+ self.assertIs(type(ev['giraffe']), bt2_field._SignedIntegerFieldConst)
self.assertEqual(ev['gnu'], 23)
self.assertEqual(ev['mosquito'], 42)
self.assertEqual(ev['ant'], -1)
+ self.assertIs(type(ev['ant']), bt2_field._SignedIntegerFieldConst)
self.assertEqual(ev['msg'], 'hellooo')
self.assertEqual(ev['cpu_id'], 1)
+ self.assertIs(type(ev['cpu_id']), bt2_field._SignedIntegerFieldConst)
self.assertEqual(ev['stuff'], 13.194)
# Test packet fields
self.assertEqual(ev['something'], 154)
+ self.assertIs(type(ev['something']), bt2_field._UnsignedIntegerFieldConst)
self.assertEqual(ev['something_else'], 17.2)
with self.assertRaises(KeyError):
ev['yes']
+ def test_getitem(self):
+ msg = utils.get_event_message()
+ ev = msg.event
+ self.assertEqual(ev['giraffe'], 1)
+ self.assertIs(type(ev['giraffe']), bt2_field._SignedIntegerField)
+ self.assertEqual(ev['ant'], -1)
+ self.assertIs(type(ev['ant']), bt2_field._SignedIntegerField)
+ self.assertEqual(ev['cpu_id'], 1)
+ self.assertIs(type(ev['cpu_id']), bt2_field._SignedIntegerField)
+ self.assertEqual(ev['something'], 154)
+ self.assertIs(type(ev['something']), bt2_field._UnsignedIntegerField)
+
if __name__ == "__main__":
unittest.main()
import unittest
import bt2
from utils import get_default_trace_class
+from bt2 import stream_class as bt2_stream_class
+from bt2 import event_class as bt2_event_class
+from bt2 import field_class as bt2_field_class
+from bt2 import value as bt2_value
+from utils import TestOutputPortMessageIterator
+
+
+def _create_const_event_class(tc, stream_class):
+ fc1 = tc.create_structure_field_class()
+ fc2 = tc.create_structure_field_class()
+ event_class = stream_class.create_event_class(
+ payload_field_class=fc1, specific_context_field_class=fc2
+ )
+
+ class MyIter(bt2._UserMessageIterator):
+ def __init__(self, self_port_output):
+
+ trace = tc()
+ stream = trace.create_stream(stream_class)
+ self._msgs = [
+ self._create_stream_beginning_message(stream),
+ self._create_event_message(event_class, stream),
+ ]
+
+ def __next__(self):
+ if len(self._msgs) == 0:
+ raise StopIteration
+
+ return self._msgs.pop(0)
+
+ class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ def __init__(self, params, obj):
+ self._add_output_port('out', params)
+
+ graph = bt2.Graph()
+ src_comp = graph.add_component(MySrc, 'my_source', None)
+ msg_iter = TestOutputPortMessageIterator(graph, src_comp.output_ports['out'])
+
+ # Ignore first message, stream beginning
+ _ = next(msg_iter)
+
+ event_msg = next(msg_iter)
+
+ return event_msg.event.cls
class EventClassTestCase(unittest.TestCase):
def test_create_default(self):
ec = self._stream_class.create_event_class()
+ self.assertIs(type(ec), bt2_event_class._EventClass)
self.assertIsNone(ec.name, 'my_event')
self.assertTrue(type(ec.id), int)
self.assertIsNone(ec.specific_context_field_class)
fc = self._tc.create_structure_field_class()
ec = self._stream_class.create_event_class(specific_context_field_class=fc)
self.assertEqual(ec.specific_context_field_class.addr, fc.addr)
+ self.assertIs(
+ type(ec.specific_context_field_class), bt2_field_class._StructureFieldClass
+ )
+
+ def test_const_create_specific_context_field_class(self):
+ ec_const = _create_const_event_class(self._tc, self._stream_class)
+ self.assertIs(
+ type(ec_const.specific_context_field_class),
+ bt2_field_class._StructureFieldClassConst,
+ )
def test_create_invalid_specific_context_field_class(self):
with self.assertRaises(TypeError):
fc = self._tc.create_structure_field_class()
ec = self._stream_class.create_event_class(payload_field_class=fc)
self.assertEqual(ec.payload_field_class.addr, fc.addr)
+ self.assertIs(
+ type(ec.payload_field_class), bt2_field_class._StructureFieldClass
+ )
+
+ def test_const_create_payload_field_class(self):
+ ec_const = _create_const_event_class(self._tc, self._stream_class)
+ self.assertIs(
+ type(ec_const.payload_field_class),
+ bt2_field_class._StructureFieldClassConst,
+ )
def test_create_invalid_payload_field_class(self):
with self.assertRaises(TypeError):
def test_create_user_attributes(self):
ec = self._stream_class.create_event_class(user_attributes={'salut': 23})
self.assertEqual(ec.user_attributes, {'salut': 23})
+ self.assertIs(type(ec.user_attributes), bt2_value.MapValue)
+
+ def test_const_create_user_attributes(self):
+ ec_const = _create_const_event_class(self._tc, self._stream_class)
+ self.assertIs(type(ec_const.user_attributes), bt2_value._MapValueConst)
def test_create_invalid_user_attributes(self):
with self.assertRaises(TypeError):
def test_stream_class(self):
ec = self._stream_class.create_event_class()
self.assertEqual(ec.stream_class.addr, self._stream_class.addr)
+ self.assertIs(type(ec.stream_class), bt2_stream_class._StreamClass)
+
+ def test_const_stream_class(self):
+ ec_const = _create_const_event_class(self._tc, self._stream_class)
+ self.assertIs(type(ec_const.stream_class), bt2_stream_class._StreamClassConst)
import itertools
import collections
import bt2
-from utils import get_default_trace_class
+from utils import get_default_trace_class, TestOutputPortMessageIterator
_COMP_BINOPS = (operator.eq, operator.ne)
return packet.context_field[field_name]
+# Create a const field of the given field class.
+#
+# The field is part of a dummy stream, itself part of a dummy trace created
+# from trace class `tc`.
+
+
+def _create_const_field(tc, field_class, field_value_setter_fn):
+ field_name = 'const field'
+
+ class MyIter(bt2._UserMessageIterator):
+ def __init__(self, self_port_output):
+ nonlocal field_class
+ nonlocal field_value_setter_fn
+ stream = _create_stream(tc, [(field_name, field_class)])
+ packet = stream.create_packet()
+
+ field_value_setter_fn(packet.context_field[field_name])
+
+ self._msgs = [
+ self._create_stream_beginning_message(stream),
+ self._create_packet_beginning_message(packet),
+ ]
+
+ def __next__(self):
+ if len(self._msgs) == 0:
+ raise StopIteration
+
+ return self._msgs.pop(0)
+
+ class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ def __init__(self, params, obj):
+ self._add_output_port('out', params)
+
+ graph = bt2.Graph()
+ src_comp = graph.add_component(MySrc, 'my_source', None)
+ msg_iter = TestOutputPortMessageIterator(graph, src_comp.output_ports['out'])
+
+ # Ignore first message, stream beginning
+ _ = next(msg_iter)
+ packet_beg_msg = next(msg_iter)
+
+ return packet_beg_msg.packet.context_field[field_name]
+
+
# Create a field of type string.
#
# The field is part of a dummy stream, itself part of a dummy trace created
class BoolFieldTestCase(_TestNumericField, unittest.TestCase):
+ @staticmethod
+ def _const_value_setter(field):
+ field.value = True
+
def _create_fc(self, tc):
return tc.create_bool_field_class()
self._def = _create_field(self._tc, self._create_fc(self._tc))
self._def.value = True
self._def_value = True
+ self._def_const = _create_const_field(
+ self._tc, self._tc.create_bool_field_class(), self._const_value_setter
+ )
self._def_new_value = False
+ def test_classes(self):
+ self.assertIs(type(self._def), bt2._BoolField)
+ self.assertIs(type(self._def_const), bt2._BoolFieldConst)
+
def test_assign_true(self):
raw = True
self._def.value = raw
class StringFieldTestCase(unittest.TestCase):
+ @staticmethod
+ def _const_value_setter(field):
+ field.value = 'Hello, World!'
+
def setUp(self):
self._tc = get_default_trace_class()
self._def_value = 'Hello, World!'
self._def = _create_string_field(self._tc)
+ self._def_const = _create_const_field(
+ self._tc, self._tc.create_string_field_class(), self._const_value_setter
+ )
self._def.value = self._def_value
self._def_new_value = 'Yes!'
def test_eq(self):
self.assertEqual(self._def, self._def_value)
+ def test_const_eq(self):
+ self.assertEqual(self._def_const, self._def_value)
+
def test_not_eq(self):
self.assertNotEqual(self._def, 23)
def test_getitem(self):
self.assertEqual(self._def[5], self._def_value[5])
+ def test_const_getitem(self):
+ self.assertEqual(self._def_const[5], self._def_value[5])
+
def test_append_str(self):
to_append = 'meow meow meow'
self._def += to_append
self._def_value += to_append
self.assertEqual(self._def, self._def_value)
+ def test_const_append_str(self):
+ to_append = 'meow meow meow'
+ with self.assertRaises(TypeError):
+ self._def_const += to_append
+ self.assertEqual(self._def_const, self._def_value)
+
def test_append_string_field(self):
field = _create_string_field(self._tc)
to_append = 'meow meow meow'
self.assertIs(type(field), bt2._SignedIntegerField)
self.assertEqual(field, 1847)
+ def test_const_getitem(self):
+ field = self._def_const[1]
+ self.assertIs(type(field), bt2._SignedIntegerFieldConst)
+ self.assertEqual(field, 1847)
+
def test_eq(self):
field = _create_int_array_field(self._tc, 3)
field[0] = 45
with self.assertRaises(IndexError):
self._def[len(self._def)] = 134679
+ def test_const_setitem(self):
+ with self.assertRaises(TypeError):
+ self._def_const[0] = 134679
+
def test_iter(self):
for field, value in zip(self._def, (45, 1847, 1948754)):
self.assertEqual(field, value)
+ def test_const_iter(self):
+ for field, value in zip(self._def_const, (45, 1847, 1948754)):
+ self.assertEqual(field, value)
+
def test_value_int_field(self):
values = [45646, 145, 12145]
self._def.value = values
class StaticArrayFieldTestCase(_TestArrayFieldCommon, unittest.TestCase):
+ @staticmethod
+ def _const_value_setter(field):
+ field.value = [45, 1847, 1948754]
+
def setUp(self):
self._tc = get_default_trace_class()
self._def = _create_int_array_field(self._tc, 3)
self._def[1] = 1847
self._def[2] = 1948754
self._def_value = [45, 1847, 1948754]
+ self._def_const = _create_const_field(
+ self._tc,
+ self._tc.create_static_array_field_class(
+ self._tc.create_signed_integer_field_class(32), 3
+ ),
+ self._const_value_setter,
+ )
def test_value_wrong_len(self):
values = [45, 1847]
class DynamicArrayFieldTestCase(_TestArrayFieldCommon, unittest.TestCase):
+ @staticmethod
+ def _const_value_setter(field):
+ field.value = [45, 1847, 1948754]
+
def setUp(self):
self._tc = get_default_trace_class()
self._def = _create_dynamic_array(self._tc)
self._def[1] = 1847
self._def[2] = 1948754
self._def_value = [45, 1847, 1948754]
+ self._def_const = _create_const_field(
+ self._tc,
+ self._tc.create_dynamic_array_field_class(
+ self._tc.create_signed_integer_field_class(32)
+ ),
+ self._const_value_setter,
+ )
def test_value_resize(self):
new_values = [1, 2, 3, 4]
self._def[3] = 0
self.assertEqual(len(self._def), 4)
+ def test_const_set_length(self):
+ with self.assertRaises(AttributeError):
+ self._def_const.length = 4
+ self.assertEqual(len(self._def), 3)
+
def test_set_invalid_length(self):
with self.assertRaises(TypeError):
self._def.length = 'cheval'
class StructureFieldTestCase(unittest.TestCase):
+ @staticmethod
+ def _const_value_setter(field):
+ field.value = {
+ 'A': -1872,
+ 'B': 'salut',
+ 'C': 17.5,
+ 'D': 16497,
+ 'E': {},
+ 'F': {'F_1': 52},
+ }
+
def _create_fc(self, tc):
fc = tc.create_structure_field_class()
fc.append_member('A', self._fc0_fn())
'F': {'F_1': 52},
}
+ self._def_const = _create_const_field(
+ self._tc, self._create_fc(self._tc), self._const_value_setter
+ )
+
def _modify_def(self):
self._def['B'] = 'hola'
self.assertEqual(len(self._def), len(self._def_value))
def test_getitem(self):
- field = self._def['A']
- self.assertIs(type(field), bt2._SignedIntegerField)
- self.assertEqual(field, -1872)
+ field1 = self._def['A']
+ field2 = self._def['B']
+ field3 = self._def['C']
+ field4 = self._def['D']
+ field5 = self._def['E']
+ field6 = self._def['F']
+
+ self.assertIs(type(field1), bt2._SignedIntegerField)
+ self.assertEqual(field1, -1872)
+
+ self.assertIs(type(field2), bt2._StringField)
+ self.assertEqual(field2, 'salut')
+
+ self.assertIs(type(field3), bt2._RealField)
+ self.assertEqual(field3, 17.5)
+
+ self.assertIs(type(field4), bt2._SignedIntegerField)
+ self.assertEqual(field4, 16497)
+
+ self.assertIs(type(field5), bt2._StructureField)
+ self.assertEqual(field5, {})
+
+ self.assertIs(type(field6), bt2._StructureField)
+ self.assertEqual(field6, {'F_1': 52})
+
+ def test_const_getitem(self):
+ field1 = self._def_const['A']
+ field2 = self._def_const['B']
+ field3 = self._def_const['C']
+ field4 = self._def_const['D']
+ field5 = self._def_const['E']
+ field6 = self._def_const['F']
+
+ self.assertIs(type(field1), bt2._SignedIntegerFieldConst)
+ self.assertEqual(field1, -1872)
+
+ self.assertIs(type(field2), bt2._StringFieldConst)
+ self.assertEqual(field2, 'salut')
+
+ self.assertIs(type(field3), bt2._RealFieldConst)
+ self.assertEqual(field3, 17.5)
+
+ self.assertIs(type(field4), bt2._SignedIntegerFieldConst)
+ self.assertEqual(field4, 16497)
+
+ self.assertIs(type(field5), bt2._StructureFieldConst)
+ self.assertEqual(field5, {})
+
+ self.assertIs(type(field6), bt2._StructureFieldConst)
+ self.assertEqual(field6, {'F_1': 52})
def test_member_at_index_out_of_bounds_after(self):
with self.assertRaises(IndexError):
field['F'] = {'F_1': 52}
self.assertEqual(self._def, field)
+ def test_const_eq(self):
+ field = _create_field(self._tc, self._create_fc(self._tc))
+ field['A'] = -1872
+ field['B'] = 'salut'
+ field['C'] = 17.5
+ field['D'] = 16497
+ field['E'] = {}
+ field['F'] = {'F_1': 52}
+ self.assertEqual(self._def_const, field)
+
def test_eq_invalid_type(self):
self.assertNotEqual(self._def, 23)
self._def['C'] = -18.47
self.assertEqual(self._def['C'], -18.47)
+ def test_const_setitem(self):
+ with self.assertRaises(TypeError):
+ self._def_const['A'] = 134679
+
def test_setitem_int_field(self):
int_fc = self._tc.create_signed_integer_field_class(32)
int_field = _create_field(self._tc, int_fc)
def test_member_at_index(self):
self.assertEqual(self._def.member_at_index(1), 'salut')
+ def test_const_member_at_index(self):
+ self.assertEqual(self._def_const.member_at_index(1), 'salut')
+
def test_iter(self):
orig_values = {
'A': -1872,
class OptionFieldTestCase(unittest.TestCase):
+ @staticmethod
+ def _const_value_setter(field):
+ field.value = {'opt_field': 'hiboux'}
+
def _create_fc(self, tc):
fc = tc.create_option_field_class(tc.create_string_field_class())
top_fc = tc.create_structure_field_class()
self._tc = get_default_trace_class()
fld = _create_field(self._tc, self._create_fc(self._tc))
self._def = fld['opt_field']
+ self._def_value = 'hiboux'
+ self._def_const = _create_const_field(
+ self._tc, self._create_fc(self._tc), self._const_value_setter
+ )['opt_field']
def test_value_prop(self):
self._def.value = 'hiboux'
self.assertEqual(self._def.field, 'hiboux')
+ self.assertIs(type(self._def), bt2._OptionField)
+ self.assertIs(type(self._def.field), bt2._StringField)
self.assertTrue(self._def.has_field)
+ def test_const_value_prop(self):
+ self.assertEqual(self._def_const.field, 'hiboux')
+ self.assertIs(type(self._def_const), bt2._OptionFieldConst)
+ self.assertIs(type(self._def_const.field), bt2._StringFieldConst)
+ self.assertTrue(self._def_const.has_field)
+
def test_has_field_prop_true(self):
self._def.has_field = True
self.assertTrue(self._def.has_field)
field = self._def.field
self.assertIsNone(field)
+ def test_const_field_prop(self):
+ with self.assertRaises(AttributeError):
+ self._def_const.has_field = False
+
+ self.assertEqual(self._def_const, self._def_value)
+ self.assertTrue(self._def_const.has_field)
+
def test_field_prop_existing_then_none(self):
self._def.value = 'meow'
field = self._def.field
self._def.value = 'walk'
self.assertEqual(self._def, field)
+ def test_const_eq(self):
+ field = _create_field(self._tc, self._create_fc(self._tc))
+ field = field['opt_field']
+ field.value = 'hiboux'
+ self.assertEqual(self._def_const, field)
+ self.assertEqual(self._def_const, self._def_value)
+
def test_eq_invalid_type(self):
self._def.value = 'gerry'
self.assertNotEqual(self._def, 23)
class VariantFieldTestCase(unittest.TestCase):
+ @staticmethod
+ def _const_value_setter(field):
+ field.selected_option_index = 3
+ field.value = 1334
+
def _create_fc(self, tc):
ft0 = tc.create_signed_integer_field_class(32)
ft1 = tc.create_string_field_class()
fld = _create_field(self._tc, self._create_fc(self._tc))
self._def = fld['variant_field']
+ self._def_value = 1334
+ self._def_selected_index = 3
+ const_fc = self._create_fc(self._tc)['variant_field']
+
+ fld_const = _create_const_field(
+ self._tc, const_fc.field_class, self._const_value_setter
+ )
+ self._def_const = fld_const
+
def test_bool_op(self):
self._def.selected_option_index = 2
self._def.value = -17.34
with self.assertRaises(IndexError):
self._def.selected_option_index = -1
+ def test_const_selected_option_index(self):
+ with self.assertRaises(AttributeError):
+ self._def_const.selected_option_index = 2
+ self.assertEqual(self._def_const.selected_option_index, 3)
+
def test_selected_option(self):
self._def.selected_option_index = 2
self._def.value = -17.34
self.assertEqual(self._def.selected_option, -17.34)
+ self.assertEqual(type(self._def.selected_option), bt2._RealField)
self._def.selected_option_index = 3
self._def.value = 1921
self.assertEqual(self._def.selected_option, 1921)
+ self.assertEqual(type(self._def.selected_option), bt2._SignedIntegerField)
+
+ def test_const_selected_option(self):
+ self.assertEqual(self._def_const.selected_option, 1334)
+ self.assertEqual(
+ type(self._def_const.selected_option), bt2._SignedIntegerFieldConst
+ )
def test_eq(self):
field = _create_field(self._tc, self._create_fc(self._tc))
self._def.value = 1774
self.assertEqual(self._def, field)
+ def test_const_eq(self):
+ field = _create_field(self._tc, self._create_fc(self._tc))
+ field = field['variant_field']
+ field.selected_option_index = 3
+ field.value = 1334
+ self.assertEqual(self._def_const, field)
+
def test_len(self):
self.assertEqual(len(self._def), 4)
import unittest
import bt2
-from utils import get_default_trace_class
+from utils import get_default_trace_class, TestOutputPortMessageIterator
+from bt2 import value as bt2_value
+from bt2 import field_class as bt2_field_class
+
+
+def _create_stream(tc, ctx_field_classes):
+ packet_context_fc = tc.create_structure_field_class()
+ for name, fc in ctx_field_classes:
+ packet_context_fc.append_member(name, fc)
+
+ trace = tc()
+ stream_class = tc.create_stream_class(
+ packet_context_field_class=packet_context_fc, supports_packets=True
+ )
+
+ stream = trace.create_stream(stream_class)
+ return stream
+
+
+def _create_const_field_class(tc, field_class, value_setter_fn):
+ field_name = 'const field'
+
+ class MyIter(bt2._UserMessageIterator):
+ def __init__(self, self_port_output):
+ nonlocal field_class
+ nonlocal value_setter_fn
+ stream = _create_stream(tc, [(field_name, field_class)])
+ packet = stream.create_packet()
+
+ value_setter_fn(packet.context_field[field_name])
+
+ self._msgs = [
+ self._create_stream_beginning_message(stream),
+ self._create_packet_beginning_message(packet),
+ ]
+
+ def __next__(self):
+ if len(self._msgs) == 0:
+ raise StopIteration
+
+ return self._msgs.pop(0)
+
+ class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ def __init__(self, params, obj):
+ self._add_output_port('out', params)
+
+ graph = bt2.Graph()
+ src_comp = graph.add_component(MySrc, 'my_source', None)
+ msg_iter = TestOutputPortMessageIterator(graph, src_comp.output_ports['out'])
+
+ # Ignore first message, stream beginning
+ _ = next(msg_iter)
+ packet_beg_msg = next(msg_iter)
+
+ return packet_beg_msg.packet.context_field[field_name].cls
class _TestFieldClass:
def test_create_user_attributes(self):
fc = self._create_default_field_class(user_attributes={'salut': 23})
self.assertEqual(fc.user_attributes, {'salut': 23})
+ self.assertIs(type(fc.user_attributes), bt2_value.MapValue)
+
+ def test_const_create_user_attributes(self):
+ fc = self._create_default_const_field_class(user_attributes={'salut': 23})
+ self.assertEqual(fc.user_attributes, {'salut': 23})
+ self.assertIs(type(fc.user_attributes), bt2_value._MapValueConst)
def test_create_invalid_user_attributes(self):
with self.assertRaises(TypeError):
class BoolFieldClassTestCase(_TestFieldClass, unittest.TestCase):
+ @staticmethod
+ def _const_value_setter(field):
+ field = False
+
def _create_default_field_class(self, **kwargs):
tc = get_default_trace_class()
return tc.create_bool_field_class(**kwargs)
+ def _create_default_const_field_class(self, *args, **kwargs):
+ tc = get_default_trace_class()
+ fc = tc.create_bool_field_class(*args, **kwargs)
+ return _create_const_field_class(tc, fc, self._const_value_setter)
+
def setUp(self):
self._fc = self._create_default_field_class()
+ self._fc_const = self._create_default_const_field_class()
def test_create_default(self):
self.assertIsNotNone(self._fc)
class BitArrayFieldClassTestCase(_TestFieldClass, unittest.TestCase):
+ @staticmethod
+ def _const_value_setter(field):
+ field = []
+
def _create_field_class(self, *args, **kwargs):
tc = get_default_trace_class()
return tc.create_bit_array_field_class(*args, **kwargs)
def _create_default_field_class(self, **kwargs):
return self._create_field_class(17, **kwargs)
+ def _create_default_const_field_class(self, *args, **kwargs):
+ tc = get_default_trace_class()
+ fc = tc.create_bit_array_field_class(17, **kwargs)
+ return _create_const_field_class(tc, fc, self._const_value_setter)
+
def setUp(self):
self._fc = self._create_default_field_class()
class SignedIntegerFieldClassTestCase(
_TestIntegerFieldClassProps, _TestFieldClass, unittest.TestCase
):
+ @staticmethod
+ def _const_value_setter(field):
+ field = -18
+
def _create_field_class(self, *args, **kwargs):
tc = get_default_trace_class()
return tc.create_signed_integer_field_class(*args, **kwargs)
+ def _create_default_const_field_class(self, *args, **kwargs):
+ tc = get_default_trace_class()
+ fc = tc.create_signed_integer_field_class(*args, **kwargs)
+ return _create_const_field_class(tc, fc, self._const_value_setter)
+
_create_default_field_class = _create_field_class
class UnsignedIntegerFieldClassTestCase(
_TestIntegerFieldClassProps, _TestFieldClass, unittest.TestCase
):
+ @staticmethod
+ def _const_value_setter(field):
+ field = 18
+
def _create_field_class(self, *args, **kwargs):
tc = get_default_trace_class()
return tc.create_unsigned_integer_field_class(*args, **kwargs)
+ def _create_default_const_field_class(self, *args, **kwargs):
+ tc = get_default_trace_class()
+ fc = tc.create_signed_integer_field_class(*args, **kwargs)
+ return _create_const_field_class(tc, fc, self._const_value_setter)
+
_create_default_field_class = _create_field_class
class RealFieldClassTestCase(_TestFieldClass, unittest.TestCase):
+ @staticmethod
+ def _const_value_setter(field):
+ field = -18
+
def _create_field_class(self, *args, **kwargs):
tc = get_default_trace_class()
return tc.create_real_field_class(*args, **kwargs)
+ def _create_default_const_field_class(self, *args, **kwargs):
+ tc = get_default_trace_class()
+ fc = tc.create_real_field_class(*args, **kwargs)
+ return _create_const_field_class(tc, fc, self._const_value_setter)
+
_create_default_field_class = _create_field_class
def test_create_default(self):
def setUp(self):
self._spec_set_up()
self._fc = self._create_default_field_class()
+ self._fc_const = self._create_default_const_field_class()
def test_create_from_invalid_type(self):
with self.assertRaises(TypeError):
self.assertEqual(mapping.label, 'hello')
self.assertEqual(mapping.ranges, self._ranges1)
+ def test_const_add_mapping(self):
+ with self.assertRaises(AttributeError):
+ self._fc_const.add_mapping('hello', self._ranges1)
+
def test_add_mapping_simple_kwargs(self):
self._fc.add_mapping(label='hello', ranges=self._ranges1)
mapping = self._fc['hello']
self.assertEqual(self._fc['e'].label, 'e')
self.assertEqual(self._fc['e'].ranges, self._ranges3)
+ def test_const_iadd(self):
+ with self.assertRaises(TypeError):
+ self._fc_const += [('d', self._ranges2), ('e', self._ranges3)]
+
def test_bool_op(self):
self.assertFalse(self._fc)
self._fc.add_mapping('a', self._ranges1)
self._inval_ranges = bt2.SignedIntegerRangeSet([(-8, -5), (48, 1928)])
self._value_in_range_1_and_3 = 20
+ @staticmethod
+ def _const_value_setter(field):
+ field = 0
+
def _create_field_class(self, *args, **kwargs):
tc = get_default_trace_class()
return tc.create_unsigned_enumeration_field_class(*args, **kwargs)
+ def _create_default_const_field_class(self, *args, **kwargs):
+ tc = get_default_trace_class()
+ fc = tc.create_unsigned_enumeration_field_class(*args, **kwargs)
+ return _create_const_field_class(tc, fc, self._const_value_setter)
+
_create_default_field_class = _create_field_class
self._inval_ranges = bt2.UnsignedIntegerRangeSet([(8, 16), (48, 99)])
self._value_in_range_1_and_3 = -7
+ @staticmethod
+ def _const_value_setter(field):
+ field = 0
+
def _create_field_class(self, *args, **kwargs):
tc = get_default_trace_class()
return tc.create_signed_enumeration_field_class(*args, **kwargs)
+ def _create_default_const_field_class(self, *args, **kwargs):
+ tc = get_default_trace_class()
+ fc = tc.create_signed_enumeration_field_class(*args, **kwargs)
+ return _create_const_field_class(tc, fc, self._const_value_setter)
+
_create_default_field_class = _create_field_class
class StringFieldClassTestCase(_TestFieldClass, unittest.TestCase):
+ @staticmethod
+ def _const_value_setter(field):
+ field = 'chaine'
+
def _create_field_class(self, *args, **kwargs):
tc = get_default_trace_class()
return tc.create_string_field_class(*args, **kwargs)
+ def _create_default_const_field_class(self, *args, **kwargs):
+ tc = get_default_trace_class()
+ fc = tc.create_string_field_class(*args, **kwargs)
+ return _create_const_field_class(tc, fc, self._const_value_setter)
+
_create_default_field_class = _create_field_class
def setUp(self):
def setUp(self):
self._tc = get_default_trace_class()
self._fc = self._create_default_field_class()
+ self._fc_const = self._create_default_const_field_class()
def test_create_default(self):
self.assertIsNotNone(self._fc)
self._append_element_method(self._fc, 'yes', sub_fc1)
self._append_element_method(self._fc, 'yes', sub_fc2)
+ def test_attr_field_class(self):
+ int_field_class = self._tc.create_signed_integer_field_class(32)
+ self._append_element_method(self._fc, 'int32', int_field_class)
+ field_class = self._fc['int32'].field_class
+
+ self.assertIs(type(field_class), bt2_field_class._SignedIntegerFieldClass)
+
+ def test_const_attr_field_class(self):
+ int_field_class = self._tc.create_signed_integer_field_class(32)
+ self._append_element_method(self._fc, 'int32', int_field_class)
+ field_class = self._fc['int32'].field_class
+ const_fc = _create_const_field_class(
+ self._tc, self._fc, self._const_value_setter
+ )
+ field_class = const_fc['int32'].field_class
+
+ self.assertIs(type(field_class), bt2_field_class._SignedIntegerFieldClassConst)
+
def test_iadd(self):
a_field_class = self._tc.create_real_field_class()
b_field_class = self._tc.create_signed_integer_field_class(17)
self.assertEqual(self._fc['e_struct'].field_class.addr, e_field_class.addr)
self.assertEqual(self._fc['e_struct'].name, 'e_struct')
+ def test_const_iadd(self):
+ a_field_class = self._tc.create_real_field_class()
+ with self.assertRaises(TypeError):
+ self._fc_const += a_field_class
+
def test_bool_op(self):
self.assertFalse(self._fc)
self._append_element_method(self._fc, 'a', self._tc.create_string_field_class())
user_attributes={'salut': 23},
)
self.assertEqual(self._fc['c'].user_attributes, {'salut': 23})
+ self.assertIs(type(self._fc.user_attributes), bt2_value.MapValue)
+ self.assertIs(type(self._fc['c'].user_attributes), bt2_value.MapValue)
def test_invalid_user_attributes(self):
with self.assertRaises(TypeError):
_append_element_method = staticmethod(bt2._StructureFieldClass.append_member)
_at_index_method = staticmethod(bt2._StructureFieldClass.member_at_index)
+ @staticmethod
+ def _const_value_setter(field):
+ field.value = {}
+
def _create_field_class(self, *args, **kwargs):
tc = get_default_trace_class()
return tc.create_structure_field_class(*args, **kwargs)
+ def _create_default_const_field_class(self, *args, **kwargs):
+ tc = get_default_trace_class()
+ fc = tc.create_structure_field_class(*args, **kwargs)
+ return _create_const_field_class(tc, fc, self._const_value_setter)
+
_create_default_field_class = _create_field_class
+ def test_const_member_field_class(self):
+ def _real_value_setter(field):
+ field.value = {'real': 0}
+
+ tc = get_default_trace_class()
+ fc = tc.create_structure_field_class()
+ member_fc = self._tc.create_real_field_class()
+ fc.append_member('real', member_fc)
+ const_fc = _create_const_field_class(tc, fc, _real_value_setter)
+
+ self.assertIs(
+ type(const_fc['real'].field_class), bt2_field_class._RealFieldClassConst
+ )
+
+ def test_member_field_class(self):
+ tc = get_default_trace_class()
+ fc = tc.create_structure_field_class()
+ member_fc = self._tc.create_real_field_class()
+ fc.append_member('real', member_fc)
+
+ self.assertIs(type(fc['real'].field_class), bt2_field_class._RealFieldClass)
+
class OptionFieldClassTestCase(_TestFieldClass, unittest.TestCase):
+ @staticmethod
+ def _const_value_setter(field):
+ field.has_field = True
+ field.value = 12
+
def _create_default_field_class(self, *args, **kwargs):
return self._tc.create_option_field_class(self._content_fc, **kwargs)
+ def _create_default_const_field_class(self, *args, **kwargs):
+ fc = self._tc.create_option_field_class(self._content_fc, **kwargs)
+ return _create_const_field_class(self._tc, fc, self._const_value_setter)
+
def setUp(self):
self._tc = get_default_trace_class()
self._content_fc = self._tc.create_signed_integer_field_class(23)
self.assertIsNone(fc.selector_field_path, None)
self.assertEqual(len(fc.user_attributes), 0)
+ def test_attr_field_class(self):
+ fc = self._create_default_field_class()
+ self.assertIs(type(fc.field_class), bt2_field_class._SignedIntegerFieldClass)
+
+ def test_const_attr_field_class(self):
+ fc = self._create_default_const_field_class()
+ self.assertIs(
+ type(fc.field_class), bt2_field_class._SignedIntegerFieldClassConst
+ )
+
def _create_field_class_for_field_path_test(self):
fc = self._create_default_field_class(selector_fc=self._tag_fc)
bt2._VariantFieldClassWithoutSelector.option_at_index
)
+ @staticmethod
+ def _const_value_setter(variant_field):
+ variant_field.selected_option_index = 0
+ variant_field.value = 12
+
def _create_field_class(self, *args, **kwargs):
tc = get_default_trace_class()
return tc.create_variant_field_class(*args, **kwargs)
+ def _create_default_const_field_class(self, *args, **kwargs):
+ tc = get_default_trace_class()
+ fc = tc.create_variant_field_class(*args, **kwargs)
+ int_field_class = self._tc.create_signed_integer_field_class(32)
+ fc.append_option('int32', int_field_class)
+
+ return _create_const_field_class(tc, fc, self._const_value_setter)
+
_create_default_field_class = _create_field_class
class _VariantFieldClassWithSelectorTestCase:
+ @staticmethod
+ def _const_value_setter(field):
+ field['variant'].selected_option_index = 0
+ field['variant'] = 12
+
def _create_default_field_class(self, *args, **kwargs):
return self._tc.create_variant_field_class(
*args, selector_fc=self._selector_fc, **kwargs
)
+ def _create_default_const_field_class(self, *args, **kwargs):
+ # Create a struct to contain the variant and its selector else we can't
+ # create the non-const field necessary to get the the const field_class
+ struct_fc = self._tc.create_structure_field_class()
+ struct_fc.append_member('selecteux', self._selector_fc)
+ variant_fc = self._tc.create_variant_field_class(
+ *args, selector_fc=self._selector_fc
+ )
+ variant_fc.append_option(
+ 'a', self._tc.create_signed_integer_field_class(32), self._ranges1
+ )
+ struct_fc.append_member('variant', variant_fc, **kwargs)
+
+ return _create_const_field_class(self._tc, struct_fc, self._const_value_setter)[
+ 'variant'
+ ].field_class
+
def setUp(self):
self._tc = get_default_trace_class()
self._spec_set_up()
self.assertEqual(opt.name, 'str')
self.assertEqual(opt.ranges.addr, self._ranges1.addr)
+ def test_const_append(self):
+ fc_const = self._create_default_const_field_class()
+ with self.assertRaises(AttributeError):
+ fc_const.append_option('str', str_field_class, self._ranges1)
+
def test_append_element_kwargs(self):
int_field_class = self._tc.create_signed_integer_field_class(32)
self._fc.append_option(
user_attributes={'salut': 23},
)
self.assertEqual(self._fc['c'].user_attributes, {'salut': 23})
+ self.assertIs(type(self._fc.user_attributes), bt2_value.MapValue)
+
+ def test_const_user_attributes(self):
+ fc_const = self._create_default_const_field_class()
+ self.assertIs(type(fc_const.user_attributes), bt2_value._MapValueConst)
def test_invalid_user_attributes(self):
with self.assertRaises(TypeError):
self.assertEqual(self._fc['d_enum'].name, 'd_enum')
self.assertEqual(self._fc['d_enum'].ranges, self._ranges3)
+ def test_const_iadd(self):
+ fc_const = self._create_default_const_field_class()
+ a_field_class = self._tc.create_real_field_class()
+ with self.assertRaises(TypeError):
+ fc_const += [('a_float', a_field_class, self._ranges1)]
+
def test_bool_op(self):
self.assertFalse(self._fc)
self._fc.append_option('a', self._tc.create_string_field_class(), self._ranges1)
self.assertEqual(self._fc['b'].name, 'b')
self.assertEqual(self._fc['b'].ranges.addr, self._ranges2.addr)
+ def test_option_field_class(self):
+ a_fc = self._tc.create_signed_integer_field_class(32)
+ self._fc.append_option('a', a_fc, self._ranges1)
+ self.assertIs(
+ type(self._fc['a'].field_class), bt2_field_class._SignedIntegerFieldClass
+ )
+
+ def test_const_option_field_class(self):
+ fc_const = self._create_default_const_field_class()
+ self.assertIs(
+ type(fc_const['a'].field_class),
+ bt2_field_class._SignedIntegerFieldClassConst,
+ )
+
def test_getitem_invalid_key_type(self):
with self.assertRaises(TypeError):
self._fc[0]
self._selector_fc = self._tc.create_signed_integer_field_class()
-class StaticArrayFieldClassTestCase(unittest.TestCase):
+class _ArrayFieldClassTestCase:
+ def test_attr_element_field_class(self):
+ fc = self._create_array()
+ self.assertIs(
+ type(fc.element_field_class), bt2_field_class._SignedIntegerFieldClass
+ )
+
+ def test_const_attr_element_field_class(self):
+ fc = self._create_const_array()
+ self.assertIs(
+ type(fc.element_field_class), bt2_field_class._SignedIntegerFieldClassConst
+ )
+
+
+class StaticArrayFieldClassTestCase(_ArrayFieldClassTestCase, unittest.TestCase):
+ @staticmethod
+ def _const_value_setter(field):
+ field = []
+
+ def _create_array(self):
+ return self._tc.create_static_array_field_class(self._elem_fc, 45)
+
+ def _create_const_array(self):
+ fc = self._tc.create_static_array_field_class(self._elem_fc, 45)
+ return _create_const_field_class(self._tc, fc, self._const_value_setter)
+
def setUp(self):
self._tc = get_default_trace_class()
self._elem_fc = self._tc.create_signed_integer_field_class(23)
)
-class DynamicArrayFieldClassTestCase(unittest.TestCase):
+class DynamicArrayFieldClassTestCase(_ArrayFieldClassTestCase, unittest.TestCase):
+ @staticmethod
+ def _const_value_setter(field):
+ field = []
+
+ def _create_array(self):
+ return self._tc.create_dynamic_array_field_class(self._elem_fc)
+
+ def _create_const_array(self):
+ fc = self._tc.create_dynamic_array_field_class(self._elem_fc)
+ return _create_const_field_class(self._tc, fc, self._const_value_setter)
+
def setUp(self):
self._tc = get_default_trace_class()
self._elem_fc = self._tc.create_signed_integer_field_class(23)
msg = next(comp_self._msg_iter)
if comp_self._at == 0:
- self.assertIsInstance(msg, bt2._StreamBeginningMessage)
+ self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
elif comp_self._at == 1:
- self.assertIsInstance(msg, bt2._PacketBeginningMessage)
+ self.assertIs(type(msg), bt2._PacketBeginningMessageConst)
elif comp_self._at >= 2 and comp_self._at <= 6:
- self.assertIsInstance(msg, bt2._EventMessage)
+ self.assertIs(type(msg), bt2._EventMessageConst)
self.assertEqual(msg.event.cls.name, 'salut')
elif comp_self._at == 7:
- self.assertIsInstance(msg, bt2._PacketEndMessage)
+ self.assertIs(type(msg), bt2._PacketEndMessageConst)
elif comp_self._at == 8:
- self.assertIsInstance(msg, bt2._StreamEndMessage)
+ self.assertIs(type(msg), bt2._StreamEndMessageConst)
comp_self._at += 1
def _user_consume(comp_self):
msg = next(comp_self._msg_iter)
if comp_self._at == 0:
- self.assertIsInstance(msg, bt2._StreamBeginningMessage)
+ self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
elif comp_self._at == 1:
- self.assertIsInstance(msg, bt2._PacketBeginningMessage)
+ self.assertIs(type(msg), bt2._PacketBeginningMessageConst)
elif comp_self._at == 2:
- self.assertIsInstance(msg, bt2._EventMessage)
+ self.assertIs(type(msg), bt2._EventMessageConst)
raise bt2.TryAgain
else:
pass
def _user_consume(comp_self):
msg = next(comp_self._msg_iter)
if comp_self._at == 0:
- self.assertIsInstance(msg, bt2._StreamBeginningMessage)
+ self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
elif comp_self._at == 1:
- self.assertIsInstance(msg, bt2._PacketBeginningMessage)
+ self.assertIs(type(msg), bt2._PacketBeginningMessageConst)
elif comp_self._at == 2:
- self.assertIsInstance(msg, bt2._EventMessage)
+ self.assertIs(type(msg), bt2._EventMessageConst)
elif comp_self._at == 3:
nonlocal raised_in_sink
raised_in_sink = True
import unittest
import bt2
+import utils
from utils import TestOutputPortMessageIterator
+from bt2 import clock_snapshot as bt2_clock_snapshot
+from bt2 import event as bt2_event
+from bt2 import event_class as bt2_event_class
+from bt2 import field as bt2_field
+from bt2 import packet as bt2_packet
+from bt2 import stream as bt2_stream
+from bt2 import stream_class as bt2_stream_class
+from bt2 import trace as bt2_trace
+from bt2 import trace_class as bt2_trace_class
class AllMessagesTestCase(unittest.TestCase):
msg = self._create_stream_beginning_message(
test_obj._stream
)
+ test_obj.assertIs(type(msg), bt2._StreamBeginningMessage)
elif self._at == 1:
msg = self._create_packet_beginning_message(
test_obj._packet, self._at
)
+ test_obj.assertIs(type(msg), bt2._PacketBeginningMessage)
elif self._at == 2:
msg = self._create_event_message(
test_obj._event_class, test_obj._packet, self._at
)
+ test_obj.assertIs(type(msg), bt2._EventMessage)
elif self._at == 3:
msg = self._create_message_iterator_inactivity_message(
test_obj._clock_class, self._at
msg = self._create_discarded_events_message(
test_obj._stream, 890, self._at, self._at
)
+ test_obj.assertIs(type(msg), bt2._DiscardedEventsMessage)
elif self._at == 5:
msg = self._create_packet_end_message(
test_obj._packet, self._at
)
+ test_obj.assertIs(type(msg), bt2._PacketEndMessage)
elif self._at == 6:
msg = self._create_discarded_packets_message(
test_obj._stream, 678, self._at, self._at
)
+ test_obj.assertIs(type(msg), bt2._DiscardedPacketsMessage)
elif self._at == 7:
if self._with_stream_msgs_clock_snapshots:
msg = self._create_stream_end_message(
)
else:
msg = self._create_stream_end_message(test_obj._stream)
+ test_obj.assertIs(type(msg), bt2._StreamEndMessage)
elif self._at >= 8:
raise bt2.Stop
else:
payload_fc = tc.create_structure_field_class()
payload_fc += [('my_int', my_int_fc)]
- ec = sc.create_event_class(name='salut', payload_field_class=payload_fc)
+ # Create specific context field class
+ my_int_fc = tc.create_signed_integer_field_class(32)
+ specific_fc = tc.create_structure_field_class()
+ specific_fc += [('my_int', my_int_fc)]
+
+ ec = sc.create_event_class(
+ name='salut',
+ payload_field_class=payload_fc,
+ specific_context_field_class=specific_fc,
+ )
trace = tc()
stream = trace.create_stream(sc)
for i, msg in enumerate(self._msg_iter):
if i == 0:
- self.assertIsInstance(msg, bt2._StreamBeginningMessage)
+ self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
+ self.assertIs(type(msg.stream), bt2_stream._StreamConst)
self.assertEqual(msg.stream.addr, self._stream.addr)
self.assertIsInstance(
msg.default_clock_snapshot, bt2._UnknownClockSnapshot
)
elif i == 1:
- self.assertIsInstance(msg, bt2._PacketBeginningMessage)
+ self.assertIs(type(msg), bt2._PacketBeginningMessageConst)
+ self.assertIs(type(msg.packet), bt2_packet._PacketConst)
+ self.assertIs(
+ type(msg.default_clock_snapshot),
+ bt2_clock_snapshot._ClockSnapshotConst,
+ )
self.assertEqual(msg.packet.addr, self._packet.addr)
self.assertEqual(msg.default_clock_snapshot.value, i)
elif i == 2:
- self.assertIsInstance(msg, bt2._EventMessage)
+ self.assertIs(type(msg), bt2._EventMessageConst)
+ self.assertIs(type(msg.event), bt2_event._EventConst)
+ self.assertIs(
+ type(msg.default_clock_snapshot),
+ bt2_clock_snapshot._ClockSnapshotConst,
+ )
+ self.assertIs(
+ type(msg.event.payload_field), bt2_field._StructureFieldConst
+ )
+ self.assertIs(
+ type(msg.event.payload_field['my_int']),
+ bt2_field._SignedIntegerFieldConst,
+ )
+
self.assertEqual(msg.event.cls.addr, self._event_class.addr)
self.assertEqual(msg.default_clock_snapshot.value, i)
elif i == 3:
- self.assertIsInstance(msg, bt2._MessageIteratorInactivityMessage)
+ self.assertIs(type(msg), bt2._MessageIteratorInactivityMessageConst)
+ self.assertIs(
+ type(msg.default_clock_snapshot),
+ bt2_clock_snapshot._ClockSnapshotConst,
+ )
self.assertEqual(msg.default_clock_snapshot.value, i)
elif i == 4:
- self.assertIsInstance(msg, bt2._DiscardedEventsMessage)
+ self.assertIs(type(msg), bt2._DiscardedEventsMessageConst)
+ self.assertIs(type(msg.stream), bt2_stream._StreamConst)
+ self.assertIs(type(msg.stream.cls), bt2_stream_class._StreamClassConst)
+ self.assertIs(
+ type(msg.beginning_default_clock_snapshot),
+ bt2_clock_snapshot._ClockSnapshotConst,
+ )
+ self.assertIs(
+ type(msg.end_default_clock_snapshot),
+ bt2_clock_snapshot._ClockSnapshotConst,
+ )
+
self.assertEqual(msg.stream.addr, self._stream.addr)
self.assertEqual(msg.count, 890)
self.assertEqual(
self.assertEqual(msg.beginning_default_clock_snapshot.value, i)
self.assertEqual(msg.end_default_clock_snapshot.value, i)
elif i == 5:
- self.assertIsInstance(msg, bt2._PacketEndMessage)
+ self.assertIs(type(msg), bt2._PacketEndMessageConst)
+ self.assertIs(type(msg.packet), bt2_packet._PacketConst)
+ self.assertIs(
+ type(msg.default_clock_snapshot),
+ bt2_clock_snapshot._ClockSnapshotConst,
+ )
self.assertEqual(msg.packet.addr, self._packet.addr)
self.assertEqual(msg.default_clock_snapshot.value, i)
elif i == 6:
- self.assertIsInstance(msg, bt2._DiscardedPacketsMessage)
+ self.assertIs(type(msg), bt2._DiscardedPacketsMessageConst)
+ self.assertIs(type(msg.stream), bt2_stream._StreamConst)
+ self.assertIs(type(msg.stream.trace), bt2_trace._TraceConst)
+ self.assertIs(
+ type(msg.stream.trace.cls), bt2_trace_class._TraceClassConst
+ )
+ self.assertIs(
+ type(msg.beginning_default_clock_snapshot),
+ bt2_clock_snapshot._ClockSnapshotConst,
+ )
+ self.assertIs(
+ type(msg.end_default_clock_snapshot),
+ bt2_clock_snapshot._ClockSnapshotConst,
+ )
self.assertEqual(msg.stream.addr, self._stream.addr)
self.assertEqual(msg.count, 678)
self.assertEqual(
self.assertEqual(msg.beginning_default_clock_snapshot.value, i)
self.assertEqual(msg.end_default_clock_snapshot.value, i)
elif i == 7:
- self.assertIsInstance(msg, bt2._StreamEndMessage)
+ self.assertIs(type(msg), bt2._StreamEndMessageConst)
+ self.assertIs(type(msg.stream), bt2_stream._StreamConst)
self.assertEqual(msg.stream.addr, self._stream.addr)
- self.assertIsInstance(
- msg.default_clock_snapshot, bt2._UnknownClockSnapshot
+ self.assertIs(
+ type(msg.default_clock_snapshot), bt2._UnknownClockSnapshot
)
else:
raise Exception
for i, msg in enumerate(self._msg_iter):
if i == 0:
- self.assertIsInstance(msg, bt2._StreamBeginningMessage)
+ self.assertIsInstance(msg, bt2._StreamBeginningMessageConst)
+ self.assertIs(type(msg.stream), bt2_stream._StreamConst)
self.assertEqual(msg.stream.addr, self._stream.addr)
with self.assertRaisesRegex(
ValueError, 'stream class has no default clock class'
):
msg.default_clock_snapshot
elif i == 1:
- self.assertIsInstance(msg, bt2._PacketBeginningMessage)
+ self.assertIsInstance(msg, bt2._PacketBeginningMessageConst)
+ self.assertIs(type(msg.packet), bt2_packet._PacketConst)
self.assertEqual(msg.packet.addr, self._packet.addr)
elif i == 2:
- self.assertIsInstance(msg, bt2._EventMessage)
+ self.assertIsInstance(msg, bt2._EventMessageConst)
+ self.assertIs(type(msg.event), bt2_event._EventConst)
+ self.assertIs(type(msg.event.cls), bt2_event_class._EventClassConst)
self.assertEqual(msg.event.cls.addr, self._event_class.addr)
with self.assertRaisesRegex(
ValueError, 'stream class has no default clock class'
):
msg.default_clock_snapshot
elif i == 3:
- self.assertIsInstance(msg, bt2._DiscardedEventsMessage)
+ self.assertIsInstance(msg, bt2._DiscardedEventsMessageConst)
+ self.assertIs(type(msg.stream), bt2_stream._StreamConst)
+ self.assertIs(type(msg.stream.cls), bt2_stream_class._StreamClassConst)
self.assertEqual(msg.stream.addr, self._stream.addr)
self.assertEqual(msg.count, 890)
self.assertIsNone(msg.stream.cls.default_clock_class)
):
msg.end_default_clock_snapshot
elif i == 4:
- self.assertIsInstance(msg, bt2._PacketEndMessage)
+ self.assertIsInstance(msg, bt2._PacketEndMessageConst)
self.assertEqual(msg.packet.addr, self._packet.addr)
+ self.assertIs(type(msg.packet), bt2_packet._PacketConst)
elif i == 5:
- self.assertIsInstance(msg, bt2._DiscardedPacketsMessage)
+ self.assertIsInstance(msg, bt2._DiscardedPacketsMessageConst)
+ self.assertIs(type(msg.stream), bt2_stream._StreamConst)
+ self.assertIs(type(msg.stream.cls), bt2_stream_class._StreamClassConst)
+ self.assertIs(
+ type(msg.stream.cls.trace_class), bt2_trace_class._TraceClassConst
+ )
self.assertEqual(msg.stream.addr, self._stream.addr)
self.assertEqual(msg.count, 678)
self.assertIsNone(msg.stream.cls.default_clock_class)
):
msg.end_default_clock_snapshot
elif i == 6:
- self.assertIsInstance(msg, bt2._StreamEndMessage)
+ self.assertIsInstance(msg, bt2._StreamEndMessageConst)
+ self.assertIs(type(msg.stream), bt2_stream._StreamConst)
self.assertEqual(msg.stream.addr, self._stream.addr)
with self.assertRaisesRegex(
ValueError, 'stream class has no default clock class'
msgs = list(self._msg_iter)
msg_stream_beg = msgs[0]
- self.assertIsInstance(msg_stream_beg, bt2._StreamBeginningMessage)
+ self.assertIsInstance(msg_stream_beg, bt2._StreamBeginningMessageConst)
+ self.assertIs(
+ type(msg_stream_beg.default_clock_snapshot),
+ bt2_clock_snapshot._ClockSnapshotConst,
+ )
self.assertEqual(msg_stream_beg.default_clock_snapshot.value, 0)
msg_stream_end = msgs[7]
- self.assertIsInstance(msg_stream_end, bt2._StreamEndMessage)
+ self.assertIsInstance(msg_stream_end, bt2._StreamEndMessageConst)
+ self.assertIs(
+ type(msg_stream_end.default_clock_snapshot),
+ bt2_clock_snapshot._ClockSnapshotConst,
+ )
self.assertEqual(msg_stream_end.default_clock_snapshot.value, 7)
+
+ def test_stream_beg_msg(self):
+ msg = utils.get_stream_beginning_message()
+ self.assertIs(type(msg.stream), bt2_stream._Stream)
+
+ def test_stream_end_msg(self):
+ msg = utils.get_stream_end_message()
+ self.assertIs(type(msg.stream), bt2_stream._Stream)
+
+ def test_packet_beg_msg(self):
+ msg = utils.get_packet_beginning_message()
+ self.assertIs(type(msg.packet), bt2_packet._Packet)
+
+ def test_packet_end_msg(self):
+ msg = utils.get_packet_end_message()
+ self.assertIs(type(msg.packet), bt2_packet._Packet)
+
+ def test_event_msg(self):
+ msg = utils.get_event_message()
+ self.assertIs(type(msg.event), bt2_event._Event)
# Skip beginning messages.
msg = next(it)
- self.assertIsInstance(msg, bt2._StreamBeginningMessage)
+ self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
msg = next(it)
- self.assertIsInstance(msg, bt2._PacketBeginningMessage)
+ self.assertIs(type(msg), bt2._PacketBeginningMessageConst)
msg_ev1 = next(it)
msg_ev2 = next(it)
- self.assertIsInstance(msg_ev1, bt2._EventMessage)
- self.assertIsInstance(msg_ev2, bt2._EventMessage)
+ self.assertIs(type(msg_ev1), bt2._EventMessageConst)
+ self.assertIs(type(msg_ev2), bt2._EventMessageConst)
self.assertEqual(msg_ev1.addr, msg_ev2.addr)
@staticmethod
msg = None
MySourceIter, graph = self._setup_seek_beginning_test(MySink)
graph.run_once()
- self.assertIsInstance(msg, bt2._StreamBeginningMessage)
+ self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
graph.run_once()
- self.assertIsInstance(msg, bt2._PacketBeginningMessage)
+ self.assertIs(type(msg), bt2._PacketBeginningMessageConst)
do_seek_beginning = True
graph.run_once()
do_seek_beginning = False
graph.run_once()
- self.assertIsInstance(msg, bt2._StreamBeginningMessage)
+ self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
def test_seek_beginning_user_error(self):
class MySink(bt2._UserSinkComponent):
def test_has__DynamicArrayField(self):
self._assert_in_bt2('_DynamicArrayField')
+ def test_has__BoolFieldConst(self):
+ self._assert_in_bt2('_BoolFieldConst')
+
+ def test_has__BitArrayFieldConst(self):
+ self._assert_in_bt2('_BitArrayFieldConst')
+
+ def test_has__IntegerFieldConst(self):
+ self._assert_in_bt2('_IntegerFieldConst')
+
+ def test_has__UnsignedIntegerFieldConst(self):
+ self._assert_in_bt2('_UnsignedIntegerFieldConst')
+
+ def test_has__SignedIntegerFieldConst(self):
+ self._assert_in_bt2('_SignedIntegerFieldConst')
+
+ def test_has__RealFieldConst(self):
+ self._assert_in_bt2('_RealFieldConst')
+
+ def test_has__EnumerationFieldConst(self):
+ self._assert_in_bt2('_EnumerationFieldConst')
+
+ def test_has__UnsignedEnumerationFieldConst(self):
+ self._assert_in_bt2('_UnsignedEnumerationFieldConst')
+
+ def test_has__SignedEnumerationFieldConst(self):
+ self._assert_in_bt2('_SignedEnumerationFieldConst')
+
+ def test_has__StringFieldConst(self):
+ self._assert_in_bt2('_StringFieldConst')
+
+ def test_has__StructureFieldConst(self):
+ self._assert_in_bt2('_StructureFieldConst')
+
+ def test_has__OptionFieldConst(self):
+ self._assert_in_bt2('_VariantFieldConst')
+
+ def test_has__VariantFieldConst(self):
+ self._assert_in_bt2('_VariantFieldConst')
+
+ def test_has__ArrayFieldConst(self):
+ self._assert_in_bt2('_ArrayFieldConst')
+
+ def test_has__StaticArrayFieldConst(self):
+ self._assert_in_bt2('_StaticArrayFieldConst')
+
+ def test_has__DynamicArrayFieldConst(self):
+ self._assert_in_bt2('_DynamicArrayFieldConst')
+
def test_has_IntegerDisplayBase(self):
self._assert_in_bt2('IntegerDisplayBase')
def test_has__DynamicArrayFieldClass(self):
self._assert_in_bt2('_DynamicArrayFieldClass')
+ def test_has__BoolFieldClassConst(self):
+ self._assert_in_bt2('_BoolFieldClassConst')
+
+ def test_has__BitArrayFieldClassConst(self):
+ self._assert_in_bt2('_BitArrayFieldClassConst')
+
+ def test_has__IntegerFieldClassConst(self):
+ self._assert_in_bt2('_IntegerFieldClassConst')
+
+ def test_has__UnsignedIntegerFieldClassConst(self):
+ self._assert_in_bt2('_UnsignedIntegerFieldClassConst')
+
+ def test_has__SignedIntegerFieldClassConst(self):
+ self._assert_in_bt2('_SignedIntegerFieldClassConst')
+
+ def test_has__RealFieldClassConst(self):
+ self._assert_in_bt2('_RealFieldClassConst')
+
+ def test_has__EnumerationFieldClassConst(self):
+ self._assert_in_bt2('_EnumerationFieldClassConst')
+
+ def test_has__UnsignedEnumerationFieldClassConst(self):
+ self._assert_in_bt2('_UnsignedEnumerationFieldClassConst')
+
+ def test_has__SignedEnumerationFieldClassConst(self):
+ self._assert_in_bt2('_SignedEnumerationFieldClassConst')
+
+ def test_has__StringFieldClassConst(self):
+ self._assert_in_bt2('_StringFieldClassConst')
+
+ def test_has__StructureFieldClassConst(self):
+ self._assert_in_bt2('_StructureFieldClassConst')
+
+ def test_has__OptionFieldClassConst(self):
+ self._assert_in_bt2('_OptionFieldClassConst')
+
+ def test_has__VariantFieldClassConst(self):
+ self._assert_in_bt2('_VariantFieldClassConst')
+
+ def test_has__VariantFieldClassWithoutSelectorConst(self):
+ self._assert_in_bt2('_VariantFieldClassWithoutSelectorConst')
+
+ def test_has__VariantFieldClassWithSelectorConst(self):
+ self._assert_in_bt2('_VariantFieldClassWithSelectorConst')
+
+ def test_has__VariantFieldClassWithUnsignedSelectorConst(self):
+ self._assert_in_bt2('_VariantFieldClassWithUnsignedSelectorConst')
+
+ def test_has__VariantFieldClassWithSignedSelectorConst(self):
+ self._assert_in_bt2('_VariantFieldClassWithSignedSelectorConst')
+
+ def test_has__ArrayFieldClassConst(self):
+ self._assert_in_bt2('_ArrayFieldClassConst')
+
+ def test_has__StaticArrayFieldClassConst(self):
+ self._assert_in_bt2('_StaticArrayFieldClassConst')
+
+ def test_has__DynamicArrayFieldClassConst(self):
+ self._assert_in_bt2('_DynamicArrayFieldClassConst')
+
def test_has_FieldPathScope(self):
self._assert_in_bt2('FieldPathScope')
def test_has__DiscardedPacketsMessage(self):
self._assert_in_bt2('_DiscardedPacketsMessage')
+ def test_has__EventMessageConst(self):
+ self._assert_in_bt2('_EventMessageConst')
+
+ def test_has__PacketBeginningMessageConst(self):
+ self._assert_in_bt2('_PacketBeginningMessageConst')
+
+ def test_has__PacketEndMessageConst(self):
+ self._assert_in_bt2('_PacketEndMessageConst')
+
+ def test_has__StreamBeginningMessageConst(self):
+ self._assert_in_bt2('_StreamBeginningMessageConst')
+
+ def test_has__StreamEndMessageConst(self):
+ self._assert_in_bt2('_StreamEndMessageConst')
+
+ def test_has__MessageIteratorInactivityMessageConst(self):
+ self._assert_in_bt2('_MessageIteratorInactivityMessageConst')
+
+ def test_has__DiscardedEventsMessageConst(self):
+ self._assert_in_bt2('_DiscardedEventsMessageConst')
+
+ def test_has__DiscardedPacketsMessageConst(self):
+ self._assert_in_bt2('_DiscardedPacketsMessageConst')
+
def test_has__UserMessageIterator(self):
self._assert_in_bt2('_UserMessageIterator')
#
import unittest
+import utils
from utils import run_in_component_init
+from bt2 import stream as bt2_stream
+from bt2 import field as bt2_field
class PacketTestCase(unittest.TestCase):
def test_attr_stream(self):
packet, stream, _ = self._create_packet(with_pc=True)
self.assertEqual(packet.stream.addr, stream.addr)
+ self.assertIs(type(packet.stream), bt2_stream._Stream)
+
+ def test_const_attr_stream(self):
+ packet = utils.get_const_packet_beginning_message().packet
+ self.assertIs(type(packet.stream), bt2_stream._StreamConst)
def test_context_field(self):
packet, stream, pc_fc = self._create_packet(with_pc=True)
self.assertEqual(packet.context_field.cls.addr, pc_fc.addr)
+ self.assertIs(type(packet.context_field), bt2_field._StructureField)
+
+ def test_const_context_field(self):
+ packet = utils.get_const_packet_beginning_message().packet
+ self.assertIs(type(packet.context_field), bt2_field._StructureFieldConst)
def test_no_context_field(self):
packet, _, _ = self._create_packet(with_pc=False)
#
import unittest
+import utils
from utils import run_in_component_init
+from bt2 import trace as bt2_trace
+from bt2 import stream as bt2_stream
+from bt2 import value as bt2_value
+from bt2 import stream_class as bt2_stream_class
class StreamTestCase(unittest.TestCase):
def test_create_default(self):
stream = self._tr.create_stream(self._sc)
self.assertIsNone(stream.name)
+ self.assertIs(type(stream), bt2_stream._Stream)
self.assertEqual(len(stream.user_attributes), 0)
def test_name(self):
def test_create_user_attributes(self):
stream = self._tr.create_stream(self._sc, user_attributes={'salut': 23})
self.assertEqual(stream.user_attributes, {'salut': 23})
+ self.assertIs(type(stream.user_attributes), bt2_value.MapValue)
+
+ def test_const_user_attributes(self):
+ stream = utils.get_const_stream_beginning_message().stream
+ self.assertEqual(stream.user_attributes, {'salut': 23})
+ self.assertIs(type(stream.user_attributes), bt2_value._MapValueConst)
def test_create_invalid_user_attributes(self):
with self.assertRaises(TypeError):
def test_stream_class(self):
stream = self._tr.create_stream(self._sc)
self.assertEqual(stream.cls, self._sc)
+ self.assertIs(type(stream.cls), bt2_stream_class._StreamClass)
+
+ def test_const_stream_class(self):
+ stream = utils.get_const_stream_beginning_message().stream
+ self.assertIs(type(stream.cls), bt2_stream_class._StreamClassConst)
def test_trace(self):
stream = self._tr.create_stream(self._sc)
self.assertEqual(stream.trace.addr, self._tr.addr)
+ self.assertIs(type(stream.trace), bt2_trace._Trace)
+
+ def test_const_trace(self):
+ stream = utils.get_const_stream_beginning_message().stream
+ self.assertIs(type(stream.trace), bt2_trace._TraceConst)
def test_invalid_id(self):
sc = self._tc.create_stream_class(assigns_automatic_stream_id=False)
import unittest
from utils import run_in_component_init
+from bt2 import stream_class as bt2_stream_class
+from bt2 import trace_class as bt2_trace_class
+from bt2 import clock_class as bt2_clock_class
+from bt2 import event_class as bt2_event_class
+from bt2 import field_class as bt2_field_class
class StreamClassTestCase(unittest.TestCase):
def test_create_default(self):
sc = self._tc.create_stream_class()
+ self.assertIs(type(sc), bt2_stream_class._StreamClass)
self.assertIsNone(sc.name)
self.assertIsNone(sc.packet_context_field_class)
self.assertIsNone(sc.event_common_context_field_class)
packet_context_field_class=fc, supports_packets=True
)
self.assertEqual(sc.packet_context_field_class, fc)
+ self.assertIs(
+ type(sc.packet_context_field_class), bt2_field_class._StructureFieldClass
+ )
def test_create_invalid_packet_context_field_class(self):
with self.assertRaises(TypeError):
fc = self._tc.create_structure_field_class()
sc = self._tc.create_stream_class(event_common_context_field_class=fc)
self.assertEqual(sc.event_common_context_field_class, fc)
+ self.assertIs(
+ type(sc.event_common_context_field_class),
+ bt2_field_class._StructureFieldClass,
+ )
def test_create_invalid_event_common_context_field_class(self):
with self.assertRaises(TypeError):
def test_create_default_clock_class(self):
sc = self._tc.create_stream_class(default_clock_class=self._cc)
self.assertEqual(sc.default_clock_class.addr, self._cc.addr)
+ self.assertIs(type(sc.default_clock_class), bt2_clock_class._ClockClass)
def test_create_invalid_default_clock_class(self):
with self.assertRaises(TypeError):
def test_trace_class(self):
sc = self._tc.create_stream_class()
self.assertEqual(sc.trace_class.addr, self._tc.addr)
+ self.assertIs(type(sc.trace_class), bt2_trace_class._TraceClass)
def _create_stream_class_with_event_classes(self):
sc = self._tc.create_stream_class(assigns_automatic_event_class_id=False)
sc, ec1, ec2 = self._create_stream_class_with_event_classes()
self.assertEqual(sc[23].addr, ec1.addr)
+ self.assertEqual(type(sc[23]), bt2_event_class._EventClass)
self.assertEqual(sc[17].addr, ec2.addr)
+ self.assertEqual(type(sc[17]), bt2_event_class._EventClass)
def test_getitem_wrong_key_type(self):
sc, _, _ = self._create_stream_class_with_event_classes()
import uuid
import unittest
+import utils
from utils import get_default_trace_class
+from bt2 import trace_class as bt2_trace_class
+from bt2 import value as bt2_value
+from bt2 import trace as bt2_trace
+from bt2 import stream as bt2_stream
class TraceTestCase(unittest.TestCase):
def test_create_user_attributes(self):
trace = self._tc(user_attributes={'salut': 23})
self.assertEqual(trace.user_attributes, {'salut': 23})
+ self.assertIs(type(trace.user_attributes), bt2_value.MapValue)
def test_create_invalid_user_attributes(self):
with self.assertRaises(TypeError):
def test_attr_trace_class(self):
trace = self._tc()
self.assertEqual(trace.cls.addr, self._tc.addr)
+ self.assertIs(type(trace.cls), bt2_trace_class._TraceClass)
+
+ def test_const_attr_trace_class(self):
+ trace = utils.get_const_stream_beginning_message().stream.trace
+ self.assertIs(type(trace.cls), bt2_trace_class._TraceClassConst)
def test_attr_name(self):
trace = self._tc(name='mein trace')
def test_env_get(self):
trace = self._tc(environment={'hello': 'you', 'foo': -5})
+ self.assertIs(type(trace.environment), bt2_trace._TraceEnvironment)
+ self.assertIs(type(trace.environment['foo']), bt2_value.SignedIntegerValue)
self.assertEqual(trace.environment['hello'], 'you')
self.assertEqual(trace.environment['foo'], -5)
+ def test_env_iter(self):
+ trace = self._tc(environment={'hello': 'you', 'foo': -5})
+ values = set(trace.environment)
+ self.assertEqual(values, {'hello', 'foo'})
+
+ def test_const_env_get(self):
+ trace = utils.get_const_stream_beginning_message().stream.trace
+ self.assertIs(type(trace.environment), bt2_trace._TraceEnvironmentConst)
+ self.assertIs(
+ type(trace.environment['patate']), bt2_value._SignedIntegerValueConst
+ )
+
+ def test_env_iter(self):
+ trace = utils.get_const_stream_beginning_message().stream.trace
+ values = set(trace.environment)
+ self.assertEqual(values, {'patate'})
+
+ def test_const_env_set(self):
+ trace = utils.get_const_stream_beginning_message().stream.trace
+ with self.assertRaises(TypeError):
+ trace.environment['patate'] = 33
+
def test_env_get_non_existent(self):
trace = self._tc(environment={'hello': 'you', 'foo': -5})
def test_getitem(self):
trace = self._create_trace_with_some_streams()
-
self.assertEqual(trace[12].id, 12)
+ self.assertIs(type(trace[12]), bt2_stream._Stream)
+
+ def test_const_getitem(self):
+ trace = utils.get_const_stream_beginning_message().stream.trace
+ self.assertIs(type(trace[0]), bt2_stream._StreamConst)
def test_getitem_invalid_key(self):
trace = self._create_trace_with_some_streams()
#
import unittest
-from utils import run_in_component_init, get_default_trace_class
+from utils import (
+ run_in_component_init,
+ get_default_trace_class,
+ get_const_stream_beginning_message,
+)
+from bt2 import stream_class as bt2_stream_class
+from bt2 import trace_class as bt2_trace_class
class TraceClassTestCase(unittest.TestCase):
tc = run_in_component_init(f)
self.assertEqual(len(tc), 0)
+ self.assertIs(type(tc), bt2_trace_class._TraceClass)
self.assertTrue(tc.assigns_automatic_stream_class_id)
self.assertEqual(len(tc.user_attributes), 0)
sc1 = tc.create_stream_class()
sc2 = tc.create_stream_class()
+ self.assertIs(type(sc1), bt2_stream_class._StreamClass)
+ self.assertIs(type(sc2), bt2_stream_class._StreamClass)
self.assertNotEqual(sc1.id, sc2.id)
def test_automatic_stream_class_id_raises(self):
def test_getitem(self):
tc, _, _, sc3 = self._create_trace_class_with_some_stream_classes()
+ self.assertIs(type(tc[2018]), bt2_stream_class._StreamClass)
self.assertEqual(tc[2018].addr, sc3.addr)
+ def test_const_getitem(self):
+ const_tc = get_const_stream_beginning_message().stream.trace.cls
+ self.assertIs(type(const_tc[0]), bt2_stream_class._StreamClassConst)
+
def test_getitem_wrong_key_type(self):
tc, _, _, _ = self._create_trace_class_with_some_stream_classes()
with self.assertRaises(TypeError):
for sc_id, stream_class in tc.items():
if sc_id == 12:
+ self.assertIs(type(stream_class), bt2_stream_class._StreamClass)
self.assertEqual(stream_class.addr, sc1.addr)
elif sc_id == 54:
self.assertEqual(stream_class.addr, sc2.addr)
elif sc_id == 2018:
self.assertEqual(stream_class.addr, sc3.addr)
+ def test_const_iter(self):
+ const_tc = get_const_stream_beginning_message().stream.trace.cls
+ const_sc = list(const_tc.values())[0]
+ self.assertIs(type(const_sc), bt2_stream_class._StreamClassConst)
+
def test_destruction_listener(self):
def on_trace_class_destruction(trace_class):
nonlocal trace_class_destroyed
msgs = list(msg_iter)
self.assertEqual(len(msgs), 28)
hist = _count_msgs_by_type(msgs)
- self.assertEqual(hist[bt2._EventMessage], 8)
+ self.assertEqual(hist[bt2._EventMessageConst], 8)
# Same as the above, but we pass a single spec instead of a spec list.
def test_iter_specs_not_list(self):
msgs = list(msg_iter)
self.assertEqual(len(msgs), 28)
hist = _count_msgs_by_type(msgs)
- self.assertEqual(hist[bt2._EventMessage], 8)
+ self.assertEqual(hist[bt2._EventMessageConst], 8)
def test_iter_custom_filter(self):
src_spec = bt2.ComponentSpec.from_named_plugin_and_component_class(
)
msg_iter = bt2.TraceCollectionMessageIterator(src_spec, flt_spec)
hist = _count_msgs_by_type(msg_iter)
- self.assertEqual(hist[bt2._EventMessage], 5)
+ self.assertEqual(hist[bt2._EventMessageConst], 5)
def test_iter_intersection(self):
specs = [
msgs = list(msg_iter)
self.assertEqual(len(msgs), 15)
hist = _count_msgs_by_type(msgs)
- self.assertEqual(hist[bt2._EventMessage], 3)
+ self.assertEqual(hist[bt2._EventMessageConst], 3)
def test_iter_intersection_params(self):
# Check that all params used to create the source component are passed
specs, stream_intersection_mode=True
)
- event_msgs = [x for x in msg_iter if type(x) is bt2._EventMessage]
+ event_msgs = [x for x in msg_iter if type(x) is bt2._EventMessageConst]
self.assertEqual(len(event_msgs), 3)
self.assertEqual(
event_msgs[0].default_clock_snapshot.ns_from_origin, 13516309000000071
msgs = list(msg_iter)
self.assertEqual(len(msgs), 56)
hist = _count_msgs_by_type(msgs)
- self.assertEqual(hist[bt2._EventMessage], 16)
+ self.assertEqual(hist[bt2._EventMessageConst], 16)
def test_iter_no_intersection_begin(self):
specs = [
]
msg_iter = bt2.TraceCollectionMessageIterator(specs, begin=13515309.000000023)
hist = _count_msgs_by_type(msg_iter)
- self.assertEqual(hist[bt2._EventMessage], 6)
+ self.assertEqual(hist[bt2._EventMessageConst], 6)
def test_iter_no_intersection_end(self):
specs = [
]
msg_iter = bt2.TraceCollectionMessageIterator(specs, end=13515309.000000075)
hist = _count_msgs_by_type(msg_iter)
- self.assertEqual(hist[bt2._EventMessage], 5)
+ self.assertEqual(hist[bt2._EventMessageConst], 5)
def test_iter_auto_source_component_spec(self):
specs = [bt2.AutoSourceComponentSpec(_3EVENTS_INTERSECT_TRACE_PATH)]
msgs = list(msg_iter)
self.assertEqual(len(msgs), 28)
hist = _count_msgs_by_type(msgs)
- self.assertEqual(hist[bt2._EventMessage], 8)
+ self.assertEqual(hist[bt2._EventMessageConst], 8)
def test_iter_auto_source_component_spec_list_of_strings(self):
msg_iter = bt2.TraceCollectionMessageIterator([_3EVENTS_INTERSECT_TRACE_PATH])
msgs = list(msg_iter)
self.assertEqual(len(msgs), 28)
hist = _count_msgs_by_type(msgs)
- self.assertEqual(hist[bt2._EventMessage], 8)
+ self.assertEqual(hist[bt2._EventMessageConst], 8)
def test_iter_auto_source_component_spec_string(self):
msg_iter = bt2.TraceCollectionMessageIterator(_3EVENTS_INTERSECT_TRACE_PATH)
msgs = list(msg_iter)
self.assertEqual(len(msgs), 28)
hist = _count_msgs_by_type(msgs)
- self.assertEqual(hist[bt2._EventMessage], 8)
+ self.assertEqual(hist[bt2._EventMessageConst], 8)
def test_iter_mixed_inputs(self):
msg_iter = bt2.TraceCollectionMessageIterator(
msgs = list(msg_iter)
self.assertEqual(len(msgs), 76)
hist = _count_msgs_by_type(msgs)
- self.assertEqual(hist[bt2._EventMessage], 24)
+ self.assertEqual(hist[bt2._EventMessageConst], 24)
def test_auto_source_component_non_existent(self):
with self.assertRaisesRegex(
bt2.AutoSourceComponentSpec(_AUTO_SOURCE_DISCOVERY_GROUPING_PATH),
]
it = bt2.TraceCollectionMessageIterator(specs)
- msgs = [x for x in it if type(x) is bt2._StreamBeginningMessage]
+ msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
self.assertEqual(len(msgs), 8)
)
]
it = bt2.TraceCollectionMessageIterator(specs)
- msgs = [x for x in it if type(x) is bt2._StreamBeginningMessage]
+ msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
self.assertEqual(len(msgs), 2)
),
]
it = bt2.TraceCollectionMessageIterator(specs)
- msgs = [x for x in it if type(x) is bt2._StreamBeginningMessage]
+ msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
self.assertEqual(len(msgs), 2)
),
]
it = bt2.TraceCollectionMessageIterator(specs)
- msgs = [x for x in it if type(x) is bt2._StreamBeginningMessage]
+ msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
self.assertEqual(len(msgs), 2)
),
]
it = bt2.TraceCollectionMessageIterator(specs)
- msgs = [x for x in it if type(x) is bt2._StreamBeginningMessage]
+ msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
self.assertEqual(len(msgs), 2)
),
]
it = bt2.TraceCollectionMessageIterator(specs)
- msgs = [x for x in it if type(x) is bt2._StreamBeginningMessage]
+ msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
self.assertEqual(len(msgs), 2)
self.assertEqual(msgs[0].stream.name, "TestSourceA: None")
bt2.AutoSourceComponentSpec(self._dir_a, params={'what': 'python-obj'}),
]
it = bt2.TraceCollectionMessageIterator(specs)
- msgs = [x for x in it if type(x) is bt2._StreamBeginningMessage]
+ msgs = [x for x in it if type(x) is bt2._StreamBeginningMessageConst]
self.assertEqual(len(msgs), 2)
self.assertEqual(msgs[0].stream.name, "TestSourceA: deore")
return run_in_component_init(f)
+# Create a pair of list, one containing non-const messages and the other
+# containing const messages
+def _get_all_message_types(with_packet=True):
+ _msgs = None
+
+ class MyIter(bt2._UserMessageIterator):
+ def __init__(self, self_output_port):
+
+ nonlocal _msgs
+ self._at = 0
+ self._msgs = [
+ self._create_stream_beginning_message(
+ self_output_port.user_data['stream']
+ )
+ ]
+
+ if with_packet:
+ assert self_output_port.user_data['packet']
+ self._msgs.append(
+ self._create_packet_beginning_message(
+ self_output_port.user_data['packet']
+ )
+ )
+
+ default_clock_snapshot = 789
+
+ if with_packet:
+ assert self_output_port.user_data['packet']
+ ev_parent = self_output_port.user_data['packet']
+ else:
+ assert self_output_port.user_data['stream']
+ ev_parent = self_output_port.user_data['stream']
+
+ msg = self._create_event_message(
+ self_output_port.user_data['event_class'],
+ ev_parent,
+ default_clock_snapshot,
+ )
+
+ msg.event.payload_field['giraffe'] = 1
+ msg.event.specific_context_field['ant'] = -1
+ msg.event.common_context_field['cpu_id'] = 1
+ self._msgs.append(msg)
+
+ if with_packet:
+ self._msgs.append(
+ self._create_packet_end_message(
+ self_output_port.user_data['packet']
+ )
+ )
+
+ self._msgs.append(
+ self._create_stream_end_message(self_output_port.user_data['stream'])
+ )
+
+ _msgs = self._msgs
+
+ def __next__(self):
+ if self._at == len(self._msgs):
+ raise bt2.Stop
+
+ msg = self._msgs[self._at]
+ self._at += 1
+ return msg
+
+ class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ def __init__(self, params, obj):
+ tc = self._create_trace_class()
+ clock_class = self._create_clock_class(frequency=1000)
+
+ # event common context (stream-class-defined)
+ cc = tc.create_structure_field_class()
+ cc += [('cpu_id', tc.create_signed_integer_field_class(8))]
+
+ # packet context (stream-class-defined)
+ pc = None
+
+ if with_packet:
+ pc = tc.create_structure_field_class()
+ pc += [('something', tc.create_unsigned_integer_field_class(8))]
+
+ stream_class = tc.create_stream_class(
+ default_clock_class=clock_class,
+ event_common_context_field_class=cc,
+ packet_context_field_class=pc,
+ supports_packets=with_packet,
+ )
+
+ # specific context (event-class-defined)
+ sc = tc.create_structure_field_class()
+ sc += [('ant', tc.create_signed_integer_field_class(16))]
+
+ # event payload
+ ep = tc.create_structure_field_class()
+ ep += [('giraffe', tc.create_signed_integer_field_class(32))]
+
+ event_class = stream_class.create_event_class(
+ name='garou', specific_context_field_class=sc, payload_field_class=ep
+ )
+
+ trace = tc(environment={'patate': 12})
+ stream = trace.create_stream(stream_class, user_attributes={'salut': 23})
+
+ if with_packet:
+ packet = stream.create_packet()
+ packet.context_field['something'] = 154
+ else:
+ packet = None
+
+ self._add_output_port(
+ 'out',
+ {
+ 'tc': tc,
+ 'stream': stream,
+ 'event_class': event_class,
+ 'trace': trace,
+ 'packet': packet,
+ },
+ )
+
+ _graph = bt2.Graph()
+ _src_comp = _graph.add_component(MySrc, 'my_source')
+ _msg_iter = TestOutputPortMessageIterator(_graph, _src_comp.output_ports['out'])
+
+ const_msgs = list(_msg_iter)
+
+ return _msgs, const_msgs
+
+
+def get_stream_beginning_message():
+ msgs, _ = _get_all_message_types()
+ for m in msgs:
+ if type(m) is bt2._StreamBeginningMessage:
+ return m
+
+
+def get_const_stream_beginning_message():
+ _, const_msgs = _get_all_message_types()
+ for m in const_msgs:
+ if type(m) is bt2._StreamBeginningMessageConst:
+ return m
+
+
+def get_stream_end_message():
+ msgs, _ = _get_all_message_types()
+ for m in msgs:
+ if type(m) is bt2._StreamEndMessage:
+ return m
+
+
+def get_packet_beginning_message():
+ msgs, _ = _get_all_message_types(with_packet=True)
+ for m in msgs:
+ if type(m) is bt2._PacketBeginningMessage:
+ return m
+
+
+def get_const_packet_beginning_message():
+ _, const_msgs = _get_all_message_types(with_packet=True)
+ for m in const_msgs:
+ if type(m) is bt2._PacketBeginningMessageConst:
+ return m
+
+
+def get_packet_end_message():
+ msgs, _ = _get_all_message_types(with_packet=True)
+ for m in msgs:
+ if type(m) is bt2._PacketEndMessage:
+ return m
+
+
+def get_event_message():
+ msgs, _ = _get_all_message_types()
+ for m in msgs:
+ if type(m) is bt2._EventMessage:
+ return m
+
+
+def get_const_event_message():
+ _, const_msgs = _get_all_message_types()
+ for m in const_msgs:
+ if type(m) is bt2._EventMessageConst:
+ return m
+
+
# Proxy sink component class.
#
# This sink accepts a list of a single item as its initialization
projects / babeltrace.git / commitdiff
