# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
+
class CTFStringEncoding:
"""
CTF string encodings.
from babeltrace.reader_trace_collection import *
from babeltrace.reader_trace_handle import *
+
# Based on enum bt_clock_type in clock-type.h
class _ClockType:
CLOCK_CYCLES = 0
import datetime
import collections
+
def _create_event(event_notification, trace_handle=None, trace_collection=None):
event = Event.__new__(Event)
event._event_notification = event_notification
event._trace_collection = trace_collection
return event
+
class Event(collections.Mapping):
"""
An :class:`Event` object represents a trace event. :class:`Event`
import babeltrace.reader_field_declaration as field_declaration
import collections
+
def _create_event_declaration(event_class):
event_declaration = EventDeclaration.__new__(EventDeclaration)
event_declaration._event_class = event_class
return event_declaration
+
class EventDeclaration:
"""
An event declaration contains the properties of a class of events,
yield field_declaration._create_field_declaration(field_type, name,
scope)
+
# Priority of the scopes when searching for event fields
_SCOPES = [
common.CTFScope.EVENT_FIELDS,
import bt2
import babeltrace.common as common
+
def _create_field_declaration(field_type, name, scope):
try:
if type(field_type) == bt2.IntegerFieldType:
import babeltrace.common as common
import babeltrace.reader_field_declaration as reader_field_declaration
+
class FieldError(Exception):
"""
Field error, raised when the value of a field cannot be accessed.
def __str__(self):
return repr(self.value)
+
class _Definition:
def __init__(self, scope_id, field, name):
self._scope_id = scope_id
import babeltrace.common as common
import bt2
+
class EnumerationMapping:
"""
Mapping from an enumeration label to a range of integers.
common.CTFStringEncoding.UTF8: bt2.Encoding.UTF8,
}
+
class FieldDeclaration:
"""
Base class of all field declarations. This class is not meant to
raise TypeError(
"Could not get Floating point exponent digit count")
-
@exponent_digits.setter
def exponent_digits(self, exponent_digits):
try:
raise TypeError(
"Could not get Floating point mantissa digit count")
-
@mantissa_digits.setter
def mantissa_digits(self, mantissa_digits):
try:
raise TypeError("Invalid tag type; must be of type EnumerationFieldDeclaration.")
self._field_type = bt2.VariantFieldType(tag_name=tag_name,
- tag_field_type=enum_tag._field_type)
+ tag_field_type=enum_tag._field_type)
super().__init__()
@property
except:
raise ValueError("Invalid stream context field.")
+
class StreamClass:
"""
A stream class contains the properties of specific
self._obj = obj
-import bt2.native_bt as _native_bt
-import atexit
-
-atexit.register(_native_bt.py3_cc_exit_handler)
-version = (_native_bt.version_get_major(), _native_bt.version_get_minor(),
- _native_bt.version_get_patch(), _native_bt.version_get_extra())
-_native_bt.py3_cc_init_from_bt2()
-del _native_bt
+def _init_and_register_exit():
+ import bt2.native_bt as _native_bt
+ import atexit
+
+ atexit.register(_native_bt.py3_cc_exit_handler)
+ version = (_native_bt.version_get_major(), _native_bt.version_get_minor(),
+ _native_bt.version_get_patch(), _native_bt.version_get_extra())
+ _native_bt.py3_cc_init_from_bt2()
+
+
+_init_and_register_exit()
+
try:
del native_bt
import bt2
import bt2.clock_value as clock_value
+
class ClockClassOffset:
def __init__(self, seconds=0, cycles=0):
utils._check_int64(seconds)
def __call__(self, cycles):
return clock_value._ClockValue(self._ptr, cycles)
-
stream_event_context_ptr = stream_event_context._ptr
ret = native_bt.event_set_stream_event_context(self._ptr,
- stream_event_context_ptr)
+ stream_event_context_ptr)
utils._handle_ret(ret, "cannot set event object's stream event context field")
@property
def _get_clock_value_cycles(self, clock_class_ptr):
clock_value_ptr = native_bt.event_get_clock_value(self._ptr,
- clock_class_ptr)
+ clock_class_ptr)
if clock_value_ptr is None:
return
def clock_value(self, clock_class):
utils._check_type(clock_class, bt2.ClockClass)
clock_value_ptr = native_bt.event_get_clock_value(self._ptr,
- clock_class._ptr)
+ clock_class._ptr)
if clock_value_ptr is None:
return
def add_clock_value(self, clock_value):
utils._check_type(clock_value, bt2.clock_value._ClockValue)
ret = native_bt.event_set_clock_value(self._ptr,
- clock_value._ptr)
+ clock_value._ptr)
utils._handle_ret(ret, "cannot set event object's clock value")
def __getitem__(self, key):
def _add_field(self, ptr, name):
return native_bt.field_type_structure_add_field(self._ptr, ptr,
- name)
+ name)
def _at(self, index):
if index < 0 or index >= len(self):
raise StopIteration
ret, name, field_type_ptr = native_bt.field_type_variant_get_field_by_index(self._variant_field_type._ptr,
- self._at)
+ self._at)
assert(ret == 0)
native_bt.put(field_type_ptr)
self._at += 1
tag_ft_ptr = tag_field_type._ptr
ptr = native_bt.field_type_variant_create(tag_ft_ptr,
- tag_name)
+ tag_name)
self._check_create_status(ptr)
super().__init__(ptr)
utils._check_type(element_field_type, _FieldType)
utils._check_str(length_name)
ptr = native_bt.field_type_sequence_create(element_field_type._ptr,
- length_name)
+ length_name)
self._check_create_status(ptr)
super().__init__(ptr)
value = property(fset=_set_value)
+
class _FloatingPointNumberField(_RealField):
_NAME = 'Floating point number'
value = property(fset=_set_value)
+
class _EnumerationField(_IntegerField):
_NAME = 'Enumeration'
value = property(fset=_set_value)
+
class _VariantField(_Field):
_NAME = 'Variant'
value = property(fset=_set_value)
+
class _ArraySequenceField(_ContainerField, collections.abc.MutableSequence):
def __getitem__(self, index):
if not isinstance(index, numbers.Integral):
value = property(fset=_set_value)
+
_TYPE_ID_TO_OBJ = {
native_bt.FIELD_TYPE_ID_INTEGER: _IntegerField,
native_bt.FIELD_TYPE_ID_FLOAT: _FloatingPointNumberField,
raise StopIteration
ec_ptr = native_bt.stream_class_get_event_class_by_index(self._stream_class._ptr,
- self._at)
+ self._at)
assert(ec_ptr)
ev_id = native_bt.event_class_get_id(ec_ptr)
native_bt.put(ec_ptr)
def __getitem__(self, key):
utils._check_int64(key)
ec_ptr = native_bt.stream_class_get_event_class_by_id(self._ptr,
- key)
+ key)
if ec_ptr is None:
raise KeyError(key)
packet_context_field_type_ptr = packet_context_field_type._ptr
ret = native_bt.stream_class_set_packet_context_type(self._ptr,
- packet_context_field_type_ptr)
+ packet_context_field_type_ptr)
utils._handle_ret(ret, "cannot set stream class object's packet context field type")
@property
event_header_field_type_ptr = event_header_field_type._ptr
ret = native_bt.stream_class_set_event_header_type(self._ptr,
- event_header_field_type_ptr)
+ event_header_field_type_ptr)
utils._handle_ret(ret, "cannot set stream class object's event header field type")
@property
event_context_field_type_ptr = event_context_field_type._ptr
ret = native_bt.stream_class_set_event_context_type(self._ptr,
- event_context_field_type_ptr)
+ event_context_field_type_ptr)
utils._handle_ret(ret, "cannot set stream class object's event context field type")
def __call__(self, name=None, id=None):
raise StopIteration
sc_ptr = native_bt.trace_get_stream_class_by_index(self._trace._ptr,
- self._at)
+ self._at)
assert(sc_ptr)
id = native_bt.stream_class_get_id(sc_ptr)
native_bt.put(sc_ptr)
raise IndexError
stream_ptr = native_bt.trace_get_stream_by_index(self._trace._ptr,
- index)
+ index)
assert(stream_ptr)
return bt2.stream._create_from_ptr(stream_ptr)
def __getitem__(self, key):
utils._check_str(key)
- cc_ptr = native_bt.trace_get_clock_class_by_name(self._trace._ptr,
- key)
+ cc_ptr = native_bt.trace_get_clock_class_by_name(self._trace._ptr, key)
if cc_ptr is None:
raise KeyError(key)
trace_ptr = self._trace_env._trace._ptr
entry_name = native_bt.trace_get_environment_field_name_by_index(trace_ptr,
- self._at)
+ self._at)
assert(entry_name is not None)
self._at += 1
return entry_name
def __getitem__(self, key):
utils._check_str(key)
value_ptr = native_bt.trace_get_environment_field_value_by_name(self._trace._ptr,
- key)
+ key)
if value_ptr is None:
raise KeyError(key)
utils._check_str(key)
value = bt2.create_value(value)
ret = native_bt.trace_set_environment_field(self._trace._ptr,
- key, value._ptr)
+ key, value._ptr)
utils._handle_ret(ret, "cannot set trace class object's environment entry")
def __delitem__(self, key):
packet_header_field_type_ptr = packet_header_field_type._ptr
ret = native_bt.trace_set_packet_header_type(self._ptr,
- packet_header_field_type_ptr)
+ packet_header_field_type_ptr)
utils._handle_ret(ret, "cannot set trace class object's packet header field type")
def __eq__(self, other):
else:
raise bt2.Error(msg)
+
def _handle_ret(ret, msg=None):
if int(ret) < 0:
_raise_bt2_error(msg)
projects / babeltrace.git / commitdiff
