def _create_component_from_ptr(ptr, comp_cls_type):
return _COMP_CLS_TYPE_TO_GENERIC_COMP_PYCLS[comp_cls_type]._create_from_ptr(ptr)
+
+# Same as the above, but acquire a new reference instead of stealing the
+# reference from the caller.
+
+def _create_component_from_ptr_and_get_ref(ptr, comp_cls_type):
+ return _COMP_CLS_TYPE_TO_GENERIC_COMP_PYCLS[comp_cls_type]._create_from_ptr_and_get_ref(ptr)
+
+
# Create a component class Python object of type
# _GenericSourceComponentClass, _GenericFilterComponentClass or
# _GenericSinkComponentClass, depending on comp_cls_type.
other_port_ptr, other_port_type)
self._port_connected(port, other_port)
+ def _graph_is_configured_from_native(self):
+ self._graph_is_configured()
+
+ def _create_trace_class(self, env=None, uuid=None,
+ assigns_automatic_stream_class_id=True):
+ ptr = self._as_self_component_ptr(self._ptr)
+ tc_ptr = native_bt.trace_class_create(ptr)
+
+ if tc_ptr is None:
+ raise bt2.CreationError('could not create trace class')
+
+ tc = bt2.TraceClass._create_from_ptr(tc_ptr)
+
+ if env is not None:
+ for key, value in env.items():
+ tc.env[key] = value
+
+ if uuid is not None:
+ tc._uuid = uuid
+
+ tc._assigns_automatic_stream_class_id = assigns_automatic_stream_class_id
+
+ return tc
+
+ def _create_clock_class(self, frequency=None, name=None, description=None,
+ precision=None, offset=None, origin_is_unix_epoch=True,
+ uuid=None):
+ ptr = self._as_self_component_ptr(self._ptr)
+ cc_ptr = native_bt.clock_class_create(ptr)
+
+ if cc_ptr is None:
+ raise bt2.CreationError('could not create clock class')
+
+ cc = bt2.clock_class._ClockClass._create_from_ptr(cc_ptr)
+
+ if frequency is not None:
+ cc._frequency = frequency
+
+ if name is not None:
+ cc._name = name
+
+ if description is not None:
+ cc._description = description
+
+ if precision is not None:
+ cc._precision = precision
+
+ if offset is not None:
+ cc._offset = offset
+
+ cc._origin_is_unix_epoch = origin_is_unix_epoch
+
+ if uuid is not None:
+ cc._uuid = uuid
+
+ return cc
+
class _UserSourceComponent(_UserComponent, _SourceComponent):
_as_not_self_specific_component_ptr = staticmethod(native_bt.self_component_source_as_component_source)
+ _as_self_component_ptr = staticmethod(native_bt.self_component_source_as_self_component)
@property
def _output_ports(self):
class _UserFilterComponent(_UserComponent, _FilterComponent):
_as_not_self_specific_component_ptr = staticmethod(native_bt.self_component_filter_as_component_filter)
+ _as_self_component_ptr = staticmethod(native_bt.self_component_filter_as_self_component)
@property
def _output_ports(self):
class _UserSinkComponent(_UserComponent, _SinkComponent):
_as_not_self_specific_component_ptr = staticmethod(native_bt.self_component_sink_as_component_sink)
+ _as_self_component_ptr = staticmethod(native_bt.self_component_sink_as_self_component)
@property
def _input_ports(self):
def _add_input_port(self, name):
utils._check_str(name)
fn = native_bt.self_component_sink_add_input_port
- comp_status, priv_port_ptr = fn(self._ptr, name, None)
+ comp_status, self_port_ptr = fn(self._ptr, name, None)
_handle_component_status(comp_status,
'cannot add input port to sink component object')
- assert priv_port_ptr
- return bt2.port._UserComponentInputPort._create_from_ptr(priv_port_ptr)
+ assert self_port_ptr
+ return bt2.port._UserComponentInputPort._create_from_ptr(self_port_ptr)