bt2/trace.py \
bt2/trace_class.py \
bt2/trace_collection_message_iterator.py \
+ bt2/user_attributes.py \
bt2/utils.py \
bt2/value.py
import uuid as uuidp
from bt2 import utils as bt2_utils
-from bt2 import value as bt2_value
from bt2 import object as bt2_object
from bt2 import native_bt
+from bt2 import user_attributes as bt2_user_attrs
class ClockClassOffset:
return (self.seconds, self.cycles) == (other.seconds, other.cycles)
-class _ClockClassConst(bt2_object._SharedObject):
+class _ClockClassConst(bt2_object._SharedObject, bt2_user_attrs._WithUserAttrsConst):
@staticmethod
def _get_ref(ptr):
native_bt.clock_class_get_ref(ptr)
def _put_ref(ptr):
native_bt.clock_class_put_ref(ptr)
- _create_value_from_ptr_and_get_ref = staticmethod(
- bt2_value._create_from_const_ptr_and_get_ref
- )
- _borrow_user_attributes_ptr = staticmethod(
- native_bt.clock_class_borrow_user_attributes_const
- )
-
- @property
- def user_attributes(self):
- ptr = self._borrow_user_attributes_ptr(self._ptr)
- assert ptr is not None
- return self._create_value_from_ptr_and_get_ref(ptr)
+ @staticmethod
+ def _borrow_user_attributes_ptr(ptr):
+ return native_bt.clock_class_borrow_user_attributes_const(ptr)
@property
def name(self):
return ns
-class _ClockClass(_ClockClassConst):
- _create_value_from_ptr_and_get_ref = staticmethod(
- bt2_value._create_from_ptr_and_get_ref
- )
- _borrow_user_attributes_ptr = staticmethod(
- native_bt.clock_class_borrow_user_attributes
- )
-
- def _user_attributes(self, user_attributes):
- value = bt2_value.create_value(user_attributes)
- bt2_utils._check_type(value, bt2_value.MapValue)
- native_bt.clock_class_set_user_attributes(self._ptr, value._ptr)
+class _ClockClass(bt2_user_attrs._WithUserAttrs, _ClockClassConst):
+ @staticmethod
+ def _borrow_user_attributes_ptr(ptr):
+ return native_bt.clock_class_borrow_user_attributes(ptr)
- _user_attributes = property(fset=_user_attributes)
+ @staticmethod
+ def _set_user_attributes_ptr(obj_ptr, value_ptr):
+ native_bt.clock_class_set_user_attributes(obj_ptr, value_ptr)
def _name(self, name):
bt2_utils._check_str(name)
from bt2 import object as bt2_object
from bt2 import native_bt
from bt2 import field_class as bt2_field_class
+from bt2 import user_attributes as bt2_user_attrs
def _bt2_stream_class():
DEBUG = native_bt.EVENT_CLASS_LOG_LEVEL_DEBUG
-class _EventClassConst(bt2_object._SharedObject):
+class _EventClassConst(bt2_object._SharedObject, bt2_user_attrs._WithUserAttrsConst):
@staticmethod
def _get_ref(ptr):
native_bt.event_class_get_ref(ptr)
_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
if sc_ptr is not None:
return self._stream_class_pycls._create_from_ptr_and_get_ref(sc_ptr)
- @property
- def user_attributes(self):
- ptr = self._borrow_user_attributes_ptr(self._ptr)
- assert ptr is not None
- return self._create_value_from_ptr_and_get_ref(ptr)
-
@property
def name(self):
return native_bt.event_class_get_name(self._ptr)
return self._create_field_class_from_ptr_and_get_ref(fc_ptr)
-class _EventClass(_EventClassConst):
+class _EventClass(bt2_user_attrs._WithUserAttrs, _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
_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)
- native_bt.event_class_set_user_attributes(self._ptr, value._ptr)
-
- _user_attributes = property(fset=_user_attributes)
+ @staticmethod
+ def _set_user_attributes_ptr(obj_ptr, value_ptr):
+ native_bt.event_class_set_user_attributes(obj_ptr, value_ptr)
def _name(self, name):
return native_bt.event_class_set_name(self._ptr, name)
from bt2 import object as bt2_object
from bt2 import native_bt
from bt2 import field_path as bt2_field_path
+from bt2 import user_attributes as bt2_user_attrs
from bt2 import integer_range_set as bt2_integer_range_set
HEXADECIMAL = native_bt.FIELD_CLASS_INTEGER_PREFERRED_DISPLAY_BASE_HEXADECIMAL
-class _FieldClassConst(bt2_object._SharedObject):
+class _FieldClassConst(bt2_object._SharedObject, bt2_user_attrs._WithUserAttrsConst):
@staticmethod
def _get_ref(ptr):
native_bt.field_class_get_ref(ptr)
def _put_ref(ptr):
native_bt.field_class_put_ref(ptr)
- _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
- )
+ @staticmethod
+ def _borrow_user_attributes_ptr(ptr):
+ return native_bt.field_class_borrow_user_attributes_const(ptr)
def _check_create_status(self, ptr):
if ptr is None:
"cannot create {} field class object".format(self._NAME.lower())
)
- @property
- def user_attributes(self):
- ptr = self._borrow_user_attributes_ptr(self._ptr)
- assert ptr is not None
- 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)
- bt2_utils._check_type(value, bt2_value.MapValue)
- native_bt.field_class_set_user_attributes(self._ptr, value._ptr)
+class _FieldClass(bt2_user_attrs._WithUserAttrs, _FieldClassConst):
+ @staticmethod
+ def _borrow_user_attributes_ptr(ptr):
+ return native_bt.field_class_borrow_user_attributes(ptr)
- _user_attributes = property(fset=_user_attributes)
+ @staticmethod
+ def _set_user_attributes_ptr(obj_ptr, value_ptr):
+ native_bt.field_class_set_user_attributes(obj_ptr, value_ptr)
class _BoolFieldClassConst(_FieldClassConst):
_NAME = "String"
-class _StructureFieldClassMemberConst:
+class _StructureFieldClassMemberConst(bt2_user_attrs._WithUserAttrsConst):
+ @property
+ def _ptr(self):
+ return self._member_ptr
+
_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
- )
+
+ @staticmethod
+ def _borrow_user_attributes_ptr(ptr):
+ return native_bt.field_class_structure_member_borrow_user_attributes_const(ptr)
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
self._owning_struct_fc = owning_struct_fc
- self._ptr = member_ptr
+ self._member_ptr = member_ptr
@property
def name(self):
- name = native_bt.field_class_structure_member_get_name(self._ptr)
+ name = native_bt.field_class_structure_member_get_name(self._member_ptr)
assert name is not None
return name
@property
def field_class(self):
- fc_ptr = self._borrow_field_class_ptr(self._ptr)
+ fc_ptr = self._borrow_field_class_ptr(self._member_ptr)
assert fc_ptr is not None
return self._create_field_class_from_ptr_and_get_ref(fc_ptr)
- @property
- def user_attributes(self):
- ptr = self._borrow_user_attributes_ptr(self._ptr)
- assert ptr is not None
- return self._create_value_from_ptr_and_get_ref(ptr)
-
-class _StructureFieldClassMember(_StructureFieldClassMemberConst):
+class _StructureFieldClassMember(
+ bt2_user_attrs._WithUserAttrs, _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
- )
+
+ @staticmethod
+ def _borrow_user_attributes_ptr(ptr):
+ return native_bt.field_class_structure_member_borrow_user_attributes(ptr)
+
_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)
- bt2_utils._check_type(value, bt2_value.MapValue)
- native_bt.field_class_structure_member_set_user_attributes(
- self._ptr, value._ptr
- )
- _user_attributes = property(fset=_user_attributes)
+ @staticmethod
+ def _set_user_attributes_ptr(obj_ptr, value_ptr):
+ native_bt.field_class_structure_member_set_user_attributes(obj_ptr, value_ptr)
class _StructureFieldClassConst(_FieldClassConst, collections.abc.Mapping):
_NAME = "Option (with signed integer selector)"
-class _VariantFieldClassOptionConst:
+class _VariantFieldClassOptionConst(bt2_user_attrs._WithUserAttrsConst):
+ @property
+ def _ptr(self):
+ return self._opt_ptr
+
_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
- )
+
+ @staticmethod
+ def _borrow_user_attributes_ptr(ptr):
+ return native_bt.field_class_variant_option_borrow_user_attributes_const(ptr)
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
self._owning_var_fc = owning_var_fc
- self._ptr = option_ptr
+ self._opt_ptr = option_ptr
@property
def name(self):
- name = native_bt.field_class_variant_option_get_name(self._ptr)
+ name = native_bt.field_class_variant_option_get_name(self._opt_ptr)
assert name is not None
return name
@property
def field_class(self):
- fc_ptr = self._borrow_field_class_ptr(self._ptr)
+ fc_ptr = self._borrow_field_class_ptr(self._opt_ptr)
assert fc_ptr is not None
return self._create_field_class_from_ptr_and_get_ref(fc_ptr)
- @property
- def user_attributes(self):
- ptr = self._borrow_user_attributes_ptr(self._ptr)
- assert ptr is not None
- return self._create_value_from_ptr_and_get_ref(ptr)
-
-class _VariantFieldClassOption(_VariantFieldClassOptionConst):
+class _VariantFieldClassOption(
+ bt2_user_attrs._WithUserAttrs, _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
- )
- _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)
- bt2_utils._check_type(value, bt2_value.MapValue)
- native_bt.field_class_variant_option_set_user_attributes(self._ptr, value._ptr)
+ @staticmethod
+ def _borrow_user_attributes_ptr(ptr):
+ return native_bt.field_class_variant_option_borrow_user_attributes(ptr)
- _user_attributes = property(fset=_user_attributes)
+ @staticmethod
+ def _set_user_attributes_ptr(obj_ptr, value_ptr):
+ native_bt.field_class_variant_option_set_user_attributes(obj_ptr, value_ptr)
class _VariantFieldClassWithIntegerSelectorOptionConst(_VariantFieldClassOptionConst):
from bt2 import error as bt2_error
from bt2 import utils as bt2_utils
-from bt2 import value as bt2_value
from bt2 import object as bt2_object
from bt2 import packet as bt2_packet
from bt2 import native_bt
from bt2 import stream_class as bt2_stream_class
+from bt2 import user_attributes as bt2_user_attrs
def _bt2_trace():
return bt2_trace
-class _StreamConst(bt2_object._SharedObject):
+class _StreamConst(bt2_object._SharedObject, bt2_user_attrs._WithUserAttrsConst):
@staticmethod
def _get_ref(ptr):
native_bt.stream_get_ref(ptr)
native_bt.stream_put_ref(ptr)
_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
- )
+
+ @staticmethod
+ def _borrow_user_attributes_ptr(ptr):
+ return native_bt.stream_borrow_user_attributes_const(ptr)
+
_borrow_trace_ptr = staticmethod(native_bt.stream_borrow_trace_const)
_stream_class_pycls = bt2_stream_class._StreamClassConst
_trace_pycls = property(lambda _: _bt2_trace()._TraceConst)
def name(self):
return native_bt.stream_get_name(self._ptr)
- @property
- def user_attributes(self):
- ptr = self._borrow_user_attributes_ptr(self._ptr)
- assert ptr is not None
- return self._create_value_from_ptr_and_get_ref(ptr)
-
@property
def id(self):
id = native_bt.stream_get_id(self._ptr)
return self._trace_pycls._create_from_ptr_and_get_ref(trace_ptr)
-class _Stream(_StreamConst):
+class _Stream(bt2_user_attrs._WithUserAttrs, _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
- )
+
+ @staticmethod
+ def _borrow_user_attributes_ptr(ptr):
+ return native_bt.stream_borrow_user_attributes(ptr)
+
_borrow_trace_ptr = staticmethod(native_bt.stream_borrow_trace)
_stream_class_pycls = bt2_stream_class._StreamClass
_trace_pycls = property(lambda _: _bt2_trace()._Trace)
return bt2_packet._Packet._create_from_ptr(packet_ptr)
- def _user_attributes(self, user_attributes):
- value = bt2_value.create_value(user_attributes)
- bt2_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
- )
+ @staticmethod
+ def _set_user_attributes_ptr(obj_ptr, value_ptr):
+ native_bt.stream_set_user_attributes(obj_ptr, value_ptr)
def _name(self, name):
bt2_utils._check_str(name)
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
+from bt2 import user_attributes as bt2_user_attrs
def _bt2_trace_class():
return bt2_trace_class
-class _StreamClassConst(bt2_object._SharedObject, collections.abc.Mapping):
+class _StreamClassConst(
+ bt2_object._SharedObject,
+ bt2_user_attrs._WithUserAttrsConst,
+ collections.abc.Mapping,
+):
@staticmethod
def _get_ref(ptr):
native_bt.stream_class_get_ref(ptr)
_borrow_default_clock_class_ptr = staticmethod(
native_bt.stream_class_borrow_default_clock_class_const
)
- _borrow_user_attributes_ptr = staticmethod(
- native_bt.stream_class_borrow_user_attributes_const
- )
+
+ @staticmethod
+ def _borrow_user_attributes_ptr(ptr):
+ return native_bt.stream_class_borrow_user_attributes_const(ptr)
_event_class_cls = property(lambda _: bt2_event_class._EventClassConst)
_trace_class_cls = property(lambda _: _bt2_trace_class()._TraceClassConst)
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 = self._borrow_user_attributes_ptr(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)
return self._clock_class_cls._create_from_ptr_and_get_ref(cc_ptr)
-class _StreamClass(_StreamClassConst):
+class _StreamClass(bt2_user_attrs._WithUserAttrs, _StreamClassConst):
@staticmethod
def _get_ref(ptr):
native_bt.stream_class_get_ref(ptr)
_borrow_default_clock_class_ptr = staticmethod(
native_bt.stream_class_borrow_default_clock_class
)
- _borrow_user_attributes_ptr = staticmethod(
- native_bt.stream_class_borrow_user_attributes
- )
+
+ @staticmethod
+ def _borrow_user_attributes_ptr(ptr):
+ return native_bt.stream_class_borrow_user_attributes(ptr)
_event_class_cls = property(lambda s: bt2_event_class._EventClass)
_trace_class_cls = property(lambda s: _bt2_trace_class()._TraceClass)
return event_class
- def _user_attributes(self, user_attributes):
- value = bt2_value.create_value(user_attributes)
- native_bt.stream_class_set_user_attributes(self._ptr, value._ptr)
-
- _user_attributes = property(fset=_user_attributes)
+ @staticmethod
+ def _set_user_attributes_ptr(obj_ptr, value_ptr):
+ native_bt.stream_class_set_user_attributes(obj_ptr, value_ptr)
def _name(self, name):
status = native_bt.stream_class_set_name(self._ptr, name)
from bt2 import stream as bt2_stream
from bt2 import native_bt
from bt2 import stream_class as bt2_stream_class
+from bt2 import user_attributes as bt2_user_attrs
def _bt2_trace_class():
raise NotImplementedError
-class _TraceConst(bt2_object._SharedObject, collections.abc.Mapping):
+class _TraceConst(
+ bt2_object._SharedObject,
+ bt2_user_attrs._WithUserAttrsConst,
+ collections.abc.Mapping,
+):
@staticmethod
def _get_ref(ptr):
native_bt.trace_get_ref(ptr)
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
- )
+
+ @staticmethod
+ def _borrow_user_attributes_ptr(ptr):
+ return native_bt.trace_borrow_user_attributes_const(ptr)
+
_stream_pycls = property(lambda _: bt2_stream._StreamConst)
_trace_class_pycls = property(lambda _: _bt2_trace_class()._TraceClassConst)
_trace_env_pycls = property(lambda _: _TraceEnvironmentConst)
assert trace_class_ptr is not None
return self._trace_class_pycls._create_from_ptr_and_get_ref(trace_class_ptr)
- @property
- def user_attributes(self):
- ptr = self._borrow_user_attributes_ptr(self._ptr)
- assert ptr is not None
- return self._create_value_from_ptr_and_get_ref(ptr)
-
@property
def name(self):
return native_bt.trace_get_name(self._ptr)
listener_handle._invalidate()
-class _Trace(_TraceConst):
+class _Trace(bt2_user_attrs._WithUserAttrs, _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
- )
+
+ @staticmethod
+ def _borrow_user_attributes_ptr(ptr):
+ return native_bt.trace_borrow_user_attributes(ptr)
+
_stream_pycls = property(lambda _: bt2_stream._Stream)
_trace_class_pycls = property(lambda _: _bt2_trace_class()._TraceClass)
_trace_env_pycls = property(lambda _: _TraceEnvironment)
_name = property(fset=_name)
- def _user_attributes(self, user_attributes):
- value = bt2_value.create_value(user_attributes)
- bt2_utils._check_type(value, bt2_value.MapValue)
- native_bt.trace_set_user_attributes(self._ptr, value._ptr)
-
- _user_attributes = property(fset=_user_attributes)
+ @staticmethod
+ def _set_user_attributes_ptr(obj_ptr, value_ptr):
+ native_bt.trace_set_user_attributes(obj_ptr, value_ptr)
def _uuid(self, uuid):
bt2_utils._check_type(uuid, uuidp.UUID)
from bt2 import error as bt2_error
from bt2 import trace as bt2_trace
from bt2 import utils as bt2_utils
-from bt2 import value as bt2_value
from bt2 import object as bt2_object
from bt2 import native_bt
from bt2 import field_class as bt2_field_class
from bt2 import stream_class as bt2_stream_class
+from bt2 import user_attributes as bt2_user_attrs
from bt2 import integer_range_set as bt2_integer_range_set
handle._invalidate()
-class _TraceClassConst(bt2_object._SharedObject, collections.abc.Mapping):
+class _TraceClassConst(
+ bt2_object._SharedObject,
+ bt2_user_attrs._WithUserAttrsConst,
+ collections.abc.Mapping,
+):
@staticmethod
def _get_ref(ptr):
native_bt.trace_class_get_ref(ptr)
_borrow_stream_class_ptr_by_id = staticmethod(
native_bt.trace_class_borrow_stream_class_by_id_const
)
- _borrow_user_attributes_ptr = staticmethod(
- native_bt.trace_class_borrow_user_attributes_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
- )
- @property
- def user_attributes(self):
- ptr = self._borrow_user_attributes_ptr(self._ptr)
- assert ptr is not None
- return self._create_value_from_ptr_and_get_ref(ptr)
+ @staticmethod
+ def _borrow_user_attributes_ptr(ptr):
+ return native_bt.trace_class_borrow_user_attributes_const(ptr)
+
+ _stream_class_pycls = bt2_stream_class._StreamClassConst
# Number of stream classes in this trace class.
listener_handle._invalidate()
-class _TraceClass(_TraceClassConst):
+class _TraceClass(bt2_user_attrs._WithUserAttrs, _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
)
- _borrow_user_attributes_ptr = staticmethod(
- native_bt.trace_class_borrow_user_attributes
- )
+
+ @staticmethod
+ def _borrow_user_attributes_ptr(ptr):
+ return native_bt.trace_class_borrow_user_attributes(ptr)
+
_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.
)
return sc
- def _user_attributes(self, user_attributes):
- value = bt2_value.create_value(user_attributes)
- bt2_utils._check_type(value, bt2_value.MapValue)
- native_bt.trace_class_set_user_attributes(self._ptr, value._ptr)
-
- _user_attributes = property(fset=_user_attributes)
+ @staticmethod
+ def _set_user_attributes_ptr(obj_ptr, value_ptr):
+ native_bt.trace_class_set_user_attributes(obj_ptr, value_ptr)
def _assigns_automatic_stream_class_id(self, auto_id):
bt2_utils._check_bool(auto_id)
--- /dev/null
+# SPDX-License-Identifier: MIT
+#
+# Copyright (c) 2023 EfficiOS, Inc.
+
+import abc
+
+from bt2 import utils as bt2_utils
+from bt2 import value as bt2_value
+
+
+class _WithUserAttrsBase(abc.ABC):
+ @staticmethod
+ @abc.abstractmethod
+ def _borrow_user_attributes_ptr(ptr):
+ raise NotImplementedError
+
+ @property
+ @abc.abstractmethod
+ def _ptr(self):
+ raise NotImplementedError
+
+
+# Mixin class for objects with user attributes (const version).
+class _WithUserAttrsConst(_WithUserAttrsBase):
+ @property
+ def user_attributes(self) -> bt2_value._MapValueConst:
+ return bt2_value._MapValueConst._create_from_ptr_and_get_ref(
+ self._borrow_user_attributes_ptr(self._ptr)
+ )
+
+
+# Mixin class for objects with user attributes (non-const version).
+class _WithUserAttrs(_WithUserAttrsBase, abc.ABC):
+ @property
+ def user_attributes(self) -> bt2_value.MapValue:
+ return bt2_value.MapValue._create_from_ptr_and_get_ref(
+ self._borrow_user_attributes_ptr(self._ptr)
+ )
+
+ @staticmethod
+ @abc.abstractmethod
+ def _set_user_attributes_ptr(obj_ptr, value_ptr):
+ raise NotImplementedError
+
+ def _user_attributes(self, user_attributes):
+ value = bt2_value.create_value(user_attributes)
+ bt2_utils._check_type(value, bt2_value.MapValue)
+ self._set_user_attributes_ptr(self._ptr, value._ptr)
+
+ _user_attributes = property(fset=_user_attributes)
import unittest
import bt2
-import utils
from bt2 import value as bt2_value
from bt2 import clock_class as bt2_clock_class
-from utils import TestOutputPortMessageIterator, run_in_component_init
+from utils import (
+ TestOutputPortMessageIterator,
+ run_in_component_init,
+ get_const_event_message,
+)
class ClockClassOffsetTestCase(unittest.TestCase):
self.assertEqual(cc.user_attributes, {"salut": 23})
self.assertIs(type(cc.user_attributes), bt2_value.MapValue)
+ def test_const_user_attributes(self):
+ cc = get_const_event_message().default_clock_snapshot.clock_class
+ self.assertEqual(cc.user_attributes, {"a-clock-class-attribute": 1})
+ self.assertIs(type(cc.user_attributes), bt2._MapValueConst)
+
def test_create_invalid_user_attributes(self):
def f(comp_self):
return comp_self._create_clock_class(user_attributes=object())
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):
def test_const_user_attributes(self):
stream = utils.get_const_stream_beginning_message().stream
- self.assertEqual(stream.user_attributes, {"salut": 23})
+ self.assertEqual(stream.user_attributes, {"a-stream-attribute": 1})
self.assertIs(type(stream.user_attributes), bt2_value._MapValueConst)
def test_create_invalid_user_attributes(self):
import unittest
+from bt2 import value as bt2_value
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
from bt2 import trace_class as bt2_trace_class
from bt2 import stream_class as bt2_stream_class
-from utils import run_in_component_init
+from utils import run_in_component_init, get_const_stream_beginning_message
class StreamClassTestCase(unittest.TestCase):
def test_create_user_attributes(self):
sc = self._tc.create_stream_class(user_attributes={"salut": 23})
self.assertEqual(sc.user_attributes, {"salut": 23})
+ self.assertIs(type(sc.user_attributes), bt2_value.MapValue)
+
+ def test_const_user_attributes(self):
+ sc = get_const_stream_beginning_message().stream.cls
+ self.assertEqual(sc.user_attributes, {"a-stream-class-attribute": 1})
+ self.assertIs(type(sc.user_attributes), bt2_value._MapValueConst)
def test_create_invalid_user_attributes(self):
with self.assertRaisesRegex(
from bt2 import value as bt2_value
from bt2 import stream as bt2_stream
from bt2 import trace_class as bt2_trace_class
-from utils import get_default_trace_class
+from utils import get_default_trace_class, get_const_stream_beginning_message
class TraceTestCase(unittest.TestCase):
self.assertEqual(trace.user_attributes, {"salut": 23})
self.assertIs(type(trace.user_attributes), bt2_value.MapValue)
+ def test_const_user_attributes(self):
+ trace = get_const_stream_beginning_message().stream.trace
+ self.assertEqual(trace.user_attributes, {"a-trace-attribute": 1})
+ self.assertIs(type(trace.user_attributes), bt2_value._MapValueConst)
+
def test_create_invalid_user_attributes(self):
with self.assertRaises(TypeError):
self._tc(user_attributes=object())
import unittest
from bt2 import utils as bt2_utils
+from bt2 import value as bt2_value
from bt2 import trace_class as bt2_trace_class
from bt2 import stream_class as bt2_stream_class
from utils import (
tc = run_in_component_init(f)
self.assertEqual(tc.user_attributes, {"salut": 23})
+ self.assertIs(type(tc.user_attributes), bt2_value.MapValue)
+
+ def test_const_user_attributes(self):
+ tc = get_const_stream_beginning_message().stream.trace.cls
+ self.assertEqual(tc.user_attributes, {"a-trace-class-attribute": 1})
+ self.assertIs(type(tc.user_attributes), bt2_value._MapValueConst)
def test_create_invalid_user_attributes(self):
def f(comp_self):
class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
def __init__(self, config, params, obj):
- tc = self._create_trace_class()
- clock_class = self._create_clock_class(frequency=1000)
+ tc = self._create_trace_class(
+ user_attributes={"a-trace-class-attribute": 1}
+ )
+ clock_class = self._create_clock_class(
+ frequency=1000, user_attributes={"a-clock-class-attribute": 1}
+ )
# event common context (stream-class-defined)
cc = tc.create_structure_field_class()
event_common_context_field_class=cc,
packet_context_field_class=pc,
supports_packets=with_packet,
+ user_attributes={"a-stream-class-attribute": 1},
)
# specific context (event-class-defined)
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
+ name="garou",
+ specific_context_field_class=sc,
+ payload_field_class=ep,
+ user_attributes={"an-event-class-attribute": 1},
)
- trace = tc(environment={"patate": 12})
- stream = trace.create_stream(stream_class, user_attributes={"salut": 23})
+ trace = tc(
+ environment={"patate": 12}, user_attributes={"a-trace-attribute": 1}
+ )
+ stream = trace.create_stream(
+ stream_class, user_attributes={"a-stream-attribute": 1}
+ )
if with_packet:
packet = stream.create_packet()
projects / babeltrace.git / commitdiff
