import os
-_env_var = os.environ.get('BABELTRACE_PYTHON_BT2_NO_TRACEBACK')
-_NO_PRINT_TRACEBACK = _env_var == '1'
-
-
# This class wraps a component class pointer. This component class could
# have been created by Python code, but since we only have the pointer,
# we can only wrap it in a generic way and lose the original Python
_as_component_class_ptr = staticmethod(native_bt.component_class_sink_as_component_class)
-def _handle_component_status(status, gen_error_msg):
- if status == native_bt.SELF_COMPONENT_STATUS_END:
- raise bt2.Stop
- elif status == native_bt.SELF_COMPONENT_STATUS_AGAIN:
- raise bt2.TryAgain
- elif status == native_bt.SELF_COMPONENT_STATUS_REFUSE_PORT_CONNECTION:
- raise bt2.PortConnectionRefused
- elif status < 0:
- raise bt2.Error(gen_error_msg)
-
-
class _PortIterator(collections.abc.Iterator):
def __init__(self, comp_ports):
self._comp_ports = comp_ports
assert name is not None
return name
+ @property
+ def logging_level(self):
+ ptr = self._as_component_ptr(self._ptr)
+ return native_bt.component_get_logging_level(ptr)
+
@property
def cls(self):
cc_ptr = self._borrow_component_class_ptr(self._ptr)
if _UserSourceComponent in bases:
_UserComponentType._set_iterator_class(cls, iter_cls)
- cc_ptr = native_bt.py3_component_class_source_create(cls,
+ cc_ptr = native_bt.bt2_component_class_source_create(cls,
comp_cls_name,
comp_cls_descr,
comp_cls_help)
elif _UserFilterComponent in bases:
_UserComponentType._set_iterator_class(cls, iter_cls)
- cc_ptr = native_bt.py3_component_class_filter_create(cls,
+ cc_ptr = native_bt.bt2_component_class_filter_create(cls,
comp_cls_name,
comp_cls_descr,
comp_cls_help)
if not hasattr(cls, '_consume'):
raise bt2.IncompleteUserClass("cannot create component class '{}': missing a _consume() method".format(class_name))
- cc_ptr = native_bt.py3_component_class_sink_create(cls,
+ cc_ptr = native_bt.bt2_component_class_sink_create(cls,
comp_cls_name,
comp_cls_descr,
comp_cls_help)
def addr(cls):
return int(cls._cc_ptr)
- def _query_from_native(cls, query_exec_ptr, obj, params_ptr):
+ def _query_from_native(cls, query_exec_ptr, obj, params_ptr, log_level):
# this can raise, in which case the native call to
# bt_component_class_query() returns NULL
if params_ptr is not None:
query_exec_ptr)
# this can raise, but the native side checks the exception
- results = cls._query(query_exec, obj, params)
+ results = cls._query(query_exec, obj, params, log_level)
# this can raise, but the native side checks the exception
results = bt2.create_value(results)
return results_addr
- def _query(cls, query_executor, obj, params):
+ def _query(cls, query_executor, obj, params, log_level):
raise NotImplementedError
def _component_class_ptr(self):
assert name is not None
return name
+ @property
+ def logging_level(self):
+ ptr = self._as_not_self_specific_component_ptr(self._ptr)
+ ptr = self._as_component_ptr(ptr)
+ return native_bt.component_get_logging_level(ptr)
+
@property
def cls(self):
comp_ptr = self._as_not_self_specific_component_ptr(self._ptr)
def _finalize(self):
pass
- def _accept_port_connection(self, port, other_port):
- return True
-
- def _accept_port_connection_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)
-
- 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_ptr, other_port_type)
- res = self._accept_port_connection(port, other_port_ptr)
-
- if type(res) is not bool:
- raise TypeError("'{}' is not a 'bool' object")
-
- return res
-
def _port_connected(self, port, other_port):
pass
utils._check_str(name)
fn = native_bt.self_component_source_add_output_port
comp_status, self_port_ptr = fn(self._ptr, name, user_data)
- _handle_component_status(comp_status,
- 'cannot add output port to source component object')
+ utils._handle_func_status(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)
utils._check_str(name)
fn = native_bt.self_component_filter_add_output_port
comp_status, self_port_ptr = fn(self._ptr, name, user_data)
- _handle_component_status(comp_status,
- 'cannot add output port to filter component object')
+ utils._handle_func_status(comp_status,
+ 'cannot add output port to filter component object')
assert self_port_ptr
return bt2.port._UserComponentOutputPort._create_from_ptr(self_port_ptr)
utils._check_str(name)
fn = native_bt.self_component_filter_add_input_port
comp_status, self_port_ptr = fn(self._ptr, name, user_data)
- _handle_component_status(comp_status,
- 'cannot add input port to filter component object')
+ utils._handle_func_status(comp_status,
+ 'cannot add input port to filter component object')
assert self_port_ptr
return bt2.port._UserComponentInputPort._create_from_ptr(self_port_ptr)
utils._check_str(name)
fn = native_bt.self_component_sink_add_input_port
comp_status, self_port_ptr = fn(self._ptr, name, user_data)
- _handle_component_status(comp_status,
- 'cannot add input port to sink component object')
+ utils._handle_func_status(comp_status,
+ 'cannot add input port to sink component object')
assert self_port_ptr
return bt2.port._UserComponentInputPort._create_from_ptr(self_port_ptr)