__version__ = '@PACKAGE_VERSION@'
+# import all public names
from bt2.clock_class import ClockClassOffset
from bt2.clock_snapshot import _ClockSnapshot
from bt2.clock_snapshot import _UnknownClockSnapshot
from bt2.value import MapValue
+def _del_global_name(name):
+ if name in globals():
+ del globals()[name]
+
+
+# remove private module names from the package
+_del_global_name('_native_bt')
+_del_global_name('clock_class')
+_del_global_name('clock_snapshot')
+_del_global_name('component')
+_del_global_name('connection')
+_del_global_name('error')
+_del_global_name('event')
+_del_global_name('event_class')
+_del_global_name('field')
+_del_global_name('field_class')
+_del_global_name('field_path')
+_del_global_name('graph')
+_del_global_name('integer_range_set')
+_del_global_name('interrupter')
+_del_global_name('logging')
+_del_global_name('message')
+_del_global_name('message_iterator')
+_del_global_name('native_bt')
+_del_global_name('object')
+_del_global_name('packet')
+_del_global_name('plugin')
+_del_global_name('port')
+_del_global_name('py_plugin')
+_del_global_name('query_executor')
+_del_global_name('stream')
+_del_global_name('stream_class')
+_del_global_name('trace')
+_del_global_name('trace_class')
+_del_global_name('trace_collection_message_iterator')
+_del_global_name('utils')
+_del_global_name('value')
+
+# remove private `_del_global_name` name from the package
+del _del_global_name
+
+
class _MemoryError(_Error):
'''Raised when an operation fails due to memory issues.'''
def _init_and_register_exit():
- import bt2.native_bt as _native_bt
+ from bt2 import native_bt
import atexit
- atexit.register(_native_bt.bt2_cc_exit_handler)
+ atexit.register(native_bt.bt2_cc_exit_handler)
version = (
- _native_bt.version_get_major(),
- _native_bt.version_get_minor(),
- _native_bt.version_get_patch(),
- _native_bt.version_get_extra(),
+ native_bt.version_get_major(),
+ native_bt.version_get_minor(),
+ native_bt.version_get_patch(),
+ native_bt.version_get_extra(),
)
- _native_bt.bt2_cc_init_from_bt2()
+ native_bt.bt2_cc_init_from_bt2()
_init_and_register_exit()
-
-try:
- del native_bt
-except:
- pass
+# remove private `_init_and_register_exit` name from the package
+del _init_and_register_exit
from bt2 import native_bt, object, utils
import numbers
import bt2
+from bt2 import clock_class as bt2_clock_class
import functools
def clock_class(self):
cc_ptr = native_bt.clock_snapshot_borrow_clock_class_const(self._ptr)
assert cc_ptr is not None
- return bt2.clock_class._ClockClass._create_from_ptr_and_get_ref(cc_ptr)
+ return bt2_clock_class._ClockClass._create_from_ptr_and_get_ref(cc_ptr)
@property
def value(self):
# THE SOFTWARE.
from bt2 import native_bt, object, utils
-import bt2.message_iterator
+from bt2 import message_iterator as bt2_message_iterator
import collections.abc
-import bt2.value
-import bt2.trace_class
+from bt2 import value as bt2_value
+from bt2 import trace_class as bt2_trace_class
+from bt2 import clock_class as bt2_clock_class
+from bt2 import query_executor as bt2_query_executor
import traceback
-import bt2.port
+from bt2 import port as bt2_port
import sys
import bt2
import os
native_bt.component_source_borrow_output_port_by_name_const,
native_bt.component_source_borrow_output_port_by_index_const,
native_bt.component_source_get_output_port_count,
- bt2.port._OutputPort,
+ bt2_port._OutputPort,
)
native_bt.component_filter_borrow_output_port_by_name_const,
native_bt.component_filter_borrow_output_port_by_index_const,
native_bt.component_filter_get_output_port_count,
- bt2.port._OutputPort,
+ bt2_port._OutputPort,
)
@property
native_bt.component_filter_borrow_input_port_by_name_const,
native_bt.component_filter_borrow_input_port_by_index_const,
native_bt.component_filter_get_input_port_count,
- bt2.port._InputPort,
+ bt2_port._InputPort,
)
native_bt.component_sink_borrow_input_port_by_name_const,
native_bt.component_sink_borrow_input_port_by_index_const,
native_bt.component_sink_get_input_port_count,
- bt2.port._InputPort,
+ bt2_port._InputPort,
)
# call user's __init__() method
if params_ptr is not None:
- params = bt2.value._create_from_ptr_and_get_ref(params_ptr)
+ params = bt2_value._create_from_ptr_and_get_ref(params_ptr)
else:
params = None
)
)
- if not issubclass(iter_cls, bt2.message_iterator._UserMessageIterator):
+ if not issubclass(iter_cls, bt2_message_iterator._UserMessageIterator):
raise bt2._IncompleteUserClass(
"cannot create component class '{}': message iterator class does not inherit bt2._UserMessageIterator".format(
cls.__name__
# this can raise, in which case the native call to
# bt_component_class_query() returns NULL
if params_ptr is not None:
- params = bt2.value._create_from_ptr_and_get_ref(params_ptr)
+ params = bt2_value._create_from_ptr_and_get_ref(params_ptr)
else:
params = None
- query_exec = bt2.QueryExecutor._create_from_ptr_and_get_ref(query_exec_ptr)
+ query_exec = bt2_query_executor.QueryExecutor._create_from_ptr_and_get_ref(
+ query_exec_ptr
+ )
# this can raise, but the native side checks the exception
results = cls._query(query_exec, obj, params, log_level)
results_ptr = results._ptr
# We return a new reference.
- bt2.value._Value._get_ref(results_ptr)
+ bt2_value._Value._get_ref(results_ptr)
return int(results_ptr)
def _bt_port_connected_from_native(
self, self_port_ptr, self_port_type, other_port_ptr
):
- port = bt2.port._create_self_from_ptr_and_get_ref(self_port_ptr, self_port_type)
+ port = bt2_port._create_self_from_ptr_and_get_ref(self_port_ptr, self_port_type)
if self_port_type == native_bt.PORT_TYPE_OUTPUT:
other_port_type = native_bt.PORT_TYPE_INPUT
else:
other_port_type = native_bt.PORT_TYPE_OUTPUT
- other_port = bt2.port._create_from_ptr_and_get_ref(
+ other_port = bt2_port._create_from_ptr_and_get_ref(
other_port_ptr, other_port_type
)
self._port_connected(port, other_port)
if tc_ptr is None:
raise bt2._MemoryError('could not create trace class')
- tc = bt2.trace_class._TraceClass._create_from_ptr(tc_ptr)
+ tc = bt2_trace_class._TraceClass._create_from_ptr(tc_ptr)
tc._assigns_automatic_stream_class_id = assigns_automatic_stream_class_id
return tc
if cc_ptr is None:
raise bt2._MemoryError('could not create clock class')
- cc = bt2.clock_class._ClockClass._create_from_ptr(cc_ptr)
+ cc = bt2_clock_class._ClockClass._create_from_ptr(cc_ptr)
if frequency is not None:
cc._frequency = frequency
native_bt.self_component_source_borrow_output_port_by_name,
native_bt.self_component_source_borrow_output_port_by_index,
get_output_port_count,
- bt2.port._UserComponentOutputPort,
+ bt2_port._UserComponentOutputPort,
)
def _add_output_port(self, name, user_data=None):
comp_status, 'cannot add output port to source component object'
)
assert self_port_ptr is not None
- return bt2.port._UserComponentOutputPort._create_from_ptr(self_port_ptr)
+ return bt2_port._UserComponentOutputPort._create_from_ptr(self_port_ptr)
class _UserFilterComponent(_UserComponent, _FilterComponent):
native_bt.self_component_filter_borrow_output_port_by_name,
native_bt.self_component_filter_borrow_output_port_by_index,
get_output_port_count,
- bt2.port._UserComponentOutputPort,
+ bt2_port._UserComponentOutputPort,
)
@property
native_bt.self_component_filter_borrow_input_port_by_name,
native_bt.self_component_filter_borrow_input_port_by_index,
get_input_port_count,
- bt2.port._UserComponentInputPort,
+ bt2_port._UserComponentInputPort,
)
def _add_output_port(self, name, user_data=None):
comp_status, 'cannot add output port to filter component object'
)
assert self_port_ptr
- return bt2.port._UserComponentOutputPort._create_from_ptr(self_port_ptr)
+ return bt2_port._UserComponentOutputPort._create_from_ptr(self_port_ptr)
def _add_input_port(self, name, user_data=None):
utils._check_str(name)
comp_status, 'cannot add input port to filter component object'
)
assert self_port_ptr
- return bt2.port._UserComponentInputPort._create_from_ptr(self_port_ptr)
+ return bt2_port._UserComponentInputPort._create_from_ptr(self_port_ptr)
class _UserSinkComponent(_UserComponent, _SinkComponent):
native_bt.self_component_sink_borrow_input_port_by_name,
native_bt.self_component_sink_borrow_input_port_by_index,
get_input_port_count,
- bt2.port._UserComponentInputPort,
+ bt2_port._UserComponentInputPort,
)
def _add_input_port(self, name, user_data=None):
comp_status, 'cannot add input port to sink component object'
)
assert self_port_ptr
- return bt2.port._UserComponentInputPort._create_from_ptr(self_port_ptr)
+ return bt2_port._UserComponentInputPort._create_from_ptr(self_port_ptr)
def _create_input_port_message_iterator(self, input_port):
- utils._check_type(input_port, bt2.port._UserComponentInputPort)
+ utils._check_type(input_port, bt2_port._UserComponentInputPort)
msg_iter_ptr = native_bt.self_component_port_input_message_iterator_create_from_sink_component(
self._bt_ptr, input_port._ptr
if msg_iter_ptr is None:
raise bt2.CreationError('cannot create message iterator object')
- return bt2.message_iterator._UserComponentInputPortMessageIterator(msg_iter_ptr)
+ return bt2_message_iterator._UserComponentInputPortMessageIterator(msg_iter_ptr)
@property
def _is_interrupted(self):
# THE SOFTWARE.
from bt2 import native_bt, utils
-import bt2.message_iterator
-import bt2.port
+from bt2 import message_iterator as bt2_message_iterator
+from bt2 import port as bt2_port
+from bt2 import object as bt2_object
import bt2
-class _Connection(bt2.object._SharedObject):
+class _Connection(bt2_object._SharedObject):
_get_ref = staticmethod(native_bt.connection_get_ref)
_put_ref = staticmethod(native_bt.connection_put_ref)
@property
def downstream_port(self):
port_ptr = native_bt.connection_borrow_downstream_port_const(self._ptr)
- return bt2.port._create_from_ptr_and_get_ref(
+ return bt2_port._create_from_ptr_and_get_ref(
port_ptr, native_bt.PORT_TYPE_INPUT
)
@property
def upstream_port(self):
port_ptr = native_bt.connection_borrow_upstream_port_const(self._ptr)
- return bt2.port._create_from_ptr_and_get_ref(
+ return bt2_port._create_from_ptr_and_get_ref(
port_ptr, native_bt.PORT_TYPE_OUTPUT
)
# THE SOFTWARE.
from bt2 import native_bt, object, utils
-import bt2.clock_class
-import bt2.event_class
-import bt2.packet
-import bt2.stream
-import bt2.field
-import bt2.clock_snapshot
+from bt2 import clock_class as bt2_clock_class
+from bt2 import event_class as bt2_event_class
+from bt2 import packet as bt2_packet
+from bt2 import stream as bt2_stream
+from bt2 import field as bt2_field
+from bt2 import clock_snapshot as bt2_clock_snapshot
import bt2
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)
+ return bt2_event_class._EventClass._create_from_ptr_and_get_ref(event_class_ptr)
@property
def name(self):
if packet_ptr is None:
return
- return bt2.packet._Packet._create_from_ptr_and_get_ref(packet_ptr)
+ return bt2_packet._Packet._create_from_ptr_and_get_ref(packet_ptr)
@property
def stream(self):
stream_ptr = native_bt.event_borrow_stream(self._ptr)
assert stream_ptr is not None
- return bt2.stream._Stream._create_from_ptr_and_get_ref(stream_ptr)
+ return bt2_stream._Stream._create_from_ptr_and_get_ref(stream_ptr)
@property
def common_context_field(self):
if field_ptr is None:
return
- return bt2.field._create_field_from_ptr(
+ return bt2_field._create_field_from_ptr(
field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
)
if field_ptr is None:
return
- return bt2.field._create_field_from_ptr(
+ return bt2_field._create_field_from_ptr(
field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
)
if field_ptr is None:
return
- return bt2.field._create_field_from_ptr(
+ return bt2_field._create_field_from_ptr(
field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
)
# THE SOFTWARE.
from bt2 import native_bt, object, utils
-import bt2.field_class
-import bt2.value
-import bt2.event
+from bt2 import field_class as bt2_field_class
+from bt2 import value as bt2_value
+from bt2 import event as bt2_event
+from bt2 import stream_class as bt2_stream_class
import bt2
sc_ptr = native_bt.event_class_borrow_stream_class(self._ptr)
if sc_ptr is not None:
- return bt2.stream_class._StreamClass._create_from_ptr_and_get_ref(sc_ptr)
+ return bt2_stream_class._StreamClass._create_from_ptr_and_get_ref(sc_ptr)
@property
def name(self):
if fc_ptr is None:
return
- return bt2.field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
+ 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)
+ utils._check_type(context_field_class, bt2_field_class._StructureFieldClass)
status = native_bt.event_class_set_specific_context_field_class(
self._ptr, context_field_class._ptr
)
if fc_ptr is None:
return
- return bt2.field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
+ 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)
+ utils._check_type(payload_field_class, bt2_field_class._StructureFieldClass)
status = native_bt.event_class_set_payload_field_class(
self._ptr, payload_field_class._ptr
)
# THE SOFTWARE.
from bt2 import native_bt, object, utils
-import bt2.field_class
+from bt2 import field_class as bt2_field_class
import collections.abc
import functools
import numbers
def field_class(self):
field_class_ptr = native_bt.field_borrow_class_const(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 bt2_field_class._create_field_class_from_ptr_and_get_ref(field_class_ptr)
def _repr(self):
raise NotImplementedError
from bt2 import native_bt, object, utils
import collections.abc
-import bt2.field
-import bt2.field_path
-import bt2.integer_range_set
+from bt2 import field as bt2_field
+from bt2 import field_path as bt2_field_path
+from bt2 import integer_range_set as bt2_integer_range_set
import bt2
class _UnsignedEnumerationFieldClassMapping(_EnumerationFieldClassMapping):
- _ranges_type = bt2.integer_range_set.UnsignedIntegerRangeSet
+ _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):
- _ranges_type = bt2.integer_range_set.SignedIntegerRangeSet
+ _ranges_type = bt2_integer_range_set.SignedIntegerRangeSet
_as_enumeration_field_class_mapping_ptr = staticmethod(
native_bt.field_class_enumeration_signed_mapping_as_mapping_const
)
_EnumerationFieldClass, _UnsignedIntegerFieldClass
):
_NAME = 'Unsigned enumeration'
- _range_set_type = bt2.integer_range_set.UnsignedIntegerRangeSet
+ _range_set_type = bt2_integer_range_set.UnsignedIntegerRangeSet
_add_mapping = staticmethod(native_bt.field_class_enumeration_unsigned_add_mapping)
@staticmethod
class _SignedEnumerationFieldClass(_EnumerationFieldClass, _SignedIntegerFieldClass):
_NAME = 'Signed enumeration'
- _range_set_type = bt2.integer_range_set.SignedIntegerRangeSet
+ _range_set_type = bt2_integer_range_set.SignedIntegerRangeSet
_add_mapping = staticmethod(native_bt.field_class_enumeration_signed_add_mapping)
@staticmethod
if ptr is None:
return
- return bt2.field_path._FieldPath._create_from_ptr_and_get_ref(ptr)
+ return bt2_field_path._FieldPath._create_from_ptr_and_get_ref(ptr)
def append_option(self, name, field_class, ranges):
utils._check_str(name)
_option_borrow_ranges_ptr = staticmethod(
native_bt.field_class_variant_with_selector_unsigned_option_borrow_ranges_const
)
- _range_set_type = bt2.integer_range_set.UnsignedIntegerRangeSet
+ _range_set_type = bt2_integer_range_set.UnsignedIntegerRangeSet
class _VariantFieldClassWithSignedSelector(_VariantFieldClassWithSelector):
_option_borrow_ranges_ptr = staticmethod(
native_bt.field_class_variant_with_selector_signed_option_borrow_ranges_const
)
- _range_set_type = bt2.integer_range_set.SignedIntegerRangeSet
+ _range_set_type = bt2_integer_range_set.SignedIntegerRangeSet
class _ArrayFieldClass(_FieldClass):
if ptr is None:
return
- return bt2.field_path._FieldPath._create_from_ptr_and_get_ref(ptr)
+ return bt2_field_path._FieldPath._create_from_ptr_and_get_ref(ptr)
_FIELD_CLASS_TYPE_TO_OBJ = {
# THE SOFTWARE.
from bt2 import native_bt, object, utils
-import bt2.interrupter
-import bt2.connection
-import bt2.component
+from bt2 import interrupter as bt2_interrupter
+from bt2 import connection as bt2_connection
+from bt2 import component as bt2_component
+from bt2 import message_iterator as bt2_message_iterator
import functools
-import bt2.port
-import bt2.logging
+from bt2 import port as bt2_port
+from bt2 import logging as bt2_logging
import bt2
def _graph_port_added_listener_from_native(
user_listener, component_ptr, component_type, port_ptr, port_type
):
- component = bt2.component._create_component_from_ptr_and_get_ref(
+ component = bt2_component._create_component_from_ptr_and_get_ref(
component_ptr, component_type
)
- port = bt2.port._create_from_ptr_and_get_ref(port_ptr, port_type)
+ port = bt2_port._create_from_ptr_and_get_ref(port_ptr, port_type)
user_listener(component, port)
downstream_component_type,
downstream_port_ptr,
):
- upstream_component = bt2.component._create_component_from_ptr_and_get_ref(
+ upstream_component = bt2_component._create_component_from_ptr_and_get_ref(
upstream_component_ptr, upstream_component_type
)
- upstream_port = bt2.port._create_from_ptr_and_get_ref(
+ upstream_port = bt2_port._create_from_ptr_and_get_ref(
upstream_port_ptr, native_bt.PORT_TYPE_OUTPUT
)
- downstream_component = bt2.component._create_component_from_ptr_and_get_ref(
+ downstream_component = bt2_component._create_component_from_ptr_and_get_ref(
downstream_component_ptr, downstream_component_type
)
- downstream_port = bt2.port._create_from_ptr_and_get_ref(
+ downstream_port = bt2_port._create_from_ptr_and_get_ref(
downstream_port_ptr, native_bt.PORT_TYPE_INPUT
)
user_listener(
component_class,
name,
params=None,
- logging_level=bt2.logging.LoggingLevel.NONE,
+ logging_level=bt2_logging.LoggingLevel.NONE,
):
- if isinstance(component_class, bt2.component._SourceComponentClass):
+ if isinstance(component_class, bt2_component._SourceComponentClass):
cc_ptr = component_class._ptr
add_fn = native_bt.graph_add_source_component
cc_type = native_bt.COMPONENT_CLASS_TYPE_SOURCE
- elif isinstance(component_class, bt2.component._FilterComponentClass):
+ elif isinstance(component_class, bt2_component._FilterComponentClass):
cc_ptr = component_class._ptr
add_fn = native_bt.graph_add_filter_component
cc_type = native_bt.COMPONENT_CLASS_TYPE_FILTER
- elif isinstance(component_class, bt2.component._SinkComponentClass):
+ elif isinstance(component_class, bt2_component._SinkComponentClass):
cc_ptr = component_class._ptr
add_fn = native_bt.graph_add_sink_component
cc_type = native_bt.COMPONENT_CLASS_TYPE_SINK
- elif issubclass(component_class, bt2.component._UserSourceComponent):
+ elif issubclass(component_class, bt2_component._UserSourceComponent):
cc_ptr = component_class._bt_cc_ptr
add_fn = native_bt.graph_add_source_component
cc_type = native_bt.COMPONENT_CLASS_TYPE_SOURCE
- elif issubclass(component_class, bt2.component._UserSinkComponent):
+ elif issubclass(component_class, bt2_component._UserSinkComponent):
cc_ptr = component_class._bt_cc_ptr
add_fn = native_bt.graph_add_sink_component
cc_type = native_bt.COMPONENT_CLASS_TYPE_SINK
- elif issubclass(component_class, bt2.component._UserFilterComponent):
+ elif issubclass(component_class, bt2_component._UserFilterComponent):
cc_ptr = component_class._bt_cc_ptr
add_fn = native_bt.graph_add_filter_component
cc_type = native_bt.COMPONENT_CLASS_TYPE_FILTER
status, comp_ptr = add_fn(self._ptr, cc_ptr, name, params_ptr, logging_level)
utils._handle_func_status(status, 'cannot add component to graph')
assert comp_ptr
- return bt2.component._create_component_from_ptr(comp_ptr, cc_type)
+ return bt2_component._create_component_from_ptr(comp_ptr, cc_type)
def connect_ports(self, upstream_port, downstream_port):
- utils._check_type(upstream_port, bt2.port._OutputPort)
- utils._check_type(downstream_port, bt2.port._InputPort)
+ utils._check_type(upstream_port, bt2_port._OutputPort)
+ utils._check_type(downstream_port, bt2_port._InputPort)
status, conn_ptr = native_bt.graph_connect_ports(
self._ptr, upstream_port._ptr, downstream_port._ptr
)
utils._handle_func_status(status, 'cannot connect component ports within graph')
assert conn_ptr
- return bt2.connection._Connection._create_from_ptr(conn_ptr)
+ return bt2_connection._Connection._create_from_ptr(conn_ptr)
def add_port_added_listener(self, listener):
if not callable(listener):
if listener_ids is None:
raise bt2._Error('cannot add listener to graph object')
- return bt2.utils._ListenerHandle(listener_ids, self)
+ return utils._ListenerHandle(listener_ids, self)
def add_ports_connected_listener(self, listener):
if not callable(listener):
if listener_ids is None:
raise bt2._Error('cannot add listener to graph object')
- return bt2.utils._ListenerHandle(listener_ids, self)
+ return utils._ListenerHandle(listener_ids, self)
def run(self):
status = native_bt.graph_run(self._ptr)
raise
def add_interrupter(self, interrupter):
- utils._check_type(interrupter, bt2.interrupter.Interrupter)
+ utils._check_type(interrupter, bt2_interrupter.Interrupter)
native_bt.graph_add_interrupter(self._ptr, interrupter._ptr)
def interrupt(self):
native_bt.graph_interrupt(self._ptr)
def create_output_port_message_iterator(self, output_port):
- utils._check_type(output_port, bt2.port._OutputPort)
+ utils._check_type(output_port, bt2_port._OutputPort)
msg_iter_ptr = native_bt.port_output_message_iterator_create(
self._ptr, output_port._ptr
)
if msg_iter_ptr is None:
raise bt2._MemoryError('cannot create output port message iterator')
- return bt2.message_iterator._OutputPortMessageIterator(msg_iter_ptr)
+ return bt2_message_iterator._OutputPortMessageIterator(msg_iter_ptr)
# THE SOFTWARE.
from bt2 import native_bt, object, utils
-import bt2.clock_snapshot
-import bt2.packet
-import bt2.stream
-import bt2.event
+from bt2 import clock_snapshot as bt2_clock_snapshot
+from bt2 import packet as bt2_packet
+from bt2 import stream as bt2_stream
+from bt2 import event as bt2_event
import bt2
def _get_default_clock_snapshot(self, borrow_clock_snapshot_ptr):
snapshot_ptr = borrow_clock_snapshot_ptr(self._ptr)
- return bt2.clock_snapshot._ClockSnapshot._create_from_ptr_and_get_ref(
+ return bt2_clock_snapshot._ClockSnapshot._create_from_ptr_and_get_ref(
snapshot_ptr, self._ptr, self._get_ref, self._put_ref
)
def event(self):
event_ptr = native_bt.message_event_borrow_event(self._ptr)
assert event_ptr is not None
- return bt2.event._Event._create_from_ptr_and_get_ref(
+ return bt2_event._Event._create_from_ptr_and_get_ref(
event_ptr, self._ptr, self._get_ref, self._put_ref
)
def packet(self):
packet_ptr = self._borrow_packet_ptr(self._ptr)
assert packet_ptr is not None
- return bt2.packet._Packet._create_from_ptr_and_get_ref(packet_ptr)
+ return bt2_packet._Packet._create_from_ptr_and_get_ref(packet_ptr)
class _PacketBeginningMessage(_PacketMessage):
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 bt2_stream._Stream._create_from_ptr_and_get_ref(stream_ptr)
@property
def default_clock_snapshot(self):
status, snapshot_ptr = self._borrow_default_clock_snapshot_ptr(self._ptr)
if status == native_bt.MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_UNKNOWN:
- return bt2.clock_snapshot._UnknownClockSnapshot()
+ return bt2_clock_snapshot._UnknownClockSnapshot()
- return bt2.clock_snapshot._ClockSnapshot._create_from_ptr_and_get_ref(
+ return bt2_clock_snapshot._ClockSnapshot._create_from_ptr_and_get_ref(
snapshot_ptr, self._ptr, self._get_ref, self._put_ref
)
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 bt2_stream._Stream._create_from_ptr_and_get_ref(stream_ptr)
@property
def count(self):
# THE SOFTWARE.
from bt2 import native_bt, object, utils
-import bt2.message
+from bt2 import message as bt2_message
import collections.abc
-import bt2.component
-import bt2.port
+from bt2 import component as bt2_component
+from bt2 import stream as bt2_stream
+from bt2 import event_class as bt2_event_class
+from bt2 import packet as bt2_packet
+from bt2 import port as bt2_port
+from bt2 import clock_class as bt2_clock_class
import bt2
msg_ptr = self._current_msgs[self._at]
self._at += 1
- return bt2.message._create_from_ptr(msg_ptr)
+ return bt2_message._create_from_ptr(msg_ptr)
@property
def can_seek_beginning(self):
return self
def _bt_init_from_native(self, self_output_port_ptr):
- self_output_port = bt2.port._create_self_from_ptr_and_get_ref(
+ self_output_port = bt2_port._create_self_from_ptr_and_get_ref(
self_output_port_ptr, native_bt.PORT_TYPE_OUTPUT
)
self.__init__(self_output_port)
except:
raise
- utils._check_type(msg, bt2.message._Message)
+ utils._check_type(msg, bt2_message._Message)
# The reference we return will be given to the message array.
# However, the `msg` Python object may stay alive, if the user has kept
self._seek_beginning()
def _create_input_port_message_iterator(self, input_port):
- utils._check_type(input_port, bt2.port._UserComponentInputPort)
+ utils._check_type(input_port, bt2_port._UserComponentInputPort)
msg_iter_ptr = native_bt.self_component_port_input_message_iterator_create_from_message_iterator(
self._bt_ptr, input_port._ptr
def _create_event_message(
self, event_class, parent=None, default_clock_snapshot=None
):
- utils._check_type(event_class, bt2.event_class._EventClass)
+ utils._check_type(event_class, bt2_event_class._EventClass)
if event_class.stream_class.supports_packets:
- utils._check_type(parent, bt2.packet._Packet)
+ utils._check_type(parent, bt2_packet._Packet)
else:
- utils._check_type(parent, bt2.stream._Stream)
+ utils._check_type(parent, bt2_stream._Stream)
if default_clock_snapshot is not None:
if event_class.stream_class.default_clock_class is None:
if ptr is None:
raise bt2._MemoryError('cannot create event message object')
- return bt2.message._EventMessage(ptr)
+ return bt2_message._EventMessage(ptr)
def _create_message_iterator_inactivity_message(self, clock_class, clock_snapshot):
- utils._check_type(clock_class, bt2.clock_class._ClockClass)
+ utils._check_type(clock_class, bt2_clock_class._ClockClass)
ptr = native_bt.message_message_iterator_inactivity_create(
self._bt_ptr, clock_class._ptr, clock_snapshot
)
if ptr is None:
raise bt2._MemoryError('cannot create inactivity message object')
- return bt2.message._MessageIteratorInactivityMessage(ptr)
+ return bt2_message._MessageIteratorInactivityMessage(ptr)
def _create_stream_beginning_message(self, stream, default_clock_snapshot=None):
- utils._check_type(stream, bt2.stream._Stream)
+ utils._check_type(stream, bt2_stream._Stream)
ptr = native_bt.message_stream_beginning_create(self._bt_ptr, stream._ptr)
if ptr is None:
raise bt2._MemoryError('cannot create stream beginning message object')
- msg = bt2.message._StreamBeginningMessage(ptr)
+ msg = bt2_message._StreamBeginningMessage(ptr)
if default_clock_snapshot is not None:
msg._default_clock_snapshot = default_clock_snapshot
return msg
def _create_stream_end_message(self, stream, default_clock_snapshot=None):
- utils._check_type(stream, bt2.stream._Stream)
+ utils._check_type(stream, bt2_stream._Stream)
ptr = native_bt.message_stream_end_create(self._bt_ptr, stream._ptr)
if ptr is None:
raise bt2._MemoryError('cannot create stream end message object')
- msg = bt2.message._StreamEndMessage(ptr)
+ msg = bt2_message._StreamEndMessage(ptr)
if default_clock_snapshot is not None:
msg._default_clock_snapshot = default_clock_snapshot
return msg
def _create_packet_beginning_message(self, packet, default_clock_snapshot=None):
- utils._check_type(packet, bt2.packet._Packet)
+ utils._check_type(packet, bt2_packet._Packet)
if packet.stream.cls.packets_have_beginning_default_clock_snapshot:
if default_clock_snapshot is None:
if ptr is None:
raise bt2._MemoryError('cannot create packet beginning message object')
- return bt2.message._PacketBeginningMessage(ptr)
+ return bt2_message._PacketBeginningMessage(ptr)
def _create_packet_end_message(self, packet, default_clock_snapshot=None):
- utils._check_type(packet, bt2.packet._Packet)
+ utils._check_type(packet, bt2_packet._Packet)
if packet.stream.cls.packets_have_end_default_clock_snapshot:
if default_clock_snapshot is None:
if ptr is None:
raise bt2._MemoryError('cannot create packet end message object')
- return bt2.message._PacketEndMessage(ptr)
+ return bt2_message._PacketEndMessage(ptr)
def _create_discarded_events_message(
self, stream, count=None, beg_clock_snapshot=None, end_clock_snapshot=None
):
- utils._check_type(stream, bt2.stream._Stream)
+ utils._check_type(stream, bt2_stream._Stream)
if not stream.cls.supports_discarded_events:
raise ValueError('stream class does not support discarded events')
if ptr is None:
raise bt2._MemoryError('cannot discarded events message object')
- msg = bt2.message._DiscardedEventsMessage(ptr)
+ msg = bt2_message._DiscardedEventsMessage(ptr)
if count is not None:
msg._count = count
def _create_discarded_packets_message(
self, stream, count=None, beg_clock_snapshot=None, end_clock_snapshot=None
):
- utils._check_type(stream, bt2.stream._Stream)
+ utils._check_type(stream, bt2_stream._Stream)
if not stream.cls.supports_discarded_packets:
raise ValueError('stream class does not support discarded packets')
if ptr is None:
raise bt2._MemoryError('cannot discarded packets message object')
- msg = bt2.message._DiscardedPacketsMessage(ptr)
+ msg = bt2_message._DiscardedPacketsMessage(ptr)
if count is not None:
msg._count = count
# THE SOFTWARE.
from bt2 import native_bt, object
-import bt2.field
+from bt2 import field as bt2_field
+from bt2 import stream as bt2_stream
import bt2
def stream(self):
stream_ptr = native_bt.packet_borrow_stream(self._ptr)
assert stream_ptr is not None
- return bt2.stream._Stream._create_from_ptr_and_get_ref(stream_ptr)
+ return bt2_stream._Stream._create_from_ptr_and_get_ref(stream_ptr)
@property
def context_field(self):
if field_ptr is None:
return
- return bt2.field._create_field_from_ptr(
+ return bt2_field._create_field_from_ptr(
field_ptr, self._ptr, self._get_ref, self._put_ref
)
from bt2 import native_bt, object, utils
import collections.abc
-import bt2.component
+from bt2 import component as bt2_component
import os.path
import bt2
self._at += 1
comp_cls_type = self._plugin_comp_cls._comp_cls_type
- comp_cls_pycls = bt2.component._COMP_CLS_TYPE_TO_GENERIC_COMP_CLS_PYCLS[
+ comp_cls_pycls = bt2_component._COMP_CLS_TYPE_TO_GENERIC_COMP_CLS_PYCLS[
comp_cls_type
]
comp_cls_ptr = comp_cls_pycls._bt_as_component_class_ptr(comp_cls_ptr)
if cc_ptr is None:
raise KeyError(key)
- return bt2.component._create_component_class_from_ptr_and_get_ref(
+ return bt2_component._create_component_class_from_ptr_and_get_ref(
cc_ptr, self._comp_cls_type
)
# THE SOFTWARE.
from bt2 import native_bt, object
-import bt2.component
-import bt2.connection
-import bt2.message_iterator
-import bt2.message
+from bt2 import component as bt2_component
+from bt2 import connection as bt2_connection
+from bt2 import message_iterator as bt2_message_iterator
+from bt2 import message as bt2_message
import bt2
if conn_ptr is None:
return
- return bt2.connection._Connection._create_from_ptr_and_get_ref(conn_ptr)
+ return bt2_connection._Connection._create_from_ptr_and_get_ref(conn_ptr)
@property
def is_connected(self):
if conn_ptr is None:
return
- return bt2.connection._Connection._create_from_ptr_and_get_ref(conn_ptr)
+ return bt2_connection._Connection._create_from_ptr_and_get_ref(conn_ptr)
@property
def user_data(self):
# THE SOFTWARE.
from bt2 import utils
-import bt2.component
+from bt2 import component as bt2_component
def plugin_component_class(component_class):
- if not issubclass(component_class, bt2.component._UserComponent):
+ if not issubclass(component_class, bt2_component._UserComponent):
raise TypeError('component class is not a subclass of a user component class')
component_class._bt_plugin_component_class = None
# THE SOFTWARE.
from bt2 import native_bt, object, utils
-import bt2.interrupter
-import bt2.component
-import bt2.logging
+from bt2 import interrupter as bt2_interrupter
+from bt2 import component as bt2_component
+from bt2 import logging as bt2_logging
+from bt2 import value as bt2_value
import bt2
super().__init__(ptr)
def add_interrupter(self, interrupter):
- utils._check_type(interrupter, bt2.interrupter.Interrupter)
+ utils._check_type(interrupter, bt2_interrupter.Interrupter)
native_bt.query_executor_add_interrupter(self._ptr, interrupter._ptr)
def interrupt(self):
component_class,
object,
params=None,
- logging_level=bt2.logging.LoggingLevel.NONE,
+ logging_level=bt2_logging.LoggingLevel.NONE,
):
- if not isinstance(component_class, bt2.component._ComponentClass):
+ if not isinstance(component_class, bt2_component._ComponentClass):
err = False
try:
- if not issubclass(component_class, bt2.component._UserComponent):
+ if not issubclass(component_class, bt2_component._UserComponent):
err = True
except TypeError:
err = True
)
utils._handle_func_status(status, 'cannot query component class')
assert result_ptr
- return bt2.value._create_from_ptr(result_ptr)
+ return bt2_value._create_from_ptr(result_ptr)
# THE SOFTWARE.
from bt2 import native_bt, utils
-import bt2.packet
-import bt2.event
+from bt2 import object as bt2_object
+from bt2 import packet as bt2_packet
+from bt2 import event as bt2_event
+from bt2 import stream_class as bt2_stream_class
import bt2
-class _Stream(bt2.object._SharedObject):
+class _Stream(bt2_object._SharedObject):
_get_ref = staticmethod(native_bt.stream_get_ref)
_put_ref = staticmethod(native_bt.stream_put_ref)
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(
+ return bt2_stream_class._StreamClass._create_from_ptr_and_get_ref(
stream_class_ptr
)
if packet_ptr is None:
raise bt2._MemoryError('cannot create packet object')
- return bt2.packet._Packet._create_from_ptr(packet_ptr)
+ return bt2_packet._Packet._create_from_ptr(packet_ptr)
# THE SOFTWARE.
from bt2 import native_bt, object, utils
-import bt2.field_class
-import bt2.event_class
-import bt2.trace_class
+from bt2 import field_class as bt2_field_class
+from bt2 import event_class as bt2_event_class
+from bt2 import trace_class as bt2_trace_class
+from bt2 import clock_class as bt2_clock_class
import collections.abc
-import bt2.stream
+from bt2 import stream as bt2_stream
import bt2
if ec_ptr is None:
raise KeyError(key)
- return bt2.event_class._EventClass._create_from_ptr_and_get_ref(ec_ptr)
+ return bt2_event_class._EventClass._create_from_ptr_and_get_ref(ec_ptr)
def __len__(self):
count = native_bt.stream_class_get_event_class_count(self._ptr)
utils._check_uint64(id)
ec_ptr = native_bt.event_class_create_with_id(self._ptr, id)
- event_class = bt2.event_class._EventClass._create_from_ptr(ec_ptr)
+ event_class = bt2_event_class._EventClass._create_from_ptr(ec_ptr)
if name is not None:
event_class._name = name
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)
+ return bt2_trace_class._TraceClass._create_from_ptr_and_get_ref(tc_ptr)
@property
def name(self):
if fc_ptr is None:
return
- return bt2.field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
+ return bt2_field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
def _packet_context_field_class(self, packet_context_field_class):
if packet_context_field_class is not None:
utils._check_type(
- packet_context_field_class, bt2.field_class._StructureFieldClass
+ packet_context_field_class, bt2_field_class._StructureFieldClass
)
if not self.supports_packets:
if fc_ptr is None:
return
- return bt2.field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
+ 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, bt2.field_class._StructureFieldClass
+ event_common_context_field_class, bt2_field_class._StructureFieldClass
)
set_context_fn = native_bt.stream_class_set_event_common_context_field_class
if cc_ptr is None:
return
- return bt2.clock_class._ClockClass._create_from_ptr_and_get_ref(cc_ptr)
+ 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)
+ utils._check_type(clock_class, bt2_clock_class._ClockClass)
native_bt.stream_class_set_default_clock_class(self._ptr, clock_class._ptr)
_default_clock_class = property(fset=_default_clock_class)
# THE SOFTWARE.
from bt2 import native_bt, object, utils
-import bt2.field_class
+from bt2 import field_class as bt2_field_class
import collections.abc
-import bt2.value
-import bt2.stream
-import bt2.trace_class
+from bt2 import value as bt2_value
+from bt2 import stream as bt2_stream
+from bt2 import trace_class as bt2_trace_class
+from bt2 import stream_class as bt2_stream_class
import bt2
import functools
import uuid as uuidp
if value_ptr is None:
raise KeyError(key)
- return bt2.value._create_from_ptr_and_get_ref(value_ptr)
+ return bt2_value._create_from_ptr_and_get_ref(value_ptr)
def __setitem__(self, key, value):
if isinstance(value, str):
def _trace_destruction_listener_from_native(user_listener, trace_ptr):
- trace = bt2.trace._Trace._create_from_ptr_and_get_ref(trace_ptr)
+ trace = _Trace._create_from_ptr_and_get_ref(trace_ptr)
user_listener(trace)
if stream_ptr is None:
raise KeyError(id)
- return bt2.stream._Stream._create_from_ptr_and_get_ref(stream_ptr)
+ return bt2_stream._Stream._create_from_ptr_and_get_ref(stream_ptr)
def __iter__(self):
for idx in range(len(self)):
def cls(self):
trace_class_ptr = native_bt.trace_borrow_class(self._ptr)
assert trace_class_ptr is not None
- return bt2.trace_class._TraceClass._create_from_ptr_and_get_ref(trace_class_ptr)
+ return bt2_trace_class._TraceClass._create_from_ptr_and_get_ref(trace_class_ptr)
@property
def name(self):
return _TraceEnv(self)
def create_stream(self, stream_class, id=None, name=None):
- utils._check_type(stream_class, bt2.stream_class._StreamClass)
+ utils._check_type(stream_class, bt2_stream_class._StreamClass)
if stream_class.assigns_automatic_stream_id:
if id is not None:
if stream_ptr is None:
raise bt2._MemoryError('cannot create stream object')
- stream = bt2.stream._Stream._create_from_ptr(stream_ptr)
+ stream = bt2_stream._Stream._create_from_ptr(stream_ptr)
if name is not None:
stream._name = name
status, 'cannot add destruction listener to trace object'
)
- return bt2.utils._ListenerHandle(listener_id, self)
+ return utils._ListenerHandle(listener_id, self)
import bt2
from bt2 import native_bt, utils, object
-import bt2.stream_class
-import bt2.field_class
-import bt2.trace
+from bt2 import stream_class as bt2_stream_class
+from bt2 import field_class as bt2_field_class
+from bt2 import trace as bt2_trace
+from bt2 import trace_class as bt2_trace_class
import collections.abc
import functools
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 = bt2_trace_class._TraceClass._create_from_ptr_and_get_ref(
trace_class_ptr
)
user_listener(trace_class)
if trace_ptr is None:
raise bt2._MemoryError('cannot create trace class object')
- trace = bt2.trace._Trace._create_from_ptr(trace_ptr)
+ trace = bt2_trace._Trace._create_from_ptr(trace_ptr)
if name is not None:
trace._name = name
if sc_ptr is None:
raise KeyError(key)
- return bt2.stream_class._StreamClass._create_from_ptr_and_get_ref(sc_ptr)
+ return bt2_stream_class._StreamClass._create_from_ptr_and_get_ref(sc_ptr)
def __iter__(self):
for idx in range(len(self)):
utils._check_uint64(id)
sc_ptr = native_bt.stream_class_create_with_id(self._ptr, id)
- sc = bt2.stream_class._StreamClass._create_from_ptr(sc_ptr)
+ sc = bt2_stream_class._StreamClass._create_from_ptr(sc_ptr)
if name is not None:
sc._name = name
):
return self._create_integer_field_class(
native_bt.field_class_integer_signed_create,
- bt2.field_class._SignedIntegerFieldClass,
+ bt2_field_class._SignedIntegerFieldClass,
'signed integer',
field_value_range,
preferred_display_base,
):
return self._create_integer_field_class(
native_bt.field_class_integer_unsigned_create,
- bt2.field_class._UnsignedIntegerFieldClass,
+ bt2_field_class._UnsignedIntegerFieldClass,
'unsigned integer',
field_value_range,
preferred_display_base,
):
return self._create_integer_field_class(
native_bt.field_class_enumeration_signed_create,
- bt2.field_class._SignedEnumerationFieldClass,
+ bt2_field_class._SignedEnumerationFieldClass,
'signed enumeration',
field_value_range,
preferred_display_base,
):
return self._create_integer_field_class(
native_bt.field_class_enumeration_unsigned_create,
- bt2.field_class._UnsignedEnumerationFieldClass,
+ bt2_field_class._UnsignedEnumerationFieldClass,
'unsigned enumeration',
field_value_range,
preferred_display_base,
field_class_ptr = native_bt.field_class_real_create(self._ptr)
self._check_create_status(field_class_ptr, 'real')
- field_class = bt2.field_class._RealFieldClass._create_from_ptr(field_class_ptr)
+ field_class = bt2_field_class._RealFieldClass._create_from_ptr(field_class_ptr)
field_class._is_single_precision = is_single_precision
field_class_ptr = native_bt.field_class_structure_create(self._ptr)
self._check_create_status(field_class_ptr, 'structure')
- return bt2.field_class._StructureFieldClass._create_from_ptr(field_class_ptr)
+ return bt2_field_class._StructureFieldClass._create_from_ptr(field_class_ptr)
def create_string_field_class(self):
field_class_ptr = native_bt.field_class_string_create(self._ptr)
self._check_create_status(field_class_ptr, 'string')
- return bt2.field_class._StringFieldClass._create_from_ptr(field_class_ptr)
+ return bt2_field_class._StringFieldClass._create_from_ptr(field_class_ptr)
def create_static_array_field_class(self, elem_fc, length):
- utils._check_type(elem_fc, bt2.field_class._FieldClass)
+ utils._check_type(elem_fc, bt2_field_class._FieldClass)
utils._check_uint64(length)
ptr = native_bt.field_class_array_static_create(self._ptr, elem_fc._ptr, length)
self._check_create_status(ptr, 'static array')
- return bt2.field_class._StaticArrayFieldClass._create_from_ptr_and_get_ref(ptr)
+ return bt2_field_class._StaticArrayFieldClass._create_from_ptr_and_get_ref(ptr)
def create_dynamic_array_field_class(self, elem_fc, length_fc=None):
- utils._check_type(elem_fc, bt2.field_class._FieldClass)
+ utils._check_type(elem_fc, bt2_field_class._FieldClass)
length_fc_ptr = None
if length_fc is not None:
- utils._check_type(length_fc, bt2.field_class._UnsignedIntegerFieldClass)
+ utils._check_type(length_fc, bt2_field_class._UnsignedIntegerFieldClass)
length_fc_ptr = length_fc._ptr
ptr = native_bt.field_class_array_dynamic_create(
self._ptr, elem_fc._ptr, length_fc_ptr
)
self._check_create_status(ptr, 'dynamic array')
- return bt2.field_class._DynamicArrayFieldClass._create_from_ptr(ptr)
+ return bt2_field_class._DynamicArrayFieldClass._create_from_ptr(ptr)
def create_variant_field_class(self, selector_fc=None):
selector_fc_ptr = None
if selector_fc is not None:
- utils._check_type(selector_fc, bt2.field_class._IntegerFieldClass)
+ utils._check_type(selector_fc, bt2_field_class._IntegerFieldClass)
selector_fc_ptr = selector_fc._ptr
ptr = native_bt.field_class_variant_create(self._ptr, selector_fc_ptr)
self._check_create_status(ptr, 'variant')
- return bt2.field_class._create_field_class_from_ptr_and_get_ref(ptr)
+ return bt2_field_class._create_field_class_from_ptr_and_get_ref(ptr)
# Add a listener to be called when the trace class is destroyed.
status, 'cannot add destruction listener to trace class object'
)
- return bt2.utils._ListenerHandle(listener_id, self)
+ return utils._ListenerHandle(listener_id, self)
from bt2 import utils
import bt2
import itertools
-import bt2.message_iterator
+from bt2 import message_iterator as bt2_message_iterator
+from bt2 import logging as bt2_logging
+from bt2 import port as bt2_port
import datetime
from collections import namedtuple
import numbers
plugin_name,
class_name,
params=None,
- logging_level=bt2.logging.LoggingLevel.NONE,
+ logging_level=bt2_logging.LoggingLevel.NONE,
):
utils._check_str(plugin_name)
utils._check_str(class_name)
FILTER = 1
-class TraceCollectionMessageIterator(bt2.message_iterator._MessageIterator):
+class TraceCollectionMessageIterator(bt2_message_iterator._MessageIterator):
def __init__(
self,
source_component_specs,
if not self._connect_ports:
return
- if type(port) is bt2.port._InputPort:
+ if type(port) is bt2_port._InputPort:
return
if component not in [comp.comp for comp in self._src_comps_and_specs]:
# THE SOFTWARE.
import bt2
-import bt2.logging
+from bt2 import logging as bt2_logging
from bt2 import native_bt
_check_int(log_level)
log_levels = (
- bt2.logging.LoggingLevel.TRACE,
- bt2.logging.LoggingLevel.DEBUG,
- bt2.logging.LoggingLevel.INFO,
- bt2.logging.LoggingLevel.WARNING,
- bt2.logging.LoggingLevel.ERROR,
- bt2.logging.LoggingLevel.FATAL,
- bt2.logging.LoggingLevel.NONE,
+ bt2_logging.LoggingLevel.TRACE,
+ bt2_logging.LoggingLevel.DEBUG,
+ bt2_logging.LoggingLevel.INFO,
+ bt2_logging.LoggingLevel.WARNING,
+ bt2_logging.LoggingLevel.ERROR,
+ bt2_logging.LoggingLevel.FATAL,
+ bt2_logging.LoggingLevel.NONE,
)
if log_level not in log_levels:
# to it that we are not going to manage anymore, since we don't create a
# Python wrapper for it. Therefore put that reference immediately.
if ptr == native_bt.value_null:
- bt2.value._Value._put_ref(ptr)
+ _Value._put_ref(ptr)
return
typeid = native_bt.value_get_type(ptr)
graph = bt2.Graph()
comp = graph.add_component(MySink, 'salut')
self._comp_cls = comp.cls
- self.assertTrue(issubclass(type(self._comp_cls), bt2.component._ComponentClass))
+ self.assertTrue(issubclass(type(self._comp_cls), bt2._SinkComponentClass))
def tearDown(self):
del self._py_comp_cls
for c in exc:
# Each cause is an instance of _ErrorCause (including subclasses).
- self.assertIsInstance(c, bt2.error._ErrorCause)
+ self.assertIsInstance(c, bt2._ErrorCause)
def test_getitem(self):
exc = self._run_failing_graph(SourceWithFailingIter, WorkingSink)
for i in range(len(exc)):
c = exc[i]
# Each cause is an instance of _ErrorCause (including subclasses).
- self.assertIsInstance(c, bt2.error._ErrorCause)
+ self.assertIsInstance(c, bt2._ErrorCause)
def test_getitem_indexerror(self):
exc = self._run_failing_graph(SourceWithFailingIter, WorkingSink)
self.assertEqual(len(exc), 5)
- self.assertIsInstance(exc[0], bt2.error._MessageIteratorErrorCause)
+ self.assertIsInstance(exc[0], bt2._MessageIteratorErrorCause)
self.assertEqual(exc[0].component_class_name, 'SourceWithFailingIter')
self.assertIn('ValueError: User message iterator is failing', exc[0].message)
- self.assertIsInstance(exc[1], bt2.error._ErrorCause)
+ self.assertIsInstance(exc[1], bt2._ErrorCause)
- self.assertIsInstance(exc[2], bt2.error._ComponentErrorCause)
+ self.assertIsInstance(exc[2], bt2._ComponentErrorCause)
self.assertEqual(exc[2].component_class_name, 'SinkWithExceptionChaining')
self.assertIn(
- 'bt2.error._Error: unexpected error: cannot advance the message iterator',
- exc[2].message,
+ 'unexpected error: cannot advance the message iterator', exc[2].message
)
- self.assertIsInstance(exc[3], bt2.error._ComponentErrorCause)
+ self.assertIsInstance(exc[3], bt2._ComponentErrorCause)
self.assertEqual(exc[3].component_class_name, 'SinkWithExceptionChaining')
self.assertIn('ValueError: oops', exc[3].message)
- self.assertIsInstance(exc[4], bt2.error._ErrorCause)
+ self.assertIsInstance(exc[4], bt2._ErrorCause)
def _common_cause_tests(self, cause):
self.assertIsInstance(cause.module_name, str)
def test_unknown_error_cause(self):
exc = self._run_failing_graph(SourceWithFailingIter, SinkWithExceptionChaining)
cause = exc[-1]
- self.assertIs(type(cause), bt2.error._ErrorCause)
+ self.assertIs(type(cause), bt2._ErrorCause)
self._common_cause_tests(cause)
def test_component_error_cause(self):
exc = self._run_failing_graph(SourceWithFailingInit, SinkWithExceptionChaining)
cause = exc[0]
- self.assertIs(type(cause), bt2.error._ComponentErrorCause)
+ self.assertIs(type(cause), bt2._ComponentErrorCause)
self._common_cause_tests(cause)
self.assertIn('Source is failing', cause.message)
q.query(SinkWithFailingQuery, 'hello')
cause = ctx.exception[0]
- self.assertIs(type(cause), bt2.error._ComponentClassErrorCause)
+ self.assertIs(type(cause), bt2._ComponentClassErrorCause)
self._common_cause_tests(cause)
self.assertIn('Query is failing', cause.message)
def test_message_iterator_error_cause(self):
exc = self._run_failing_graph(SourceWithFailingIter, SinkWithExceptionChaining)
cause = exc[0]
- self.assertIs(type(cause), bt2.error._MessageIteratorErrorCause)
+ self.assertIs(type(cause), bt2._MessageIteratorErrorCause)
self._common_cause_tests(cause)
self.assertIn('User message iterator is failing', cause.message)
def test_getitem(self):
field = self._def[1]
- self.assertIs(type(field), bt2.field._SignedIntegerField)
+ self.assertIs(type(field), bt2._SignedIntegerField)
self.assertEqual(field, 1847)
def test_eq(self):
def test_getitem(self):
field = self._def['A']
- self.assertIs(type(field), bt2.field._SignedIntegerField)
+ self.assertIs(type(field), bt2._SignedIntegerField)
self.assertEqual(field, -1872)
def test_member_at_index_out_of_bounds_after(self):
class StructureFieldClassTestCase(_TestElementContainer, unittest.TestCase):
- _append_element_method = staticmethod(
- bt2.field_class._StructureFieldClass.append_member
- )
- _at_index_method = staticmethod(
- bt2.field_class._StructureFieldClass.member_at_index
- )
+ _append_element_method = staticmethod(bt2._StructureFieldClass.append_member)
+ _at_index_method = staticmethod(bt2._StructureFieldClass.member_at_index)
def _create_default_fc(self):
return self._tc.create_structure_field_class()
_TestElementContainer, unittest.TestCase
):
_append_element_method = staticmethod(
- bt2.field_class._VariantFieldClassWithoutSelector.append_option
+ bt2._VariantFieldClassWithoutSelector.append_option
)
_at_index_method = staticmethod(
- bt2.field_class._VariantFieldClassWithoutSelector.option_at_index
+ bt2._VariantFieldClassWithoutSelector.option_at_index
)
def _create_default_fc(self):
self.assertEqual(len(path_items), 3)
- self.assertIsInstance(path_items[0], bt2.field_path._IndexFieldPathItem)
+ self.assertIsInstance(path_items[0], bt2._IndexFieldPathItem)
self.assertEqual(path_items[0].index, 1)
- self.assertIsInstance(
- path_items[1], bt2.field_path._CurrentArrayElementFieldPathItem
- )
+ self.assertIsInstance(path_items[1], bt2._CurrentArrayElementFieldPathItem)
- self.assertIsInstance(path_items[2], bt2.field_path._IndexFieldPathItem)
+ self.assertIsInstance(path_items[2], bt2._IndexFieldPathItem)
self.assertEqual(path_items[2].index, 0)
def test_selector_field_path_root_scope(self):
self._fill_default_fc_for_field_path_test()
self.assertEqual(
- self._fc.selector_field_path.root_scope, bt2.field_path.Scope.PACKET_CONTEXT
+ self._fc.selector_field_path.root_scope, bt2.Scope.PACKET_CONTEXT
)
self.assertEqual(len(path_items), 3)
- self.assertIsInstance(path_items[0], bt2.field_path._IndexFieldPathItem)
+ self.assertIsInstance(path_items[0], bt2._IndexFieldPathItem)
self.assertEqual(path_items[0].index, 1)
- self.assertIsInstance(
- path_items[1], bt2.field_path._CurrentArrayElementFieldPathItem
- )
+ self.assertIsInstance(path_items[1], bt2._CurrentArrayElementFieldPathItem)
- self.assertIsInstance(path_items[2], bt2.field_path._IndexFieldPathItem)
+ self.assertIsInstance(path_items[2], bt2._IndexFieldPathItem)
self.assertEqual(path_items[2].index, 2)
def test_field_path_root_scope(self):
fc = self._create_field_class_for_field_path_test()
- self.assertEqual(
- fc.length_field_path.root_scope, bt2.field_path.Scope.PACKET_CONTEXT
- )
+ self.assertEqual(fc.length_field_path.root_scope, bt2.Scope.PACKET_CONTEXT)
def test_create_invalid_field_class(self):
with self.assertRaises(TypeError):
msg = next(comp_self._msg_iter)
if comp_self._at == 0:
- self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
+ self.assertIsInstance(msg, bt2._StreamBeginningMessage)
elif comp_self._at == 1:
- self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
+ self.assertIsInstance(msg, bt2._PacketBeginningMessage)
elif comp_self._at >= 2 and comp_self._at <= 6:
- self.assertIsInstance(msg, bt2.message._EventMessage)
+ self.assertIsInstance(msg, bt2._EventMessage)
self.assertEqual(msg.event.cls.name, 'salut')
elif comp_self._at == 7:
- self.assertIsInstance(msg, bt2.message._PacketEndMessage)
+ self.assertIsInstance(msg, bt2._PacketEndMessage)
elif comp_self._at == 8:
- self.assertIsInstance(msg, bt2.message._StreamEndMessage)
+ self.assertIsInstance(msg, bt2._StreamEndMessage)
comp_self._at += 1
def _consume(comp_self):
msg = next(comp_self._msg_iter)
if comp_self._at == 0:
- self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
+ self.assertIsInstance(msg, bt2._StreamBeginningMessage)
elif comp_self._at == 1:
- self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
+ self.assertIsInstance(msg, bt2._PacketBeginningMessage)
elif comp_self._at == 2:
- self.assertIsInstance(msg, bt2.message._EventMessage)
+ self.assertIsInstance(msg, bt2._EventMessage)
raise bt2.TryAgain
else:
pass
def _consume(comp_self):
msg = next(comp_self._msg_iter)
if comp_self._at == 0:
- self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
+ self.assertIsInstance(msg, bt2._StreamBeginningMessage)
elif comp_self._at == 1:
- self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
+ self.assertIsInstance(msg, bt2._PacketBeginningMessage)
elif comp_self._at == 2:
- self.assertIsInstance(msg, bt2.message._EventMessage)
+ self.assertIsInstance(msg, bt2._EventMessage)
elif comp_self._at == 3:
nonlocal raised_in_sink
raised_in_sink = True
for i, msg in enumerate(self._msg_iter):
if i == 0:
- self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
+ self.assertIsInstance(msg, bt2._StreamBeginningMessage)
self.assertEqual(msg.stream.addr, self._stream.addr)
self.assertIsInstance(
- msg.default_clock_snapshot, bt2.clock_snapshot._UnknownClockSnapshot
+ msg.default_clock_snapshot, bt2._UnknownClockSnapshot
)
elif i == 1:
- self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
+ self.assertIsInstance(msg, bt2._PacketBeginningMessage)
self.assertEqual(msg.packet.addr, self._packet.addr)
self.assertEqual(msg.default_clock_snapshot.value, i)
elif i == 2:
- self.assertIsInstance(msg, bt2.message._EventMessage)
+ self.assertIsInstance(msg, bt2._EventMessage)
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.message._MessageIteratorInactivityMessage
- )
+ self.assertIsInstance(msg, bt2._MessageIteratorInactivityMessage)
self.assertEqual(msg.default_clock_snapshot.value, i)
elif i == 4:
- self.assertIsInstance(msg, bt2.message._DiscardedEventsMessage)
+ self.assertIsInstance(msg, bt2._DiscardedEventsMessage)
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.message._PacketEndMessage)
+ self.assertIsInstance(msg, bt2._PacketEndMessage)
self.assertEqual(msg.packet.addr, self._packet.addr)
self.assertEqual(msg.default_clock_snapshot.value, i)
elif i == 6:
- self.assertIsInstance(msg, bt2.message._DiscardedPacketsMessage)
+ self.assertIsInstance(msg, bt2._DiscardedPacketsMessage)
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.message._StreamEndMessage)
+ self.assertIsInstance(msg, bt2._StreamEndMessage)
self.assertEqual(msg.stream.addr, self._stream.addr)
self.assertIsInstance(
- msg.default_clock_snapshot, bt2.clock_snapshot._UnknownClockSnapshot
+ msg.default_clock_snapshot, bt2._UnknownClockSnapshot
)
else:
raise Exception
for i, msg in enumerate(self._msg_iter):
if i == 0:
- self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
+ self.assertIsInstance(msg, bt2._StreamBeginningMessage)
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.message._PacketBeginningMessage)
+ self.assertIsInstance(msg, bt2._PacketBeginningMessage)
self.assertEqual(msg.packet.addr, self._packet.addr)
elif i == 2:
- self.assertIsInstance(msg, bt2.message._EventMessage)
+ self.assertIsInstance(msg, bt2._EventMessage)
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.message._DiscardedEventsMessage)
+ self.assertIsInstance(msg, bt2._DiscardedEventsMessage)
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.message._PacketEndMessage)
+ self.assertIsInstance(msg, bt2._PacketEndMessage)
self.assertEqual(msg.packet.addr, self._packet.addr)
elif i == 5:
- self.assertIsInstance(msg, bt2.message._DiscardedPacketsMessage)
+ self.assertIsInstance(msg, bt2._DiscardedPacketsMessage)
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.message._StreamEndMessage)
+ self.assertIsInstance(msg, bt2._StreamEndMessage)
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.message._StreamBeginningMessage)
+ self.assertIsInstance(msg_stream_beg, bt2._StreamBeginningMessage)
self.assertEqual(msg_stream_beg.default_clock_snapshot.value, 0)
msg_stream_end = msgs[7]
- self.assertIsInstance(msg_stream_end, bt2.message._StreamEndMessage)
+ self.assertIsInstance(msg_stream_end, bt2._StreamEndMessage)
self.assertEqual(msg_stream_end.default_clock_snapshot.value, 7)
# Skip beginning messages.
msg = next(it)
- self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
+ self.assertIsInstance(msg, bt2._StreamBeginningMessage)
msg = next(it)
- self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
+ self.assertIsInstance(msg, bt2._PacketBeginningMessage)
msg_ev1 = next(it)
msg_ev2 = next(it)
- self.assertIsInstance(msg_ev1, bt2.message._EventMessage)
- self.assertIsInstance(msg_ev2, bt2.message._EventMessage)
+ self.assertIsInstance(msg_ev1, bt2._EventMessage)
+ self.assertIsInstance(msg_ev2, bt2._EventMessage)
self.assertEqual(msg_ev1.addr, msg_ev2.addr)
@staticmethod
it, MySourceIter = self._setup_seek_beginning_test()
msg = next(it)
- self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
+ self.assertIsInstance(msg, bt2._StreamBeginningMessage)
msg = next(it)
- self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
+ self.assertIsInstance(msg, bt2._PacketBeginningMessage)
it.seek_beginning()
msg = next(it)
- self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
+ self.assertIsInstance(msg, bt2._StreamBeginningMessage)
# Verify that we can seek beginning after having reached the end.
#
#
# it.seek_beginning()
# msg = next(it)
- # self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
+ # self.assertIsInstance(msg, bt2._StreamBeginningMessage)
def test_seek_beginning_user_error(self):
it, MySourceIter = self._setup_seek_beginning_test()
for at, msg in enumerate(msg_iter):
if at == 0:
- self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
+ self.assertIsInstance(msg, bt2._StreamBeginningMessage)
elif at == 1:
- self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
+ self.assertIsInstance(msg, bt2._PacketBeginningMessage)
elif at == 5:
- self.assertIsInstance(msg, bt2.message._PacketEndMessage)
+ self.assertIsInstance(msg, bt2._PacketEndMessage)
elif at == 6:
- self.assertIsInstance(msg, bt2.message._StreamEndMessage)
+ self.assertIsInstance(msg, bt2._StreamEndMessage)
else:
- self.assertIsInstance(msg, bt2.message._EventMessage)
+ self.assertIsInstance(msg, bt2._EventMessage)
self.assertEqual(msg.event.cls.name, 'salut')
field = msg.event.payload_field['my_int']
self.assertEqual(field, at * 3)
import unittest
import bt2
-import bt2.plugin
import os
def test_find_existing(self):
plugin = bt2.find_plugin('ctf', find_in_user_dir=False, find_in_sys_dir=False)
- self.assertIsInstance(plugin, bt2.plugin._Plugin)
class PluginTestCase(unittest.TestCase):
exc = ctx.exception
self.assertEqual(len(exc), 2)
cause = exc[0]
- self.assertIsInstance(cause, bt2.error._ComponentClassErrorCause)
+ self.assertIsInstance(cause, bt2._ComponentClassErrorCause)
self.assertIn('raise ValueError', cause.message)
self.assertEqual(cause.component_class_type, bt2.ComponentClassType.SINK)
self.assertEqual(cause.component_class_name, 'MySink')
tc, sc1, sc2, sc3 = self._create_trace_class_with_some_stream_classes()
for sc_id, stream_class in tc.items():
- self.assertIsInstance(stream_class, bt2.stream_class._StreamClass)
-
if sc_id == 12:
self.assertEqual(stream_class.addr, sc1.addr)
elif sc_id == 54:
msgs = list(msg_iter)
self.assertEqual(len(msgs), 28)
hist = _count_msgs_by_type(msgs)
- self.assertEqual(hist[bt2.message._EventMessage], 8)
+ self.assertEqual(hist[bt2._EventMessage], 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.message._EventMessage], 8)
+ self.assertEqual(hist[bt2._EventMessage], 8)
def test_iter_custom_filter(self):
src_spec = bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)
flt_spec = bt2.ComponentSpec('utils', 'trimmer', {'end': '13515309.000000075'})
msg_iter = bt2.TraceCollectionMessageIterator(src_spec, flt_spec)
hist = _count_msgs_by_type(msg_iter)
- self.assertEqual(hist[bt2.message._EventMessage], 5)
+ self.assertEqual(hist[bt2._EventMessage], 5)
def test_iter_intersection(self):
specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
msgs = list(msg_iter)
self.assertEqual(len(msgs), 15)
hist = _count_msgs_by_type(msgs)
- self.assertEqual(hist[bt2.message._EventMessage], 3)
+ self.assertEqual(hist[bt2._EventMessage], 3)
def test_iter_intersection_no_inputs_param(self):
specs = [bt2.ComponentSpec('text', 'dmesg', {'read-from-stdin': True})]
msgs = list(msg_iter)
self.assertEqual(len(msgs), 56)
hist = _count_msgs_by_type(msgs)
- self.assertEqual(hist[bt2.message._EventMessage], 16)
+ self.assertEqual(hist[bt2._EventMessage], 16)
def test_iter_no_intersection_begin(self):
specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
msg_iter = bt2.TraceCollectionMessageIterator(specs, begin=13515309.000000023)
hist = _count_msgs_by_type(msg_iter)
- self.assertEqual(hist[bt2.message._EventMessage], 6)
+ self.assertEqual(hist[bt2._EventMessage], 6)
def test_iter_no_intersection_end(self):
specs = [bt2.ComponentSpec('ctf', 'fs', _3EVENTS_INTERSECT_TRACE_PATH)]
msg_iter = bt2.TraceCollectionMessageIterator(specs, end=13515309.000000075)
hist = _count_msgs_by_type(msg_iter)
- self.assertEqual(hist[bt2.message._EventMessage], 5)
+ self.assertEqual(hist[bt2._EventMessage], 5)
projects / babeltrace.git / commitdiff
