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)
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.
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)
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)
- other_port_type = native_bt.PORT_TYPE_INPUT if self_port_type == native_bt.PORT_TYPE_OUTPUT else native_bt.PORT_TYPE_OUTPUT
+
+ 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)
def _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)
- other_port_type = native_bt.PORT_TYPE_INPUT if self_port_type == native_bt.PORT_TYPE_OUTPUT else native_bt.PORT_TYPE_OUTPUT
+
+ 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)
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):
get_output_port_count,
bt2.port._UserComponentOutputPort)
- def _add_output_port(self, name):
+ def _add_output_port(self, name, user_data=None):
utils._check_str(name)
fn = native_bt.self_component_source_add_output_port
- comp_status, self_port_ptr = fn(self._ptr, name, None)
+ comp_status, self_port_ptr = fn(self._ptr, name, user_data)
_handle_component_status(comp_status,
'cannot add output port to source component object')
assert self_port_ptr is not None
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):
get_input_port_count,
bt2.port._UserComponentInputPort)
- def _add_output_port(self, name):
+ def _add_output_port(self, name, user_data=None):
utils._check_str(name)
fn = native_bt.self_component_filter_add_output_port
- comp_status, self_port_ptr = fn(self._ptr, name, None)
+ comp_status, self_port_ptr = fn(self._ptr, name, user_data)
_handle_component_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)
- def _add_input_port(self, name):
+ def _add_input_port(self, name, user_data=None):
utils._check_str(name)
fn = native_bt.self_component_filter_add_input_port
- comp_status, self_port_ptr = fn(self._ptr, name, None)
+ comp_status, self_port_ptr = fn(self._ptr, name, user_data)
_handle_component_status(comp_status,
'cannot add input port to filter component object')
assert self_port_ptr
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):
get_input_port_count,
bt2.port._UserComponentInputPort)
- def _add_input_port(self, name):
+ def _add_input_port(self, name, user_data=None):
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, user_data)
_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)
projects / babeltrace.git / blobdiff