This prevents name collisions with the upcoming CTFWriter classes.
Signed-off-by: Jérémie Galarneau <jeremie.galarneau@efficios.com>
-class ctf:
-
- #enum equivalent, accessible constants
- #These are taken directly from ctf/events.h
- #All changes to enums must also be made here
- class type_id:
- UNKNOWN = 0
- INTEGER = 1
- FLOAT = 2
- ENUM = 3
- STRING = 4
- STRUCT = 5
- UNTAGGED_VARIANT = 6
- VARIANT = 7
- ARRAY = 8
- SEQUENCE = 9
- NR_CTF_TYPES = 10
-
- def get_type_id_name(id):
- name = "UNKNOWN"
- constants = [attr for attr in dir(ctf.type_id) if not callable(getattr(ctf.type_id, attr)) and not attr.startswith("__")]
- for attr in constants:
- if getattr(ctf.type_id, attr) == id:
- name = attr
- break
- return name
-
+class CTFStringEncoding:
+ NONE = 0
+ UTF8 = 1
+ ASCII = 2
+ UNKNOWN = 3
+
+#enum equivalent, accessible constants
+#These are taken directly from ctf/events.h
+#All changes to enums must also be made here
+class CTFTypeId:
+ UNKNOWN = 0
+ INTEGER = 1
+ FLOAT = 2
+ ENUM = 3
+ STRING = 4
+ STRUCT = 5
+ UNTAGGED_VARIANT = 6
+ VARIANT = 7
+ ARRAY = 8
+ SEQUENCE = 9
+ NR_CTF_TYPES = 10
+
+ def get_type_name(id):
+ name = "UNKNOWN"
+ constants = [attr for attr in dir(CTFTypeId) if not callable(getattr(CTFTypeId, attr)) and not attr.startswith("__")]
+ for attr in constants:
+ if getattr(CTFTypeId, attr) == id:
+ name = attr
+ break
+ return name
+
+class CTFReader:
class scope:
TRACE_PACKET_HEADER = 0
STREAM_PACKET_CONTEXT = 1
class scope:
TRACE_PACKET_HEADER = 0
STREAM_PACKET_CONTEXT = 1
def __new__(cls, context, begin_pos = None, end_pos = None):
# __new__ is used to control the return value
def __new__(cls, context, begin_pos = None, end_pos = None):
# __new__ is used to control the return value
- # as the ctf.Iterator class should return None
+ # as the CTFReader.Iterator class should return None
# if bt_ctf_iter_create returns NULL
if begin_pos is None:
# if bt_ctf_iter_create returns NULL
if begin_pos is None:
if it is None:
return None
if it is None:
return None
- ret_class = super(ctf.Iterator, cls).__new__(cls)
+ ret_class = super(CTFReader.Iterator, cls).__new__(cls)
ret_class._i = it
return ret_class
ret_class._i = it
return ret_class
ret = _bt_ctf_iter_read_event(self._i)
if ret is None:
return ret
ret = _bt_ctf_iter_read_event(self._i)
if ret is None:
return ret
- ev = ctf.Event.__new__(ctf.Event)
+ ev = CTFReader.Event.__new__(CTFReader.Event)
class Event(object):
"""
This class represents an event from the trace.
class Event(object):
"""
This class represents an event from the trace.
- It is obtained with read_event() from ctf.Iterator.
+ It is obtained with read_event() from CTFReader.Iterator.
Do not instantiate.
"""
def __init__(self):
Do not instantiate.
"""
def __init__(self):
- raise NotImplementedError("ctf.Event cannot be instantiated")
+ raise NotImplementedError("CTFReader.Event cannot be instantiated")
def get_top_level_scope(self, scope):
"""
Return a definition of the top-level scope
def get_top_level_scope(self, scope):
"""
Return a definition of the top-level scope
- Top-level scopes are defined in ctf.scope.
+ Top-level scopes are defined in CTFReader.scope.
In order to get a field or a field list, the user needs to pass a
scope as argument, this scope can be a top-level scope or a scope
relative to an arbitrary field. This function provides the mapping
between the scope and the actual definition of top-level scopes.
On error return None.
"""
In order to get a field or a field list, the user needs to pass a
scope as argument, this scope can be a top-level scope or a scope
relative to an arbitrary field. This function provides the mapping
between the scope and the actual definition of top-level scopes.
On error return None.
"""
- evDef = ctf.Definition.__new__(ctf.Definition)
+ evDef = CTFReader.Definition.__new__(CTFReader.Definition)
evDef._d = _bt_ctf_get_top_level_scope(self._e, scope)
if evDef._d is None:
return None
evDef._d = _bt_ctf_get_top_level_scope(self._e, scope)
if evDef._d is None:
return None
Return the definition of a specific field.
Return None on error.
"""
Return the definition of a specific field.
Return None on error.
"""
- evDef = ctf.Definition.__new__(ctf.Definition)
+ evDef = CTFReader.Definition.__new__(CTFReader.Definition)
try:
evDef._d = _bt_ctf_get_field(self._e, scope._d, field)
except AttributeError:
try:
evDef._d = _bt_ctf_get_field(self._e, scope._d, field)
except AttributeError:
Return the definition of fields by a name
Return None on error
"""
Return the definition of fields by a name
Return None on error
"""
- eventScope = self.get_top_level_scope(ctf.scope.EVENT_FIELDS)
- streamScope = self.get_top_level_scope(ctf.scope.STREAM_EVENT_CONTEXT)
+ eventScope = self.get_top_level_scope(CTFReader.scope.EVENT_FIELDS)
+ streamScope = self.get_top_level_scope(CTFReader.scope.STREAM_EVENT_CONTEXT)
fields_by_name = []
if eventScope is not None:
fields_by_name = []
if eventScope is not None:
def_list = []
for i in range(count):
def_list = []
for i in range(count):
- tmp = ctf.Definition.__new__(ctf.Definition)
+ tmp = CTFReader.Definition.__new__(CTFReader.Definition)
tmp._d = _bt_python_field_one_from_list(field_lc, i)
tmp._s = scope
def_list.append(tmp)
tmp._d = _bt_python_field_one_from_list(field_lc, i)
tmp._s = scope
def_list.append(tmp)
def get_field_list(self):
"""Return a list of Definitions or None on error."""
def get_field_list(self):
"""Return a list of Definitions or None on error."""
- eventScope = self.get_top_level_scope(ctf.scope.EVENT_FIELDS)
- streamScope = self.get_top_level_scope(ctf.scope.STREAM_EVENT_CONTEXT)
+ eventScope = self.get_top_level_scope(CTFReader.scope.EVENT_FIELDS)
+ streamScope = self.get_top_level_scope(CTFReader.scope.STREAM_EVENT_CONTEXT)
def_list = []
if eventScope is not None:
def_list = []
if eventScope is not None:
If the field is an array or a sequence, return the element
at position index, otherwise return None
"""
If the field is an array or a sequence, return the element
at position index, otherwise return None
"""
- evDef = ctf.Definition.__new__(ctf.Definition)
+ evDef = CTFReader.Definition.__new__(CTFReader.Definition)
try:
evDef._d = _bt_ctf_get_index(self._e, field._d, index)
except AttributeError:
try:
evDef._d = _bt_ctf_get_index(self._e, field._d, index)
except AttributeError:
"""Definition class. Do not instantiate."""
def __init__(self):
"""Definition class. Do not instantiate."""
def __init__(self):
- raise NotImplementedError("ctf.Definition cannot be instantiated")
+ raise NotImplementedError("CTFReader.Definition cannot be instantiated")
def __repr__(self):
return "Babeltrace Definition: name('{0}'), type({1})".format(
def __repr__(self):
return "Babeltrace Definition: name('{0}'), type({1})".format(
if array is None:
return None
if array is None:
return None
- element = ctf.Definition.__new__(ctf.Definition)
+ element = CTFReader.Definition.__new__(CTFReader.Definition)
element._d = _bt_array_index(array, index)
if element._d is None:
return None
element._d = _bt_array_index(array, index)
if element._d is None:
return None
"""
seq = _bt_python_get_sequence_from_def(self._d)
if seq is not None:
"""
seq = _bt_python_get_sequence_from_def(self._d)
if seq is not None:
- element = ctf.Definition.__new__(ctf.Definition)
+ element = CTFReader.Definition.__new__(CTFReader.Definition)
element._d = _bt_sequence_index(seq, index)
if element._d is not None:
return element
element._d = _bt_sequence_index(seq, index)
if element._d is not None:
return element
Return the value associated with the field.
If the field does not exist or is not of the type requested,
the value returned is undefined. To check if an error occured,
Return the value associated with the field.
If the field does not exist or is not of the type requested,
the value returned is undefined. To check if an error occured,
- use the ctf.field_error() function after accessing a field.
+ use the CTFReader.field_error() function after accessing a field.
"""
return _bt_ctf_get_uint64(self._d)
"""
return _bt_ctf_get_uint64(self._d)
Return the value associated with the field.
If the field does not exist or is not of the type requested,
the value returned is undefined. To check if an error occured,
Return the value associated with the field.
If the field does not exist or is not of the type requested,
the value returned is undefined. To check if an error occured,
- use the ctf.field_error() function after accessing a field.
+ use the CTFReader.field_error() function after accessing a field.
"""
return _bt_ctf_get_int64(self._d)
"""
return _bt_ctf_get_int64(self._d)
Return the value associated with the field.
If the field does not exist or is not of the type requested,
the value returned is undefined. To check if an error occured,
Return the value associated with the field.
If the field does not exist or is not of the type requested,
the value returned is undefined. To check if an error occured,
- use the ctf.field_error() function after accessing a field.
+ use the CTFReader.field_error() function after accessing a field.
"""
return _bt_ctf_get_char_array(self._d)
"""
return _bt_ctf_get_char_array(self._d)
Return the value associated with the field.
If the field does not exist or is not of the type requested,
the value returned is undefined. To check if an error occured,
Return the value associated with the field.
If the field does not exist or is not of the type requested,
the value returned is undefined. To check if an error occured,
- use the ctf.field_error() function after accessing a field.
+ use the CTFReader.field_error() function after accessing a field.
"""
return _bt_ctf_get_string(self._d)
"""
return _bt_ctf_get_string(self._d)
Return the value associated with the field.
If the field does not exist or is not of the type requested,
the value returned is undefined. To check if an error occured,
Return the value associated with the field.
If the field does not exist or is not of the type requested,
the value returned is undefined. To check if an error occured,
- use the ctf.field_error() function after accessing a field.
+ use the CTFReader.field_error() function after accessing a field.
"""
return _bt_ctf_get_float(self._d)
"""
return _bt_ctf_get_float(self._d)
Return the variant's selected field.
If the field does not exist or is not of the type requested,
the value returned is undefined. To check if an error occured,
Return the variant's selected field.
If the field does not exist or is not of the type requested,
the value returned is undefined. To check if an error occured,
- use the ctf.field_error() function after accessing a field.
+ use the CTFReader.field_error() function after accessing a field.
"""
return _bt_ctf_get_variant(self._d)
"""
return _bt_ctf_get_variant(self._d)
Return the structure's field at position i.
If the field does not exist or is not of the type requested,
the value returned is undefined. To check if an error occured,
Return the structure's field at position i.
If the field does not exist or is not of the type requested,
the value returned is undefined. To check if an error occured,
- use the ctf.field_error() function after accessing a field.
+ use the CTFReader.field_error() function after accessing a field.
"""
return _bt_ctf_get_struct_field_index(self._d, i)
"""
return _bt_ctf_get_struct_field_index(self._d, i)
"""
id = self.field_type()
value = None
"""
id = self.field_type()
value = None
- if id == ctf.type_id.STRING:
+ if id == CTFTypeId.STRING:
- elif id == ctf.type_id.ARRAY:
+ elif id == CTFTypeId.ARRAY:
value = []
for i in range(self.get_array_len()):
element = self.get_array_element_at(i)
value.append(element.get_value())
value = []
for i in range(self.get_array_len()):
element = self.get_array_element_at(i)
value.append(element.get_value())
- elif id == ctf.type_id.INTEGER:
+ elif id == CTFTypeId.INTEGER:
if self.get_int_signedness() == 0:
value = self.get_uint64()
else:
value = self.get_int64()
if self.get_int_signedness() == 0:
value = self.get_uint64()
else:
value = self.get_int64()
- elif id == ctf.type_id.ENUM:
+ elif id == CTFTypeId.ENUM:
value = self.get_enum_str()
value = self.get_enum_str()
- elif id == ctf.type_id.SEQUENCE:
+ elif id == CTFTypeId.SEQUENCE:
seq_len = self.get_sequence_len()
value = []
for i in range(seq_len):
evDef = self.get_sequence_element_at(i)
value.append(evDef.get_value())
seq_len = self.get_sequence_len()
value = []
for i in range(seq_len):
evDef = self.get_sequence_element_at(i)
value.append(evDef.get_value())
- elif id == ctf.type_id.FLOAT:
+ elif id == CTFTypeId.FLOAT:
- elif id == ctf.type_id.VARIANT:
- variant = ctf.Definition.__new__(ctf.Definition)
+ elif id == CTFTypeId.VARIANT:
+ variant = CTFReader.Definition.__new__(CTFReader.Definition)
variant._d = self.get_variant();
value = variant.get_value()
variant._d = self.get_variant();
value = variant.get_value()
- elif id == ctf.type_id.STRUCT:
+ elif id == CTFTypeId.STRUCT:
value = {}
for i in range(self.get_struct_field_count()):
value = {}
for i in range(self.get_struct_field_count()):
- member = ctf.Definition.__new__(ctf.Definition)
+ member = CTFReader.Definition.__new__(CTFReader.Definition)
member._d = self.get_struct_field_at(i);
value[member.field_name()] = member.get_value()
member._d = self.get_struct_field_at(i);
value[member.field_name()] = member.get_value()
- if ctf.field_error():
- raise ctf.FieldError("Error occured while accessing field {} of type {}".format(self.field_name(), ctf.type_id.get_type_id_name(self.field_type())))
+ if CTFReader.field_error():
+ raise CTFReader.FieldError("Error occured while accessing field {} of type {}".format(self.field_name(), CTFTypeId.get_type_name(self.field_type())))
return value
def get_scope(self):
return value
def get_scope(self):
"""Event declaration class. Do not instantiate."""
def __init__(self):
"""Event declaration class. Do not instantiate."""
def __init__(self):
- raise NotImplementedError("ctf.EventDecl cannot be instantiated")
+ raise NotImplementedError("CTFReader.EventDecl cannot be instantiated")
def __repr__(self):
return "Babeltrace EventDecl: name {0}".format(self.get_name())
def __repr__(self):
return "Babeltrace EventDecl: name {0}".format(self.get_name())
def get_decl_fields(self, scope):
"""
def get_decl_fields(self, scope):
"""
- Return a list of ctf.FieldDecl
+ Return a list of CTFReader.FieldDecl
Return None on error.
"""
ptr_list = _by_python_field_decl_listcaller(self._d, scope)
Return None on error.
"""
ptr_list = _by_python_field_decl_listcaller(self._d, scope)
decl_list = []
i = 0
while True:
decl_list = []
i = 0
while True:
- tmp = ctf.FieldDecl.__new__(ctf.FieldDecl)
+ tmp = CTFReader.FieldDecl.__new__(CTFReader.FieldDecl)
tmp._d = _bt_python_field_decl_one_from_list(
ptr_list, i)
tmp._d = _bt_python_field_decl_one_from_list(
ptr_list, i)
"""Field declaration class. Do not instantiate."""
def __init__(self):
"""Field declaration class. Do not instantiate."""
def __init__(self):
- raise NotImplementedError("ctf.FieldDecl cannot be instantiated")
+ raise NotImplementedError("CTFReader.FieldDecl cannot be instantiated")
def __repr__(self):
return "Babeltrace FieldDecl: name {0}".format(self.get_name())
def __repr__(self):
return "Babeltrace FieldDecl: name {0}".format(self.get_name())
@staticmethod
def get_event_decl_list(trace_handle, context):
"""
@staticmethod
def get_event_decl_list(trace_handle, context):
"""
- Return a list of ctf.EventDecl
+ Return a list of CTFReader.EventDecl
Return None on error.
"""
try:
Return None on error.
"""
try:
decl_list = []
for i in range(count):
decl_list = []
for i in range(count):
- tmp = ctf.EventDecl.__new__(ctf.EventDecl)
+ tmp = CTFReader.EventDecl.__new__(CTFReader.EventDecl)
tmp._d = _bt_python_decl_one_from_list(ptr_list, i)
decl_list.append(tmp)
tmp._d = _bt_python_decl_one_from_list(ptr_list, i)
decl_list.append(tmp)
# Iterator setup
bp = babeltrace.IterPos(babeltrace.SEEK_BEGIN)
# Iterator setup
bp = babeltrace.IterPos(babeltrace.SEEK_BEGIN)
-ctf_it = babeltrace.ctf.Iterator(ctx,bp)
+ctf_it = babeltrace.CTFReader.Iterator(ctx,bp)
# Reading events from trace
# and outputting timestamps and event names
# Reading events from trace
# and outputting timestamps and event names
raise IOError("Error adding trace")
# Listing events
raise IOError("Error adding trace")
# Listing events
-lst = ctf.get_event_decl_list(trace_handle, ctx)
+lst = CTFReader.get_event_decl_list(trace_handle, ctx)
print("--- Event list ---")
for item in lst:
print("event : {}".format(item.get_name()))
print("--- Event list ---")
for item in lst:
print("event : {}".format(item.get_name()))
# Iter trace
bp = IterPos(SEEK_BEGIN)
# Iter trace
bp = IterPos(SEEK_BEGIN)
-ctf_it = ctf.Iterator(ctx,bp)
+ctf_it = CTFReader.Iterator(ctx,bp)
event = ctf_it.read_event()
while(event is not None):
event = ctf_it.read_event()
while(event is not None):
# Setting iterator
bp = IterPos(SEEK_BEGIN)
# Setting iterator
bp = IterPos(SEEK_BEGIN)
-ctf_it = ctf.Iterator(ctx, bp)
+ctf_it = CTFReader.Iterator(ctx, bp)
# Reading events
event = ctf_it.read_event()
# Reading events
event = ctf_it.read_event()
while True:
if event.get_name() == "sched_switch":
# Getting scope definition
while True:
if event.get_name() == "sched_switch":
# Getting scope definition
- sco = event.get_top_level_scope(ctf.scope.STREAM_EVENT_CONTEXT)
+ sco = event.get_top_level_scope(CTFReader.scope.STREAM_EVENT_CONTEXT)
if sco is None:
print("ERROR: Cannot get definition scope for sched_switch")
break # Next event
if sco is None:
print("ERROR: Cannot get definition scope for sched_switch")
break # Next event
if usePID and (pid != long(sys.argv[1])):
break # Next event
if usePID and (pid != long(sys.argv[1])):
break # Next event
- sco = event.get_top_level_scope(ctf.scope.EVENT_FIELDS)
+ sco = event.get_top_level_scope(CTFReader.scope.EVENT_FIELDS)
# prev_comm
field = event.get_field_with_scope(sco, "prev_comm")
# prev_comm
field = event.get_field_with_scope(sco, "prev_comm")
raise IOError("Error adding trace")
# Listing events
raise IOError("Error adding trace")
# Listing events
-lst = ctf.get_event_decl_list(trace_handle, ctx)
+lst = CTFReader.get_event_decl_list(trace_handle, ctx)
print("--- Event list ---")
for item in lst:
print("event : {}".format(item.get_name()))
print("--- Event list ---")
for item in lst:
print("event : {}".format(item.get_name()))
# Iter trace
bp = IterPos(SEEK_BEGIN)
# Iter trace
bp = IterPos(SEEK_BEGIN)
-ctf_it = ctf.Iterator(ctx,bp)
+ctf_it = CTFReader.Iterator(ctx,bp)
event = ctf_it.read_event()
while(event is not None):
event = ctf_it.read_event()
while(event is not None):
projects / babeltrace.git / commitdiff
