-# The MIT License (MIT)
+# SPDX-License-Identifier: MIT
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-#
-# Permission is hereby granted, free of charge, to any person obtaining a copy
-# of this software and associated documentation files (the "Software"), to deal
-# in the Software without restriction, including without limitation the rights
-# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-# copies of the Software, and to permit persons to whom the Software is
-# furnished to do so, subject to the following conditions:
-#
-# The above copyright notice and this permission notice shall be included in
-# all copies or substantial portions of the Software.
-#
-# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
-# THE SOFTWARE.
from bt2 import native_bt, object, utils
from bt2 import message_iterator as bt2_message_iterator
def _trim_docstring(docstring):
lines = docstring.expandtabs().splitlines()
+
+ if len(lines) == 0:
+ return ''
+
indent = sys.maxsize
- for line in lines[1:]:
- stripped = line.lstrip()
+ if len(lines) > 1:
+ for line in lines[1:]:
+ stripped = line.lstrip()
- if stripped:
- indent = min(indent, len(line) - len(stripped))
+ if stripped:
+ indent = min(indent, len(line) - len(stripped))
trimmed = [lines[0].strip()]
- if indent < sys.maxsize:
+ if indent < sys.maxsize and len(lines) > 1:
for line in lines[1:]:
trimmed.append(line[indent:].rstrip())
# ...
#
# A user-defined Python component class can have an __init__() method
-# which must at least accept the `params` and `name` arguments:
+# which must accept the following parameters:
#
-# def __init__(self, params, name, something_else):
+# def __init__(self, config, params, obj):
# ...
#
-# The user-defined component class can also have a _finalize() method
-# (do NOT use __del__()) to be notified when the component object is
-# finalized.
+# The value of the `obj` parameter is what was passed as the `obj`
+# parameter if the component was instantiated from Python with
+# Graph.add_component(). If the component was not instantiated from
+# Python, is is always `None`.
+#
+# The user-defined component class can also have a _user_finalize()
+# method (do NOT use __del__()) to be notified when the component object
+# is finalized.
#
# User-defined source and filter component classes must use the
# `message_iterator_class` class parameter to specify the
# message_iterator_class=MyMessageIterator):
# ...
#
-# This message iterator class must inherit
-# bt2._UserMessageIterator, and it must define the _get() and
-# _next() methods. The message iterator class can also define an
-# __init__() method: this method has access to the original Python
-# component object which was used to create it as the `component`
-# property. The message iterator class can also define a
-# _finalize() method (again, do NOT use __del__()): this is called when
-# the message iterator is (really) destroyed.
+# This message iterator class must inherit bt2._UserMessageIterator.
+# It can implement the __init__() method, which must accept the
+# following parameters:
+#
+# def __init__(self, config, port):
+# ...
+#
+# It can also implement the __next__() and _user_finalize() methods
+# (again, do NOT use __del__()), which don't accept any parameters
+# other than `self`.
#
# When the user-defined class is destroyed, this metaclass's __del__()
# method is called: the native BT component class pointer is put (not
)
)
+ if hasattr(iter_cls, '_user_can_seek_ns_from_origin') and not hasattr(
+ iter_cls, '_user_seek_ns_from_origin'
+ ):
+ raise bt2._IncompleteUserClass(
+ "cannot create component class '{}': message iterator class implements _user_can_seek_ns_from_origin but not _user_seek_ns_from_origin".format(
+ cls.__name__
+ )
+ )
+
+ if hasattr(iter_cls, '_user_can_seek_beginning') and not hasattr(
+ iter_cls, '_user_seek_beginning'
+ ):
+ raise bt2._IncompleteUserClass(
+ "cannot create component class '{}': message iterator class implements _user_can_seek_beginning but not _user_seek_beginning".format(
+ cls.__name__
+ )
+ )
+
cls._iter_cls = iter_cls
@property
def _user_get_supported_mip_versions(cls, params, obj, log_level):
return [0]
- def _bt_query_from_native(cls, priv_query_exec_ptr, object, params_ptr, method_obj):
+ def _bt_query_from_native(
+ cls, priv_query_exec_ptr, object_name, params_ptr, method_obj
+ ):
# this can raise, but the native side checks the exception
if params_ptr is not None:
params = bt2_value._create_from_const_ptr_and_get_ref(params_ptr)
try:
# this can raise, but the native side checks the exception
- results = cls._user_query(priv_query_exec, object, params, method_obj)
+ results = cls._user_query(priv_query_exec, object_name, params, method_obj)
finally:
# the private query executor is a private view on the query
# executor; it's not a shared object (the library does not
bt2_value._Value._get_ref(results_ptr)
return int(results_ptr)
- def _user_query(cls, priv_query_executor, object, params, method_obj):
+ def _user_query(cls, priv_query_executor, object_name, params, method_obj):
raise bt2.UnknownObject
def _bt_component_class_ptr(self):
def _add_output_port(self, name, user_data=None):
utils._check_str(name)
+
+ if name in self._output_ports:
+ raise ValueError(
+ 'source component `{}` already contains an output port named `{}`'.format(
+ self.name, 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'
)
assert self_port_ptr is not None
- return bt2_port._UserComponentOutputPort._create_from_ptr(self_port_ptr)
+ return bt2_port._UserComponentOutputPort._create_from_ptr_and_get_ref(
+ self_port_ptr
+ )
class _UserFilterComponent(_UserComponent, _FilterComponentConst):
def _add_output_port(self, name, user_data=None):
utils._check_str(name)
+
+ if name in self._output_ports:
+ raise ValueError(
+ 'filter component `{}` already contains an output port named `{}`'.format(
+ self.name, 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'
)
assert self_port_ptr
- return bt2_port._UserComponentOutputPort._create_from_ptr(self_port_ptr)
+ return bt2_port._UserComponentOutputPort._create_from_ptr_and_get_ref(
+ self_port_ptr
+ )
def _add_input_port(self, name, user_data=None):
utils._check_str(name)
+
+ if name in self._input_ports:
+ raise ValueError(
+ 'filter component `{}` already contains an input port named `{}`'.format(
+ self.name, 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'
)
assert self_port_ptr
- return bt2_port._UserComponentInputPort._create_from_ptr(self_port_ptr)
+ return bt2_port._UserComponentInputPort._create_from_ptr_and_get_ref(
+ self_port_ptr
+ )
class _UserSinkComponent(_UserComponent, _SinkComponentConst):
def _add_input_port(self, name, user_data=None):
utils._check_str(name)
+
+ if name in self._input_ports:
+ raise ValueError(
+ 'sink component `{}` already contains an input port named `{}`'.format(
+ self.name, 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'
)
assert self_port_ptr
- return bt2_port._UserComponentInputPort._create_from_ptr(self_port_ptr)
+ return bt2_port._UserComponentInputPort._create_from_ptr_and_get_ref(
+ self_port_ptr
+ )
- def _create_input_port_message_iterator(self, input_port):
+ def _create_message_iterator(self, input_port):
utils._check_type(input_port, bt2_port._UserComponentInputPort)
- status, msg_iter_ptr = native_bt.bt2_self_component_port_input_message_iterator_create_from_sink_component(
+ if not input_port.is_connected:
+ raise ValueError('input port is not connected')
+
+ (
+ status,
+ msg_iter_ptr,
+ ) = native_bt.bt2_message_iterator_create_from_sink_component(
self._bt_ptr, input_port._ptr
)
utils._handle_func_status(status, 'cannot create message iterator object')