#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object, utils
+from bt2 import native_bt
+from bt2 import object as bt2_object
+from bt2 import utils as bt2_utils
from bt2 import value as bt2_value
import uuid as uuidp
class ClockClassOffset:
def __init__(self, seconds=0, cycles=0):
- utils._check_int64(seconds)
- utils._check_int64(cycles)
+ bt2_utils._check_int64(seconds)
+ bt2_utils._check_int64(cycles)
self._seconds = seconds
self._cycles = cycles
return (self.seconds, self.cycles) == (other.seconds, other.cycles)
-class _ClockClassConst(object._SharedObject):
+class _ClockClassConst(bt2_object._SharedObject):
@staticmethod
def _get_ref(ptr):
native_bt.clock_class_get_ref(ptr)
return uuidp.UUID(bytes=uuid_bytes)
def cycles_to_ns_from_origin(self, cycles):
- utils._check_uint64(cycles)
+ bt2_utils._check_uint64(cycles)
status, ns = native_bt.clock_class_cycles_to_ns_from_origin(self._ptr, cycles)
error_msg = "cannot convert clock value to nanoseconds from origin for given clock class"
- utils._handle_func_status(status, error_msg)
+ bt2_utils._handle_func_status(status, error_msg)
return ns
def _user_attributes(self, user_attributes):
value = bt2_value.create_value(user_attributes)
- utils._check_type(value, bt2_value.MapValue)
+ bt2_utils._check_type(value, bt2_value.MapValue)
native_bt.clock_class_set_user_attributes(self._ptr, value._ptr)
_user_attributes = property(fset=_user_attributes)
def _name(self, name):
- utils._check_str(name)
+ bt2_utils._check_str(name)
status = native_bt.clock_class_set_name(self._ptr, name)
- utils._handle_func_status(status, "cannot set clock class object's name")
+ bt2_utils._handle_func_status(status, "cannot set clock class object's name")
_name = property(fset=_name)
def _description(self, description):
- utils._check_str(description)
+ bt2_utils._check_str(description)
status = native_bt.clock_class_set_description(self._ptr, description)
- utils._handle_func_status(status, "cannot set clock class object's description")
+ bt2_utils._handle_func_status(
+ status, "cannot set clock class object's description"
+ )
_description = property(fset=_description)
def _frequency(self, frequency):
- utils._check_uint64(frequency)
+ bt2_utils._check_uint64(frequency)
native_bt.clock_class_set_frequency(self._ptr, frequency)
_frequency = property(fset=_frequency)
def _precision(self, precision):
- utils._check_uint64(precision)
+ bt2_utils._check_uint64(precision)
native_bt.clock_class_set_precision(self._ptr, precision)
_precision = property(fset=_precision)
def _offset(self, offset):
- utils._check_type(offset, ClockClassOffset)
+ bt2_utils._check_type(offset, ClockClassOffset)
native_bt.clock_class_set_offset(self._ptr, offset.seconds, offset.cycles)
_offset = property(fset=_offset)
def _origin_is_unix_epoch(self, origin_is_unix_epoch):
- utils._check_bool(origin_is_unix_epoch)
+ bt2_utils._check_bool(origin_is_unix_epoch)
native_bt.clock_class_set_origin_is_unix_epoch(
self._ptr, int(origin_is_unix_epoch)
)
_origin_is_unix_epoch = property(fset=_origin_is_unix_epoch)
def _uuid(self, uuid):
- utils._check_type(uuid, uuidp.UUID)
+ bt2_utils._check_type(uuid, uuidp.UUID)
native_bt.clock_class_set_uuid(self._ptr, uuid.bytes)
_uuid = property(fset=_uuid)
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object, utils
+from bt2 import native_bt
+from bt2 import object as bt2_object
+from bt2 import utils as bt2_utils
import numbers
from bt2 import clock_class as bt2_clock_class
import functools
@functools.total_ordering
-class _ClockSnapshotConst(object._UniqueObject):
+class _ClockSnapshotConst(bt2_object._UniqueObject):
@property
def clock_class(self):
cc_ptr = native_bt.clock_snapshot_borrow_clock_class_const(self._ptr)
@property
def ns_from_origin(self):
status, ns = native_bt.clock_snapshot_get_ns_from_origin(self._ptr)
- utils._handle_func_status(
+ bt2_utils._handle_func_status(
status, "cannot get clock snapshot's nanoseconds from origin"
)
return ns
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object, utils
+from bt2 import native_bt
+from bt2 import object as bt2_object
+from bt2 import utils as bt2_utils
from bt2 import message_iterator as bt2_message_iterator
import collections.abc
from bt2 import value as bt2_value
# 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)
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):
+class _GenericSourceComponentConst(bt2_object._SharedObject, _SourceComponentConst):
@staticmethod
def _get_ref(ptr):
native_bt.component_source_get_ref(ptr)
# This is analogous to _FilterComponentClassConst, but for filter
# component objects.
-class _GenericFilterComponentConst(object._SharedObject, _FilterComponentConst):
+class _GenericFilterComponentConst(bt2_object._SharedObject, _FilterComponentConst):
@staticmethod
def _get_ref(ptr):
native_bt.component_filter_get_ref(ptr)
# This is analogous to _SinkComponentClassConst, but for sink
# component objects.
-class _GenericSinkComponentConst(object._SharedObject, _SinkComponentConst):
+class _GenericSinkComponentConst(bt2_object._SharedObject, _SinkComponentConst):
@staticmethod
def _get_ref(ptr):
native_bt.component_sink_get_ref(ptr)
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()
)
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)
#
# Copyright (c) 2016-2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object, utils
+from bt2 import native_bt
+from bt2 import object as bt2_object
+from bt2 import utils as bt2_utils
from bt2 import event_class as bt2_event_class
from bt2 import packet as bt2_packet
from bt2 import stream as bt2_stream
import collections.abc
-class _EventConst(object._UniqueObject, collections.abc.Mapping):
+class _EventConst(bt2_object._UniqueObject, collections.abc.Mapping):
_borrow_class_ptr = staticmethod(native_bt.event_borrow_class_const)
_borrow_packet_ptr = staticmethod(native_bt.event_borrow_packet_const)
_borrow_stream_ptr = staticmethod(native_bt.event_borrow_stream_const)
)
def __getitem__(self, key):
- utils._check_str(key)
+ bt2_utils._check_str(key)
payload_field = self.payload_field
if payload_field is not None and key in payload_field:
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object, utils
+from bt2 import native_bt
+from bt2 import object as bt2_object
+from bt2 import utils as bt2_utils
from bt2 import field_class as bt2_field_class
from bt2 import value as bt2_value
DEBUG = native_bt.EVENT_CLASS_LOG_LEVEL_DEBUG
-class _EventClassConst(object._SharedObject):
+class _EventClassConst(bt2_object._SharedObject):
@staticmethod
def _get_ref(ptr):
native_bt.event_class_get_ref(ptr)
def _emf_uri(self, emf_uri):
status = native_bt.event_class_set_emf_uri(self._ptr, emf_uri)
- utils._handle_func_status(status, "cannot set event class object's EMF URI")
+ bt2_utils._handle_func_status(status, "cannot set event class object's EMF URI")
_emf_uri = property(fset=_emf_uri)
status = native_bt.event_class_set_specific_context_field_class(
self._ptr, context_field_class._ptr
)
- utils._handle_func_status(
+ bt2_utils._handle_func_status(
status, "cannot set event class object's context field class"
)
status = native_bt.event_class_set_payload_field_class(
self._ptr, payload_field_class._ptr
)
- utils._handle_func_status(
+ bt2_utils._handle_func_status(
status, "cannot set event class object's payload field class"
)
payload_field_class,
):
if name is not None:
- utils._check_str(name)
+ bt2_utils._check_str(name)
if user_attributes is not None:
value = bt2_value.create_value(user_attributes)
- utils._check_type(value, bt2_value.MapValue)
+ bt2_utils._check_type(value, bt2_value.MapValue)
if log_level is not None:
log_levels = (
raise ValueError("'{}' is not a valid log level".format(log_level))
if emf_uri is not None:
- utils._check_str(emf_uri)
+ bt2_utils._check_str(emf_uri)
if specific_context_field_class is not None:
- utils._check_type(
+ bt2_utils._check_type(
specific_context_field_class, bt2_field_class._StructureFieldClass
)
if payload_field_class is not None:
- utils._check_type(payload_field_class, bt2_field_class._StructureFieldClass)
+ bt2_utils._check_type(
+ payload_field_class, bt2_field_class._StructureFieldClass
+ )
_EVENT_CLASS_LOG_LEVEL_TO_OBJ = {
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object, utils
+from bt2 import native_bt
+from bt2 import object as bt2_object
+from bt2 import utils as bt2_utils
from bt2 import field_class as bt2_field_class
import collections.abc
import functools
return field
-class _FieldConst(object._UniqueObject):
+class _FieldConst(bt2_object._UniqueObject):
_create_field_from_ptr = staticmethod(_create_field_from_const_ptr)
_create_field_class_from_ptr_and_get_ref = staticmethod(
bt2_field_class._create_field_class_from_const_ptr_and_get_ref
_NAME = "Bit array"
def _value_as_integer(self, value):
- utils._check_uint64(value)
+ bt2_utils._check_uint64(value)
native_bt.field_bit_array_set_value_as_integer(self._ptr, value)
value_as_integer = property(
@property
def labels(self):
status, labels = self._get_mapping_labels(self._ptr)
- utils._handle_func_status(status, "cannot get label for enumeration field")
+ bt2_utils._handle_func_status(status, "cannot get label for enumeration field")
assert labels is not None
return labels
def __iadd__(self, value):
value = self._value_to_str(value)
status = native_bt.field_string_append(self._ptr, value)
- utils._handle_func_status(
+ bt2_utils._handle_func_status(
status, "cannot append to string field object's value"
)
return self
return "{{{}}}".format(", ".join(items))
def __getitem__(self, key):
- utils._check_str(key)
+ bt2_utils._check_str(key)
field_ptr = self._borrow_member_field_ptr_by_name(self._ptr, key)
if field_ptr is None:
)
def member_at_index(self, index):
- utils._check_uint64(index)
+ bt2_utils._check_uint64(index)
if index >= len(self):
raise IndexError
_borrow_field_ptr = staticmethod(native_bt.field_option_borrow_field)
def _has_field(self, value):
- utils._check_bool(value)
+ bt2_utils._check_bool(value)
native_bt.field_option_set_has_field(self._ptr, value)
has_field = property(fget=_OptionFieldConst.has_field.fget, fset=_has_field)
_NAME = "Dynamic array"
def _set_length(self, length):
- utils._check_uint64(length)
+ bt2_utils._check_uint64(length)
status = native_bt.field_array_dynamic_set_length(self._ptr, length)
- utils._handle_func_status(status, "cannot set dynamic array length")
+ bt2_utils._handle_func_status(status, "cannot set dynamic array length")
length = property(fget=_ArrayField._get_length, fset=_set_length)
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object, utils
+from bt2 import native_bt
+from bt2 import object as bt2_object
+from bt2 import utils as bt2_utils
import collections.abc
from bt2 import field_path as bt2_field_path
from bt2 import integer_range_set as bt2_integer_range_set
HEXADECIMAL = native_bt.FIELD_CLASS_INTEGER_PREFERRED_DISPLAY_BASE_HEXADECIMAL
-class _FieldClassConst(object._SharedObject):
+class _FieldClassConst(bt2_object._SharedObject):
@staticmethod
def _get_ref(ptr):
native_bt.field_class_get_ref(ptr)
def _user_attributes(self, user_attributes):
value = bt2_value.create_value(user_attributes)
- utils._check_type(value, bt2_value.MapValue)
+ bt2_utils._check_type(value, bt2_value.MapValue)
native_bt.field_class_set_user_attributes(self._ptr, value._ptr)
_user_attributes = property(fset=_user_attributes)
_field_value_range = property(fset=_field_value_range)
def _preferred_display_base(self, base):
- utils._check_uint64(base)
+ bt2_utils._check_uint64(base)
if base not in (
IntegerDisplayBase.BINARY,
self._check_int_type(value)
status, labels = self._get_mapping_labels_for_value(self._ptr, value)
- utils._handle_func_status(
+ bt2_utils._handle_func_status(
status, "cannot get mapping labels for value {}".format(value)
)
return [self[label] for label in labels]
yield self._mapping_pycls(mapping_ptr).label
def __getitem__(self, label):
- utils._check_str(label)
+ bt2_utils._check_str(label)
mapping_ptr = self._borrow_mapping_ptr_by_label(self._ptr, label)
if mapping_ptr is None:
class _EnumerationFieldClass(_EnumerationFieldClassConst, _IntegerFieldClass):
def add_mapping(self, label, ranges):
- utils._check_str(label)
- utils._check_type(ranges, self._range_set_pycls)
+ bt2_utils._check_str(label)
+ bt2_utils._check_type(ranges, self._range_set_pycls)
if label in self:
raise ValueError("duplicate mapping label '{}'".format(label))
status = self._add_mapping(self._ptr, label, ranges._ptr)
- utils._handle_func_status(
+ bt2_utils._handle_func_status(
status, "cannot add mapping to enumeration field class object"
)
_get_mapping_labels_for_value = staticmethod(
native_bt.field_class_enumeration_unsigned_get_mapping_labels_for_value
)
- _check_int_type = staticmethod(utils._check_uint64)
+ _check_int_type = staticmethod(bt2_utils._check_uint64)
class _UnsignedEnumerationFieldClass(
_get_mapping_labels_for_value = staticmethod(
native_bt.field_class_enumeration_signed_get_mapping_labels_for_value
)
- _check_int_type = staticmethod(utils._check_int64)
+ _check_int_type = staticmethod(bt2_utils._check_int64)
class _SignedEnumerationFieldClass(
def _user_attributes(self, user_attributes):
value = bt2_value.create_value(user_attributes)
- utils._check_type(value, bt2_value.MapValue)
+ bt2_utils._check_type(value, bt2_value.MapValue)
native_bt.field_class_structure_member_set_user_attributes(
self._ptr, value._ptr
)
yield native_bt.field_class_structure_member_get_name(member_ptr)
def member_at_index(self, index):
- utils._check_uint64(index)
+ bt2_utils._check_uint64(index)
if index >= len(self):
raise IndexError
_structure_member_field_class_pycls = property(lambda _: _StructureFieldClassMember)
def append_member(self, name, field_class, user_attributes=None):
- utils._check_str(name)
- utils._check_type(field_class, _FieldClass)
+ bt2_utils._check_str(name)
+ bt2_utils._check_type(field_class, _FieldClass)
if name in self:
raise ValueError("duplicate member name '{}'".format(name))
status = native_bt.field_class_structure_append_member(
self._ptr, name, field_class._ptr
)
- utils._handle_func_status(
+ bt2_utils._handle_func_status(
status, "cannot append member to structure field class object"
)
_NAME = "Option (with boolean selector)"
def _selector_is_reversed(self, selector_is_reversed):
- utils._check_bool(selector_is_reversed)
+ bt2_utils._check_bool(selector_is_reversed)
native_bt.field_class_option_with_selector_field_bool_set_selector_is_reversed(
self._ptr, selector_is_reversed
)
def _user_attributes(self, user_attributes):
value = bt2_value.create_value(user_attributes)
- utils._check_type(value, bt2_value.MapValue)
+ bt2_utils._check_type(value, bt2_value.MapValue)
native_bt.field_class_variant_option_set_user_attributes(self._ptr, value._ptr)
_user_attributes = property(fset=_user_attributes)
yield native_bt.field_class_variant_option_get_name(base_opt_ptr)
def option_at_index(self, index):
- utils._check_uint64(index)
+ bt2_utils._check_uint64(index)
if index >= len(self):
raise IndexError
_NAME = "Variant (without selector)"
def append_option(self, name, field_class, user_attributes=None):
- utils._check_str(name)
- utils._check_type(field_class, _FieldClass)
+ bt2_utils._check_str(name)
+ bt2_utils._check_type(field_class, _FieldClass)
if name in self:
raise ValueError("duplicate option name '{}'".format(name))
status = native_bt.field_class_variant_without_selector_append_option(
self._ptr, name, field_class._ptr
)
- utils._handle_func_status(
+ bt2_utils._handle_func_status(
status, "cannot append option to variant field class object"
)
_NAME = "Variant (with selector)"
def append_option(self, name, field_class, ranges, user_attributes=None):
- utils._check_str(name)
- utils._check_type(field_class, _FieldClass)
- utils._check_type(ranges, self._variant_option_pycls._range_set_pycls)
+ bt2_utils._check_str(name)
+ bt2_utils._check_type(field_class, _FieldClass)
+ bt2_utils._check_type(ranges, self._variant_option_pycls._range_set_pycls)
if name in self:
raise ValueError("duplicate option name '{}'".format(name))
# TODO: check overlaps (precondition of self._append_option())
status = self._append_option(self._ptr, name, field_class._ptr, ranges._ptr)
- utils._handle_func_status(
+ bt2_utils._handle_func_status(
status, "cannot append option to variant field class object"
)
# Copyright (c) 2018 Francis Deslauriers <francis.deslauriers@efficios.com>
import collections
-from bt2 import native_bt, object
+from bt2 import native_bt
+from bt2 import object as bt2_object
class FieldPathScope:
pass
-class _FieldPathConst(object._SharedObject, collections.abc.Iterable):
+class _FieldPathConst(bt2_object._SharedObject, collections.abc.Iterable):
@staticmethod
def _get_ref(ptr):
native_bt.field_path_get_ref(ptr)
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object, utils
+from bt2 import native_bt
+from bt2 import object as bt2_object
+from bt2 import utils as bt2_utils
from bt2 import interrupter as bt2_interrupter
from bt2 import connection as bt2_connection
from bt2 import component as bt2_component
user_listener(component, port)
-class Graph(object._SharedObject):
+class Graph(bt2_object._SharedObject):
@staticmethod
def _get_ref(ptr):
native_bt.graph_get_ref(ptr)
native_bt.graph_put_ref(ptr)
def __init__(self, mip_version=0):
- utils._check_uint64(mip_version)
+ bt2_utils._check_uint64(mip_version)
if mip_version > bt2.get_maximal_mip_version():
raise ValueError("unknown MIP version {}".format(mip_version))
)
)
- utils._check_str(name)
- utils._check_log_level(logging_level)
+ bt2_utils._check_str(name)
+ bt2_utils._check_log_level(logging_level)
base_cc_ptr = component_class._bt_component_class_ptr()
if obj is not None and not native_bt.bt2_is_python_component_class(base_cc_ptr):
status, comp_ptr = add_fn(
self._ptr, cc_ptr, name, params_ptr, obj, logging_level
)
- utils._handle_func_status(status, "cannot add component to graph")
+ bt2_utils._handle_func_status(status, "cannot add component to graph")
assert comp_ptr
return bt2_component._create_component_from_const_ptr_and_get_ref(
comp_ptr, cc_type
)
def connect_ports(self, upstream_port, downstream_port):
- utils._check_type(upstream_port, bt2_port._OutputPortConst)
- utils._check_type(downstream_port, bt2_port._InputPortConst)
+ bt2_utils._check_type(upstream_port, bt2_port._OutputPortConst)
+ bt2_utils._check_type(downstream_port, bt2_port._InputPortConst)
status, conn_ptr = native_bt.graph_connect_ports(
self._ptr, upstream_port._ptr, downstream_port._ptr
)
- utils._handle_func_status(status, "cannot connect component ports within graph")
+ bt2_utils._handle_func_status(
+ status, "cannot connect component ports within graph"
+ )
assert conn_ptr
return bt2_connection._ConnectionConst._create_from_ptr_and_get_ref(conn_ptr)
def run_once(self):
status = native_bt.graph_run_once(self._ptr)
- utils._handle_func_status(status, "graph object could not run once")
+ bt2_utils._handle_func_status(status, "graph object could not run once")
def run(self):
status = native_bt.graph_run(self._ptr)
- utils._handle_func_status(status, "graph object stopped running")
+ bt2_utils._handle_func_status(status, "graph object stopped running")
def add_interrupter(self, interrupter):
- utils._check_type(interrupter, bt2_interrupter.Interrupter)
+ bt2_utils._check_type(interrupter, bt2_interrupter.Interrupter)
native_bt.graph_add_interrupter(self._ptr, interrupter._ptr)
@property
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object, utils
+from bt2 import native_bt
+from bt2 import object as bt2_object
+from bt2 import utils as bt2_utils
import collections.abc
import bt2
class _SignedIntegerRangeConst(_IntegerRangeConst):
- _is_type = staticmethod(utils._is_int64)
- _check_type = staticmethod(utils._check_int64)
+ _is_type = staticmethod(bt2_utils._is_int64)
+ _check_type = staticmethod(bt2_utils._check_int64)
class SignedIntegerRange(_SignedIntegerRangeConst, _IntegerRange):
class _UnsignedIntegerRangeConst(_IntegerRangeConst):
- _is_type = staticmethod(utils._is_uint64)
- _check_type = staticmethod(utils._check_uint64)
+ _is_type = staticmethod(bt2_utils._is_uint64)
+ _check_type = staticmethod(bt2_utils._check_uint64)
class UnsignedIntegerRange(_UnsignedIntegerRangeConst, _IntegerRange):
pass
-class _IntegerRangeSetConst(object._SharedObject, collections.abc.Set):
+class _IntegerRangeSetConst(bt2_object._SharedObject, collections.abc.Set):
def __len__(self):
range_set_ptr = self._as_range_set_ptr(self._ptr)
count = native_bt.integer_range_set_get_range_count(range_set_ptr)
rg = self._range_pycls(rg[0], rg[1])
status = self._add_range(self._ptr, rg.lower, rg.upper)
- utils._handle_func_status(status, "cannot add range to range set object")
+ bt2_utils._handle_func_status(status, "cannot add range to range set object")
def discard(self, rg):
raise NotImplementedError
#
# Copyright (c) 2019 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object
+from bt2 import native_bt
+from bt2 import object as bt2_object
import bt2
-class Interrupter(object._SharedObject):
+class Interrupter(bt2_object._SharedObject):
@staticmethod
def _get_ref(ptr):
native_bt.interrupter_get_ref(ptr)
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object, utils
+from bt2 import native_bt
+from bt2 import object as bt2_object
+from bt2 import utils as bt2_utils
from bt2 import clock_snapshot as bt2_clock_snapshot
from bt2 import packet as bt2_packet
from bt2 import stream as bt2_stream
return _MESSAGE_TYPE_TO_CLS[msg_type]._create_from_ptr(ptr)
-class _MessageConst(object._SharedObject):
+class _MessageConst(bt2_object._SharedObject):
@staticmethod
def _get_ref(ptr):
native_bt.message_get_ref(ptr)
class _StreamMessage(_StreamMessageConst, _Message):
def _default_clock_snapshot(self, raw_value):
- utils._check_uint64(raw_value)
+ bt2_utils._check_uint64(raw_value)
self._set_default_clock_snapshot(self._ptr, raw_value)
_default_clock_snapshot = property(
_stream_pycls = property(lambda _: bt2_stream._Stream)
def _set_count(self, count):
- utils._check_uint64(count)
+ bt2_utils._check_uint64(count)
if count == 0:
raise ValueError("discarded {} count is 0".format(self._item_name))
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object, utils
+from bt2 import native_bt
+from bt2 import object as bt2_object
+from bt2 import utils as bt2_utils
from bt2 import message as bt2_message
import collections.abc
from bt2 import stream as bt2_stream
raise NotImplementedError
-class _UserComponentInputPortMessageIterator(object._SharedObject, _MessageIterator):
+class _UserComponentInputPortMessageIterator(
+ bt2_object._SharedObject, _MessageIterator
+):
@staticmethod
def _get_ref(ptr):
native_bt.message_iterator_get_ref(ptr)
status, msgs = native_bt.bt2_self_component_port_input_get_msg_range(
self._ptr
)
- utils._handle_func_status(
+ bt2_utils._handle_func_status(
status, "unexpected error: cannot advance the message iterator"
)
self._current_msgs = msgs
def can_seek_beginning(self):
(status, res) = native_bt.message_iterator_can_seek_beginning(self._ptr)
- utils._handle_func_status(
+ bt2_utils._handle_func_status(
status,
"cannot check whether or not message iterator can seek its beginning",
)
self._at = 0
status = native_bt.message_iterator_seek_beginning(self._ptr)
- utils._handle_func_status(status, "cannot seek message iterator beginning")
+ bt2_utils._handle_func_status(status, "cannot seek message iterator beginning")
def can_seek_ns_from_origin(self, ns_from_origin):
- utils._check_int64(ns_from_origin)
+ bt2_utils._check_int64(ns_from_origin)
(status, res) = native_bt.message_iterator_can_seek_ns_from_origin(
self._ptr, ns_from_origin
)
- utils._handle_func_status(
+ bt2_utils._handle_func_status(
status,
"cannot check whether or not message iterator can seek given ns from origin",
)
return res != 0
def seek_ns_from_origin(self, ns_from_origin):
- utils._check_int64(ns_from_origin)
+ bt2_utils._check_int64(ns_from_origin)
# Forget about buffered messages, they won't be valid after seeking.
self._current_msgs.clear()
status = native_bt.message_iterator_seek_ns_from_origin(
self._ptr, ns_from_origin
)
- utils._handle_func_status(
+ bt2_utils._handle_func_status(
status, "message iterator cannot seek given ns from origin"
)
self._ptr = ptr
def can_seek_forward(self, value):
- utils._check_bool(value)
+ bt2_utils._check_bool(value)
native_bt.self_message_iterator_configuration_set_can_seek_forward(
self._ptr, value
)
except Exception:
raise
- utils._check_type(msg, bt2_message._MessageConst)
+ bt2_utils._check_type(msg, bt2_message._MessageConst)
# The reference we return will be given to the message array.
# However, the `msg` Python object may stay alive, if the user has kept
# method indicates whether the iterator can seek beginning.
if hasattr(self, "_user_can_seek_beginning"):
can_seek_beginning = self._user_can_seek_beginning()
- utils._check_bool(can_seek_beginning)
+ bt2_utils._check_bool(can_seek_beginning)
return can_seek_beginning
else:
return hasattr(self, "_user_seek_beginning")
if hasattr(self, "_user_can_seek_ns_from_origin"):
can_seek_ns_from_origin = self._user_can_seek_ns_from_origin(ns_from_origin)
- utils._check_bool(can_seek_ns_from_origin)
+ bt2_utils._check_bool(can_seek_ns_from_origin)
return can_seek_ns_from_origin
else:
return hasattr(self, "_user_seek_ns_from_origin")
self._user_seek_ns_from_origin(ns_from_origin)
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_message_iterator(
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 _UserComponentInputPortMessageIterator(msg_iter_ptr)
def _create_event_message(self, event_class, parent, default_clock_snapshot=None):
- utils._check_type(event_class, bt2_event_class._EventClass)
+ bt2_utils._check_type(event_class, bt2_event_class._EventClass)
if event_class.stream_class.supports_packets:
- utils._check_type(parent, bt2_packet._Packet)
+ bt2_utils._check_type(parent, bt2_packet._Packet)
else:
- utils._check_type(parent, bt2_stream._Stream)
+ bt2_utils._check_type(parent, bt2_stream._Stream)
if default_clock_snapshot is not None:
if event_class.stream_class.default_clock_class is None:
"event messages in this stream must not have a default clock snapshot"
)
- utils._check_uint64(default_clock_snapshot)
+ bt2_utils._check_uint64(default_clock_snapshot)
if event_class.stream_class.supports_packets:
ptr = native_bt.message_event_create_with_packet_and_default_clock_snapshot(
return bt2_message._EventMessage(ptr)
def _create_message_iterator_inactivity_message(self, clock_class, clock_snapshot):
- utils._check_type(clock_class, bt2_clock_class._ClockClass)
+ bt2_utils._check_type(clock_class, bt2_clock_class._ClockClass)
ptr = native_bt.message_message_iterator_inactivity_create(
self._bt_ptr, clock_class._ptr, clock_snapshot
)
return bt2_message._MessageIteratorInactivityMessage(ptr)
def _create_stream_beginning_message(self, stream, default_clock_snapshot=None):
- utils._check_type(stream, bt2_stream._Stream)
+ bt2_utils._check_type(stream, bt2_stream._Stream)
ptr = native_bt.message_stream_beginning_create(self._bt_ptr, stream._ptr)
if ptr is None:
return msg
def _create_stream_end_message(self, stream, default_clock_snapshot=None):
- utils._check_type(stream, bt2_stream._Stream)
+ bt2_utils._check_type(stream, bt2_stream._Stream)
ptr = native_bt.message_stream_end_create(self._bt_ptr, stream._ptr)
if ptr is None:
return msg
def _create_packet_beginning_message(self, packet, default_clock_snapshot=None):
- utils._check_type(packet, bt2_packet._Packet)
+ bt2_utils._check_type(packet, bt2_packet._Packet)
if packet.stream.cls.packets_have_beginning_default_clock_snapshot:
if default_clock_snapshot is None:
"packet beginning messages in this stream must have a default clock snapshot"
)
- utils._check_uint64(default_clock_snapshot)
+ bt2_utils._check_uint64(default_clock_snapshot)
ptr = native_bt.message_packet_beginning_create_with_default_clock_snapshot(
self._bt_ptr, packet._ptr, default_clock_snapshot
)
return bt2_message._PacketBeginningMessage(ptr)
def _create_packet_end_message(self, packet, default_clock_snapshot=None):
- utils._check_type(packet, bt2_packet._Packet)
+ bt2_utils._check_type(packet, bt2_packet._Packet)
if packet.stream.cls.packets_have_end_default_clock_snapshot:
if default_clock_snapshot is None:
"packet end messages in this stream must have a default clock snapshot"
)
- utils._check_uint64(default_clock_snapshot)
+ bt2_utils._check_uint64(default_clock_snapshot)
ptr = native_bt.message_packet_end_create_with_default_clock_snapshot(
self._bt_ptr, packet._ptr, default_clock_snapshot
)
def _create_discarded_events_message(
self, stream, count=None, beg_clock_snapshot=None, end_clock_snapshot=None
):
- utils._check_type(stream, bt2_stream._Stream)
+ bt2_utils._check_type(stream, bt2_stream._Stream)
if not stream.cls.supports_discarded_events:
raise ValueError("stream class does not support discarded events")
"discarded events have default clock snapshots for this stream class"
)
- utils._check_uint64(beg_clock_snapshot)
- utils._check_uint64(end_clock_snapshot)
+ bt2_utils._check_uint64(beg_clock_snapshot)
+ bt2_utils._check_uint64(end_clock_snapshot)
if beg_clock_snapshot > end_clock_snapshot:
raise ValueError(
def _create_discarded_packets_message(
self, stream, count=None, beg_clock_snapshot=None, end_clock_snapshot=None
):
- utils._check_type(stream, bt2_stream._Stream)
+ bt2_utils._check_type(stream, bt2_stream._Stream)
if not stream.cls.supports_discarded_packets:
raise ValueError("stream class does not support discarded packets")
"discarded packets have default clock snapshots for this stream class"
)
- utils._check_uint64(beg_clock_snapshot)
- utils._check_uint64(end_clock_snapshot)
+ bt2_utils._check_uint64(beg_clock_snapshot)
+ bt2_utils._check_uint64(end_clock_snapshot)
if beg_clock_snapshot > end_clock_snapshot:
raise ValueError(
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, utils
+from bt2 import native_bt
+from bt2 import utils as bt2_utils
import bt2
def get_greatest_operative_mip_version(
component_descriptors, log_level=bt2.LoggingLevel.NONE
):
- utils._check_log_level(log_level)
+ bt2_utils._check_log_level(log_level)
comp_descr_set_ptr = native_bt.component_descriptor_set_create()
if comp_descr_set_ptr is None:
status = native_bt.bt2_component_descriptor_set_add_descriptor_with_initialize_method_data(
comp_descr_set_ptr, base_cc_ptr, params_ptr, descr.obj
)
- utils._handle_func_status(
+ bt2_utils._handle_func_status(
status, "cannot add descriptor to component descriptor set"
)
if status == native_bt.__BT_FUNC_STATUS_NO_MATCH:
return None
- utils._handle_func_status(status, "cannot get greatest operative MIP version")
+ bt2_utils._handle_func_status(
+ status, "cannot get greatest operative MIP version"
+ )
return version
finally:
native_bt.component_descriptor_set_put_ref(comp_descr_set_ptr)
#
# Copyright (c) 2016-2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object
+from bt2 import native_bt
+from bt2 import object as bt2_object
from bt2 import field as bt2_field
return bt2_stream
-class _PacketConst(object._SharedObject):
+class _PacketConst(bt2_object._SharedObject):
@staticmethod
def _get_ref(ptr):
native_bt.packet_get_ref(ptr)
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object, utils
+from bt2 import native_bt
+from bt2 import object as bt2_object
+from bt2 import utils as bt2_utils
import collections.abc
from bt2 import component as bt2_component
import os.path
def find_plugins_in_path(path, recurse=True, fail_on_load_error=False):
- utils._check_str(path)
- utils._check_bool(recurse)
- utils._check_bool(fail_on_load_error)
+ bt2_utils._check_str(path)
+ bt2_utils._check_bool(recurse)
+ bt2_utils._check_bool(fail_on_load_error)
plugin_set_ptr = None
if os.path.isfile(path):
if status == native_bt.__BT_FUNC_STATUS_NOT_FOUND:
return
- utils._handle_func_status(status, "failed to find plugins")
+ bt2_utils._handle_func_status(status, "failed to find plugins")
assert plugin_set_ptr is not None
return _PluginSet._create_from_ptr(plugin_set_ptr)
find_in_static=True,
fail_on_load_error=False,
):
- utils._check_bool(find_in_std_env_var)
- utils._check_bool(find_in_user_dir)
- utils._check_bool(find_in_sys_dir)
- utils._check_bool(find_in_static)
- utils._check_bool(fail_on_load_error)
+ bt2_utils._check_bool(find_in_std_env_var)
+ bt2_utils._check_bool(find_in_user_dir)
+ bt2_utils._check_bool(find_in_sys_dir)
+ bt2_utils._check_bool(find_in_static)
+ bt2_utils._check_bool(fail_on_load_error)
plugin_set_ptr = None
status, plugin_set_ptr = native_bt.bt2_plugin_find_all(
if status == native_bt.__BT_FUNC_STATUS_NOT_FOUND:
return
- utils._handle_func_status(status, "failed to find plugins")
+ bt2_utils._handle_func_status(status, "failed to find plugins")
assert plugin_set_ptr is not None
return _PluginSet._create_from_ptr(plugin_set_ptr)
find_in_static=True,
fail_on_load_error=False,
):
- utils._check_str(name)
- utils._check_bool(fail_on_load_error)
+ bt2_utils._check_str(name)
+ bt2_utils._check_bool(fail_on_load_error)
status, ptr = native_bt.bt2_plugin_find(
name,
int(find_in_std_env_var),
if status == native_bt.__BT_FUNC_STATUS_NOT_FOUND:
return
- utils._handle_func_status(status, "failed to find plugin")
+ bt2_utils._handle_func_status(status, "failed to find plugin")
assert ptr is not None
return _Plugin._create_from_ptr(ptr)
-class _PluginSet(object._SharedObject, collections.abc.Sequence):
+class _PluginSet(bt2_object._SharedObject, collections.abc.Sequence):
@staticmethod
def _put_ref(ptr):
native_bt.plugin_set_put_ref(ptr)
return count
def __getitem__(self, index):
- utils._check_uint64(index)
+ bt2_utils._check_uint64(index)
if index >= len(self):
raise IndexError
self._plugin = plugin
def __getitem__(self, key):
- utils._check_str(key)
+ bt2_utils._check_str(key)
cc_ptr = self._borrow_component_class_by_name(self._plugin._ptr, key)
if cc_ptr is None:
_comp_cls_type = native_bt.COMPONENT_CLASS_TYPE_SINK
-class _Plugin(object._SharedObject):
+class _Plugin(bt2_object._SharedObject):
@staticmethod
def _put_ref(ptr):
native_bt.plugin_put_ref(ptr)
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object
+from bt2 import native_bt
+from bt2 import object as bt2_object
def _bt2_connection():
return cls._create_from_ptr_and_get_ref(ptr)
-class _PortConst(object._SharedObject):
+class _PortConst(bt2_object._SharedObject):
@classmethod
def _get_ref(cls, ptr):
ptr = cls._as_port_ptr(ptr)
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import utils
+from bt2 import utils as bt2_utils
from bt2 import component as bt2_component
import sys
"cannot find module '{}' in loaded modules".format(module_name)
)
- utils._check_str(name)
+ bt2_utils._check_str(name)
if description is not None:
- utils._check_str(description)
+ bt2_utils._check_str(description)
if author is not None:
- utils._check_str(author)
+ bt2_utils._check_str(author)
if license is not None:
- utils._check_str(license)
+ bt2_utils._check_str(license)
if version is not None:
if not _validate_version(version):
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object, utils
+from bt2 import native_bt
+from bt2 import object as bt2_object
+from bt2 import utils as bt2_utils
from bt2 import interrupter as bt2_interrupter
from bt2 import value as bt2_value
import bt2
return native_bt.query_executor_get_logging_level(self._common_ptr)
-class QueryExecutor(object._SharedObject, _QueryExecutorCommon):
+class QueryExecutor(bt2_object._SharedObject, _QueryExecutorCommon):
@staticmethod
def _get_ref(ptr):
native_bt.query_executor_get_ref(ptr)
o = component_class
raise TypeError("'{}' is not a component class object".format(o))
- utils._check_str(object_name)
+ bt2_utils._check_str(object_name)
if params is None:
params_ptr = native_bt.value_null
self._method_obj = method_obj
def add_interrupter(self, interrupter):
- utils._check_type(interrupter, bt2_interrupter.Interrupter)
+ bt2_utils._check_type(interrupter, bt2_interrupter.Interrupter)
native_bt.query_executor_add_interrupter(self._ptr, interrupter._ptr)
@property
return bt2_interrupter.Interrupter._create_from_ptr_and_get_ref(ptr)
def _set_logging_level(self, log_level):
- utils._check_log_level(log_level)
+ bt2_utils._check_log_level(log_level)
status = native_bt.query_executor_set_logging_level(self._ptr, log_level)
- utils._handle_func_status(status, "cannot set query executor's logging level")
+ bt2_utils._handle_func_status(
+ status, "cannot set query executor's logging level"
+ )
logging_level = property(
fget=_QueryExecutorCommon.logging_level, fset=_set_logging_level
def query(self):
status, result_ptr = native_bt.query_executor_query(self._ptr)
- utils._handle_func_status(status, "cannot query component class")
+ bt2_utils._handle_func_status(status, "cannot query component class")
assert result_ptr is not None
return bt2_value._create_from_const_ptr(result_ptr)
#
# Copyright (c) 2016-2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, utils
+from bt2 import native_bt
+from bt2 import utils as bt2_utils
from bt2 import object as bt2_object
from bt2 import packet as bt2_packet
from bt2 import stream_class as bt2_stream_class
def _user_attributes(self, user_attributes):
value = bt2_value.create_value(user_attributes)
- utils._check_type(value, bt2_value.MapValue)
+ bt2_utils._check_type(value, bt2_value.MapValue)
native_bt.stream_set_user_attributes(self._ptr, value._ptr)
_user_attributes = property(
)
def _name(self, name):
- utils._check_str(name)
+ bt2_utils._check_str(name)
native_bt.stream_set_name(self._ptr, name)
_name = property(fget=_StreamConst.name.fget, fset=_name)
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object, utils
+from bt2 import native_bt
+from bt2 import object as bt2_object
+from bt2 import utils as bt2_utils
from bt2 import field_class as bt2_field_class
from bt2 import event_class as bt2_event_class
from bt2 import clock_class as bt2_clock_class
return bt2_trace_class
-class _StreamClassConst(object._SharedObject, collections.abc.Mapping):
+class _StreamClassConst(bt2_object._SharedObject, collections.abc.Mapping):
@staticmethod
def _get_ref(ptr):
native_bt.stream_class_get_ref(ptr)
_clock_class_cls = property(lambda _: bt2_clock_class._ClockClassConst)
def __getitem__(self, key):
- utils._check_int64(key)
+ bt2_utils._check_int64(key)
ec_ptr = self._borrow_event_class_ptr_by_id(self._ptr, key)
if ec_ptr is None:
"id not provided, but stream class does not assign automatic event class ids"
)
- utils._check_uint64(id)
+ bt2_utils._check_uint64(id)
ec_ptr = native_bt.event_class_create_with_id(self._ptr, id)
event_class = bt2_event_class._EventClass._create_from_ptr(ec_ptr)
def _name(self, name):
status = native_bt.stream_class_set_name(self._ptr, name)
- utils._handle_func_status(status, "cannot set stream class object's name")
+ bt2_utils._handle_func_status(status, "cannot set stream class object's name")
_name = property(fset=_name)
status = native_bt.stream_class_set_packet_context_field_class(
self._ptr, packet_context_field_class._ptr
)
- utils._handle_func_status(
+ bt2_utils._handle_func_status(
status, "cannot set stream class' packet context field class"
)
def _event_common_context_field_class(self, event_common_context_field_class):
set_context_fn = native_bt.stream_class_set_event_common_context_field_class
status = set_context_fn(self._ptr, event_common_context_field_class._ptr)
- utils._handle_func_status(
+ bt2_utils._handle_func_status(
status, "cannot set stream class' event context field type"
)
):
# Name
if name is not None:
- utils._check_str(name)
+ bt2_utils._check_str(name)
# User attributes
if user_attributes is not None:
value = bt2_value.create_value(user_attributes)
- utils._check_type(value, bt2_value.MapValue)
+ bt2_utils._check_type(value, bt2_value.MapValue)
# Packet context field class
if packet_context_field_class is not None:
"cannot have a packet context field class without supporting packets"
)
- utils._check_type(
+ bt2_utils._check_type(
packet_context_field_class, bt2_field_class._StructureFieldClass
)
# Event common context field class
if event_common_context_field_class is not None:
- utils._check_type(
+ bt2_utils._check_type(
event_common_context_field_class, bt2_field_class._StructureFieldClass
)
# Default clock class
if default_clock_class is not None:
- utils._check_type(default_clock_class, bt2_clock_class._ClockClass)
+ bt2_utils._check_type(default_clock_class, bt2_clock_class._ClockClass)
# Assigns automatic event class id
- utils._check_bool(assigns_automatic_event_class_id)
+ bt2_utils._check_bool(assigns_automatic_event_class_id)
# Assigns automatic stream id
- utils._check_bool(assigns_automatic_stream_id)
+ bt2_utils._check_bool(assigns_automatic_stream_id)
# Packets
- utils._check_bool(supports_packets)
- utils._check_bool(packets_have_beginning_default_clock_snapshot)
- utils._check_bool(packets_have_end_default_clock_snapshot)
+ bt2_utils._check_bool(supports_packets)
+ bt2_utils._check_bool(packets_have_beginning_default_clock_snapshot)
+ bt2_utils._check_bool(packets_have_end_default_clock_snapshot)
if not supports_packets:
if packets_have_beginning_default_clock_snapshot:
)
# Discarded events
- utils._check_bool(supports_discarded_events)
- utils._check_bool(discarded_events_have_default_clock_snapshots)
+ bt2_utils._check_bool(supports_discarded_events)
+ bt2_utils._check_bool(discarded_events_have_default_clock_snapshots)
if discarded_events_have_default_clock_snapshots:
if not supports_discarded_events:
)
# Discarded packets
- utils._check_bool(supports_discarded_packets)
- utils._check_bool(discarded_packets_have_default_clock_snapshots)
+ bt2_utils._check_bool(supports_discarded_packets)
+ bt2_utils._check_bool(discarded_packets_have_default_clock_snapshots)
if supports_discarded_packets and not supports_packets:
raise ValueError(
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object, utils
+from bt2 import native_bt
+from bt2 import object as bt2_object
+from bt2 import utils as bt2_utils
import collections.abc
from bt2 import value as bt2_value
from bt2 import stream as bt2_stream
self._trace = trace
def __getitem__(self, key):
- utils._check_str(key)
+ bt2_utils._check_str(key)
borrow_entry_fn = native_bt.trace_borrow_environment_entry_value_by_name_const
value_ptr = borrow_entry_fn(self._trace._ptr, key)
raise TypeError("expected str or int, got {}".format(type(value)))
status = set_env_entry_fn(self._trace._ptr, key, value)
- utils._handle_func_status(status, "cannot set trace object's environment entry")
+ bt2_utils._handle_func_status(
+ status, "cannot set trace object's environment entry"
+ )
def __delitem__(self, key):
raise NotImplementedError
-class _TraceConst(object._SharedObject, collections.abc.Mapping):
+class _TraceConst(bt2_object._SharedObject, collections.abc.Mapping):
@staticmethod
def _get_ref(ptr):
native_bt.trace_get_ref(ptr)
return count
def __getitem__(self, id):
- utils._check_uint64(id)
+ bt2_utils._check_uint64(id)
stream_ptr = self._borrow_stream_ptr_by_id(self._ptr, id)
if not callable(listener):
raise TypeError("'listener' parameter is not callable")
- handle = utils._ListenerHandle(self.addr)
+ handle = bt2_utils._ListenerHandle(self.addr)
fn = native_bt.bt2_trace_add_destruction_listener
listener_from_native = functools.partial(
)
status, listener_id = fn(self._ptr, listener_from_native)
- utils._handle_func_status(
+ bt2_utils._handle_func_status(
status, "cannot add destruction listener to trace object"
)
return handle
def remove_destruction_listener(self, listener_handle):
- utils._check_type(listener_handle, utils._ListenerHandle)
+ bt2_utils._check_type(listener_handle, bt2_utils._ListenerHandle)
if listener_handle._addr != self.addr:
raise ValueError(
status = native_bt.trace_remove_destruction_listener(
self._ptr, listener_handle._listener_id
)
- utils._handle_func_status(status)
+ bt2_utils._handle_func_status(status)
listener_handle._invalidate()
_trace_env_pycls = property(lambda _: _TraceEnvironment)
def _name(self, name):
- utils._check_str(name)
+ bt2_utils._check_str(name)
status = native_bt.trace_set_name(self._ptr, name)
- utils._handle_func_status(status, "cannot set trace class object's name")
+ bt2_utils._handle_func_status(status, "cannot set trace class object's name")
_name = property(fset=_name)
def _user_attributes(self, user_attributes):
value = bt2_value.create_value(user_attributes)
- utils._check_type(value, bt2_value.MapValue)
+ bt2_utils._check_type(value, bt2_value.MapValue)
native_bt.trace_set_user_attributes(self._ptr, value._ptr)
_user_attributes = property(fset=_user_attributes)
def _uuid(self, uuid):
- utils._check_type(uuid, uuidp.UUID)
+ bt2_utils._check_type(uuid, uuidp.UUID)
native_bt.trace_set_uuid(self._ptr, uuid.bytes)
_uuid = property(fset=_uuid)
def create_stream(self, stream_class, id=None, name=None, user_attributes=None):
- utils._check_type(stream_class, bt2_stream_class._StreamClass)
+ bt2_utils._check_type(stream_class, bt2_stream_class._StreamClass)
if stream_class.assigns_automatic_stream_id:
if id is not None:
"id not provided, but stream class does not assign automatic stream ids"
)
- utils._check_uint64(id)
+ bt2_utils._check_uint64(id)
stream_ptr = native_bt.stream_create_with_id(
stream_class._ptr, self._ptr, id
)
# Copyright (c) 2018 Francis Deslauriers <francis.deslauriers@efficios.com>
# Copyright (c) 2019 Simon Marchi <simon.marchi@efficios.com>
-from bt2 import native_bt, utils, object
+from bt2 import native_bt
+from bt2 import utils as bt2_utils
+from bt2 import object as bt2_object
from bt2 import stream_class as bt2_stream_class
from bt2 import field_class as bt2_field_class
from bt2 import integer_range_set as bt2_integer_range_set
handle._invalidate()
-class _TraceClassConst(object._SharedObject, collections.abc.Mapping):
+class _TraceClassConst(bt2_object._SharedObject, collections.abc.Mapping):
@staticmethod
def _get_ref(ptr):
native_bt.trace_class_get_ref(ptr)
# Get a stream class by stream id.
def __getitem__(self, key):
- utils._check_uint64(key)
+ bt2_utils._check_uint64(key)
sc_ptr = self._borrow_stream_class_ptr_by_id(self._ptr, key)
if sc_ptr is None:
if not callable(listener):
raise TypeError("'listener' parameter is not callable")
- handle = utils._ListenerHandle(self.addr)
+ handle = bt2_utils._ListenerHandle(self.addr)
listener_from_native = functools.partial(
_trace_class_destruction_listener_from_native, listener, handle
fn = native_bt.bt2_trace_class_add_destruction_listener
status, listener_id = fn(self._ptr, listener_from_native)
- utils._handle_func_status(
+ bt2_utils._handle_func_status(
status, "cannot add destruction listener to trace class object"
)
return handle
def remove_destruction_listener(self, listener_handle):
- utils._check_type(listener_handle, utils._ListenerHandle)
+ bt2_utils._check_type(listener_handle, bt2_utils._ListenerHandle)
if listener_handle._addr != self.addr:
raise ValueError(
status = native_bt.trace_class_remove_destruction_listener(
self._ptr, listener_handle._listener_id
)
- utils._handle_func_status(status)
+ bt2_utils._handle_func_status(status)
listener_handle._invalidate()
"id not provided, but trace class does not assign automatic stream class ids"
)
- utils._check_uint64(id)
+ bt2_utils._check_uint64(id)
sc_ptr = native_bt.stream_class_create_with_id(self._ptr, id)
sc = bt2_stream_class._StreamClass._create_from_ptr(sc_ptr)
def _user_attributes(self, user_attributes):
value = bt2_value.create_value(user_attributes)
- utils._check_type(value, bt2_value.MapValue)
+ bt2_utils._check_type(value, bt2_value.MapValue)
native_bt.trace_class_set_user_attributes(self._ptr, value._ptr)
_user_attributes = property(fset=_user_attributes)
def _assigns_automatic_stream_class_id(self, auto_id):
- utils._check_bool(auto_id)
+ bt2_utils._check_bool(auto_id)
return native_bt.trace_class_set_assigns_automatic_stream_class_id(
self._ptr, auto_id
)
return fc
def create_bit_array_field_class(self, length, user_attributes=None):
- utils._check_uint64(length)
+ bt2_utils._check_uint64(length)
if length < 1 or length > 64:
raise ValueError(
return fc
def create_static_array_field_class(self, elem_fc, length, user_attributes=None):
- utils._check_type(elem_fc, bt2_field_class._FieldClass)
- utils._check_uint64(length)
+ bt2_utils._check_type(elem_fc, bt2_field_class._FieldClass)
+ bt2_utils._check_uint64(length)
ptr = native_bt.field_class_array_static_create(self._ptr, elem_fc._ptr, length)
self._check_field_class_create_status(ptr, "static array")
fc = bt2_field_class._StaticArrayFieldClass._create_from_ptr(ptr)
def create_dynamic_array_field_class(
self, elem_fc, length_fc=None, user_attributes=None
):
- utils._check_type(elem_fc, bt2_field_class._FieldClass)
+ bt2_utils._check_type(elem_fc, bt2_field_class._FieldClass)
length_fc_ptr = None
if length_fc is not None:
- utils._check_type(length_fc, bt2_field_class._UnsignedIntegerFieldClass)
+ bt2_utils._check_type(length_fc, bt2_field_class._UnsignedIntegerFieldClass)
length_fc_ptr = length_fc._ptr
ptr = native_bt.field_class_array_dynamic_create(
def create_option_without_selector_field_class(
self, content_fc, user_attributes=None
):
- utils._check_type(content_fc, bt2_field_class._FieldClass)
+ bt2_utils._check_type(content_fc, bt2_field_class._FieldClass)
ptr = native_bt.field_class_option_without_selector_create(
self._ptr, content_fc._ptr
)
def create_option_with_bool_selector_field_class(
self, content_fc, selector_fc, selector_is_reversed=False, user_attributes=None
):
- utils._check_type(content_fc, bt2_field_class._FieldClass)
- utils._check_bool(selector_is_reversed)
- utils._check_type(selector_fc, bt2_field_class._BoolFieldClass)
+ bt2_utils._check_type(content_fc, bt2_field_class._FieldClass)
+ bt2_utils._check_bool(selector_is_reversed)
+ bt2_utils._check_type(selector_fc, bt2_field_class._BoolFieldClass)
ptr = native_bt.field_class_option_with_selector_field_bool_create(
self._ptr, content_fc._ptr, selector_fc._ptr
)
def create_option_with_integer_selector_field_class(
self, content_fc, selector_fc, ranges, user_attributes=None
):
- utils._check_type(content_fc, bt2_field_class._FieldClass)
- utils._check_type(selector_fc, bt2_field_class._IntegerFieldClass)
+ bt2_utils._check_type(content_fc, bt2_field_class._FieldClass)
+ bt2_utils._check_type(selector_fc, bt2_field_class._IntegerFieldClass)
if len(ranges) == 0:
raise ValueError("integer range set is empty")
if isinstance(selector_fc, bt2_field_class._UnsignedIntegerFieldClass):
- utils._check_type(ranges, bt2_integer_range_set.UnsignedIntegerRangeSet)
+ bt2_utils._check_type(ranges, bt2_integer_range_set.UnsignedIntegerRangeSet)
ptr = native_bt.field_class_option_with_selector_field_integer_unsigned_create(
self._ptr, content_fc._ptr, selector_fc._ptr, ranges._ptr
)
else:
- utils._check_type(ranges, bt2_integer_range_set.SignedIntegerRangeSet)
+ bt2_utils._check_type(ranges, bt2_integer_range_set.SignedIntegerRangeSet)
ptr = (
native_bt.field_class_option_with_selector_field_integer_signed_create(
self._ptr, content_fc._ptr, selector_fc._ptr, ranges._ptr
selector_fc_ptr = None
if selector_fc is not None:
- utils._check_type(selector_fc, bt2_field_class._IntegerFieldClass)
+ bt2_utils._check_type(selector_fc, bt2_field_class._IntegerFieldClass)
selector_fc_ptr = selector_fc._ptr
ptr = native_bt.field_class_variant_create(self._ptr, selector_fc_ptr)
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import utils, native_bt
+from bt2 import native_bt
+from bt2 import utils as bt2_utils
import bt2
import itertools
from bt2 import message_iterator as bt2_message_iterator
# TraceCollectionMessageIterator.
def __init__(self, params, obj, logging_level):
if logging_level is not None:
- utils._check_log_level(logging_level)
+ bt2_utils._check_log_level(logging_level)
self._params = bt2.create_value(params)
self._obj = obj
if plugin_set is None:
plugin_set = bt2.find_plugins()
else:
- utils._check_type(plugin_set, bt2_plugin._PluginSet)
+ bt2_utils._check_type(plugin_set, bt2_plugin._PluginSet)
res_ptr = native_bt.bt2_auto_discover_source_components(
inputs._ptr, plugin_set._ptr
assert "status" in res
status = res["status"]
- utils._handle_func_status(status, "cannot auto-discover source components")
+ bt2_utils._handle_func_status(status, "cannot auto-discover source components")
comp_specs = []
comp_specs_raw = res["results"]
end=None,
plugin_set=None,
):
- utils._check_bool(stream_intersection_mode)
+ bt2_utils._check_bool(stream_intersection_mode)
self._stream_intersection_mode = stream_intersection_mode
self._begin_ns = _get_ns(begin)
self._end_ns = _get_ns(end)
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
-from bt2 import native_bt, object, utils
+from bt2 import native_bt
+from bt2 import object as bt2_object
+from bt2 import utils as bt2_utils
import collections.abc
import functools
import numbers
)
-class _ValueConst(object._SharedObject, metaclass=abc.ABCMeta):
+class _ValueConst(bt2_object._SharedObject, metaclass=abc.ABCMeta):
@staticmethod
def _get_ref(ptr):
native_bt.value_get_ref(ptr)
class UnsignedIntegerValue(_UnsignedIntegerValueConst, _IntegerValue):
_NAME = "Unsigned integer"
- _check_int_range = staticmethod(utils._check_uint64)
+ _check_int_range = staticmethod(bt2_utils._check_uint64)
_create_default_value = staticmethod(native_bt.value_integer_unsigned_create)
_create_value = staticmethod(native_bt.value_integer_unsigned_create_init)
_set_value = staticmethod(native_bt.value_integer_unsigned_set)
class SignedIntegerValue(_SignedIntegerValueConst, _IntegerValue):
_NAME = "Signed integer"
- _check_int_range = staticmethod(utils._check_int64)
+ _check_int_range = staticmethod(bt2_utils._check_int64)
_create_default_value = staticmethod(native_bt.value_integer_signed_create)
_create_value = staticmethod(native_bt.value_integer_signed_create_init)
_set_value = staticmethod(native_bt.value_integer_signed_set)
if isinstance(value, _StringValueConst):
value = value._value
- utils._check_str(value)
+ bt2_utils._check_str(value)
return value
@property
def _set_value(self, value):
status = native_bt.value_string_set(self._ptr, self._value_to_str(value))
- utils._handle_func_status(status)
+ bt2_utils._handle_func_status(status)
value = property(fset=_set_value)
ptr = value._ptr
status = native_bt.value_array_set_element_by_index(self._ptr, index, ptr)
- utils._handle_func_status(status)
+ bt2_utils._handle_func_status(status)
def append(self, value):
value = create_value(value)
ptr = value._ptr
status = native_bt.value_array_append_element(self._ptr, ptr)
- utils._handle_func_status(status)
+ bt2_utils._handle_func_status(status)
def __iadd__(self, iterable):
# Python will raise a TypeError if there's anything wrong with
return native_bt.value_map_has_entry(self._ptr, key)
def _check_key_type(self, key):
- utils._check_str(key)
+ bt2_utils._check_str(key)
def _check_key(self, key):
if key not in self:
ptr = value._ptr
status = native_bt.value_map_insert_entry(self._ptr, key, ptr)
- utils._handle_func_status(status)
+ bt2_utils._handle_func_status(status)
_TYPE_TO_OBJ = {
projects / babeltrace.git / commitdiff
