# - _bt_as_component_class_ptr: static method, convert the passed component class
# pointer to a 'bt_component_class *'.
+
class _GenericComponentClass(object._SharedObject):
@property
def name(self):
class _GenericSourceComponentClass(_GenericComponentClass):
_get_ref = staticmethod(native_bt.component_class_source_get_ref)
_put_ref = staticmethod(native_bt.component_class_source_put_ref)
- _bt_as_component_class_ptr = staticmethod(native_bt.component_class_source_as_component_class)
+ _bt_as_component_class_ptr = staticmethod(
+ native_bt.component_class_source_as_component_class
+ )
class _GenericFilterComponentClass(_GenericComponentClass):
_get_ref = staticmethod(native_bt.component_class_filter_get_ref)
_put_ref = staticmethod(native_bt.component_class_filter_put_ref)
- _bt_as_component_class_ptr = staticmethod(native_bt.component_class_filter_as_component_class)
+ _bt_as_component_class_ptr = staticmethod(
+ native_bt.component_class_filter_as_component_class
+ )
class _GenericSinkComponentClass(_GenericComponentClass):
_get_ref = staticmethod(native_bt.component_class_sink_get_ref)
_put_ref = staticmethod(native_bt.component_class_sink_put_ref)
- _bt_as_component_class_ptr = staticmethod(native_bt.component_class_sink_as_component_class)
+ _bt_as_component_class_ptr = staticmethod(
+ native_bt.component_class_sink_as_component_class
+ )
class _PortIterator(collections.abc.Iterator):
# bt_component_sink *. Its type must match the type expected by the
# functions passed as arguments.
- def __init__(self, component_ptr,
- borrow_port_ptr_by_name,
- borrow_port_ptr_at_index,
- get_port_count,
- port_pycls):
+ def __init__(
+ self,
+ component_ptr,
+ borrow_port_ptr_by_name,
+ borrow_port_ptr_at_index,
+ get_port_count,
+ port_pycls,
+ ):
self._component_ptr = component_ptr
self._borrow_port_ptr_by_name = borrow_port_ptr_by_name
self._borrow_port_ptr_at_index = borrow_port_ptr_at_index
# - _bt_as_component_ptr: static method, must return the passed specialized
# component pointer (e.g. 'bt_component_sink *') as a 'bt_component *'.
+
class _Component:
@property
def name(self):
def cls(self):
cc_ptr = self._bt_borrow_component_class_ptr(self._ptr)
assert cc_ptr is not None
- return _create_component_class_from_ptr_and_get_ref(cc_ptr, self._bt_comp_cls_type)
+ return _create_component_class_from_ptr_and_get_ref(
+ cc_ptr, self._bt_comp_cls_type
+ )
def __eq__(self, other):
if not hasattr(other, 'addr'):
class _SourceComponent(_Component):
- _bt_borrow_component_class_ptr = staticmethod(native_bt.component_source_borrow_class_const)
+ _bt_borrow_component_class_ptr = staticmethod(
+ native_bt.component_source_borrow_class_const
+ )
_bt_comp_cls_type = native_bt.COMPONENT_CLASS_TYPE_SOURCE
- _bt_as_component_class_ptr = staticmethod(native_bt.component_class_source_as_component_class)
+ _bt_as_component_class_ptr = staticmethod(
+ native_bt.component_class_source_as_component_class
+ )
_bt_as_component_ptr = staticmethod(native_bt.component_source_as_component_const)
class _FilterComponent(_Component):
- _bt_borrow_component_class_ptr = staticmethod(native_bt.component_filter_borrow_class_const)
+ _bt_borrow_component_class_ptr = staticmethod(
+ native_bt.component_filter_borrow_class_const
+ )
_bt_comp_cls_type = native_bt.COMPONENT_CLASS_TYPE_FILTER
- _bt_as_component_class_ptr = staticmethod(native_bt.component_class_filter_as_component_class)
+ _bt_as_component_class_ptr = staticmethod(
+ native_bt.component_class_filter_as_component_class
+ )
_bt_as_component_ptr = staticmethod(native_bt.component_filter_as_component_const)
class _SinkComponent(_Component):
- _bt_borrow_component_class_ptr = staticmethod(native_bt.component_sink_borrow_class_const)
+ _bt_borrow_component_class_ptr = staticmethod(
+ native_bt.component_sink_borrow_class_const
+ )
_bt_comp_cls_type = native_bt.COMPONENT_CLASS_TYPE_SINK
- _bt_as_component_class_ptr = staticmethod(native_bt.component_class_sink_as_component_class)
+ _bt_as_component_class_ptr = staticmethod(
+ native_bt.component_class_sink_as_component_class
+ )
_bt_as_component_ptr = staticmethod(native_bt.component_sink_as_component_const)
@property
def output_ports(self):
- return _ComponentPorts(self._ptr,
- 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)
+ return _ComponentPorts(
+ self._ptr,
+ 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,
+ )
# This is analogous to _GenericFilterComponentClass, but for filter
@property
def output_ports(self):
- return _ComponentPorts(self._ptr,
- 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)
+ return _ComponentPorts(
+ self._ptr,
+ 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,
+ )
@property
def input_ports(self):
- return _ComponentPorts(self._ptr,
- 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)
+ return _ComponentPorts(
+ self._ptr,
+ 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,
+ )
# This is analogous to _GenericSinkComponentClass, but for sink
@property
def input_ports(self):
- return _ComponentPorts(self._ptr,
- 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)
+ return _ComponentPorts(
+ self._ptr,
+ 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,
+ )
_COMP_CLS_TYPE_TO_GENERIC_COMP_PYCLS = {
#
# Steals the reference to ptr from the caller.
+
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)
+ 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
#
# Acquires a new reference to ptr.
+
def _create_component_class_from_ptr_and_get_ref(ptr, comp_cls_type):
- return _COMP_CLS_TYPE_TO_GENERIC_COMP_CLS_PYCLS[comp_cls_type]._create_from_ptr_and_get_ref(ptr)
+ return _COMP_CLS_TYPE_TO_GENERIC_COMP_CLS_PYCLS[
+ comp_cls_type
+ ]._create_from_ptr_and_get_ref(ptr)
def _trim_docstring(docstring):
if _UserSourceComponent in bases:
_UserComponentType._bt_set_iterator_class(cls, iter_cls)
- cc_ptr = native_bt.bt2_component_class_source_create(cls,
- comp_cls_name,
- comp_cls_descr,
- comp_cls_help)
+ cc_ptr = native_bt.bt2_component_class_source_create(
+ cls, comp_cls_name, comp_cls_descr, comp_cls_help
+ )
elif _UserFilterComponent in bases:
_UserComponentType._bt_set_iterator_class(cls, iter_cls)
- cc_ptr = native_bt.bt2_component_class_filter_create(cls,
- comp_cls_name,
- comp_cls_descr,
- comp_cls_help)
+ cc_ptr = native_bt.bt2_component_class_filter_create(
+ cls, comp_cls_name, comp_cls_descr, comp_cls_help
+ )
elif _UserSinkComponent in bases:
if not hasattr(cls, '_graph_is_configured'):
- raise bt2.IncompleteUserClass("cannot create component class '{}': missing a _graph_is_configured() method".format(class_name))
+ raise bt2.IncompleteUserClass(
+ "cannot create component class '{}': missing a _graph_is_configured() method".format(
+ class_name
+ )
+ )
if not hasattr(cls, '_consume'):
- raise bt2.IncompleteUserClass("cannot create component class '{}': missing a _consume() method".format(class_name))
-
- cc_ptr = native_bt.bt2_component_class_sink_create(cls,
- comp_cls_name,
- comp_cls_descr,
- comp_cls_help)
+ raise bt2.IncompleteUserClass(
+ "cannot create component class '{}': missing a _consume() method".format(
+ class_name
+ )
+ )
+
+ cc_ptr = native_bt.bt2_component_class_sink_create(
+ cls, comp_cls_name, comp_cls_descr, comp_cls_help
+ )
else:
- raise bt2.IncompleteUserClass("cannot find a known component class base in the bases of '{}'".format(class_name))
+ raise bt2.IncompleteUserClass(
+ "cannot find a known component class base in the bases of '{}'".format(
+ class_name
+ )
+ )
if cc_ptr is None:
- raise bt2.CreationError("cannot create component class '{}'".format(class_name))
+ raise bt2.CreationError(
+ "cannot create component class '{}'".format(class_name)
+ )
cls._bt_cc_ptr = cc_ptr
return self
def __call__(cls, *args, **kwargs):
- raise bt2.Error('cannot directly instantiate a user component from a Python module')
+ raise bt2.Error(
+ 'cannot directly instantiate a user component from a Python module'
+ )
@staticmethod
def _bt_set_iterator_class(cls, iter_cls):
if iter_cls is None:
- raise bt2.IncompleteUserClass("cannot create component class '{}': missing message iterator class".format(cls.__name__))
+ raise bt2.IncompleteUserClass(
+ "cannot create component class '{}': missing message iterator class".format(
+ cls.__name__
+ )
+ )
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__))
+ raise bt2.IncompleteUserClass(
+ "cannot create component class '{}': message iterator class does not inherit bt2._UserMessageIterator".format(
+ cls.__name__
+ )
+ )
if not hasattr(iter_cls, '__next__'):
- raise bt2.IncompleteUserClass("cannot create component class '{}': message iterator class is missing a __next__() method".format(cls.__name__))
+ raise bt2.IncompleteUserClass(
+ "cannot create component class '{}': message iterator class is missing a __next__() method".format(
+ cls.__name__
+ )
+ )
cls._iter_cls = iter_cls
else:
params = None
- query_exec = bt2.QueryExecutor._create_from_ptr_and_get_ref(
- query_exec_ptr)
+ query_exec = bt2.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)
cc_ptr = cls._bt_as_component_class_ptr(cls._bt_cc_ptr)
native_bt.component_class_put_ref(cc_ptr)
+
# Subclasses must provide these methods or property:
#
# - _bt_as_not_self_specific_component_ptr: static method, must return the passed
# - _bt_comp_cls_type: property, one of the native_bt.COMPONENT_CLASS_TYPE_*
# constants.
+
class _UserComponent(metaclass=_UserComponentType):
@property
def name(self):
def cls(self):
comp_ptr = self._bt_as_not_self_specific_component_ptr(self._bt_ptr)
cc_ptr = self._bt_borrow_component_class_ptr(comp_ptr)
- return _create_component_class_from_ptr_and_get_ref(cc_ptr, self._bt_comp_cls_type)
+ return _create_component_class_from_ptr_and_get_ref(
+ cc_ptr, self._bt_comp_cls_type
+ )
@property
def addr(self):
def _port_connected(self, port, other_port):
pass
- 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)
+ 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)
if self_port_type == native_bt.PORT_TYPE_OUTPUT:
other_port_type = native_bt.PORT_TYPE_INPUT
other_port_type = native_bt.PORT_TYPE_OUTPUT
other_port = bt2.port._create_from_ptr_and_get_ref(
- other_port_ptr, other_port_type)
+ other_port_ptr, other_port_type
+ )
self._port_connected(port, other_port)
def _bt_graph_is_configured_from_native(self):
return tc
- def _create_clock_class(self, frequency=None, name=None, description=None,
- precision=None, offset=None, origin_is_unix_epoch=True,
- uuid=None):
+ def _create_clock_class(
+ self,
+ frequency=None,
+ name=None,
+ description=None,
+ precision=None,
+ offset=None,
+ origin_is_unix_epoch=True,
+ uuid=None,
+ ):
ptr = self._bt_as_self_component_ptr(self._bt_ptr)
cc_ptr = native_bt.clock_class_create(ptr)
class _UserSourceComponent(_UserComponent, _SourceComponent):
- _bt_as_not_self_specific_component_ptr = staticmethod(native_bt.self_component_source_as_component_source)
- _bt_as_self_component_ptr = staticmethod(native_bt.self_component_source_as_self_component)
+ _bt_as_not_self_specific_component_ptr = staticmethod(
+ native_bt.self_component_source_as_component_source
+ )
+ _bt_as_self_component_ptr = staticmethod(
+ native_bt.self_component_source_as_self_component
+ )
@property
def _output_ports(self):
ptr = self._bt_as_not_self_specific_component_ptr(self_ptr)
return native_bt.component_source_get_output_port_count(ptr)
- return _ComponentPorts(self._bt_ptr,
- 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)
+ return _ComponentPorts(
+ self._bt_ptr,
+ 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,
+ )
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._bt_ptr, name, user_data)
- utils._handle_func_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)
class _UserFilterComponent(_UserComponent, _FilterComponent):
- _bt_as_not_self_specific_component_ptr = staticmethod(native_bt.self_component_filter_as_component_filter)
- _bt_as_self_component_ptr = staticmethod(native_bt.self_component_filter_as_self_component)
+ _bt_as_not_self_specific_component_ptr = staticmethod(
+ native_bt.self_component_filter_as_component_filter
+ )
+ _bt_as_self_component_ptr = staticmethod(
+ native_bt.self_component_filter_as_self_component
+ )
@property
def _output_ports(self):
ptr = self._bt_as_not_self_specific_component_ptr(self_ptr)
return native_bt.component_filter_get_output_port_count(ptr)
- return _ComponentPorts(self._bt_ptr,
- 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)
+ return _ComponentPorts(
+ self._bt_ptr,
+ 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,
+ )
@property
def _input_ports(self):
ptr = self._bt_as_not_self_specific_component_ptr(self_ptr)
return native_bt.component_filter_get_input_port_count(ptr)
- return _ComponentPorts(self._bt_ptr,
- 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)
+ return _ComponentPorts(
+ self._bt_ptr,
+ 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,
+ )
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._bt_ptr, name, user_data)
- utils._handle_func_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._bt_ptr, name, user_data)
- utils._handle_func_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)
class _UserSinkComponent(_UserComponent, _SinkComponent):
- _bt_as_not_self_specific_component_ptr = staticmethod(native_bt.self_component_sink_as_component_sink)
- _bt_as_self_component_ptr = staticmethod(native_bt.self_component_sink_as_self_component)
+ _bt_as_not_self_specific_component_ptr = staticmethod(
+ native_bt.self_component_sink_as_component_sink
+ )
+ _bt_as_self_component_ptr = staticmethod(
+ native_bt.self_component_sink_as_self_component
+ )
@property
def _input_ports(self):
ptr = self._bt_as_not_self_specific_component_ptr(self_ptr)
return native_bt.component_sink_get_input_port_count(ptr)
- return _ComponentPorts(self._bt_ptr,
- 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)
+ return _ComponentPorts(
+ self._bt_ptr,
+ 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,
+ )
def _add_input_port(self, name, user_data=None):
utils._check_str(name)
fn = native_bt.self_component_sink_add_input_port
comp_status, self_port_ptr = fn(self._bt_ptr, name, user_data)
- utils._handle_func_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)
projects / babeltrace.git / blobdiff