return name
@property
- def component_class(self):
+ def cls(self):
cc_ptr = self._borrow_component_class_ptr(self._ptr)
assert cc_ptr is not None
return _create_component_class_from_ptr_and_get_ref(cc_ptr, self._comp_cls_type)
return name
@property
- def component_class(self):
+ def cls(self):
comp_ptr = self._as_not_self_specific_component_ptr(self._ptr)
cc_ptr = self._borrow_component_class_ptr(comp_ptr)
return _create_component_class_from_ptr_and_get_ref(cc_ptr, self._comp_cls_type)
class _Event(object._UniqueObject):
@property
- def event_class(self):
+ def cls(self):
event_class_ptr = native_bt.event_borrow_class(self._ptr)
assert event_class_ptr is not None
return bt2.event_class._EventClass._create_from_ptr_and_get_ref(event_class_ptr)
@property
def name(self):
- return self.event_class.name
+ return self.cls.name
@property
def id(self):
- return self.event_class.id
+ return self.cls.id
@property
def packet(self):
@property
def default_clock_snapshot(self):
- self._check_has_default_clock_class(self.event.packet.stream.stream_class.default_clock_class)
+ self._check_has_default_clock_class(self.event.packet.stream.cls.default_clock_class)
return self._get_default_clock_snapshot(self._borrow_default_clock_snapshot_ptr)
@property
class _PacketMessage(_Message, _MessageWithDefaultClockSnapshot):
@property
def default_clock_snapshot(self):
- self._check_has_default_clock_class(self.packet.stream.stream_class.default_clock_class)
+ self._check_has_default_clock_class(self.packet.stream.cls.default_clock_class)
return self._get_default_clock_snapshot(self._borrow_default_clock_snapshot_ptr)
@property
class _StreamActivityMessage(_Message):
@property
def default_clock_snapshot(self):
- self._check_has_default_clock_class(self.stream.stream_class.default_clock_class)
+ self._check_has_default_clock_class(self.stream.cls.default_clock_class)
status, snapshot_ptr = self._borrow_default_clock_snapshot_ptr(self._ptr)
if status == native_bt.MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_KNOWN:
@property
def _has_default_clock_snapshots(self):
- return self.stream.stream_class.discarded_events_have_default_clock_snapshots
+ return self.stream.cls.discarded_events_have_default_clock_snapshots
class _DiscardedPacketsMessage(_DiscardedMessage):
@property
def _has_default_clock_snapshots(self):
- return self.stream.stream_class.discarded_packets_have_default_clock_snapshots
+ return self.stream.cls.discarded_packets_have_default_clock_snapshots
_MESSAGE_TYPE_TO_CLS = {
def _create_event_message(self, event_class, packet, default_clock_snapshot=None):
utils._check_type(event_class, bt2.event_class._EventClass)
utils._check_type(packet, bt2.packet._Packet)
- self._validate_default_clock_snapshot(packet.stream.stream_class, default_clock_snapshot)
+ self._validate_default_clock_snapshot(packet.stream.cls, default_clock_snapshot)
if default_clock_snapshot is not None:
utils._check_uint64(default_clock_snapshot)
def _create_stream_activity_beginning_message(self, stream, default_clock_snapshot=None):
utils._check_type(stream, bt2.stream._Stream)
- self._validate_default_clock_snapshot(stream.stream_class, default_clock_snapshot)
+ self._validate_default_clock_snapshot(stream.cls, default_clock_snapshot)
ptr = native_bt.message_stream_activity_beginning_create(self._ptr, stream._ptr)
def _create_stream_activity_end_message(self, stream, default_clock_snapshot=None):
utils._check_type(stream, bt2.stream._Stream)
- self._validate_default_clock_snapshot(stream.stream_class, default_clock_snapshot)
+ self._validate_default_clock_snapshot(stream.cls, default_clock_snapshot)
ptr = native_bt.message_stream_activity_end_create(self._ptr, stream._ptr)
def _create_packet_beginning_message(self, packet, default_clock_snapshot=None):
utils._check_type(packet, bt2.packet._Packet)
- if packet.stream.stream_class.packets_have_beginning_default_clock_snapshot:
+ if packet.stream.cls.packets_have_beginning_default_clock_snapshot:
if default_clock_snapshot is None:
raise ValueError("packet beginning messages in this stream must have a default clock snapshots")
def _create_packet_end_message(self, packet, default_clock_snapshot=None):
utils._check_type(packet, bt2.packet._Packet)
- if packet.stream.stream_class.packets_have_end_default_clock_snapshot:
+ if packet.stream.cls.packets_have_end_default_clock_snapshot:
if default_clock_snapshot is None:
raise ValueError("packet end messages in this stream must have a default clock snapshots")
end_clock_snapshot=None):
utils._check_type(stream, bt2.stream._Stream)
- if not stream.stream_class.supports_discarded_events:
+ if not stream.cls.supports_discarded_events:
raise ValueError('stream class does not support discarded events')
- if stream.stream_class.discarded_events_have_default_clock_snapshots:
+ if stream.cls.discarded_events_have_default_clock_snapshots:
if beg_clock_snapshot is None or end_clock_snapshot is None:
raise ValueError('discarded events have default clock snapshots for this stream class')
def _create_discarded_packets_message(self, stream, count=None, beg_clock_snapshot=None, end_clock_snapshot=None):
utils._check_type(stream, bt2.stream._Stream)
- if not stream.stream_class.supports_discarded_packets:
+ if not stream.cls.supports_discarded_packets:
raise ValueError('stream class does not support discarded packets')
- if stream.stream_class.discarded_packets_have_default_clock_snapshots:
+ if stream.cls.discarded_packets_have_default_clock_snapshots:
if beg_clock_snapshot is None or end_clock_snapshot is None:
raise ValueError('discarded packets have default clock snapshots for this stream class')
_put_ref = staticmethod(native_bt.stream_put_ref)
@property
- def stream_class(self):
+ def cls(self):
stream_class_ptr = native_bt.stream_borrow_class(self._ptr)
assert stream_class_ptr is not None
return bt2.stream_class._StreamClass._create_from_ptr_and_get_ref(stream_class_ptr)
class ComponentSpec:
- def __init__(self, plugin_name, component_class_name, params=None):
+ def __init__(self, plugin_name, class_name, params=None):
utils._check_str(plugin_name)
- utils._check_str(component_class_name)
+ utils._check_str(class_name)
self._plugin_name = plugin_name
- self._component_class_name = component_class_name
+ self._class_name = class_name
if type(params) is str:
self._params = bt2.create_value({'paths': [params]})
return self._plugin_name
@property
- def component_class_name(self):
- return self._component_class_name
+ def class_name(self):
+ return self._class_name
@property
def params(self):
# contains the stream intersection range for each exposed
# trace
query_exec = bt2.QueryExecutor()
- trace_info_res = query_exec.query(src_comp_and_spec.comp.component_class,
+ trace_info_res = query_exec.query(src_comp_and_spec.comp.cls,
'trace-info', params)
begin = None
end = None
def _get_unique_comp_name(self, comp_spec):
name = '{}-{}'.format(comp_spec.plugin_name,
- comp_spec.component_class_name)
+ comp_spec.class_name)
comps_and_specs = itertools.chain(self._src_comps_and_specs,
self._flt_comps_and_specs)
else:
comp_classes = plugin.filter_component_classes
- if comp_spec.component_class_name not in comp_classes:
+ if comp_spec.class_name not in comp_classes:
cc_type = 'source' if comp_cls_type == _CompClsType.SOURCE else 'filter'
raise bt2.Error('no such {} component class in "{}" plugin: {}'.format(cc_type,
comp_spec.plugin_name,
- comp_spec.component_class_name))
+ comp_spec.class_name))
- comp_cls = comp_classes[comp_spec.component_class_name]
+ comp_cls = comp_classes[comp_spec.class_name]
name = self._get_unique_comp_name(comp_spec)
comp = self._graph.add_component(comp_cls, name, comp_spec.params)
return comp
def test_class(self):
class MySink(bt2._UserSinkComponent):
def __init__(comp_self, params):
- self.assertEqual(comp_self.component_class, MySink)
+ self.assertEqual(comp_self.cls, MySink)
def _consume(self):
pass
pass
comp = self._create_comp(MySink)
- self.assertEqual(comp.component_class, MySink)
+ self.assertEqual(comp.cls, MySink)
def test_addr(self):
class MySink(bt2._UserSinkComponent):
self._py_comp_cls = MySink
graph = bt2.Graph()
comp = graph.add_component(MySink, 'salut')
- self._comp_cls = comp.component_class
+ self._comp_cls = comp.cls
self.assertTrue(issubclass(type(self._comp_cls),
bt2.component._GenericComponentClass))
def test_attr_event_class(self):
msg = self._create_test_event_message()
- self.assertEqual(msg.event.event_class.addr, self.event_class.addr)
+ self.assertEqual(msg.event.cls.addr, self.event_class.addr)
def test_attr_name(self):
msg = self._create_test_event_message()
comp = self._graph.add_component(MySink, 'salut')
assert comp
- comp2 = self._graph.add_component(comp.component_class, 'salut2')
+ comp2 = self._graph.add_component(comp.cls, 'salut2')
self.assertEqual(comp2.name, 'salut2')
def test_add_component_params(self):
self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
elif comp_self._at >= 2 and comp_self._at <= 6:
self.assertIsInstance(msg, bt2.message._EventMessage)
- self.assertEqual(msg.event.event_class.name, 'salut')
+ self.assertEqual(msg.event.cls.name, 'salut')
elif comp_self._at == 7:
self.assertIsInstance(msg, bt2.message._PacketEndMessage)
elif comp_self._at == 8:
self.assertEqual(msg.default_clock_snapshot.value, i)
elif i == 3:
self.assertIsInstance(msg, bt2.message._EventMessage)
- self.assertEqual(msg.event.event_class.addr, self._event_class.addr)
+ self.assertEqual(msg.event.cls.addr, self._event_class.addr)
self.assertEqual(msg.default_clock_snapshot.value, i)
elif i == 4:
self.assertIsInstance(msg, bt2.message._MessageIteratorInactivityMessage)
self.assertIsInstance(msg, bt2.message._DiscardedEventsMessage)
self.assertEqual(msg.stream.addr, self._stream.addr)
self.assertEqual(msg.count, 890)
- self.assertEqual(msg.stream.stream_class.default_clock_class.addr, self._clock_class.addr)
+ self.assertEqual(msg.stream.cls.default_clock_class.addr, self._clock_class.addr)
self.assertEqual(msg.beginning_default_clock_snapshot.value, i)
self.assertEqual(msg.end_default_clock_snapshot.value, i)
elif i == 6:
self.assertIsInstance(msg, bt2.message._DiscardedPacketsMessage)
self.assertEqual(msg.stream.addr, self._stream.addr)
self.assertEqual(msg.count, 678)
- self.assertEqual(msg.stream.stream_class.default_clock_class.addr, self._clock_class.addr)
+ self.assertEqual(msg.stream.cls.default_clock_class.addr, self._clock_class.addr)
self.assertEqual(msg.beginning_default_clock_snapshot.value, i)
self.assertEqual(msg.end_default_clock_snapshot.value, i)
elif i == 8:
self.assertEqual(msg.packet.addr, self._packet.addr)
elif i == 3:
self.assertIsInstance(msg, bt2.message._EventMessage)
- self.assertEqual(msg.event.event_class.addr, self._event_class.addr)
+ self.assertEqual(msg.event.cls.addr, self._event_class.addr)
with self.assertRaises(bt2.NonexistentClockSnapshot):
msg.default_clock_snapshot
elif i == 4:
self.assertIsInstance(msg, bt2.message._DiscardedEventsMessage)
self.assertEqual(msg.stream.addr, self._stream.addr)
self.assertEqual(msg.count, 890)
- self.assertIsNone(msg.stream.stream_class.default_clock_class)
+ self.assertIsNone(msg.stream.cls.default_clock_class)
with self.assertRaises(bt2.NonexistentClockSnapshot):
msg.beginning_default_clock_snapshot
with self.assertRaises(bt2.NonexistentClockSnapshot):
self.assertIsInstance(msg, bt2.message._DiscardedPacketsMessage)
self.assertEqual(msg.stream.addr, self._stream.addr)
self.assertEqual(msg.count, 678)
- self.assertIsNone(msg.stream.stream_class.default_clock_class)
+ self.assertIsNone(msg.stream.cls.default_clock_class)
with self.assertRaises(bt2.NonexistentClockSnapshot):
msg.beginning_default_clock_snapshot
with self.assertRaises(bt2.NonexistentClockSnapshot):
self.assertIsInstance(msg, bt2.message._StreamEndMessage)
else:
self.assertIsInstance(msg, bt2.message._EventMessage)
- self.assertEqual(msg.event.event_class.name, 'salut')
+ self.assertEqual(msg.event.cls.name, 'salut')
field = msg.event.payload_field['my_int']
self.assertEqual(field, at * 3)
def test_stream_class(self):
stream = self._tr.create_stream(self._sc)
- self.assertEqual(stream.stream_class, self._sc)
+ self.assertEqual(stream.cls, self._sc)
def test_invalid_id(self):
sc = self._tc.create_stream_class(assigns_automatic_stream_id=False)