#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object, utils
-from bt2 import message_iterator as bt2_message_iterator
+import sys
import collections.abc
+
+from bt2 import port as bt2_port
+from bt2 import error as bt2_error
+from bt2 import utils as bt2_utils
from bt2 import value as bt2_value
-from bt2 import trace_class as bt2_trace_class
+from bt2 import object as bt2_object
+from bt2 import native_bt
from bt2 import clock_class as bt2_clock_class
+from bt2 import trace_class as bt2_trace_class
from bt2 import query_executor as bt2_query_executor
-from bt2 import port as bt2_port
-import sys
-import bt2
+from bt2 import message_iterator as bt2_message_iterator
+from bt2 import integer_range_set as bt2_integer_range_set
+
+
+class _IncompleteUserClass(Exception):
+ pass
# This class wraps a component class pointer. This component class could
# pointer to a 'bt_component_class *'.
-class _ComponentClassConst(object._SharedObject):
+class _ComponentClassConst(bt2_object._SharedObject):
@property
def name(self):
ptr = self._bt_as_component_class_ptr(self._ptr)
class _SourceComponentClassConst(_ComponentClassConst):
- _get_ref = staticmethod(native_bt.component_class_source_get_ref)
- _put_ref = staticmethod(native_bt.component_class_source_put_ref)
+ @staticmethod
+ def _get_ref(ptr):
+ native_bt.component_class_source_get_ref(ptr)
+
+ @staticmethod
+ def _put_ref(ptr):
+ native_bt.component_class_source_put_ref(ptr)
+
_bt_as_component_class_ptr = staticmethod(
native_bt.component_class_source_as_component_class
)
class _FilterComponentClassConst(_ComponentClassConst):
- _get_ref = staticmethod(native_bt.component_class_filter_get_ref)
- _put_ref = staticmethod(native_bt.component_class_filter_put_ref)
+ @staticmethod
+ def _get_ref(ptr):
+ native_bt.component_class_filter_get_ref(ptr)
+
+ @staticmethod
+ def _put_ref(ptr):
+ native_bt.component_class_filter_put_ref(ptr)
+
_bt_as_component_class_ptr = staticmethod(
native_bt.component_class_filter_as_component_class
)
class _SinkComponentClassConst(_ComponentClassConst):
- _get_ref = staticmethod(native_bt.component_class_sink_get_ref)
- _put_ref = staticmethod(native_bt.component_class_sink_put_ref)
+ @staticmethod
+ def _get_ref(ptr):
+ native_bt.component_class_sink_get_ref(ptr)
+
+ @staticmethod
+ def _put_ref(ptr):
+ native_bt.component_class_sink_put_ref(ptr)
+
_bt_as_component_class_ptr = staticmethod(
native_bt.component_class_sink_as_component_class
)
class _ComponentPorts(collections.abc.Mapping):
-
# component_ptr is a bt_component_source *, bt_component_filter * or
# bt_component_sink *. Its type must match the type expected by the
# functions passed as arguments.
self._port_pycls = port_pycls
def __getitem__(self, key):
- utils._check_str(key)
+ bt2_utils._check_str(key)
port_ptr = self._borrow_port_ptr_by_name(self._component_ptr, key)
if port_ptr is None:
# This is analogous to _SourceComponentClassConst, but for source
# component objects.
-class _GenericSourceComponentConst(object._SharedObject, _SourceComponentConst):
- _get_ref = staticmethod(native_bt.component_source_get_ref)
- _put_ref = staticmethod(native_bt.component_source_put_ref)
+class _GenericSourceComponentConst(bt2_object._SharedObject, _SourceComponentConst):
+ @staticmethod
+ def _get_ref(ptr):
+ native_bt.component_source_get_ref(ptr)
+
+ @staticmethod
+ def _put_ref(ptr):
+ native_bt.component_source_put_ref(ptr)
@property
def output_ports(self):
# This is analogous to _FilterComponentClassConst, but for filter
# component objects.
-class _GenericFilterComponentConst(object._SharedObject, _FilterComponentConst):
- _get_ref = staticmethod(native_bt.component_filter_get_ref)
- _put_ref = staticmethod(native_bt.component_filter_put_ref)
+class _GenericFilterComponentConst(bt2_object._SharedObject, _FilterComponentConst):
+ @staticmethod
+ def _get_ref(ptr):
+ native_bt.component_filter_get_ref(ptr)
+
+ @staticmethod
+ def _put_ref(ptr):
+ native_bt.component_filter_put_ref(ptr)
@property
def output_ports(self):
# This is analogous to _SinkComponentClassConst, but for sink
# component objects.
-class _GenericSinkComponentConst(object._SharedObject, _SinkComponentConst):
- _get_ref = staticmethod(native_bt.component_sink_get_ref)
- _put_ref = staticmethod(native_bt.component_sink_put_ref)
+class _GenericSinkComponentConst(bt2_object._SharedObject, _SinkComponentConst):
+ @staticmethod
+ def _get_ref(ptr):
+ native_bt.component_sink_get_ref(ptr)
+
+ @staticmethod
+ def _put_ref(ptr):
+ native_bt.component_sink_put_ref(ptr)
@property
def input_ports(self):
return
comp_cls_name = kwargs.get("name", class_name)
- utils._check_str(comp_cls_name)
+ bt2_utils._check_str(comp_cls_name)
comp_cls_descr = None
comp_cls_help = None
if hasattr(cls, "__doc__") and cls.__doc__ is not None:
- utils._check_str(cls.__doc__)
+ bt2_utils._check_str(cls.__doc__)
docstring = _trim_docstring(cls.__doc__)
lines = docstring.splitlines()
)
elif _UserSinkComponent in bases:
if not hasattr(cls, "_user_consume"):
- raise bt2._IncompleteUserClass(
+ raise _IncompleteUserClass(
"cannot create component class '{}': missing a _user_consume() method".format(
class_name
)
cls, comp_cls_name, comp_cls_descr, comp_cls_help
)
else:
- raise bt2._IncompleteUserClass(
+ raise _IncompleteUserClass(
"cannot find a known component class base in the bases of '{}'".format(
class_name
)
)
if cc_ptr is None:
- raise bt2._MemoryError(
+ raise bt2_error._MemoryError(
"cannot create component class '{}'".format(class_name)
)
@staticmethod
def _bt_set_iterator_class(cls, iter_cls):
if iter_cls is None:
- raise bt2._IncompleteUserClass(
+ raise _IncompleteUserClass(
"cannot create component class '{}': missing message iterator class".format(
cls.__name__
)
)
if not issubclass(iter_cls, bt2_message_iterator._UserMessageIterator):
- raise bt2._IncompleteUserClass(
+ raise _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(
+ raise _IncompleteUserClass(
"cannot create component class '{}': message iterator class is missing a __next__() method".format(
cls.__name__
)
if hasattr(iter_cls, "_user_can_seek_ns_from_origin") and not hasattr(
iter_cls, "_user_seek_ns_from_origin"
):
- raise bt2._IncompleteUserClass(
+ raise _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(
+ raise _IncompleteUserClass(
"cannot create component class '{}': message iterator class implements _user_can_seek_beginning but not _user_seek_beginning".format(
cls.__name__
)
# this can raise, but the native side checks the exception
range_set = cls._user_get_supported_mip_versions(params, obj, log_level)
- if type(range_set) is not bt2.UnsignedIntegerRangeSet:
+ if type(range_set) is not bt2_integer_range_set.UnsignedIntegerRangeSet:
# this can raise, but the native side checks the exception
- range_set = bt2.UnsignedIntegerRangeSet(range_set)
+ range_set = bt2_integer_range_set.UnsignedIntegerRangeSet(range_set)
# return new reference
range_set._get_ref(range_set._ptr)
priv_query_exec._invalidate()
# this can raise, but the native side checks the exception
- results = bt2.create_value(results)
+ results = bt2_value.create_value(results)
if results is None:
results_ptr = native_bt.value_null
return int(results_ptr)
def _user_query(cls, priv_query_executor, object_name, params, method_obj):
- raise bt2.UnknownObject
+ raise bt2_utils.UnknownObject
def _bt_component_class_ptr(self):
return self._bt_as_component_class_ptr(self._bt_cc_ptr)
tc_ptr = native_bt.trace_class_create(ptr)
if tc_ptr is None:
- raise bt2._MemoryError("could not create trace class")
+ raise bt2_error._MemoryError("could not create trace class")
tc = bt2_trace_class._TraceClass._create_from_ptr(tc_ptr)
tc._assigns_automatic_stream_class_id = assigns_automatic_stream_class_id
cc_ptr = native_bt.clock_class_create(ptr)
if cc_ptr is None:
- raise bt2._MemoryError("could not create clock class")
+ raise bt2_error._MemoryError("could not create clock class")
cc = bt2_clock_class._ClockClass._create_from_ptr(cc_ptr)
)
def _add_output_port(self, name, user_data=None):
- utils._check_str(name)
+ bt2_utils._check_str(name)
if name in self._output_ports:
raise ValueError(
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(
+ bt2_utils._handle_func_status(
comp_status, "cannot add output port to source component object"
)
assert self_port_ptr is not None
)
def _add_output_port(self, name, user_data=None):
- utils._check_str(name)
+ bt2_utils._check_str(name)
if name in self._output_ports:
raise ValueError(
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(
+ bt2_utils._handle_func_status(
comp_status, "cannot add output port to filter component object"
)
assert self_port_ptr
)
def _add_input_port(self, name, user_data=None):
- utils._check_str(name)
+ bt2_utils._check_str(name)
if name in self._input_ports:
raise ValueError(
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(
+ bt2_utils._handle_func_status(
comp_status, "cannot add input port to filter component object"
)
assert self_port_ptr
)
def _add_input_port(self, name, user_data=None):
- utils._check_str(name)
+ bt2_utils._check_str(name)
if name in self._input_ports:
raise ValueError(
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(
+ bt2_utils._handle_func_status(
comp_status, "cannot add input port to sink component object"
)
assert self_port_ptr
)
def _create_message_iterator(self, input_port):
- utils._check_type(input_port, bt2_port._UserComponentInputPort)
+ bt2_utils._check_type(input_port, bt2_port._UserComponentInputPort)
if not input_port.is_connected:
raise ValueError("input port is not connected")
) = 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")
+ bt2_utils._handle_func_status(status, "cannot create message iterator object")
assert msg_iter_ptr is not None
return bt2_message_iterator._UserComponentInputPortMessageIterator(msg_iter_ptr)
projects / babeltrace.git / blobdiff