base_cc_ptr = component_class._bt_component_class_ptr()
if obj is not None and not native_bt.bt2_is_python_component_class(base_cc_ptr):
base_cc_ptr = component_class._bt_component_class_ptr()
if obj is not None and not native_bt.bt2_is_python_component_class(base_cc_ptr):
if params is not None and not isinstance(params, (dict, bt2.MapValue)):
raise TypeError("'params' parameter is not a 'dict' or a 'bt2.MapValue'.")
if params is not None and not isinstance(params, (dict, bt2.MapValue)):
raise TypeError("'params' parameter is not a 'dict' or a 'bt2.MapValue'.")
status, comp_ptr = add_fn(
self._ptr, cc_ptr, name, params_ptr, obj, logging_level
)
status, comp_ptr = add_fn(
self._ptr, cc_ptr, name, params_ptr, obj, logging_level
)
status, conn_ptr = native_bt.graph_connect_ports(
self._ptr, upstream_port._ptr, downstream_port._ptr
)
status, conn_ptr = native_bt.graph_connect_ports(
self._ptr, upstream_port._ptr, downstream_port._ptr
)
# keep the partial's reference
self._listener_partials.append(listener_from_native)
def run_once(self):
status = native_bt.graph_run_once(self._ptr)
# keep the partial's reference
self._listener_partials.append(listener_from_native)
def run_once(self):
status = native_bt.graph_run_once(self._ptr)
def add_interrupter(self, interrupter):
utils._check_type(interrupter, bt2_interrupter.Interrupter)
def add_interrupter(self, interrupter):
utils._check_type(interrupter, bt2_interrupter.Interrupter)