from bt2 import native_bt
from bt2 import user_attributes as bt2_user_attrs
+typing = bt2_utils._typing_mod
+
class ClockClassOffset:
- def __init__(self, seconds=0, cycles=0):
+ def __init__(self, seconds: int = 0, cycles: int = 0):
bt2_utils._check_int64(seconds)
bt2_utils._check_int64(cycles)
self._seconds = seconds
self._cycles = cycles
@property
- def seconds(self):
+ def seconds(self) -> int:
return self._seconds
@property
- def cycles(self):
+ def cycles(self) -> int:
return self._cycles
- def __eq__(self, other):
+ def __eq__(self, other: object) -> bool:
if not isinstance(other, self.__class__):
# not comparing apples to apples
return False
return native_bt.clock_class_borrow_user_attributes_const(ptr)
@property
- def name(self):
+ def name(self) -> typing.Optional[str]:
return native_bt.clock_class_get_name(self._ptr)
@property
- def description(self):
+ def description(self) -> typing.Optional[str]:
return native_bt.clock_class_get_description(self._ptr)
@property
- def frequency(self):
+ def frequency(self) -> int:
return native_bt.clock_class_get_frequency(self._ptr)
@property
- def precision(self):
+ def precision(self) -> int:
precision = native_bt.clock_class_get_precision(self._ptr)
return precision
@property
- def offset(self):
+ def offset(self) -> ClockClassOffset:
offset_s, offset_cycles = native_bt.clock_class_get_offset(self._ptr)
return ClockClassOffset(offset_s, offset_cycles)
@property
- def origin_is_unix_epoch(self):
+ def origin_is_unix_epoch(self) -> bool:
return native_bt.clock_class_origin_is_unix_epoch(self._ptr)
@property
- def uuid(self):
+ def uuid(self) -> typing.Optional[uuidp.UUID]:
uuid_bytes = native_bt.clock_class_get_uuid(self._ptr)
if uuid_bytes is None:
return uuidp.UUID(bytes=uuid_bytes)
- def cycles_to_ns_from_origin(self, cycles):
+ def cycles_to_ns_from_origin(self, cycles: int) -> int:
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"
@functools.total_ordering
class _ClockSnapshotConst(bt2_object._UniqueObject):
@property
- def clock_class(self):
+ def clock_class(self) -> bt2_clock_class._ClockClassConst:
cc_ptr = native_bt.clock_snapshot_borrow_clock_class_const(self._ptr)
assert cc_ptr is not None
return bt2_clock_class._ClockClassConst._create_from_ptr_and_get_ref(cc_ptr)
@property
- def value(self):
+ def value(self) -> int:
return native_bt.clock_snapshot_get_value(self._ptr)
@property
- def ns_from_origin(self):
+ def ns_from_origin(self) -> int:
status, ns = native_bt.clock_snapshot_get_ns_from_origin(self._ptr)
bt2_utils._handle_func_status(
status, "cannot get clock snapshot's nanoseconds from origin"
)
return ns
- def __eq__(self, other):
+ def __eq__(self, other: object) -> bool:
if not isinstance(other, numbers.Integral):
return NotImplemented
return self.value == int(other)
- def __lt__(self, other):
+ def __lt__(self, other: object) -> bool:
if not isinstance(other, numbers.Integral):
return NotImplemented
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
import sys
+import uuid as uuidp
import collections.abc
from bt2 import port as bt2_port
from bt2 import message_iterator as bt2_message_iterator
from bt2 import integer_range_set as bt2_integer_range_set
+typing = bt2_utils._typing_mod
+
class _IncompleteUserClass(Exception):
pass
class _ComponentClassConst(bt2_object._SharedObject):
@property
- def name(self):
+ def name(self) -> str:
ptr = self._bt_as_component_class_ptr(self._ptr)
name = native_bt.component_class_get_name(ptr)
assert name is not None
return name
@property
- def description(self):
+ def description(self) -> typing.Optional[str]:
ptr = self._bt_as_component_class_ptr(self._ptr)
return native_bt.component_class_get_description(ptr)
@property
- def help(self):
+ def help(self) -> typing.Optional[str]:
ptr = self._bt_as_component_class_ptr(self._ptr)
return native_bt.component_class_get_help(ptr)
def _bt_component_class_ptr(self):
return self._bt_as_component_class_ptr(self._ptr)
- def __eq__(self, other):
+ def __eq__(self, other: object) -> bool:
if not isinstance(other, _ComponentClassConst):
try:
if not issubclass(other, _UserComponent):
self._comp_ports = comp_ports
self._at = 0
- def __next__(self):
+ def __next__(self) -> str:
if self._at == len(self._comp_ports):
raise StopIteration
self._get_port_count = get_port_count
self._port_pycls = port_pycls
- def __getitem__(self, key):
+ def __getitem__(self, key: str) -> bt2_port._PortConst:
bt2_utils._check_str(key)
port_ptr = self._borrow_port_ptr_by_name(self._component_ptr, key)
return self._port_pycls._create_from_ptr_and_get_ref(port_ptr)
- def __len__(self):
+ def __len__(self) -> int:
count = self._get_port_count(self._component_ptr)
assert count >= 0
return count
- def __iter__(self):
+ def __iter__(self) -> typing.Iterator[bt2_port._PortConst]:
return _PortIterator(self)
class _ComponentConst:
@property
- def name(self):
+ def name(self) -> str:
ptr = self._bt_as_component_ptr(self._ptr)
name = native_bt.component_get_name(ptr)
assert name is not None
return name
@property
- def logging_level(self):
+ def logging_level(self) -> int:
ptr = self._bt_as_component_ptr(self._ptr)
return native_bt.component_get_logging_level(ptr)
@property
- def cls(self):
+ def cls(self) -> _ComponentClassConst:
cc_ptr = self._bt_borrow_component_class_ptr(self._ptr)
assert cc_ptr is not None
return _create_component_class_from_const_ptr_and_get_ref(
cc_ptr, self._bt_comp_cls_type
)
- def __eq__(self, other):
+ def __eq__(self, other: object) -> bool:
if not hasattr(other, "addr"):
return False
native_bt.component_source_put_ref(ptr)
@property
- def output_ports(self):
+ def output_ports(self) -> _ComponentPorts:
return _ComponentPorts(
self._ptr,
native_bt.component_source_borrow_output_port_by_name_const,
native_bt.component_filter_put_ref(ptr)
@property
- def output_ports(self):
+ def output_ports(self) -> _ComponentPorts:
return _ComponentPorts(
self._ptr,
native_bt.component_filter_borrow_output_port_by_name_const,
)
@property
- def input_ports(self):
+ def input_ports(self) -> _ComponentPorts:
return _ComponentPorts(
self._ptr,
native_bt.component_filter_borrow_input_port_by_name_const,
native_bt.component_sink_put_ref(ptr)
@property
- def input_ports(self):
+ def input_ports(self) -> _ComponentPorts:
return _ComponentPorts(
self._ptr,
native_bt.component_sink_borrow_input_port_by_name_const,
cls._iter_cls = iter_cls
@property
- def name(cls):
+ def name(cls) -> str:
ptr = cls._bt_as_component_class_ptr(cls._bt_cc_ptr)
return native_bt.component_class_get_name(ptr)
@property
- def description(cls):
+ def description(cls) -> typing.Optional[str]:
ptr = cls._bt_as_component_class_ptr(cls._bt_cc_ptr)
return native_bt.component_class_get_description(ptr)
@property
- def help(cls):
+ def help(cls) -> typing.Optional[str]:
ptr = cls._bt_as_component_class_ptr(cls._bt_cc_ptr)
return native_bt.component_class_get_help(ptr)
@property
- def addr(cls):
+ def addr(cls) -> int:
return int(cls._bt_cc_ptr)
def _bt_get_supported_mip_versions_from_native(cls, params_ptr, obj, log_level):
class _UserComponent(metaclass=_UserComponentType):
@property
- def name(self):
+ def name(self) -> str:
ptr = self._bt_as_not_self_specific_component_ptr(self._bt_ptr)
ptr = self._bt_as_component_ptr(ptr)
name = native_bt.component_get_name(ptr)
return name
@property
- def logging_level(self):
+ def logging_level(self) -> int:
ptr = self._bt_as_not_self_specific_component_ptr(self._bt_ptr)
ptr = self._bt_as_component_ptr(ptr)
return native_bt.component_get_logging_level(ptr)
@property
- def cls(self):
+ def cls(self) -> _ComponentClassConst:
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_const_ptr_and_get_ref(
)
@property
- def addr(self):
+ def addr(self) -> int:
return int(self._bt_ptr)
@property
- def _graph_mip_version(self):
+ def _graph_mip_version(self) -> int:
ptr = self._bt_as_self_component_ptr(self._bt_ptr)
return native_bt.self_component_get_graph_mip_version(ptr)
self._user_port_connected(port, other_port)
def _create_trace_class(
- self, user_attributes=None, assigns_automatic_stream_class_id=True
- ):
+ self,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ assigns_automatic_stream_class_id: bool = True,
+ ) -> bt2_trace_class._TraceClass:
ptr = self._bt_as_self_component_ptr(self._bt_ptr)
tc_ptr = native_bt.trace_class_create(ptr)
def _create_clock_class(
self,
- frequency=None,
- name=None,
- user_attributes=None,
- description=None,
- precision=None,
- offset=None,
- origin_is_unix_epoch=True,
- uuid=None,
- ):
+ frequency: typing.Optional[int] = None,
+ name: typing.Optional[str] = None,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ description: typing.Optional[str] = None,
+ precision: typing.Optional[int] = None,
+ offset: typing.Optional[int] = None,
+ origin_is_unix_epoch: bool = True,
+ uuid: typing.Optional[uuidp.UUID] = None,
+ ) -> bt2_clock_class._ClockClass:
ptr = self._bt_as_self_component_ptr(self._bt_ptr)
cc_ptr = native_bt.clock_class_create(ptr)
_config_pycls = _UserSourceComponentConfiguration
@property
- def _output_ports(self):
+ def _output_ports(self) -> _ComponentPorts:
def get_output_port_count(self_ptr):
ptr = self._bt_as_not_self_specific_component_ptr(self_ptr)
return native_bt.component_source_get_output_port_count(ptr)
bt2_port._UserComponentOutputPort,
)
- def _add_output_port(self, name, user_data=None):
+ def _add_output_port(
+ self, name: str, user_data: object = None
+ ) -> bt2_port._UserComponentOutputPort:
bt2_utils._check_str(name)
if name in self._output_ports:
_config_pycls = _UserFilterComponentConfiguration
@property
- def _output_ports(self):
+ def _output_ports(self) -> _ComponentPorts:
def get_output_port_count(self_ptr):
ptr = self._bt_as_not_self_specific_component_ptr(self_ptr)
return native_bt.component_filter_get_output_port_count(ptr)
)
@property
- def _input_ports(self):
+ def _input_ports(self) -> _ComponentPorts:
def get_input_port_count(self_ptr):
ptr = self._bt_as_not_self_specific_component_ptr(self_ptr)
return native_bt.component_filter_get_input_port_count(ptr)
bt2_port._UserComponentInputPort,
)
- def _add_output_port(self, name, user_data=None):
+ def _add_output_port(
+ self, name: str, user_data: object = None
+ ) -> bt2_port._UserComponentOutputPort:
bt2_utils._check_str(name)
if name in self._output_ports:
self_port_ptr
)
- def _add_input_port(self, name, user_data=None):
+ def _add_input_port(
+ self, name: str, user_data: object = None
+ ) -> bt2_port._UserComponentInputPort:
bt2_utils._check_str(name)
if name in self._input_ports:
pass
@property
- def _input_ports(self):
+ def _input_ports(self) -> _ComponentPorts:
def get_input_port_count(self_ptr):
ptr = self._bt_as_not_self_specific_component_ptr(self_ptr)
return native_bt.component_sink_get_input_port_count(ptr)
bt2_port._UserComponentInputPort,
)
- def _add_input_port(self, name, user_data=None):
+ def _add_input_port(
+ self, name: str, user_data: object = None
+ ) -> bt2_port._UserComponentInputPort:
bt2_utils._check_str(name)
if name in self._input_ports:
self_port_ptr
)
- def _create_message_iterator(self, input_port):
+ def _create_message_iterator(
+ self, input_port: bt2_port._UserComponentInputPort
+ ) -> bt2_message_iterator._UserComponentInputPortMessageIterator:
bt2_utils._check_type(input_port, bt2_port._UserComponentInputPort)
if not input_port.is_connected:
return bt2_message_iterator._UserComponentInputPortMessageIterator(msg_iter_ptr)
@property
- def _is_interrupted(self):
+ def _is_interrupted(self) -> bool:
return bool(native_bt.self_component_sink_is_interrupted(self._bt_ptr))
native_bt.connection_put_ref(ptr)
@property
- def downstream_port(self):
+ def downstream_port(self) -> bt2_port._InputPortConst:
port_ptr = native_bt.connection_borrow_downstream_port_const(self._ptr)
return bt2_port._create_from_const_ptr_and_get_ref(
port_ptr, native_bt.PORT_TYPE_INPUT
)
@property
- def upstream_port(self):
+ def upstream_port(self) -> bt2_port._OutputPortConst:
port_ptr = native_bt.connection_borrow_upstream_port_const(self._ptr)
return bt2_port._create_from_const_ptr_and_get_ref(
port_ptr, native_bt.PORT_TYPE_OUTPUT
from bt2 import native_bt
+try:
+ import typing as _typing_mod # noqa: F401
+except ImportError:
+ from bt2 import local_typing as _typing_mod # noqa: F401
+
+typing = _typing_mod
+
class ComponentClassType:
SOURCE = native_bt.COMPONENT_CLASS_TYPE_SOURCE
self._line_number = native_bt.error_cause_get_line_number(ptr)
self._str = native_bt.bt2_format_bt_error_cause(ptr)
- def __str__(self):
+ def __str__(self) -> str:
return self._str
@property
- def message(self):
+ def message(self) -> str:
return self._message
@property
- def module_name(self):
+ def module_name(self) -> str:
return self._module_name
@property
- def file_name(self):
+ def file_name(self) -> str:
return self._file_name
@property
- def line_number(self):
+ def line_number(self) -> int:
return self._line_number
self._plugin_name = native_bt.error_cause_component_actor_get_plugin_name(ptr)
@property
- def component_name(self):
+ def component_name(self) -> str:
return self._component_name
@property
- def component_class_type(self):
+ def component_class_type(self) -> int:
return self._component_class_type
@property
- def component_class_name(self):
+ def component_class_name(self) -> str:
return self._component_class_name
@property
- def plugin_name(self):
+ def plugin_name(self) -> typing.Optional[str]:
return self._plugin_name
)
@property
- def component_class_type(self):
+ def component_class_type(self) -> int:
return self._component_class_type
@property
- def component_class_name(self):
+ def component_class_name(self) -> int:
return self._component_class_name
@property
- def plugin_name(self):
+ def plugin_name(self) -> typing.Optional[str]:
return self._plugin_name
)
@property
- def component_name(self):
+ def component_name(self) -> str:
return self._component_name
@property
- def component_output_port_name(self):
+ def component_output_port_name(self) -> str:
return self._component_output_port_name
@property
- def component_class_type(self):
+ def component_class_type(self) -> int:
return self._component_class_type
@property
- def component_class_name(self):
+ def component_class_name(self) -> str:
return self._component_class_name
@property
- def plugin_name(self):
+ def plugin_name(self) -> typing.Optional[str]:
return self._plugin_name
if self._ptr is not None:
native_bt.error_release(self._ptr)
- def __getitem__(self, index):
+ def __getitem__(self, index: int) -> _ErrorCause:
return self._causes[index]
- def __len__(self):
+ def __len__(self) -> int:
return len(self._causes)
- def __str__(self):
+ def __str__(self) -> str:
return self._str
from bt2 import native_bt
from bt2 import event_class as bt2_event_class
+typing = bt2_utils._typing_mod
+
class _EventConst(bt2_object._UniqueObject, collections.abc.Mapping):
_borrow_class_ptr = staticmethod(native_bt.event_borrow_class_const)
_stream_pycls = property(lambda _: bt2_stream._StreamConst)
@property
- def cls(self):
+ def cls(self) -> bt2_event_class._EventClassConst:
event_class_ptr = self._borrow_class_ptr(self._ptr)
assert event_class_ptr is not None
return self._event_class_pycls._create_from_ptr_and_get_ref(event_class_ptr)
@property
- def name(self):
+ def name(self) -> typing.Optional[str]:
return self.cls.name
@property
- def id(self):
+ def id(self) -> typing.Optional[int]:
return self.cls.id
@property
- def packet(self):
+ def packet(self) -> typing.Optional[bt2_packet._PacketConst]:
packet_ptr = self._borrow_packet_ptr(self._ptr)
if packet_ptr is None:
return self._packet_pycls._create_from_ptr_and_get_ref(packet_ptr)
@property
- def stream(self):
+ def stream(self) -> typing.Optional[bt2_stream._StreamConst]:
stream_ptr = self._borrow_stream_ptr(self._ptr)
assert stream_ptr is not None
return self._stream_pycls._create_from_ptr_and_get_ref(stream_ptr)
@property
- def common_context_field(self):
+ def common_context_field(self) -> typing.Optional[bt2_field._StructureFieldConst]:
field_ptr = self._borrow_common_context_field_ptr(self._ptr)
if field_ptr is None:
)
@property
- def specific_context_field(self):
+ def specific_context_field(self) -> typing.Optional[bt2_field._StructureFieldConst]:
field_ptr = self._borrow_specific_context_field_ptr(self._ptr)
if field_ptr is None:
)
@property
- def payload_field(self):
+ def payload_field(self) -> typing.Optional[bt2_field._StructureFieldConst]:
field_ptr = self._borrow_payload_field_ptr(self._ptr)
if field_ptr is None:
from bt2 import field_class as bt2_field_class
from bt2 import user_attributes as bt2_user_attrs
+typing = bt2_utils._typing_mod
+
+if typing.TYPE_CHECKING:
+ from bt2 import stream_class as bt2_stream_class
+
def _bt2_stream_class():
from bt2 import stream_class as bt2_stream_class
_stream_class_pycls = property(lambda s: _bt2_stream_class()._StreamClassConst)
@property
- def stream_class(self):
+ def stream_class(self) -> "bt2_stream_class._StreamClassConst":
sc_ptr = self._borrow_stream_class_ptr(self._ptr)
if sc_ptr is not None:
return self._stream_class_pycls._create_from_ptr_and_get_ref(sc_ptr)
@property
- def name(self):
+ def name(self) -> typing.Optional[str]:
return native_bt.event_class_get_name(self._ptr)
@property
- def id(self):
+ def id(self) -> int:
id = native_bt.event_class_get_id(self._ptr)
return id if id >= 0 else None
@property
- def log_level(self):
+ def log_level(self) -> typing.Optional[EventClassLogLevel]:
is_available, log_level = native_bt.event_class_get_log_level(self._ptr)
if is_available != native_bt.PROPERTY_AVAILABILITY_AVAILABLE:
return _EVENT_CLASS_LOG_LEVEL_TO_OBJ[log_level]
@property
- def emf_uri(self):
+ def emf_uri(self) -> typing.Optional[str]:
return native_bt.event_class_get_emf_uri(self._ptr)
@property
- def specific_context_field_class(self):
+ def specific_context_field_class(
+ self,
+ ) -> typing.Optional[bt2_field_class._StructureFieldClassConst]:
fc_ptr = self._borrow_specific_context_field_class_ptr(self._ptr)
if fc_ptr is None:
return self._create_field_class_from_ptr_and_get_ref(fc_ptr)
@property
- def payload_field_class(self):
+ def payload_field_class(
+ self,
+ ) -> typing.Optional[bt2_field_class._StructureFieldClassConst]:
fc_ptr = self._borrow_payload_field_class_ptr(self._ptr)
if fc_ptr is None:
from bt2 import native_bt
from bt2 import field_class as bt2_field_class
+typing = bt2_utils._typing_mod
+
def _create_field_from_ptr_template(
object_map, ptr, owner_ptr, owner_get_ref, owner_put_ref
)
_borrow_class_ptr = staticmethod(native_bt.field_borrow_class_const)
- def __eq__(self, other):
+ def __eq__(self, other: object) -> bool:
other = _get_leaf_field(other)
return self._spec_eq(other)
@property
- def cls(self):
+ def cls(self) -> bt2_field_class._FieldClassConst:
field_class_ptr = self._borrow_class_ptr(self._ptr)
assert field_class_ptr is not None
return self._create_field_class_from_ptr_and_get_ref(field_class_ptr)
def _repr(self):
raise NotImplementedError
- def __repr__(self):
+ def __repr__(self) -> str:
return self._repr()
_NAME = "Const bit array"
@property
- def value_as_integer(self):
+ def value_as_integer(self) -> int:
return native_bt.field_bit_array_get_value_as_integer(self._ptr)
def _spec_eq(self, other):
def _repr(self):
return repr(self.value_as_integer)
- def __str__(self):
+ def __str__(self) -> str:
return str(self.value_as_integer)
- def __len__(self):
+ def __len__(self) -> int:
return self.cls.length
"'{}' object is not a number object".format(other.__class__.__name__)
)
- def __int__(self):
+ def __int__(self) -> int:
return int(self._value)
- def __float__(self):
+ def __float__(self) -> float:
return float(self._value)
def _repr(self):
return repr(self._value)
- def __lt__(self, other):
+ def __lt__(self, other) -> bool:
if not isinstance(other, numbers.Number):
raise TypeError(
"unorderable types: {}() < {}()".format(
except Exception:
return False
- def __hash__(self):
+ def __hash__(self) -> int:
return hash(self._value)
def __rmod__(self, other):
class _NumericField(_NumericFieldConst, _Field):
- def __hash__(self):
+ def __hash__(self) -> int:
# Non const field are not hashable as their value may be modified
# without changing the underlying Python object.
raise TypeError("unhashable type: '{}'".format(self._NAME))
class _BoolFieldConst(_IntegralFieldConst, _FieldConst):
_NAME = "Const boolean"
- def __bool__(self):
+ def __bool__(self) -> bool:
return self._value
@classmethod
return "{} ({})".format(self._value, ", ".join(self.labels))
@property
- def labels(self):
+ def labels(self) -> typing.List[str]:
status, labels = self._get_mapping_labels(self._ptr)
bt2_utils._handle_func_status(status, "cannot get label for enumeration field")
except Exception:
return False
- def __lt__(self, other):
+ def __lt__(self, other) -> bool:
return self._value < self._value_to_str(other)
- def __bool__(self):
+ def __bool__(self) -> bool:
return bool(self._value)
- def __hash__(self):
+ def __hash__(self) -> int:
return hash(self._value)
def _repr(self):
return repr(self._value)
- def __str__(self):
+ def __str__(self) -> str:
return str(self._value)
- def __getitem__(self, index):
+ def __getitem__(self, index) -> str:
return self._value[index]
- def __len__(self):
+ def __len__(self) -> int:
return native_bt.field_string_get_length(self._ptr)
value = property(fset=_set_value)
- def __iadd__(self, value):
+ def __iadd__(self, value) -> "_StringField":
value = self._value_to_str(value)
status = native_bt.field_string_append(self._ptr, value)
bt2_utils._handle_func_status(
)
return self
- def __hash__(self):
+ def __hash__(self) -> int:
# Non const field are not hashable as their value may be modified
# without changing the underlying Python object.
raise TypeError("unhashable type: '{}'".format(self._NAME))
class _ContainerFieldConst(_FieldConst):
- def __bool__(self):
+ def __bool__(self) -> bool:
return len(self) != 0
def _count(self):
return len(self.cls)
- def __len__(self):
+ def __len__(self) -> int:
count = self._count()
assert count >= 0
return count
def _count(self):
return len(self.cls)
- def __iter__(self):
+ def __iter__(self) -> typing.Iterator[str]:
# same name iterator
return iter(self.cls)
items = ["{}: {}".format(repr(k), repr(v)) for k, v in self.items()]
return "{{{}}}".format(", ".join(items))
- def __getitem__(self, key):
+ def __getitem__(self, key) -> _FieldConst:
bt2_utils._check_str(key)
field_ptr = self._borrow_member_field_ptr_by_name(self._ptr, key)
field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
)
- def member_at_index(self, index):
+ def member_at_index(self, index) -> _FieldConst:
bt2_utils._check_uint64(index)
if index >= len(self):
native_bt.field_structure_borrow_member_field_by_name
)
- def __setitem__(self, key, value):
+ def __setitem__(self, key: str, value):
# raises if key is somehow invalid
field = self[key]
_borrow_field_ptr = staticmethod(native_bt.field_option_borrow_field_const)
@property
- def field(self):
+ def field(self) -> typing.Optional[_FieldConst]:
field_ptr = self._borrow_field_ptr(self._ptr)
if field_ptr is None:
)
@property
- def has_field(self):
+ def has_field(self) -> bool:
return self.field is not None
def _spec_eq(self, other):
return _get_leaf_field(self) == other
- def __bool__(self):
+ def __bool__(self) -> bool:
return self.has_field
- def __str__(self):
+ def __str__(self) -> str:
return str(self.field)
def _repr(self):
return len(self.cls)
@property
- def selected_option_index(self):
+ def selected_option_index(self) -> int:
return native_bt.field_variant_get_selected_option_index(self._ptr)
@property
- def selected_option(self):
+ def selected_option(self) -> _FieldConst:
# TODO: Is there a way to check if the variant field has a selected_option,
# so we can raise an exception instead of hitting a pre-condition check?
# If there is something, that check should be added to selected_option_index too.
def _spec_eq(self, other):
return _get_leaf_field(self) == other
- def __bool__(self):
+ def __bool__(self) -> bool:
raise NotImplementedError
- def __str__(self):
+ def __str__(self) -> str:
return str(self.selected_option)
def _repr(self):
length = property(fget=_get_length)
- def __getitem__(self, index):
+ def __getitem__(self, index: int) -> _FieldConst:
if not isinstance(index, numbers.Integral):
raise TypeError(
"'{}' is not an integral number object: invalid index".format(
field_ptr, self._owner_ptr, self._owner_get_ref, self._owner_put_ref
)
- def insert(self, index, value):
+ def insert(self, index: int, value):
raise NotImplementedError
def _spec_eq(self, other):
native_bt.field_array_borrow_element_field_by_index
)
- def __setitem__(self, index, value):
+ def __setitem__(self, index: int, value):
# raises if index is somehow invalid
field = self[index]
from bt2 import user_attributes as bt2_user_attrs
from bt2 import integer_range_set as bt2_integer_range_set
+typing = bt2_utils._typing_mod
+
def _obj_type_from_field_class_ptr_template(type_map, ptr):
typeid = native_bt.field_class_get_type(ptr)
class _IntegerFieldClassConst(_FieldClassConst):
@property
- def field_value_range(self):
+ def field_value_range(self) -> int:
size = native_bt.field_class_integer_get_field_value_range(self._ptr)
assert size >= 1
return size
@property
- def preferred_display_base(self):
+ def preferred_display_base(self) -> IntegerDisplayBase:
base = native_bt.field_class_integer_get_preferred_display_base(self._ptr)
assert base >= 0
return base
self._ranges = self._range_set_pycls._create_from_ptr_and_get_ref(ranges_ptr)
@property
- def label(self):
+ def label(self) -> str:
return self._label
@property
- def ranges(self):
+ def ranges(self) -> bt2_integer_range_set._IntegerRangeSetConst:
return self._ranges
class _EnumerationFieldClassConst(_IntegerFieldClassConst, collections.abc.Mapping):
- def __len__(self):
+ def __len__(self) -> int:
count = native_bt.field_class_enumeration_get_mapping_count(self._ptr)
assert count >= 0
return count
- def mappings_for_value(self, value):
+ def mappings_for_value(self, value: int) -> "list[str]":
self._check_int_type(value)
status, labels = self._get_mapping_labels_for_value(self._ptr, value)
)
return [self[label] for label in labels]
- def __iter__(self):
+ def __iter__(self) -> typing.Iterator[_EnumerationFieldClassMapping]:
for idx in range(len(self)):
mapping_ptr = self._borrow_mapping_ptr_by_index(self._ptr, idx)
yield self._mapping_pycls(mapping_ptr).label
- def __getitem__(self, label):
+ def __getitem__(self, label: str) -> _EnumerationFieldClassMapping:
bt2_utils._check_str(label)
mapping_ptr = self._borrow_mapping_ptr_by_label(self._ptr, label)
class _EnumerationFieldClass(_EnumerationFieldClassConst, _IntegerFieldClass):
- def add_mapping(self, label, ranges):
+ def add_mapping(
+ self, label: str, ranges: bt2_integer_range_set._IntegerRangeSetConst
+ ):
bt2_utils._check_str(label)
bt2_utils._check_type(ranges, self._range_set_pycls)
status, "cannot add mapping to enumeration field class object"
)
- def __iadd__(self, mappings):
+ def __iadd__(
+ self,
+ mappings: typing.Iterable[
+ typing.Tuple[str, bt2_integer_range_set._IntegerRangeSetConst]
+ ],
+ ) -> "_EnumerationFieldClass":
for label, ranges in mappings:
self.add_mapping(label, ranges)
self._member_ptr = member_ptr
@property
- def name(self):
+ def name(self) -> str:
name = native_bt.field_class_structure_member_get_name(self._member_ptr)
assert name is not None
return name
@property
- def field_class(self):
+ def field_class(self) -> _FieldClassConst:
fc_ptr = self._borrow_field_class_ptr(self._member_ptr)
assert fc_ptr is not None
return self._create_field_class_from_ptr_and_get_ref(fc_ptr)
lambda _: _StructureFieldClassMemberConst
)
- def __len__(self):
+ def __len__(self) -> int:
count = native_bt.field_class_structure_get_member_count(self._ptr)
assert count >= 0
return count
- def __getitem__(self, key):
+ def __getitem__(self, key: str) -> _StructureFieldClassMemberConst:
if not isinstance(key, str):
raise TypeError(
"key must be a 'str' object, got '{}'".format(key.__class__.__name__)
return self._structure_member_field_class_pycls(self, member_ptr)
- def __iter__(self):
+ def __iter__(self) -> typing.Iterator[str]:
for idx in range(len(self)):
member_ptr = self._borrow_member_ptr_by_index(self._ptr, idx)
assert member_ptr is not None
yield native_bt.field_class_structure_member_get_name(member_ptr)
- def member_at_index(self, index):
+ def member_at_index(self, index: int) -> _StructureFieldClassMemberConst:
bt2_utils._check_uint64(index)
if index >= len(self):
)
_structure_member_field_class_pycls = property(lambda _: _StructureFieldClassMember)
- def append_member(self, name, field_class, user_attributes=None):
+ def append_member(
+ self,
+ name: str,
+ field_class: _FieldClass,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ ):
bt2_utils._check_str(name)
bt2_utils._check_type(field_class, _FieldClass)
if user_attributes is not None:
self[name]._set_user_attributes(user_attributes_value)
- def __iadd__(self, members):
+ def __iadd__(
+ self, members: typing.Iterable[typing.Tuple[str, _FieldClass]]
+ ) -> "_StructureFieldClass":
for name, field_class in members:
self.append_member(name, field_class)
)
@property
- def field_class(self):
+ def field_class(self) -> _FieldClassConst:
elem_fc_ptr = self._borrow_field_class_ptr(self._ptr)
return self._create_field_class_from_ptr_and_get_ref(elem_fc_ptr)
_NAME = "Const option (with selector)"
@property
- def selector_field_path(self):
+ def selector_field_path(self) -> typing.Optional[bt2_field_path._FieldPathConst]:
ptr = native_bt.field_class_option_with_selector_field_borrow_selector_field_path_const(
self._ptr
)
_NAME = "Const option (with boolean selector)"
@property
- def selector_is_reversed(self):
+ def selector_is_reversed(self) -> bool:
return bool(
native_bt.field_class_option_with_selector_field_bool_selector_is_reversed(
self._ptr
_NAME = "Const option (with integer selector)"
@property
- def ranges(self):
+ def ranges(self) -> bt2_integer_range_set._IntegerRangeSetConst:
range_set_ptr = self._borrow_selector_ranges_ptr(self._ptr)
assert range_set_ptr is not None
return self._range_set_pycls._create_from_ptr_and_get_ref(range_set_ptr)
self._opt_ptr = option_ptr
@property
- def name(self):
+ def name(self) -> str:
name = native_bt.field_class_variant_option_get_name(self._opt_ptr)
assert name is not None
return name
@property
- def field_class(self):
+ def field_class(self) -> _FieldClassConst:
fc_ptr = self._borrow_field_class_ptr(self._opt_ptr)
assert fc_ptr is not None
return self._create_field_class_from_ptr_and_get_ref(fc_ptr)
super().__init__(owning_var_fc, self._as_option_ptr(spec_opt_ptr))
@property
- def ranges(self):
+ def ranges(self) -> bt2_integer_range_set._IntegerRangeSetConst:
range_set_ptr = self._borrow_ranges_ptr(self._spec_ptr)
assert range_set_ptr is not None
return self._range_set_pycls._create_from_ptr_and_get_ref(range_set_ptr)
def _create_option_from_ptr(self, opt_ptr):
return self._variant_option_pycls(self, opt_ptr)
- def __len__(self):
+ def __len__(self) -> int:
count = native_bt.field_class_variant_get_option_count(self._ptr)
assert count >= 0
return count
- def __getitem__(self, key):
+ def __getitem__(self, key: str) -> _VariantFieldClassOptionConst:
if not isinstance(key, str):
raise TypeError(
"key must be a 'str' object, got '{}'".format(key.__class__.__name__)
return self._create_option_from_ptr(opt_ptr)
- def __iter__(self):
+ def __iter__(self) -> typing.Iterator[str]:
for idx in range(len(self)):
opt_ptr = self._borrow_option_ptr_by_index(self._ptr, idx)
assert opt_ptr is not None
base_opt_ptr = self._as_option_ptr(opt_ptr)
yield native_bt.field_class_variant_option_get_name(base_opt_ptr)
- def option_at_index(self, index):
+ def option_at_index(self, index: int) -> _VariantFieldClassOptionConst:
bt2_utils._check_uint64(index)
if index >= len(self):
):
_NAME = "Variant (without selector)"
- def append_option(self, name, field_class, user_attributes=None):
+ def append_option(
+ self,
+ name: str,
+ field_class: _FieldClass,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ ):
bt2_utils._check_str(name)
bt2_utils._check_type(field_class, _FieldClass)
if user_attributes is not None:
self[name]._set_user_attributes(user_attributes_value)
- def __iadd__(self, options):
+ def __iadd__(
+ self, options: typing.Iterable[typing.Tuple[str, _FieldClass]]
+ ) -> "_VariantFieldClassWithoutSelector":
for name, field_class in options:
self.append_option(name, field_class)
_NAME = "Const variant (with selector)"
@property
- def selector_field_path(self):
+ def selector_field_path(self) -> typing.Optional[bt2_field_path._FieldPathConst]:
ptr = native_bt.field_class_variant_with_selector_field_borrow_selector_field_path_const(
self._ptr
)
):
_NAME = "Variant (with selector)"
- def append_option(self, name, field_class, ranges, user_attributes=None):
+ def append_option(
+ self,
+ name: str,
+ field_class: _FieldClass,
+ ranges: bt2_integer_range_set._IntegerRangeSetConst,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ ):
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 user_attributes is not None:
self[name]._set_user_attributes(user_attributes_value)
- def __iadd__(self, options):
+ def __iadd__(
+ self,
+ options: typing.Iterable[
+ typing.Tuple[str, _FieldClass, bt2_integer_range_set._IntegerRangeSetConst]
+ ],
+ ):
for name, field_class, ranges in options:
self.append_option(name, field_class, ranges)
)
@property
- def element_field_class(self):
+ def element_field_class(self) -> _FieldClassConst:
elem_fc_ptr = self._borrow_element_field_class(self._ptr)
return self._create_field_class_from_ptr_and_get_ref(elem_fc_ptr)
_NAME = "Const static array"
@property
- def length(self):
+ def length(self) -> int:
return native_bt.field_class_array_static_get_length(self._ptr)
_NAME = "Const dynamic array (with length field)"
@property
- def length_field_path(self):
+ def length_field_path(self) -> typing.Optional[bt2_field_path._FieldPathConst]:
ptr = native_bt.field_class_array_dynamic_with_length_field_borrow_length_field_path_const(
self._ptr
)
import collections
+from bt2 import utils as bt2_utils
from bt2 import object as bt2_object
from bt2 import native_bt
+typing = bt2_utils._typing_mod
+
class FieldPathScope:
PACKET_CONTEXT = native_bt.FIELD_PATH_SCOPE_PACKET_CONTEXT
self._index = index
@property
- def index(self):
+ def index(self) -> int:
return self._index
native_bt.field_path_put_ref(ptr)
@property
- def root_scope(self):
+ def root_scope(self) -> FieldPathScope:
scope = native_bt.field_path_get_root_scope(self._ptr)
return _SCOPE_TO_OBJ[scope]
- def __len__(self):
+ def __len__(self) -> int:
return native_bt.field_path_get_item_count(self._ptr)
- def __iter__(self):
+ def __iter__(
+ self,
+ ) -> typing.Iterator[
+ typing.Union[
+ _IndexFieldPathItem,
+ _CurrentArrayElementFieldPathItem,
+ _CurrentOptionContentFieldPathItem,
+ ]
+ ]:
for idx in range(len(self)):
item_ptr = native_bt.field_path_borrow_item_by_index_const(self._ptr, idx)
assert item_ptr is not None
from bt2 import connection as bt2_connection
from bt2 import interrupter as bt2_interrupter
+typing = bt2_utils._typing_mod
+
def _graph_port_added_listener_from_native(
user_listener, component_ptr, component_type, port_ptr, port_type
def _put_ref(ptr):
native_bt.graph_put_ref(ptr)
- def __init__(self, mip_version=0):
+ def __init__(self, mip_version: int = 0):
bt2_utils._check_uint64(mip_version)
if mip_version > bt2_mip.get_maximal_mip_version():
def add_component(
self,
- component_class,
- name,
+ component_class: typing.Union[
+ bt2_component._ComponentClassConst,
+ typing.Type[bt2_component._UserComponent],
+ ],
+ name: str,
params=None,
- obj=None,
- logging_level=bt2_logging.LoggingLevel.NONE,
- ):
+ obj: object = None,
+ logging_level: int = bt2_logging.LoggingLevel.NONE,
+ ) -> bt2_component._ComponentConst:
if isinstance(component_class, bt2_component._SourceComponentClassConst):
cc_ptr = component_class._ptr
add_fn = native_bt.bt2_graph_add_source_component
comp_ptr, cc_type
)
- def connect_ports(self, upstream_port, downstream_port):
+ def connect_ports(
+ self,
+ upstream_port: bt2_port._OutputPortConst,
+ downstream_port: bt2_port._InputPortConst,
+ ) -> bt2_connection._ConnectionConst:
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(
assert conn_ptr
return bt2_connection._ConnectionConst._create_from_ptr_and_get_ref(conn_ptr)
- def add_port_added_listener(self, listener):
+ def add_port_added_listener(
+ self,
+ listener: typing.Callable[
+ [bt2_component._ComponentConst, bt2_port._PortConst], None
+ ],
+ ):
if not callable(listener):
raise TypeError("'listener' parameter is not callable")
status = native_bt.graph_run(self._ptr)
bt2_utils._handle_func_status(status, "graph object stopped running")
- def add_interrupter(self, interrupter):
+ def add_interrupter(self, interrupter: bt2_interrupter.Interrupter):
bt2_utils._check_type(interrupter, bt2_interrupter.Interrupter)
native_bt.graph_add_interrupter(self._ptr, interrupter._ptr)
@property
- def default_interrupter(self):
+ def default_interrupter(self) -> bt2_interrupter.Interrupter:
ptr = native_bt.graph_borrow_default_interrupter(self._ptr)
return bt2_interrupter.Interrupter._create_from_ptr_and_get_ref(ptr)
from bt2 import object as bt2_object
from bt2 import native_bt
+typing = bt2_utils._typing_mod
+
class _IntegerRangeConst:
def __init__(self, lower, upper=None):
self._upper = upper
@property
- def lower(self):
+ def lower(self) -> int:
return self._lower
@property
- def upper(self):
+ def upper(self) -> int:
return self._upper
- def contains(self, value):
+ def contains(self, value: int) -> bool:
self._check_type(value)
return value >= self._lower and value <= self._upper
- def __eq__(self, other):
+ def __eq__(self, other: object) -> bool:
if not isinstance(other, _IntegerRangeConst):
return False
class _IntegerRange(_IntegerRangeConst):
- def __init__(self, lower, upper=None):
+ def __init__(self, lower: int, upper: typing.Optional[int] = None):
super().__init__(lower, upper)
class _IntegerRangeSetConst(bt2_object._SharedObject, collections.abc.Set):
- def __len__(self):
+ def __len__(self) -> int:
range_set_ptr = self._as_range_set_ptr(self._ptr)
count = native_bt.integer_range_set_get_range_count(range_set_ptr)
assert count >= 0
return count
- def __contains__(self, other_range):
+ def __contains__(self, other_range: object) -> bool:
for rg in self:
if rg == other_range:
return True
return False
- def __iter__(self):
+ def __iter__(self) -> typing.Iterator[_IntegerRangeConst]:
for idx in range(len(self)):
rg_ptr = self._borrow_range_ptr_by_index(self._ptr, idx)
assert rg_ptr is not None
upper = self._range_get_upper(rg_ptr)
yield self._range_pycls(lower, upper)
- def __eq__(self, other):
+ def __eq__(self, other: object) -> bool:
if not isinstance(other, _IntegerRangeSetConst):
return False
return self._is_equal(self._ptr, other._ptr)
- def contains_value(self, value):
+ def contains_value(self, value: int) -> bool:
for rg in self:
if rg.contains(value):
return True
class _IntegerRangeSet(_IntegerRangeSetConst, collections.abc.MutableSet):
- def __init__(self, ranges=None):
+ def __init__(
+ self,
+ ranges: typing.Union[
+ None,
+ typing.Iterable[_IntegerRange],
+ typing.Iterable[typing.Union[typing.Tuple[int, int], int]],
+ ] = None,
+ ):
ptr = self._create_range_set()
if ptr is None:
for rg in ranges:
self.add(rg)
- def add(self, rg):
+ def add(
+ self, rg: typing.Union[_IntegerRange, typing.Union[typing.Tuple[int, int], int]]
+ ):
if type(rg) is not self._range_pycls:
if self._range_pycls._is_type(rg):
rg = self._range_pycls(rg)
status = self._add_range(self._ptr, rg.lower, rg.upper)
bt2_utils._handle_func_status(status, "cannot add range to range set object")
- def discard(self, rg):
+ def discard(self, rg: _IntegerRange):
raise NotImplementedError
super().__init__(ptr)
@property
- def is_set(self):
+ def is_set(self) -> bool:
return bool(native_bt.interrupter_is_set(self._ptr))
- def __bool__(self):
+ def __bool__(self) -> bool:
return self.is_set
def set(self):
NONE = native_bt.LOGGING_LEVEL_NONE
-def get_minimal_logging_level():
+def get_minimal_logging_level() -> int:
return native_bt.logging_get_minimal_level()
-def get_global_logging_level():
+def get_global_logging_level() -> int:
return native_bt.logging_get_global_level()
-def set_global_logging_level(level):
+def set_global_logging_level(level: int):
levels = (
LoggingLevel.TRACE,
LoggingLevel.DEBUG,
from bt2 import native_bt
from bt2 import clock_snapshot as bt2_clock_snapshot
+typing = bt2_utils._typing_mod
+
def _create_from_ptr(ptr):
msg_type = native_bt.message_get_type(ptr)
_event_pycls = property(lambda _: bt2_event._EventConst)
@property
- def default_clock_snapshot(self):
+ def default_clock_snapshot(self) -> bt2_clock_snapshot._ClockSnapshotConst:
self._check_has_default_clock_class(self.event.stream.cls.default_clock_class)
return self._get_default_clock_snapshot(self._borrow_default_clock_snapshot)
@property
- def event(self):
+ def event(self) -> bt2_event._EventConst:
event_ptr = self._borrow_event(self._ptr)
assert event_ptr is not None
return self._event_pycls._create_from_ptr_and_get_ref(
_packet_pycls = bt2_packet._PacketConst
@property
- def default_clock_snapshot(self):
+ def default_clock_snapshot(self) -> bt2_clock_snapshot._ClockSnapshotConst:
self._check_has_default_clock_class(self.packet.stream.cls.default_clock_class)
return self._get_default_clock_snapshot(self._borrow_default_clock_snapshot_ptr)
@property
- def packet(self):
+ def packet(self) -> bt2_packet._PacketConst:
packet_ptr = self._borrow_packet(self._ptr)
assert packet_ptr is not None
return self._packet_pycls._create_from_ptr_and_get_ref(packet_ptr)
_stream_pycls = property(lambda _: bt2_stream._StreamConst)
@property
- def stream(self):
+ def stream(self) -> bt2_stream._StreamConst:
stream_ptr = self._borrow_stream_ptr(self._ptr)
assert stream_ptr
return self._stream_pycls._create_from_ptr_and_get_ref(stream_ptr)
@property
- def default_clock_snapshot(self):
+ def default_clock_snapshot(self) -> bt2_clock_snapshot._ClockSnapshotConst:
self._check_has_default_clock_class(self.stream.cls.default_clock_class)
status, snapshot_ptr = self._borrow_default_clock_snapshot_ptr(self._ptr)
)
@property
- def clock_snapshot(self):
+ def clock_snapshot(self) -> bt2_clock_snapshot._ClockSnapshotConst:
# This kind of message always has a clock class: no
# need to call self._check_has_default_clock_class() here.
return self._get_default_clock_snapshot(self._borrow_clock_snapshot_ptr)
_stream_pycls = property(lambda _: bt2_stream._StreamConst)
@property
- def stream(self):
+ def stream(self) -> bt2_stream._StreamConst:
stream_ptr = self._borrow_stream_ptr(self._ptr)
assert stream_ptr
return self._stream_pycls._create_from_ptr_and_get_ref(stream_ptr)
@property
- def count(self):
+ def count(self) -> typing.Optional[int]:
avail, count = self._get_count(self._ptr)
if avail is native_bt.PROPERTY_AVAILABILITY_AVAILABLE:
return count
)
@property
- def beginning_default_clock_snapshot(self):
+ def beginning_default_clock_snapshot(
+ self,
+ ) -> bt2_clock_snapshot._ClockSnapshotConst:
self._check_has_default_clock_snapshots()
return self._get_default_clock_snapshot(
self._borrow_beginning_clock_snapshot_ptr
)
@property
- def end_default_clock_snapshot(self):
+ def end_default_clock_snapshot(self) -> bt2_clock_snapshot._ClockSnapshotConst:
self._check_has_default_clock_snapshots()
return self._get_default_clock_snapshot(self._borrow_end_clock_snapshot_ptr)
from bt2 import clock_class as bt2_clock_class
from bt2 import event_class as bt2_event_class
+typing = bt2_utils._typing_mod
+
+if typing.TYPE_CHECKING:
+ from bt2 import component as bt2_component
+
class _MessageIterator(collections.abc.Iterator):
- def __next__(self):
+ def __next__(self) -> bt2_message._MessageConst:
raise NotImplementedError
self._at = 0
super().__init__(ptr)
- def __next__(self):
+ def __next__(self) -> bt2_message._MessageConst:
if len(self._current_msgs) == self._at:
status, msgs = native_bt.bt2_self_component_port_input_get_msg_range(
self._ptr
return bt2_message._create_from_ptr(msg_ptr)
- def can_seek_beginning(self):
+ def can_seek_beginning(self) -> bool:
(status, res) = native_bt.message_iterator_can_seek_beginning(self._ptr)
bt2_utils._handle_func_status(
status,
status = native_bt.message_iterator_seek_beginning(self._ptr)
bt2_utils._handle_func_status(status, "cannot seek message iterator beginning")
- def can_seek_ns_from_origin(self, ns_from_origin):
+ def can_seek_ns_from_origin(self, ns_from_origin) -> bool:
bt2_utils._check_int64(ns_from_origin)
(status, res) = native_bt.message_iterator_can_seek_ns_from_origin(
self._ptr, ns_from_origin
)
return res != 0
- def seek_ns_from_origin(self, ns_from_origin):
+ def seek_ns_from_origin(self, ns_from_origin: int):
bt2_utils._check_int64(ns_from_origin)
# Forget about buffered messages, they won't be valid after seeking.
)
@property
- def can_seek_forward(self):
+ def can_seek_forward(self) -> bool:
return native_bt.message_iterator_can_seek_forward(self._ptr)
pass
@property
- def _component(self):
+ def _component(self) -> "bt2_component._UserComponent":
return native_bt.bt2_get_user_component_from_user_msg_iter(self._bt_ptr)
@property
- def _port(self):
+ def _port(self) -> bt2_port._UserComponentOutputPort:
port_ptr = native_bt.self_message_iterator_borrow_port(self._bt_ptr)
assert port_ptr is not None
return bt2_port._create_self_from_ptr_and_get_ref(
)
@property
- def addr(self):
+ def addr(self) -> int:
return int(self._bt_ptr)
@property
def _user_finalize(self):
pass
- def __next__(self):
+ def __next__(self) -> bt2_message._MessageConst:
raise bt2_utils.Stop
def _bt_next_from_native(self):
def _bt_seek_ns_from_origin_from_native(self, ns_from_origin):
self._user_seek_ns_from_origin(ns_from_origin)
- def _create_message_iterator(self, input_port):
+ def _create_message_iterator(
+ self, input_port: bt2_port._UserComponentInputPort
+ ) -> _UserComponentInputPortMessageIterator:
bt2_utils._check_type(input_port, bt2_port._UserComponentInputPort)
if not input_port.is_connected:
return _UserComponentInputPortMessageIterator(msg_iter_ptr)
- def _create_event_message(self, event_class, parent, default_clock_snapshot=None):
+ def _create_event_message(
+ self,
+ event_class: bt2_event_class._EventClassConst,
+ parent: typing.Union[bt2_stream._StreamConst, bt2_packet._PacketConst],
+ default_clock_snapshot: typing.Optional[int] = None,
+ ) -> bt2_message._EventMessage:
bt2_utils._check_type(event_class, bt2_event_class._EventClass)
if event_class.stream_class.supports_packets:
return bt2_message._EventMessage(ptr)
- def _create_message_iterator_inactivity_message(self, clock_class, clock_snapshot):
+ def _create_message_iterator_inactivity_message(
+ self, clock_class: bt2_clock_class._ClockClassConst, clock_snapshot: int
+ ) -> bt2_message._MessageIteratorInactivityMessage:
bt2_utils._check_type(clock_class, bt2_clock_class._ClockClass)
+ bt2_utils._check_uint64(clock_snapshot)
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):
+ def _create_stream_beginning_message(
+ self,
+ stream: bt2_stream._StreamConst,
+ default_clock_snapshot: typing.Optional[int] = None,
+ ) -> bt2_message._StreamBeginningMessage:
bt2_utils._check_type(stream, bt2_stream._Stream)
ptr = native_bt.message_stream_beginning_create(self._bt_ptr, stream._ptr)
return msg
- def _create_stream_end_message(self, stream, default_clock_snapshot=None):
+ def _create_stream_end_message(
+ self,
+ stream: bt2_stream._StreamConst,
+ default_clock_snapshot: typing.Optional[int] = None,
+ ) -> bt2_message._StreamEndMessage:
bt2_utils._check_type(stream, bt2_stream._Stream)
ptr = native_bt.message_stream_end_create(self._bt_ptr, stream._ptr)
return msg
- def _create_packet_beginning_message(self, packet, default_clock_snapshot=None):
+ def _create_packet_beginning_message(
+ self,
+ packet: bt2_packet._PacketConst,
+ default_clock_snapshot: typing.Optional[int] = None,
+ ) -> bt2_message._PacketBeginningMessage:
bt2_utils._check_type(packet, bt2_packet._Packet)
if packet.stream.cls.packets_have_beginning_default_clock_snapshot:
return bt2_message._PacketBeginningMessage(ptr)
- def _create_packet_end_message(self, packet, default_clock_snapshot=None):
+ def _create_packet_end_message(
+ self,
+ packet: bt2_packet._PacketConst,
+ default_clock_snapshot: typing.Optional[int] = None,
+ ) -> bt2_message._PacketEndMessage:
bt2_utils._check_type(packet, bt2_packet._Packet)
if packet.stream.cls.packets_have_end_default_clock_snapshot:
return bt2_message._PacketEndMessage(ptr)
def _create_discarded_events_message(
- self, stream, count=None, beg_clock_snapshot=None, end_clock_snapshot=None
- ):
+ self,
+ stream: bt2_stream._StreamConst,
+ count: typing.Optional[int] = None,
+ beg_clock_snapshot: typing.Optional[int] = None,
+ end_clock_snapshot: typing.Optional[int] = None,
+ ) -> bt2_message._DiscardedEventsMessage:
bt2_utils._check_type(stream, bt2_stream._Stream)
if not stream.cls.supports_discarded_events:
return msg
def _create_discarded_packets_message(
- self, stream, count=None, beg_clock_snapshot=None, end_clock_snapshot=None
- ):
+ self,
+ stream: bt2_stream._StreamConst,
+ count: typing.Optional[int] = None,
+ beg_clock_snapshot: typing.Optional[int] = None,
+ end_clock_snapshot: typing.Optional[int] = None,
+ ) -> bt2_message._DiscardedPacketsMessage:
bt2_utils._check_type(stream, bt2_stream._Stream)
if not stream.cls.supports_discarded_packets:
from bt2 import native_bt
from bt2 import component_descriptor as bt2_component_descriptor
+typing = bt2_utils._typing_mod
+
def get_greatest_operative_mip_version(
- component_descriptors, log_level=bt2_logging.LoggingLevel.NONE
-):
+ component_descriptors: typing.Sequence[
+ bt2_component_descriptor.ComponentDescriptor
+ ],
+ log_level: int = bt2_logging.LoggingLevel.NONE,
+) -> typing.Optional[int]:
bt2_utils._check_log_level(log_level)
comp_descr_set_ptr = native_bt.component_descriptor_set_create()
return self._ptr_internal
@property
- def addr(self):
+ def addr(self) -> int:
return int(self._ptr_internal)
- def __repr__(self):
+ def __repr__(self) -> str:
return "<{}.{} object @ {}>".format(
self.__class__.__module__, self.__class__.__name__, hex(self.addr)
)
#
# Copyright (c) 2016-2017 Philippe Proulx <pproulx@efficios.com>
+
from bt2 import field as bt2_field
+from bt2 import utils as bt2_utils
from bt2 import object as bt2_object
from bt2 import native_bt
+typing = bt2_utils._typing_mod
+
+if typing.TYPE_CHECKING:
+ from bt2 import stream as bt2_stream
+
def _bt2_stream():
from bt2 import stream as bt2_stream
_create_field_from_ptr = staticmethod(bt2_field._create_field_from_const_ptr)
@property
- def stream(self):
- stream_ptr = self._borrow_stream_ptr(self._ptr)
- assert stream_ptr is not None
- return self._stream_pycls._create_from_ptr_and_get_ref(stream_ptr)
+ def stream(self) -> "bt2_stream._StreamConst":
+ return self._stream_pycls._create_from_ptr_and_get_ref(
+ self._borrow_stream_ptr(self._ptr)
+ )
@property
- def context_field(self):
+ def context_field(self) -> typing.Optional[bt2_field._StructureFieldConst]:
field_ptr = self._borrow_context_field_ptr(self._ptr)
if field_ptr is None:
from bt2 import component as bt2_component
from bt2 import native_bt
-
-def find_plugins_in_path(path, recurse=True, fail_on_load_error=False):
- 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):
- status, plugin_set_ptr = native_bt.bt2_plugin_find_all_from_file(
- path, fail_on_load_error
- )
- elif os.path.isdir(path):
- status, plugin_set_ptr = native_bt.bt2_plugin_find_all_from_dir(
- path, int(recurse), int(fail_on_load_error)
- )
- else:
- raise ValueError("invalid path: '{}'".format(path))
-
- if status == native_bt.__BT_FUNC_STATUS_NOT_FOUND:
- return
-
- 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)
-
-
-def find_plugins(
- find_in_std_env_var=True,
- find_in_user_dir=True,
- find_in_sys_dir=True,
- find_in_static=True,
- fail_on_load_error=False,
-):
- 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(
- int(find_in_std_env_var),
- int(find_in_user_dir),
- int(find_in_sys_dir),
- int(find_in_static),
- int(fail_on_load_error),
- )
-
- if status == native_bt.__BT_FUNC_STATUS_NOT_FOUND:
- return
-
- 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)
-
-
-def find_plugin(
- name,
- find_in_std_env_var=True,
- find_in_user_dir=True,
- find_in_sys_dir=True,
- find_in_static=True,
- fail_on_load_error=False,
-):
- 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),
- int(find_in_user_dir),
- int(find_in_sys_dir),
- int(find_in_static),
- int(fail_on_load_error),
- )
-
- if status == native_bt.__BT_FUNC_STATUS_NOT_FOUND:
- return
-
- bt2_utils._handle_func_status(status, "failed to find plugin")
- assert ptr is not None
- return _Plugin._create_from_ptr(ptr)
-
-
-class _PluginSet(bt2_object._SharedObject, collections.abc.Sequence):
- @staticmethod
- def _put_ref(ptr):
- native_bt.plugin_set_put_ref(ptr)
-
- @staticmethod
- def _get_ref(ptr):
- native_bt.plugin_set_get_ref(ptr)
-
- def __len__(self):
- count = native_bt.plugin_set_get_plugin_count(self._ptr)
- assert count >= 0
- return count
-
- def __getitem__(self, index):
- bt2_utils._check_uint64(index)
-
- if index >= len(self):
- raise IndexError
-
- plugin_ptr = native_bt.plugin_set_borrow_plugin_by_index_const(self._ptr, index)
- assert plugin_ptr is not None
- return _Plugin._create_from_ptr_and_get_ref(plugin_ptr)
+typing = bt2_utils._typing_mod
class _PluginVersion:
self._extra = extra
@property
- def major(self):
+ def major(self) -> int:
return self._major
@property
- def minor(self):
+ def minor(self) -> int:
return self._minor
@property
- def patch(self):
+ def patch(self) -> int:
return self._patch
@property
- def extra(self):
+ def extra(self) -> typing.Optional[str]:
return self._extra
- def __str__(self):
+ def __str__(self) -> str:
extra = ""
if self._extra is not None:
self._plugin_comp_cls = plugin_comp_cls
self._at = 0
- def __next__(self):
+ def __next__(self) -> str:
plugin_ptr = self._plugin_comp_cls._plugin._ptr
total = self._plugin_comp_cls._component_class_count(plugin_ptr)
def __init__(self, plugin):
self._plugin = plugin
- def __getitem__(self, key):
+ def __getitem__(self, key: str) -> bt2_component._ComponentClassConst:
bt2_utils._check_str(key)
cc_ptr = self._borrow_component_class_by_name(self._plugin._ptr, key)
cc_ptr, self._comp_cls_type
)
- def __len__(self):
+ def __len__(self) -> int:
return self._component_class_count(self._plugin._ptr)
- def __iter__(self):
+ def __iter__(self) -> typing.Iterator[bt2_component._ComponentClassConst]:
return _PluginComponentClassesIterator(self)
native_bt.plugin_get_ref(ptr)
@property
- def name(self):
+ def name(self) -> str:
name = native_bt.plugin_get_name(self._ptr)
assert name is not None
return name
@property
- def author(self):
+ def author(self) -> typing.Optional[str]:
return native_bt.plugin_get_author(self._ptr)
@property
- def license(self):
+ def license(self) -> typing.Optional[str]:
return native_bt.plugin_get_license(self._ptr)
@property
- def description(self):
+ def description(self) -> typing.Optional[str]:
return native_bt.plugin_get_description(self._ptr)
@property
- def path(self):
+ def path(self) -> typing.Optional[str]:
return native_bt.plugin_get_path(self._ptr)
@property
- def version(self):
+ def version(self) -> typing.Optional[_PluginVersion]:
status, major, minor, patch, extra = native_bt.bt2_plugin_get_version(self._ptr)
if status == native_bt.PROPERTY_AVAILABILITY_NOT_AVAILABLE:
return _PluginVersion(major, minor, patch, extra)
@property
- def source_component_classes(self):
+ def source_component_classes(self) -> _PluginSourceComponentClasses:
return _PluginSourceComponentClasses(self)
@property
- def filter_component_classes(self):
+ def filter_component_classes(self) -> _PluginFilterComponentClasses:
return _PluginFilterComponentClasses(self)
@property
- def sink_component_classes(self):
+ def sink_component_classes(self) -> _PluginSinkComponentClasses:
return _PluginSinkComponentClasses(self)
+
+
+class _PluginSet(bt2_object._SharedObject, collections.abc.Sequence):
+ @staticmethod
+ def _put_ref(ptr):
+ native_bt.plugin_set_put_ref(ptr)
+
+ @staticmethod
+ def _get_ref(ptr):
+ native_bt.plugin_set_get_ref(ptr)
+
+ def __len__(self) -> int:
+ count = native_bt.plugin_set_get_plugin_count(self._ptr)
+ assert count >= 0
+ return count
+
+ def __getitem__(self, index: int) -> _Plugin:
+ bt2_utils._check_uint64(index)
+
+ if index >= len(self):
+ raise IndexError
+
+ plugin_ptr = native_bt.plugin_set_borrow_plugin_by_index_const(self._ptr, index)
+ assert plugin_ptr is not None
+ return _Plugin._create_from_ptr_and_get_ref(plugin_ptr)
+
+
+def find_plugins_in_path(
+ path: str, recurse: bool = True, fail_on_load_error: bool = False
+) -> typing.Optional[_PluginSet]:
+ 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):
+ status, plugin_set_ptr = native_bt.bt2_plugin_find_all_from_file(
+ path, fail_on_load_error
+ )
+ elif os.path.isdir(path):
+ status, plugin_set_ptr = native_bt.bt2_plugin_find_all_from_dir(
+ path, int(recurse), int(fail_on_load_error)
+ )
+ else:
+ raise ValueError("invalid path: '{}'".format(path))
+
+ if status == native_bt.__BT_FUNC_STATUS_NOT_FOUND:
+ return
+
+ 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)
+
+
+def find_plugins(
+ find_in_std_env_var: bool = True,
+ find_in_user_dir: bool = True,
+ find_in_sys_dir: bool = True,
+ find_in_static: bool = True,
+ fail_on_load_error: bool = False,
+) -> typing.Optional[_PluginSet]:
+ 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(
+ int(find_in_std_env_var),
+ int(find_in_user_dir),
+ int(find_in_sys_dir),
+ int(find_in_static),
+ int(fail_on_load_error),
+ )
+
+ if status == native_bt.__BT_FUNC_STATUS_NOT_FOUND:
+ return
+
+ 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)
+
+
+def find_plugin(
+ name: str,
+ find_in_std_env_var: bool = True,
+ find_in_user_dir: bool = True,
+ find_in_sys_dir: bool = True,
+ find_in_static: bool = True,
+ fail_on_load_error: bool = False,
+) -> typing.Optional[_Plugin]:
+ 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),
+ int(find_in_user_dir),
+ int(find_in_sys_dir),
+ int(find_in_static),
+ int(fail_on_load_error),
+ )
+
+ if status == native_bt.__BT_FUNC_STATUS_NOT_FOUND:
+ return
+
+ bt2_utils._handle_func_status(status, "failed to find plugin")
+ assert ptr is not None
+ return _Plugin._create_from_ptr(ptr)
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
+from bt2 import utils as bt2_utils
from bt2 import object as bt2_object
from bt2 import native_bt
+typing = bt2_utils._typing_mod
+
+if typing.TYPE_CHECKING:
+ from bt2 import connection as bt2_connection
+
def _bt2_connection():
from bt2 import connection as bt2_connection
return native_bt.port_put_ref(ptr)
@property
- def name(self):
+ def name(self) -> str:
ptr = self._as_port_ptr(self._ptr)
name = native_bt.port_get_name(ptr)
assert name is not None
return name
@property
- def connection(self):
+ def connection(self) -> typing.Optional["bt2_connection._ConnectionConst"]:
ptr = self._as_port_ptr(self._ptr)
conn_ptr = native_bt.port_borrow_connection_const(ptr)
return _bt2_connection()._ConnectionConst._create_from_ptr_and_get_ref(conn_ptr)
@property
- def is_connected(self):
+ def is_connected(self) -> bool:
return self.connection is not None
return native_bt.self_component_port_as_port(ptr)
@property
- def connection(self):
+ def connection(self) -> typing.Optional["bt2_connection._ConnectionConst"]:
ptr = self._as_port_ptr(self._ptr)
conn_ptr = native_bt.port_borrow_connection_const(ptr)
return _bt2_connection()._ConnectionConst._create_from_ptr_and_get_ref(conn_ptr)
@property
- def user_data(self):
+ def user_data(self) -> object:
ptr = self._as_self_port_ptr(self._ptr)
return native_bt.self_component_port_get_data(ptr)
from bt2 import utils as bt2_utils
from bt2 import component as bt2_component
+typing = bt2_utils._typing_mod
# Python plugin path to `_PluginInfo` (cache)
_plugin_infos = {}
-def plugin_component_class(component_class):
+def plugin_component_class(component_class: typing.Type[bt2_component._UserComponent]):
if not issubclass(component_class, bt2_component._UserComponent):
raise TypeError("component class is not a subclass of a user component class")
def register_plugin(
- module_name, name, description=None, author=None, license=None, version=None
+ module_name: str,
+ name: str,
+ description: typing.Optional[str] = None,
+ author: typing.Optional[str] = None,
+ license: typing.Optional[str] = None,
+ version: typing.Union[
+ typing.Tuple[int, int, int], typing.Tuple[int, int, int, str], None
+ ] = None,
):
if module_name not in sys.modules:
raise RuntimeError(
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
+
from bt2 import error as bt2_error
from bt2 import utils as bt2_utils
from bt2 import value as bt2_value
from bt2 import native_bt
from bt2 import interrupter as bt2_interrupter
+typing = bt2_utils._typing_mod
+
+if typing.TYPE_CHECKING:
+ from bt2 import component as bt2_component
+
def _bt2_component():
from bt2 import component as bt2_component
return self._as_query_executor_ptr()
@property
- def is_interrupted(self):
+ def is_interrupted(self) -> bool:
is_interrupted = native_bt.query_executor_is_interrupted(self._common_ptr)
return bool(is_interrupted)
@property
- def logging_level(self):
+ def logging_level(self) -> int:
return native_bt.query_executor_get_logging_level(self._common_ptr)
def _as_query_executor_ptr(self):
return self._ptr
- def __init__(self, component_class, object_name, params=None, method_obj=None):
+ def __init__(
+ self,
+ component_class: "bt2_component._ComponentClassConst",
+ object_name: str,
+ params=None,
+ method_obj: object = None,
+ ):
if not isinstance(component_class, _bt2_component()._ComponentClassConst):
err = False
# query() method is called, the Python object still exists.
self._method_obj = method_obj
- def add_interrupter(self, interrupter):
+ def add_interrupter(self, interrupter: bt2_interrupter.Interrupter):
bt2_utils._check_type(interrupter, bt2_interrupter.Interrupter)
native_bt.query_executor_add_interrupter(self._ptr, interrupter._ptr)
@property
- def default_interrupter(self):
+ def default_interrupter(self) -> bt2_interrupter.Interrupter:
ptr = native_bt.query_executor_borrow_default_interrupter(self._ptr)
return bt2_interrupter.Interrupter._create_from_ptr_and_get_ref(ptr)
)
@property
- def is_interrupted(self):
+ def is_interrupted(self) -> bool:
is_interrupted = native_bt.query_executor_is_interrupted(self._ptr)
return bool(is_interrupted)
- def query(self):
+ def query(self) -> typing.Optional[bt2_value._ValueConst]:
status, result_ptr = native_bt.query_executor_query(self._ptr)
bt2_utils._handle_func_status(status, "cannot query component class")
assert result_ptr is not None
from bt2 import stream_class as bt2_stream_class
from bt2 import user_attributes as bt2_user_attrs
+typing = bt2_utils._typing_mod
+
+if typing.TYPE_CHECKING:
+ from bt2 import trace as bt2_trace
+
def _bt2_trace():
from bt2 import trace as bt2_trace
_trace_pycls = property(lambda _: _bt2_trace()._TraceConst)
@property
- def cls(self):
+ def cls(self) -> bt2_stream_class._StreamClassConst:
stream_class_ptr = self._borrow_class_ptr(self._ptr)
assert stream_class_ptr is not None
return self._stream_class_pycls._create_from_ptr_and_get_ref(stream_class_ptr)
@property
- def name(self):
+ def name(self) -> typing.Optional[str]:
return native_bt.stream_get_name(self._ptr)
@property
- def id(self):
+ def id(self) -> int:
id = native_bt.stream_get_id(self._ptr)
return id if id >= 0 else None
@property
- def trace(self):
+ def trace(self) -> "bt2_trace._TraceConst":
trace_ptr = self._borrow_trace_ptr(self._ptr)
assert trace_ptr is not None
return self._trace_pycls._create_from_ptr_and_get_ref(trace_ptr)
_stream_class_pycls = bt2_stream_class._StreamClass
_trace_pycls = property(lambda _: _bt2_trace()._Trace)
- def create_packet(self):
+ def create_packet(self) -> bt2_packet._Packet:
if not self.cls.supports_packets:
raise ValueError(
"cannot create packet: stream class does not support packets"
from bt2 import field_class as bt2_field_class
from bt2 import user_attributes as bt2_user_attrs
+typing = bt2_utils._typing_mod
+
+if typing.TYPE_CHECKING:
+ from bt2 import trace_class as bt2_trace_class
+
def _bt2_trace_class():
from bt2 import trace_class as bt2_trace_class
_trace_class_cls = property(lambda _: _bt2_trace_class()._TraceClassConst)
_clock_class_cls = property(lambda _: bt2_clock_class._ClockClassConst)
- def __getitem__(self, key):
+ def __getitem__(self, key: int) -> bt2_event_class._EventClassConst:
bt2_utils._check_int64(key)
ec_ptr = self._borrow_event_class_ptr_by_id(self._ptr, key)
return self._event_class_cls._create_from_ptr_and_get_ref(ec_ptr)
- def __len__(self):
+ def __len__(self) -> int:
count = native_bt.stream_class_get_event_class_count(self._ptr)
assert count >= 0
return count
- def __iter__(self):
+ def __iter__(self) -> typing.Iterator[int]:
for idx in range(len(self)):
ec_ptr = self._borrow_event_class_ptr_by_index(self._ptr, idx)
assert ec_ptr is not None
yield id
@property
- def trace_class(self):
+ def trace_class(self) -> "bt2_trace_class._TraceClassConst":
tc_ptr = self._borrow_trace_class_ptr(self._ptr)
if tc_ptr is not None:
return self._trace_class_cls._create_from_ptr_and_get_ref(tc_ptr)
@property
- def name(self):
+ def name(self) -> typing.Optional[str]:
return native_bt.stream_class_get_name(self._ptr)
@property
- def assigns_automatic_event_class_id(self):
+ def assigns_automatic_event_class_id(self) -> bool:
return native_bt.stream_class_assigns_automatic_event_class_id(self._ptr)
@property
- def assigns_automatic_stream_id(self):
+ def assigns_automatic_stream_id(self) -> bool:
return native_bt.stream_class_assigns_automatic_stream_id(self._ptr)
@property
- def supports_packets(self):
+ def supports_packets(self) -> bool:
return native_bt.stream_class_supports_packets(self._ptr)
@property
- def packets_have_beginning_default_clock_snapshot(self):
+ def packets_have_beginning_default_clock_snapshot(self) -> bool:
return native_bt.stream_class_packets_have_beginning_default_clock_snapshot(
self._ptr
)
@property
- def packets_have_end_default_clock_snapshot(self):
+ def packets_have_end_default_clock_snapshot(self) -> bool:
return native_bt.stream_class_packets_have_end_default_clock_snapshot(self._ptr)
@property
- def supports_discarded_events(self):
+ def supports_discarded_events(self) -> bool:
return native_bt.stream_class_supports_discarded_events(self._ptr)
@property
- def discarded_events_have_default_clock_snapshots(self):
+ def discarded_events_have_default_clock_snapshots(self) -> bool:
return native_bt.stream_class_discarded_events_have_default_clock_snapshots(
self._ptr
)
@property
- def supports_discarded_packets(self):
+ def supports_discarded_packets(self) -> bool:
return native_bt.stream_class_supports_discarded_packets(self._ptr)
@property
- def discarded_packets_have_default_clock_snapshots(self):
+ def discarded_packets_have_default_clock_snapshots(self) -> bool:
return native_bt.stream_class_discarded_packets_have_default_clock_snapshots(
self._ptr
)
@property
- def id(self):
+ def id(self) -> int:
id = native_bt.stream_class_get_id(self._ptr)
if id < 0:
return id
@property
- def packet_context_field_class(self):
+ def packet_context_field_class(
+ self,
+ ) -> typing.Optional[bt2_field_class._StructureFieldClassConst]:
fc_ptr = self._borrow_packet_context_field_class_ptr(self._ptr)
if fc_ptr is None:
return bt2_field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
@property
- def event_common_context_field_class(self):
+ def event_common_context_field_class(
+ self,
+ ) -> typing.Optional[bt2_field_class._StructureFieldClassConst]:
fc_ptr = self._borrow_event_common_context_field_class_ptr(self._ptr)
if fc_ptr is None:
return bt2_field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
@property
- def default_clock_class(self):
+ def default_clock_class(self) -> typing.Optional[bt2_clock_class._ClockClassConst]:
cc_ptr = self._borrow_default_clock_class_ptr(self._ptr)
if cc_ptr is None:
return
def create_event_class(
self,
- id=None,
- name=None,
- user_attributes=None,
- log_level=None,
- emf_uri=None,
- specific_context_field_class=None,
- payload_field_class=None,
- ):
+ id: typing.Optional[int] = None,
+ name: typing.Optional[str] = None,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ log_level: typing.Optional[bt2_event_class.EventClassLogLevel] = None,
+ emf_uri: typing.Optional[str] = None,
+ specific_context_field_class: typing.Optional[
+ bt2_field_class._StructureFieldClass
+ ] = None,
+ payload_field_class: typing.Optional[
+ bt2_field_class._StructureFieldClass
+ ] = None,
+ ) -> bt2_event_class._EventClass:
# Validate parameters before we create the object.
bt2_event_class._EventClass._validate_create_params(
name,
from bt2 import stream_class as bt2_stream_class
from bt2 import user_attributes as bt2_user_attrs
+typing = bt2_utils._typing_mod
+
+if typing.TYPE_CHECKING:
+ from bt2 import trace_class as bt2_trace_class
+
def _bt2_trace_class():
from bt2 import trace_class as bt2_trace_class
def __init__(self, trace):
self._trace = trace
- def __getitem__(self, key):
+ def __getitem__(
+ self, key: str
+ ) -> typing.Union[bt2_value.SignedIntegerValue, bt2_value.StringValue]:
bt2_utils._check_str(key)
borrow_entry_fn = native_bt.trace_borrow_environment_entry_value_by_name_const
return self._create_value_from_ptr_and_get_ref(value_ptr)
- def __len__(self):
+ def __len__(self) -> int:
count = native_bt.trace_get_environment_entry_count(self._trace._ptr)
assert count >= 0
return count
- def __iter__(self):
+ def __iter__(self) -> typing.Iterator[str]:
trace_ptr = self._trace._ptr
for idx in range(len(self)):
bt2_value._create_from_ptr_and_get_ref
)
- def __setitem__(self, key, value):
+ def __setitem__(self, key: str, value: typing.Union[str, int]):
if isinstance(value, str):
set_env_entry_fn = native_bt.trace_set_environment_entry_string
elif isinstance(value, int):
_trace_class_pycls = property(lambda _: _bt2_trace_class()._TraceClassConst)
_trace_env_pycls = property(lambda _: _TraceEnvironmentConst)
- def __len__(self):
+ def __len__(self) -> int:
count = native_bt.trace_get_stream_count(self._ptr)
assert count >= 0
return count
- def __getitem__(self, id):
+ def __getitem__(self, id: int) -> bt2_stream._StreamConst:
bt2_utils._check_uint64(id)
stream_ptr = self._borrow_stream_ptr_by_id(self._ptr, id)
return self._stream_pycls._create_from_ptr_and_get_ref(stream_ptr)
- def __iter__(self):
+ def __iter__(self) -> typing.Iterator[bt2_stream._StreamConst]:
for idx in range(len(self)):
stream_ptr = self._borrow_stream_ptr_by_index(self._ptr, idx)
assert stream_ptr is not None
yield id
@property
- def cls(self):
+ def cls(self) -> "bt2_trace_class._TraceClassConst":
trace_class_ptr = self._borrow_class_ptr(self._ptr)
assert trace_class_ptr is not None
return self._trace_class_pycls._create_from_ptr_and_get_ref(trace_class_ptr)
@property
- def name(self):
+ def name(self) -> typing.Optional[str]:
return native_bt.trace_get_name(self._ptr)
@property
- def uuid(self):
+ def uuid(self) -> typing.Optional[uuidp.UUID]:
uuid_bytes = native_bt.trace_get_uuid(self._ptr)
if uuid_bytes is None:
return
return uuidp.UUID(bytes=uuid_bytes)
@property
- def environment(self):
+ def environment(self) -> _TraceEnvironmentConst:
return self._trace_env_pycls(self)
- def add_destruction_listener(self, listener):
+ def add_destruction_listener(
+ self, listener: typing.Callable[["_TraceConst"], None]
+ ) -> bt2_utils._ListenerHandle:
"""Add a listener to be called when the trace is destroyed."""
if not callable(listener):
raise TypeError("'listener' parameter is not callable")
return handle
- def remove_destruction_listener(self, listener_handle):
+ def remove_destruction_listener(self, listener_handle: bt2_utils._ListenerHandle):
bt2_utils._check_type(listener_handle, bt2_utils._ListenerHandle)
if listener_handle._addr != self.addr:
bt2_utils._check_type(uuid, uuidp.UUID)
native_bt.trace_set_uuid(self._ptr, uuid.bytes)
- def create_stream(self, stream_class, id=None, name=None, user_attributes=None):
+ def create_stream(
+ self,
+ stream_class: bt2_stream_class._StreamClass,
+ id: typing.Optional[int] = None,
+ name: typing.Optional[str] = None,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ ) -> bt2_stream._Stream:
bt2_utils._check_type(stream_class, bt2_stream_class._StreamClass)
if stream_class.assigns_automatic_stream_id:
# Copyright (c) 2018 Francis Deslauriers <francis.deslauriers@efficios.com>
# Copyright (c) 2019 Simon Marchi <simon.marchi@efficios.com>
-import typing
+import uuid as uuidp
import functools
import collections.abc
from bt2 import value as bt2_value
from bt2 import object as bt2_object
from bt2 import native_bt
+from bt2 import clock_class as bt2_clock_class
from bt2 import field_class as bt2_field_class
from bt2 import stream_class as bt2_stream_class
from bt2 import user_attributes as bt2_user_attrs
from bt2 import integer_range_set as bt2_integer_range_set
+typing = bt2_utils._typing_mod
+
def _trace_class_destruction_listener_from_native(
user_listener, handle, trace_class_ptr
# Instantiate a trace of this class.
- def __call__(self, name=None, user_attributes=None, uuid=None, environment=None):
+ def __call__(
+ self,
+ name: typing.Optional[str] = None,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ uuid: typing.Optional[uuidp.UUID] = None,
+ environment: typing.Optional[
+ typing.Mapping[str, typing.Union[str, int]]
+ ] = None,
+ ) -> bt2_trace._Trace:
trace_ptr = native_bt.trace_create(self._ptr)
if trace_ptr is None:
def create_stream_class(
self,
- id=None,
- name=None,
- user_attributes=None,
- packet_context_field_class=None,
- event_common_context_field_class=None,
- default_clock_class=None,
- assigns_automatic_event_class_id=True,
- assigns_automatic_stream_id=True,
- supports_packets=False,
- packets_have_beginning_default_clock_snapshot=False,
- packets_have_end_default_clock_snapshot=False,
- supports_discarded_events=False,
- discarded_events_have_default_clock_snapshots=False,
- supports_discarded_packets=False,
- discarded_packets_have_default_clock_snapshots=False,
- ):
+ id: typing.Optional[int] = None,
+ name: typing.Optional[str] = None,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ packet_context_field_class: typing.Optional[
+ bt2_field_class._StructureFieldClass
+ ] = None,
+ event_common_context_field_class: typing.Optional[
+ bt2_field_class._StructureFieldClass
+ ] = None,
+ default_clock_class: typing.Optional[bt2_clock_class._ClockClass] = None,
+ assigns_automatic_event_class_id: bool = True,
+ assigns_automatic_stream_id: bool = True,
+ supports_packets: bool = False,
+ packets_have_beginning_default_clock_snapshot: bool = False,
+ packets_have_end_default_clock_snapshot: bool = False,
+ supports_discarded_events: bool = False,
+ discarded_events_have_default_clock_snapshots: bool = False,
+ supports_discarded_packets: bool = False,
+ discarded_packets_have_default_clock_snapshots: bool = False,
+ ) -> bt2_stream_class._StreamClass:
# Validate parameters before we create the object.
bt2_stream_class._StreamClass._validate_create_params(
name,
return fc
- def create_bool_field_class(self, user_attributes=None):
+ def create_bool_field_class(
+ self, user_attributes: typing.Optional[bt2_value._MapValueConst] = None
+ ) -> bt2_field_class._BoolFieldClass:
return self._check_and_wrap_field_class(
native_bt.field_class_bool_create(self._ptr), "boolean", user_attributes
)
- def create_bit_array_field_class(self, length, user_attributes=None):
+ def create_bit_array_field_class(
+ self,
+ length: int,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ ) -> bt2_field_class._BitArrayFieldClass:
bt2_utils._check_uint64(length)
if length < 1 or length > 64:
return fc
def create_signed_integer_field_class(
- self, field_value_range=None, preferred_display_base=None, user_attributes=None
- ):
+ self,
+ field_value_range: typing.Optional[int] = None,
+ preferred_display_base: typing.Optional[
+ bt2_field_class.IntegerDisplayBase
+ ] = None,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ ) -> bt2_field_class._SignedIntegerFieldClass:
return self._create_integer_field_class(
native_bt.field_class_integer_signed_create,
"signed integer",
)
def create_unsigned_integer_field_class(
- self, field_value_range=None, preferred_display_base=None, user_attributes=None
- ):
+ self,
+ field_value_range: typing.Optional[int] = None,
+ preferred_display_base: typing.Optional[
+ bt2_field_class.IntegerDisplayBase
+ ] = None,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ ) -> bt2_field_class._UnsignedIntegerFieldClass:
return self._create_integer_field_class(
native_bt.field_class_integer_unsigned_create,
"unsigned integer",
)
def create_signed_enumeration_field_class(
- self, field_value_range=None, preferred_display_base=None, user_attributes=None
- ):
+ self,
+ field_value_range: typing.Optional[int] = None,
+ preferred_display_base: typing.Optional[
+ bt2_field_class.IntegerDisplayBase
+ ] = None,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ ) -> bt2_field_class._SignedEnumerationFieldClass:
return self._create_integer_field_class(
native_bt.field_class_enumeration_signed_create,
"signed enumeration",
)
def create_unsigned_enumeration_field_class(
- self, field_value_range=None, preferred_display_base=None, user_attributes=None
- ):
+ self,
+ field_value_range: typing.Optional[int] = None,
+ preferred_display_base: typing.Optional[
+ bt2_field_class.IntegerDisplayBase
+ ] = None,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ ) -> bt2_field_class._UnsignedEnumerationFieldClass:
return self._create_integer_field_class(
native_bt.field_class_enumeration_unsigned_create,
"unsigned enumeration",
user_attributes,
)
- def create_single_precision_real_field_class(self, user_attributes=None):
+ def create_single_precision_real_field_class(
+ self, user_attributes: typing.Optional[bt2_value._MapValueConst] = None
+ ) -> bt2_field_class._SinglePrecisionRealFieldClass:
return self._check_and_wrap_field_class(
native_bt.field_class_real_single_precision_create(self._ptr),
"single-precision real",
user_attributes,
)
- def create_double_precision_real_field_class(self, user_attributes=None):
+ def create_double_precision_real_field_class(
+ self, user_attributes: typing.Optional[bt2_value._MapValueConst] = None
+ ) -> bt2_field_class._DoublePrecisionRealFieldClass:
return self._check_and_wrap_field_class(
native_bt.field_class_real_double_precision_create(self._ptr),
"double-precision real",
user_attributes,
)
- def create_structure_field_class(self, user_attributes=None):
+ def create_structure_field_class(
+ self, user_attributes: typing.Optional[bt2_value._MapValueConst] = None
+ ) -> bt2_field_class._StructureFieldClass:
return self._check_and_wrap_field_class(
native_bt.field_class_structure_create(self._ptr),
"structure",
user_attributes,
)
- def create_string_field_class(self, user_attributes=None):
+ def create_string_field_class(
+ self, user_attributes: typing.Optional[bt2_value._MapValueConst] = None
+ ) -> bt2_field_class._StringFieldClass:
return self._check_and_wrap_field_class(
native_bt.field_class_string_create(self._ptr), "string", user_attributes
)
- def create_static_array_field_class(self, elem_fc, length, user_attributes=None):
+ def create_static_array_field_class(
+ self,
+ elem_fc: bt2_field_class._FieldClass,
+ length: int,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ ) -> bt2_field_class._StaticArrayFieldClass:
bt2_utils._check_type(elem_fc, bt2_field_class._FieldClass)
bt2_utils._check_uint64(length)
return self._check_and_wrap_field_class(
)
def create_dynamic_array_field_class(
- self, elem_fc, length_fc=None, user_attributes=None
- ):
+ self,
+ elem_fc: bt2_field_class._FieldClass,
+ length_fc: typing.Optional[bt2_field_class._FieldClass] = None,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ ) -> bt2_field_class._DynamicArrayFieldClass:
bt2_utils._check_type(elem_fc, bt2_field_class._FieldClass)
length_fc_ptr = None
)
def create_option_without_selector_field_class(
- self, content_fc, user_attributes=None
- ):
+ self,
+ content_fc: bt2_field_class._FieldClass,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ ) -> bt2_field_class._OptionFieldClass:
bt2_utils._check_type(content_fc, bt2_field_class._FieldClass)
return self._check_and_wrap_field_class(
)
def create_option_with_bool_selector_field_class(
- self, content_fc, selector_fc, selector_is_reversed=False, user_attributes=None
- ):
+ self,
+ content_fc: bt2_field_class._FieldClass,
+ selector_fc: bt2_field_class._FieldClass,
+ selector_is_reversed: bool = False,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ ) -> bt2_field_class._OptionWithBoolSelectorFieldClass:
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)
return fc
def create_option_with_integer_selector_field_class(
- self, content_fc, selector_fc, ranges, user_attributes=None
- ):
+ self,
+ content_fc: bt2_field_class._FieldClass,
+ selector_fc: bt2_field_class._FieldClass,
+ ranges: bt2_integer_range_set._IntegerRangeSetConst,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ ) -> bt2_field_class._OptionWithIntegerSelectorFieldClass:
bt2_utils._check_type(content_fc, bt2_field_class._FieldClass)
bt2_utils._check_type(selector_fc, bt2_field_class._IntegerFieldClass)
return self._check_and_wrap_field_class(ptr, "option", user_attributes)
- def create_variant_field_class(self, selector_fc=None, user_attributes=None):
+ def create_variant_field_class(
+ self,
+ selector_fc: typing.Optional[bt2_field_class._FieldClass] = None,
+ user_attributes: typing.Optional[bt2_value._MapValueConst] = None,
+ ) -> bt2_field_class._VariantFieldClass:
selector_fc_ptr = None
if selector_fc is not None:
from bt2 import value as bt2_value
from bt2 import plugin as bt2_plugin
from bt2 import logging as bt2_logging
+from bt2 import message as bt2_message
from bt2 import component as bt2_component
from bt2 import native_bt
from bt2 import query_executor as bt2_query_executor
from bt2 import message_iterator as bt2_message_iterator
from bt2 import component_descriptor as bt2_component_descriptor
+typing = bt2_utils._typing_mod
+
# a pair of component and ComponentSpec
_ComponentAndSpec = namedtuple("_ComponentAndSpec", ["comp", "spec"])
return self._params
@property
- def obj(self):
+ def obj(self) -> object:
return self._obj
@property
# A component spec with a specific component class.
def __init__(
self,
- component_class,
+ component_class: typing.Union[
+ bt2_component._SourceComponentClassConst,
+ bt2_component._FilterComponentClassConst,
+ typing.Type[bt2_component._UserSourceComponent],
+ typing.Type[bt2_component._UserFilterComponent],
+ ],
params=None,
obj=None,
logging_level=bt2_logging.LoggingLevel.NONE,
self._component_class = component_class
@property
- def component_class(self):
+ def component_class(
+ self,
+ ) -> typing.Union[
+ bt2_component._SourceComponentClassConst,
+ bt2_component._FilterComponentClassConst,
+ typing.Type[bt2_component._UserSourceComponent],
+ typing.Type[bt2_component._UserFilterComponent],
+ ]:
return self._component_class
@classmethod
def from_named_plugin_and_component_class(
cls,
- plugin_name,
- component_class_name,
+ plugin_name: str,
+ component_class_name: str,
params=None,
- obj=None,
- logging_level=bt2_logging.LoggingLevel.NONE,
+ obj: object = None,
+ logging_level: int = bt2_logging.LoggingLevel.NONE,
):
plugin = bt2_plugin.find_plugin(plugin_name)
# A component spec that does automatic source discovery.
_no_obj = object()
- def __init__(self, input, params=None, obj=_no_obj, logging_level=None):
+ def __init__(
+ self,
+ input: str,
+ params=None,
+ obj: object = _no_obj,
+ logging_level: typing.Optional[int] = None,
+ ):
super().__init__(params, obj, logging_level)
self._input = input
@property
- def input(self):
+ def input(self) -> str:
return self._input
class TraceCollectionMessageIterator(bt2_message_iterator._MessageIterator):
def __init__(
self,
- source_component_specs,
- filter_component_specs=None,
- stream_intersection_mode=False,
- begin=None,
- end=None,
- plugin_set=None,
+ source_component_specs: typing.Iterable[ComponentSpec],
+ filter_component_specs: typing.Optional[typing.Iterable[ComponentSpec]] = None,
+ stream_intersection_mode: bool = False,
+ begin: typing.Union[None, numbers.Real, datetime.datetime] = None,
+ end: typing.Union[None, numbers.Real, datetime.datetime] = None,
+ plugin_set: typing.Optional[bt2_plugin._PluginSet] = None,
):
bt2_utils._check_bool(stream_intersection_mode)
self._stream_intersection_mode = stream_intersection_mode
'"{}" object is not a ComponentSpec'.format(type(comp_spec))
)
- def __next__(self):
+ def __next__(self) -> bt2_message._MessageConst:
assert self._msg_list[0] is None
self._graph.run_once()
msg = self._msg_list[0]
from bt2 import object as bt2_object
from bt2 import native_bt
+typing = bt2_utils._typing_mod
+
def _create_from_ptr_template(ptr, object_map):
if ptr is None:
return _create_from_ptr_and_get_ref_template(ptr, _TYPE_TO_CONST_OBJ)
-def create_value(value):
+def create_value(value) -> typing.Optional["_Value"]:
if value is None:
# null value object
return
_create_from_const_ptr_and_get_ref
)
- def __ne__(self, other):
+ def __ne__(self, other) -> bool:
return not (self == other)
def _check_create_status(self, ptr):
"'{}' object is not a number object".format(other.__class__.__name__)
)
- def __int__(self):
+ def __int__(self) -> int:
return int(self._value)
- def __float__(self):
+ def __float__(self) -> float:
return float(self._value)
def __repr__(self):
return repr(self._value)
- def __lt__(self, other):
+ def __lt__(self, other) -> bool:
return self._value < self._extract_value(other)
- def __eq__(self, other):
+ def __eq__(self, other: object) -> bool:
try:
return self._value == self._extract_value(other)
except Exception:
else:
return round(self._value, ndigits)
- def __ceil__(self):
+ def __ceil__(self) -> int:
return math.ceil(self._value)
- def __floor__(self):
+ def __floor__(self) -> int:
return math.floor(self._value)
- def __trunc__(self):
+ def __trunc__(self) -> int:
return int(self._value)
def __abs__(self):
class _BoolValueConst(_IntegralValueConst):
_NAME = "Const boolean"
- def __bool__(self):
+ def __bool__(self) -> bool:
return self._value
- def __repr__(self):
+ def __repr__(self) -> str:
return repr(self._value)
@property
class BoolValue(_BoolValueConst, _IntegralValue):
_NAME = "Boolean"
- def __init__(self, value=None):
+ def __init__(self, value: typing.Union[None, bool, _BoolValueConst] = None):
if value is None:
ptr = native_bt.value_bool_create()
else:
class _IntegerValue(_IntegerValueConst, _IntegralValue):
- def __init__(self, value=None):
+ def __init__(self, value: typing.Optional[typing.SupportsInt] = None):
if value is None:
ptr = self._create_default_value()
else:
class RealValue(_RealValueConst, _NumericValue):
_NAME = "Real number"
- def __init__(self, value=None):
+ def __init__(self, value: typing.Optional[typing.SupportsFloat] = None):
if value is None:
ptr = native_bt.value_real_create()
else:
def _value(self):
return native_bt.value_string_get(self._ptr)
- def __eq__(self, other):
+ def __eq__(self, other: object) -> bool:
try:
return self._value == self._value_to_str(other)
except Exception:
return False
- def __lt__(self, other):
+ def __lt__(self, other) -> bool:
return self._value < self._value_to_str(other)
- def __bool__(self):
+ def __bool__(self) -> bool:
return bool(self._value)
- def __repr__(self):
+ def __repr__(self) -> str:
return repr(self._value)
- def __str__(self):
+ def __str__(self) -> str:
return self._value
- def __getitem__(self, index):
+ def __getitem__(self, index: int) -> str:
return self._value[index]
- def __len__(self):
+ def __len__(self) -> int:
return len(self._value)
- def __contains__(self, item):
+ def __contains__(self, item: typing.Union[str, "_StringValueConst"]) -> bool:
return self._value_to_str(item) in self._value
class StringValue(_StringValueConst, _Value):
_NAME = "String"
- def __init__(self, value=None):
+ def __init__(self, value: typing.Union[None, str, _StringValueConst] = None):
if value is None:
ptr = native_bt.value_string_create()
else:
value = property(fset=_set_value)
- def __iadd__(self, value):
+ def __iadd__(self, value: typing.Union[str, _StringValueConst]) -> "StringValue":
curvalue = self._value
curvalue += self._value_to_str(value)
self.value = curvalue
class _ContainerConst:
- def __bool__(self):
+ def __bool__(self) -> bool:
return len(self) != 0
)
_is_const = True
- def __eq__(self, other):
+ def __eq__(self, other: object) -> bool:
if not isinstance(other, collections.abc.Sequence):
return False
return True
- def __len__(self):
+ def __len__(self) -> int:
size = native_bt.value_array_get_length(self._ptr)
assert size >= 0
return size
if index < 0 or index >= len(self):
raise IndexError("array value object index is out of range")
- def __getitem__(self, index):
+ def __getitem__(self, index: int) -> typing.Optional[_ValueConst]:
self._check_index(index)
ptr = self._borrow_element_by_index(self._ptr, index)
assert ptr
return self._create_value_from_ptr_and_get_ref(ptr)
- def __repr__(self):
+ def __repr__(self) -> str:
return "[{}]".format(", ".join([repr(v) for v in self]))
native_bt.value_array_borrow_element_by_index
)
- def __init__(self, value=None):
+ def __init__(self, value: typing.Optional[typing.Iterable] = None):
ptr = native_bt.value_array_create()
self._check_create_status(ptr)
super().__init__(ptr)
for elem in value:
self.append(elem)
- def __setitem__(self, index, value):
+ def __setitem__(self, index: int, value):
self._check_index(index)
value = create_value(value)
status = native_bt.value_array_append_element(self._ptr, ptr)
bt2_utils._handle_func_status(status)
- def __iadd__(self, iterable):
+ def __iadd__(self, iterable: typing.Iterable):
# Python will raise a TypeError if there's anything wrong with
# the iterable protocol.
for elem in iterable:
self._keys = _create_from_ptr(keys_ptr)
- def __next__(self):
+ def __next__(self) -> str:
if self._at == len(self._map_obj):
raise StopIteration
_NAME = "Const map"
_borrow_entry_value_ptr = staticmethod(native_bt.value_map_borrow_entry_value_const)
- def __ne__(self, other):
+ def __ne__(self, other) -> bool:
return _Value.__ne__(self, other)
- def __eq__(self, other):
+ def __eq__(self, other: object) -> bool:
if not isinstance(other, collections.abc.Mapping):
return False
return True
- def __len__(self):
+ def __len__(self) -> int:
size = native_bt.value_map_get_size(self._ptr)
assert size >= 0
return size
- def __contains__(self, key):
+ def __contains__(self, key: str) -> bool:
self._check_key_type(key)
return native_bt.value_map_has_entry(self._ptr, key)
if key not in self:
raise KeyError(key)
- def __getitem__(self, key):
+ def __getitem__(self, key: str) -> typing.Optional[_ValueConst]:
self._check_key(key)
ptr = self._borrow_entry_value_ptr(self._ptr, key)
assert ptr
return self._create_value_from_ptr_and_get_ref(ptr)
- def __iter__(self):
+ def __iter__(self) -> typing.Iterator[str]:
return _MapValueKeyIterator(self)
- def __repr__(self):
+ def __repr__(self) -> str:
items = ["{}: {}".format(repr(k), repr(v)) for k, v in self.items()]
return "{{{}}}".format(", ".join(items))
_NAME = "Map"
_borrow_entry_value_ptr = staticmethod(native_bt.value_map_borrow_entry_value)
- def __init__(self, value=None):
+ def __init__(self, value: typing.Optional[typing.Iterable] = None):
ptr = native_bt.value_map_create()
self._check_create_status(ptr)
super().__init__(ptr)
for key, elem in value.items():
self[key] = elem
- def __setitem__(self, key, value):
+ def __setitem__(self, key: str, value):
self._check_key_type(key)
value = create_value(value)
projects / babeltrace.git / commitdiff
