* Copyright 2012 EfficiOS Inc.
*
* Author: Danny Serres <danny.serres@efficios.com>
+ * Author: Jérémie Galarneau <jeremie.galarneau@efficios.com>
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
Common Trace Format (CTF). Its main use is to pretty-print CTF
traces into a human-readable text output.
-To use this module, the first step is to create a Context and add a
+To use this module, the first step is to create a TraceCollection and add a
trace to it."
%enddef
typedef unsigned long long uint64_t;
typedef long long int64_t;
typedef int bt_intern_str;
+typedef int64_t ssize_t;
/* =================================================================
PYTHON-COMPLEMENTS.H
struct bt_definition *field);
struct definition_sequence *_bt_python_get_sequence_from_def(
struct bt_definition *field);
+struct bt_declaration *_bt_python_get_array_element_declaration(
+ struct bt_declaration *field);
+struct bt_declaration *_bt_python_get_sequence_element_declaration(
+ struct bt_declaration *field);
+const char *_bt_python_get_array_string(struct bt_definition *field);
+const char *_bt_python_get_sequence_string(struct bt_definition *field);
int _bt_python_field_integer_get_signedness(const struct bt_ctf_field *field);
enum ctf_type_id _bt_python_get_field_type(const struct bt_ctf_field *field);
void bt_context_put(struct bt_context *ctx);
struct bt_context *bt_ctf_event_get_context(const struct bt_ctf_event *event);
-// class Context to prevent direct access to struct bt_context
+// class TraceCollection to prevent direct access to struct bt_context
%pythoncode%{
-class Context:
+class TraceCollection:
"""
- The context represents the object in which a trace_collection is
- open. As long as this structure is allocated, the trace_collection
- is open and the traces it contains can be read and seeked by the
- iterators and callbacks.
+ The TraceCollection is the object that contains all currently opened traces.
"""
def __init__(self):
- self._c = _bt_context_create()
+ self._tc = _bt_context_create()
def __del__(self):
- _bt_context_put(self._c)
+ _bt_context_put(self._tc)
- def add_trace(self, path, format_str,
- packet_seek=None, stream_list=None, metadata=None):
+ def add_trace(self, path, format_str):
"""
- Add a trace by path to the context.
+ Add a trace by path to the TraceCollection.
Open a trace.
format is a string containing the format name in which the trace was
produced.
- packet_seek is not implemented for Python. Should be left None to
- use the default packet_seek handler provided by the trace format.
-
- stream_list is a linked list of streams, it is used to open a trace
- where the trace data is located in memory mapped areas instead of
- trace files, this argument should be None when path is not None.
-
- The metadata parameter acts as a metadata override when not None,
- otherwise the format handles the metadata opening.
-
Return: the corresponding TraceHandle on success or None on error.
"""
- if metadata is not None:
- metadata = metadata._file
-
- ret = _bt_context_add_trace(self._c, path, format_str, packet_seek,
- stream_list, metadata)
+ ret = _bt_context_add_trace(self._tc, path, format_str, None, None, None)
if ret < 0:
return None
th = TraceHandle.__new__(TraceHandle)
th._id = ret
+ th._trace_collection = self
return th
def add_traces_recursive(self, path, format_str):
Open a trace recursively.
Find each trace present in the subdirectory starting from the given
- path, and add them to the context.
+ path, and add them to the TraceCollection.
Return a dict of TraceHandle instances (the full path is the key).
Return None on error.
def remove_trace(self, trace_handle):
"""
- Remove a trace from the context.
+ Remove a trace from the TraceCollection.
Effectively closing the trace.
"""
try:
- _bt_context_remove_trace(self._c, trace_handle._id)
+ _bt_context_remove_trace(self._tc, trace_handle._id)
except AttributeError:
raise TypeError("in remove_trace, "
"argument 2 must be a TraceHandle instance")
+
+ @property
+ def events(self):
+ """
+ Generator function to iterate over the events of open in the current
+ TraceCollection.
+
+ Due to limitations of the native Babeltrace API, only one event
+ may be "alive" at a time (i.e. a user should never store a copy
+ of the events returned by this function for ulterior use). Users
+ shall make sure to copy the information they need from an event
+ before accessing the next one.
+
+ Furthermore, event objects become invalid when the generator goes
+ out of scope as the underlying iterator will be reclaimed. Using an
+ event after the the generator has gone out of scope may result in a
+ crash or data corruption.
+ """
+ begin_pos_ptr = _bt_iter_pos()
+ end_pos_ptr = _bt_iter_pos()
+ begin_pos_ptr.type = SEEK_BEGIN
+ end_pos_ptr.type = SEEK_LAST
+
+ for event in self._events(begin_pos_ptr, end_pos_ptr):
+ yield event
+
+ def events_timestamps(self, timestamp_begin, timestamp_end):
+ """
+ Generator function to iterate over the events of open in the current
+ TraceCollection from timestamp_begin to timestamp_end.
+ """
+ begin_pos_ptr = _bt_iter_pos()
+ end_pos_ptr = _bt_iter_pos()
+ begin_pos_ptr.type = end_pos_ptr.type = SEEK_TIME
+ begin_pos_ptr.u.seek_time = timestamp_begin
+ end_pos_ptr.u.seek_time = timestamp_end
+
+ for event in self._events(begin_pos_ptr, end_pos_ptr):
+ yield event
+
+ @property
+ def timestamp_begin(self):
+ pos_ptr = _bt_iter_pos()
+ pos_ptr.type = SEEK_BEGIN
+ return self._timestamp_at_pos(pos_ptr)
+
+ @property
+ def timestamp_end(self):
+ pos_ptr = _bt_iter_pos()
+ pos_ptr.type = SEEK_LAST
+ return self._timestamp_at_pos(pos_ptr)
+
+ def _timestamp_at_pos(self, pos_ptr):
+ ctf_it_ptr = _bt_ctf_iter_create(self._tc, pos_ptr, pos_ptr)
+ if ctf_it_ptr is None:
+ raise NotImplementedError("Creation of multiple iterators is unsupported.")
+ ev_ptr = _bt_ctf_iter_read_event(ctf_it_ptr)
+ _bt_ctf_iter_destroy(ctf_it_ptr)
+ if ev_ptr is None:
+ return None;
+
+ def _events(self, begin_pos_ptr, end_pos_ptr):
+ ctf_it_ptr = _bt_ctf_iter_create(self._tc, begin_pos_ptr, end_pos_ptr)
+ if ctf_it_ptr is None:
+ raise NotImplementedError(
+ "Creation of multiple iterators is unsupported.")
+
+ while True:
+ ev_ptr = _bt_ctf_iter_read_event(ctf_it_ptr)
+ if ev_ptr is None:
+ break
+
+ ev = Event.__new__(Event)
+ ev._e = ev_ptr
+ try:
+ yield ev
+ except GeneratorExit:
+ break
+
+ ret = _bt_iter_next(_bt_ctf_get_iter(ctf_it_ptr))
+ if ret != 0:
+ break
+
+ _bt_ctf_iter_destroy(ctf_it_ptr)
+
%}
} u;
};
-
-%pythoncode%{
-
-class IterPos:
- """This class represents the position where to set an iterator."""
-
- __can_access = False
-
- def __init__(self, seek_type, seek_time = None):
- """
- seek_type represents the type of seek to use.
- seek_time is the timestamp to seek to when using SEEK_TIME, it
- is expressed in nanoseconds
- Only use SEEK_RESTORE on IterPos obtained from the get_pos function
- in Iter class.
- """
-
- self._pos = _bt_iter_pos()
- self._pos.type = seek_type
- if seek_time and seek_type == SEEK_TIME:
- self._pos.u.seek_time = seek_time
- self.__can_access = True
-
- def __del__(self):
- if not self.__can_access:
- _bt_iter_free_pos(self._pos)
-
- def _get_type(self):
- if not __can_access:
- raise AttributeError("seek_type is not available")
- return self._pos.type
-
- def _set_type(self, seek_type):
- if not __can_access:
- raise AttributeError("seek_type is not available")
- self._pos.type = seek_type
-
- def _get_time(self):
- if not __can_access:
- raise AttributeError("seek_time is not available")
-
- elif self._pos.type is not SEEK_TIME:
- raise TypeError("seek_type is not SEEK_TIME")
-
- return self._pos.u.seek_time
-
- def _set_time(self, time):
- if not __can_access:
- raise AttributeError("seek_time is not available")
-
- elif self._pos.type is not SEEK_TIME:
- raise TypeError("seek_type is not SEEK_TIME")
-
- self._pos.u.seek_time = time
-
- def _get_pos(self):
- return self._pos
-
-
- seek_type = property(_get_type, _set_type)
- seek_time = property(_get_time, _set_time)
-
-
-class Iterator:
-
- __with_init = False
-
- def __init__(self, context, begin_pos = None, end_pos = None, _no_init = None):
- """
- Allocate a trace collection iterator.
-
- begin_pos and end_pos are optional parameters to specify the
- position at which the trace collection should be seeked upon
- iterator creation, and the position at which iteration will
- start returning "EOF".
-
- By default, if begin_pos is None, a BT_SEEK_CUR is performed at
- creation. By default, if end_pos is None, a BT_SEEK_END (end of
- trace) is the EOF criterion.
- """
- if _no_init is None:
- if begin_pos is None:
- bp = None
- else:
- try:
- bp = begin_pos._pos
- except AttributeError:
- raise TypeError("in __init__, "
- "argument 3 must be a IterPos instance")
-
- if end_pos is None:
- ep = None
- else:
- try:
- ep = end_pos._pos
- except AttributeError:
- raise TypeError("in __init__, "
- "argument 4 must be a IterPos instance")
-
- try:
- self._bi = _bt_iter_create(context._c, bp, ep)
- except AttributeError:
- raise TypeError("in __init__, "
- "argument 2 must be a Context instance")
-
- self.__with_init = True
-
- else:
- self._bi = _no_init
-
- def __del__(self):
- if self.__with_init:
- _bt_iter_destroy(self._bi)
-
- def next(self):
- """
- Move trace collection position to the next event.
- Returns 0 on success, a negative value on error.
- """
- return _bt_iter_next(self._bi)
-
- def get_pos(self):
- """Return a IterPos class of the current iterator position."""
- ret = IterPos(0)
- ret.__can_access = False
- ret._pos = _bt_iter_get_pos(self._bi)
- return ret
-
- def set_pos(self, pos):
- """
- Move the iterator to a given position.
-
- On error, the stream_heap is reinitialized and returned empty.
- Return 0 for success.
- Return EOF if the position requested is after the last event of the
- trace collection.
- Return -EINVAL when called with invalid parameter.
- Return -ENOMEM if the stream_heap could not be properly initialized.
- """
- try:
- return _bt_iter_set_pos(self._bi, pos._pos)
- except AttributeError:
- raise TypeError("in set_pos, "
- "argument 2 must be a IterPos instance")
-
- def create_time_pos(self, timestamp):
- """
- Create a position based on time
- This function allocates and returns a new IterPos to be able to
- restore an iterator position based on a timestamp.
- """
-
- if timestamp < 0:
- raise TypeError("timestamp must be an unsigned int")
-
- ret = IterPos(0)
- ret.__can_access = False
- ret._pos = _bt_iter_create_time_pos(self._bi, timestamp)
- return ret
-%}
-
-
-/* =================================================================
- CLOCK-TYPE.H
- ¯¯¯¯¯¯¯¯¯¯¯¯
- *** Enum copied from clock-type.h
- All changes must also be made here
-*/
-%rename("CLOCK_CYCLES") BT_CLOCK_CYCLES;
-%rename("CLOCK_REAL") BT_CLOCK_REAL;
-
-enum bt_clock_type {
- BT_CLOCK_CYCLES = 0,
- BT_CLOCK_REAL
-};
-
/* =================================================================
TRACE-HANDLE.H, TRACE-HANDLE-INTERNAL.H
¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯¯
%pythoncode%{
+# Based on enum bt_clock_type in clock-type.h
+class ClockType:
+ CLOCK_CYCLES = 0
+ CLOCK_REAL = 1
+
class TraceHandle(object):
"""
The TraceHandle allows the user to manipulate a trace file directly.
def __repr__(self):
return "Babeltrace TraceHandle: trace_id('{0}')".format(self._id)
- def get_id(self):
+ @property
+ def id(self):
"""Return the TraceHandle id."""
return self._id
- def get_path(self, context):
+ @property
+ def path(self):
"""Return the path of a TraceHandle."""
- try:
- return _bt_trace_handle_get_path(context._c, self._id)
- except AttributeError:
- raise TypeError("in get_path, "
- "argument 2 must be a Context instance")
+ return _bt_trace_handle_get_path(self._trace_collection._tc, self._id)
- def get_timestamp_begin(self, context, clock_type):
+ @property
+ def timestamp_begin(self):
"""Return the creation time of the buffers of a trace."""
- try:
- return _bt_trace_handle_get_timestamp_begin(
- context._c, self._id,clock_type)
- except AttributeError:
- raise TypeError("in get_timestamp_begin, "
- "argument 2 must be a Context instance")
+ return _bt_trace_handle_get_timestamp_begin(
+ self._trace_collection._tc, self._id, ClockType.CLOCK_REAL)
- def get_timestamp_end(self, context, clock_type):
+ @property
+ def timestamp_end(self):
"""Return the destruction timestamp of the buffers of a trace."""
- try:
- return _bt_trace_handle_get_timestamp_end(
- context._c, self._id, clock_type)
- except AttributeError:
- raise TypeError("in get_timestamp_end, "
- "argument 2 must be a Context instance")
+ return _bt_trace_handle_get_timestamp_end(
+ self._trace_collection._tc, self._id, ClockType.CLOCK_REAL)
+
+ @property
+ def events(self):
+ """
+ Generator returning all events (EventDeclaration) in a trace.
+ """
+ ret = _bt_python_event_decl_listcaller(self.id, self._trace_collection._tc)
+
+ if not isinstance(ret, list):
+ return
+
+ ptr_list, count = ret
+ for i in range(count):
+ tmp = EventDeclaration.__new__(EventDeclaration)
+ tmp._ed = _bt_python_decl_one_from_list(ptr_list, i)
+ yield tmp
%}
//Events
-
%rename("_bt_ctf_get_top_level_scope") bt_ctf_get_top_level_scope(const struct
bt_ctf_event *event, enum bt_ctf_scope scope);
%rename("_bt_ctf_event_name") bt_ctf_event_name(const struct bt_ctf_event *ctf_event);
const struct bt_ctf_field_decl *field);
%rename("_bt_ctf_get_decl_from_def") bt_ctf_get_decl_from_def(
const struct bt_definition *field);
+%rename("_bt_ctf_get_decl_from_field_decl") bt_ctf_get_decl_from_field_decl(
+ const struct bt_ctf_field_decl *field);
%rename("_bt_array_index") bt_array_index(struct definition_array *array, uint64_t i);
%rename("_bt_sequence_len") bt_sequence_len(struct definition_sequence *sequence);
%rename("_bt_sequence_index") bt_sequence_index(struct definition_sequence *sequence, uint64_t i);
const char *bt_ctf_get_decl_event_name(const struct bt_ctf_event_decl *event);
const char *bt_ctf_get_decl_field_name(const struct bt_ctf_field_decl *field);
const struct bt_declaration *bt_ctf_get_decl_from_def(const struct bt_definition *field);
+const struct bt_declaration *bt_ctf_get_decl_from_field_decl(const struct bt_ctf_field_decl *field);
uint64_t bt_sequence_len(struct definition_sequence *sequence);
struct bt_definition *bt_sequence_index(struct definition_sequence *sequence, uint64_t i);
uint64_t bt_ctf_get_struct_field_count(const struct bt_definition *structure);
ASCII = 2
UNKNOWN = 3
+# Based on the enum in ctf-writer/writer.h
+class ByteOrder:
+ BYTE_ORDER_NATIVE = 0
+ BYTE_ORDER_LITTLE_ENDIAN = 1
+ BYTE_ORDER_BIG_ENDIAN = 2
+ BYTE_ORDER_NETWORK = 3
+ BYTE_ORDER_UNKNOWN = 4 # Python-specific entry
+
#enum equivalent, accessible constants
#These are taken directly from ctf/events.h
#All changes to enums must also be made here
SEQUENCE = 9
NR_CTF_TYPES = 10
- def get_type_name(id):
- name = "UNKNOWN"
+ def type_name(id):
+ name = "UNKNOWN_TYPE"
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:
break
return name
-class CTFReader:
- class scope:
- TRACE_PACKET_HEADER = 0
- STREAM_PACKET_CONTEXT = 1
- STREAM_EVENT_HEADER = 2
- STREAM_EVENT_CONTEXT = 3
- EVENT_CONTEXT = 4
- EVENT_FIELDS = 5
+class CTFScope:
+ TRACE_PACKET_HEADER = 0
+ STREAM_PACKET_CONTEXT = 1
+ STREAM_EVENT_HEADER = 2
+ STREAM_EVENT_CONTEXT = 3
+ EVENT_CONTEXT = 4
+ EVENT_FIELDS = 5
+
+ def scope_name(scope):
+ name = "UNKNOWN_SCOPE"
+ constants = [attr for attr in dir(CTFScope) if not callable(getattr(CTFScope, attr)) and not attr.startswith("__")]
+ for attr in constants:
+ if getattr(CTFScope, attr) == scope:
+ name = attr
+ break
+ return name
- class Iterator(Iterator, object):
- """
- Allocate a CTF trace collection iterator.
+# Priority of the scopes when searching for event fields
+_scopes = [CTFScope.EVENT_FIELDS, CTFScope.EVENT_CONTEXT, CTFScope.STREAM_EVENT_CONTEXT,
+ CTFScope.STREAM_EVENT_HEADER, CTFScope.STREAM_PACKET_CONTEXT, CTFScope.TRACE_PACKET_HEADER]
- begin_pos and end_pos are optional parameters to specify the
- position at which the trace collection should be seeked upon
- iterator creation, and the position at which iteration will
- start returning "EOF".
+import collections
+class Event(collections.Mapping):
+ """
+ This class represents an event from the trace.
+ It is obtained using the TraceCollection generator functions.
+ Do not instantiate.
+ """
+ def __init__(self):
+ raise NotImplementedError("Event cannot be instantiated")
- By default, if begin_pos is None, a SEEK_CUR is performed at
- creation. By default, if end_pos is None, a SEEK_END (end of
- trace) is the EOF criterion.
+ @property
+ def name(self):
+ """Return the name of the event or None on error."""
+ return _bt_ctf_event_name(self._e)
- Only one iterator can be created against a context. If more than one
- iterator is being created for the same context, the second creation
- will return None. The previous iterator must be destroyed before
- creation of the new iterator for this function to succeed.
+ @property
+ def cycles(self):
"""
+ Return the timestamp of the event as written in
+ the packet (in cycles) or -1ULL on error.
+ """
+ return _bt_ctf_get_cycles(self._e)
- def __new__(cls, context, begin_pos = None, end_pos = None):
- # __new__ is used to control the return value
- # as the CTFReader.Iterator class should return None
- # if bt_ctf_iter_create returns NULL
-
- if begin_pos is None:
- bp = None
- else:
- bp = begin_pos._pos
- if end_pos is None:
- ep = None
- else:
- ep = end_pos._pos
- try:
- it = _bt_ctf_iter_create(context._c, bp, ep)
- except AttributeError:
- raise TypeError("in __init__, "
- "argument 2 must be a Context instance")
- if it is None:
- return None
-
- ret_class = super(CTFReader.Iterator, cls).__new__(cls)
- ret_class._i = it
- return ret_class
+ @property
+ def timestamp(self):
+ """
+ Return the timestamp of the event offset with the
+ system clock source or -1ULL on error.
+ """
+ return _bt_ctf_get_timestamp(self._e)
- def __init__(self, context, begin_pos = None, end_pos = None):
- Iterator.__init__(self, None, None, None,
- _bt_ctf_get_iter(self._i))
+ def field_with_scope(self, field_name, scope):
+ """
+ Get field_name's value in scope.
+ None is returned if no field matches field_name.
+ """
+ if not scope in _scopes:
+ raise ValueError("Invalid scope provided")
+ field = self._field_with_scope(field_name, scope)
+ if field is not None:
+ return field.value
+ return None
- def __del__(self):
- _bt_ctf_iter_destroy(self._i)
+ def field_list_with_scope(self, scope):
+ """Return a list of field names in scope."""
+ if not scope in _scopes:
+ raise ValueError("Invalid scope provided")
+ field_names = []
+ for field in self._field_list_with_scope(scope):
+ field_names.append(field.name)
+ return field_names
- def read_event(self):
- """
- Read the iterator's current event data.
- Return current event on success, None on end of trace.
- """
- ret = _bt_ctf_iter_read_event(self._i)
- if ret is None:
- return ret
- ev = CTFReader.Event.__new__(CTFReader.Event)
- ev._e = ret
- return ev
+ @property
+ def handle(self):
+ """
+ Get the TraceHandle associated with this event
+ Return None on error
+ """
+ ret = _bt_ctf_event_get_handle_id(self._e)
+ if ret < 0:
+ return None
+ th = TraceHandle.__new__(TraceHandle)
+ th._id = ret
+ th._trace_collection = self.get_trace_collection()
+ return th
- class Event(object):
+ @property
+ def trace_collection(self):
"""
- This class represents an event from the trace.
- It is obtained with read_event() from CTFReader.Iterator.
- Do not instantiate.
+ Get the TraceCollection associated with this event.
+ Return None on error.
"""
+ trace_collection = TraceCollection()
+ trace_collection._tc = _bt_ctf_event_get_context(self._e);
+ if trace_collection._tc is None:
+ return None
+ else:
+ return trace_collection
+
+ def __getitem__(self, field_name):
+ """
+ Get field_name's value. If the field_name exists in multiple
+ scopes, the first field found is returned. The scopes are searched
+ in the following order:
+ 1) EVENT_FIELDS
+ 2) EVENT_CONTEXT
+ 3) STREAM_EVENT_CONTEXT
+ 4) STREAM_EVENT_HEADER
+ 5) STREAM_PACKET_CONTEXT
+ 6) TRACE_PACKET_HEADER
+ None is returned if no field matches field_name.
+
+ Use field_with_scope() to explicitly access fields in a given
+ scope.
+ """
+ field = self._field(field_name)
+ if field is not None:
+ return field.value
+ raise KeyError(field_name)
+
+ def __iter__(self):
+ for key in self.keys():
+ yield key
+
+ def __len__(self):
+ count = 0
+ for scope in _scopes:
+ scope_ptr = _bt_ctf_get_top_level_scope(self._e, scope)
+ ret = _bt_python_field_listcaller(self._e, scope_ptr)
+ if isinstance(ret, list):
+ count += ret[1]
+ return count
+
+ def __contains__(self, field_name):
+ return self._field(field_name) is not None
+
+ def keys(self):
+ """Return a list of field names."""
+ field_names = set()
+ for scope in _scopes:
+ for name in self.field_list_with_scope(scope):
+ field_names.add(name)
+ return list(field_names)
+
+ def get(self, field_name, default = None):
+ field = self._field(field_name)
+ if field is None:
+ return default
+ return field.value
+
+ def items(self):
+ for field in self.keys():
+ yield (field, self[field])
+
+ def _field_with_scope(self, field_name, scope):
+ scope_ptr = _bt_ctf_get_top_level_scope(self._e, scope)
+ if scope_ptr is None:
+ return None
- def __init__(self):
- raise NotImplementedError("CTFReader.Event cannot be instantiated")
-
- def get_top_level_scope(self, scope):
- """
- Return a definition of the top-level 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.
- """
- evDef = CTFReader.Definition.__new__(CTFReader.Definition)
- evDef._d = _bt_ctf_get_top_level_scope(self._e, scope)
- if evDef._d is None:
- return None
- return evDef
-
- def get_name(self):
- """Return the name of the event or None on error."""
- return _bt_ctf_event_name(self._e)
-
- def get_cycles(self):
- """
- Return the timestamp of the event as written in
- the packet (in cycles) or -1ULL on error.
- """
- return _bt_ctf_get_cycles(self._e)
-
- def get_timestamp(self):
- """
- Return the timestamp of the event offsetted with the
- system clock source or -1ULL on error.
- """
- return _bt_ctf_get_timestamp(self._e)
-
- def get_field_with_scope(self, scope, field):
- """
- Return the definition of a specific field.
- Return None on error.
- """
- evDef = CTFReader.Definition.__new__(CTFReader.Definition)
- try:
- evDef._d = _bt_ctf_get_field(self._e, scope._d, field)
- except AttributeError:
- raise TypeError("in get_field, argument 2 must be a "
- "Definition (scope) instance")
- if evDef._d is None:
- return None
- evDef._s = scope
- return evDef
-
- def get_field(self, field):
- """
- Return the definition of fields by a name
- Return None on error
- """
- 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:
- evDef = self.get_field_with_scope(eventScope, field)
- if evDef is not None:
- fields_by_name.append(evDef)
-
- if streamScope is not None:
- evDef = self.get_field_with_scope(streamScope, field)
- if evDef is not None:
- fields_by_name.append(evDef);
-
- if not fields_by_name:
- return None
- return fields_by_name
-
- def get_field_list_with_scope(self, scope):
- """
- Return a list of Definitions associated with the scope
- Return None on error.
- """
- try:
- field_lc, count = _bt_python_field_listcaller(self._e, scope._d)
- except AttributeError:
- raise TypeError("in get_field_list, argument 2 must be a "
- "Definition (scope) instance")
-
- if field_lc is None:
- return None
-
- def_list = []
- for i in range(count):
- tmp = CTFReader.Definition.__new__(CTFReader.Definition)
- tmp._d = _bt_python_field_one_from_list(field_lc, i)
- tmp._s = scope
- def_list.append(tmp)
-
- return def_list
-
- def get_field_list(self):
- """Return a list of Definitions or None on error."""
- 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:
- event_field_list = self.get_field_list_with_scope(eventScope)
- if event_field_list is not None:
- def_list = event_field_list
-
- if streamScope is not None:
- event_field_list = self.get_field_list_with_scope(streamScope)
- if event_field_list is not None:
- def_list.extend(event_field_list)
-
- if not def_list:
- return None
- return def_list
-
- def get_index(self, field, index):
- """
- If the field is an array or a sequence, return the element
- at position index, otherwise return None
- """
- evDef = CTFReader.Definition.__new__(CTFReader.Definition)
- try:
- evDef._d = _bt_ctf_get_index(self._e, field._d, index)
- except AttributeError:
- raise TypeError("in get_index, argument 2 must be a "
- "Definition (field) instance")
-
- if evDef._d is None:
- return None
- return evDef
-
- def get_handle(self):
- """
- Get the TraceHandle associated with an event
- Return None on error
- """
- ret = _bt_ctf_event_get_handle_id(self._e)
- if ret < 0:
- return None
-
- th = TraceHandle.__new__(TraceHandle)
- th._id = ret
- return th
-
- def get_context(self):
- """
- Get the context associated with an event.
- Return None on error.
- """
- ctx = Context()
- ctx._c = _bt_ctf_event_get_context(self._e);
- if ctx._c is None:
- return None
- else:
- return ctx
+ definition_ptr = _bt_ctf_get_field(self._e, scope_ptr, field_name)
+ if definition_ptr is None:
+ return None
- class FieldError(Exception):
- def __init__(self, value):
- self.value = value
+ field = _Definition(definition_ptr, scope)
+ return field
- def __str__(self):
- return repr(self.value)
+ def _field(self, field_name):
+ field = None
+ for scope in _scopes:
+ field = self._field_with_scope(field_name, scope)
+ if field is not None:
+ break
+ return field
+
+ def _field_list_with_scope(self, scope):
+ fields = []
+ scope_ptr = _bt_ctf_get_top_level_scope(self._e, scope)
+
+ # Returns a list [list_ptr, count]. If list_ptr is NULL, SWIG will only
+ # provide the "count" return value
+ count = 0
+ list_ptr = None
+ ret = _bt_python_field_listcaller(self._e, scope_ptr)
+ if isinstance(ret, list):
+ list_ptr, count = ret
- class Definition(object):
- """Definition class. Do not instantiate."""
+ for i in range(count):
+ definition_ptr = _bt_python_field_one_from_list(list_ptr, i)
+ if definition_ptr is not None:
+ definition = _Definition(definition_ptr, scope)
+ fields.append(definition)
+ return fields
- def __init__(self):
- raise NotImplementedError("CTFReader.Definition cannot be instantiated")
-
- def __repr__(self):
- return "Babeltrace Definition: name('{0}'), type({1})".format(
- self.field_name(), self.field_type())
-
- def field_name(self):
- """Return the name of a field or None on error."""
- return _bt_ctf_field_name(self._d)
-
- def field_type(self):
- """Return the type of a field or -1 if unknown."""
- return _bt_ctf_field_type(_bt_ctf_get_decl_from_def(self._d))
-
- def get_int_signedness(self):
- """
- Return the signedness of an integer:
- 0 if unsigned; 1 if signed; -1 on error.
- """
- return _bt_ctf_get_int_signedness(_bt_ctf_get_decl_from_def(self._d))
-
- def get_int_base(self):
- """Return the base of an int or a negative value on error."""
- return _bt_ctf_get_int_base(_bt_ctf_get_decl_from_def(self._d))
-
- def get_int_byte_order(self):
- """
- Return the byte order of an int or a negative
- value on error.
- """
- return _bt_ctf_get_int_byte_order(_bt_ctf_get_decl_from_def(self._d))
-
- def get_int_len(self):
- """
- Return the size, in bits, of an int or a negative
- value on error.
- """
- return _bt_ctf_get_int_len(_bt_ctf_get_decl_from_def(self._d))
-
- def get_enum_str(self):
- """
- Return the string matching the current enumeration.
- Return None on error.
- """
- return _bt_ctf_get_enum_str(self._d)
-
- def get_encoding(self):
- """
- Return the encoding of an int or a string.
- Return a negative value on error.
- """
- return _bt_ctf_get_encoding(_bt_ctf_get_decl_from_def(self._d))
-
- def get_array_len(self):
- """
- Return the len of an array or a negative
- value on error.
- """
- return _bt_ctf_get_array_len(_bt_ctf_get_decl_from_def(self._d))
-
- def get_array_element_at(self, index):
- """
- Return the array's element at position index.
- Return None on error
- """
- array = _bt_python_get_array_from_def(self._d)
- if array is None:
- return None
-
- element = CTFReader.Definition.__new__(CTFReader.Definition)
- element._d = _bt_array_index(array, index)
- if element._d is None:
- return None
- return element
-
- def get_sequence_len(self):
- """
- Return the len of a sequence or a negative
- value on error.
- """
- seq = _bt_python_get_sequence_from_def(self._d)
- return _bt_sequence_len(seq)
-
- def get_sequence_element_at(self, index):
- """
- Return the sequence's element at position index,
- otherwise return None
- """
- seq = _bt_python_get_sequence_from_def(self._d)
- if seq is not None:
- element = CTFReader.Definition.__new__(CTFReader.Definition)
- element._d = _bt_sequence_index(seq, index)
- if element._d is not None:
- return element
- return None
+class FieldError(Exception):
+ def __init__(self, value):
+ self.value = value
- def get_uint64(self):
- """
- 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 CTFReader.field_error() function after accessing a field.
- """
- return _bt_ctf_get_uint64(self._d)
-
- def get_int64(self):
- """
- 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 CTFReader.field_error() function after accessing a field.
- """
- return _bt_ctf_get_int64(self._d)
-
- def get_char_array(self):
- """
- 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 CTFReader.field_error() function after accessing a field.
- """
- return _bt_ctf_get_char_array(self._d)
-
- def get_str(self):
- """
- 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 CTFReader.field_error() function after accessing a field.
- """
- return _bt_ctf_get_string(self._d)
-
- def get_float(self):
- """
- 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 CTFReader.field_error() function after accessing a field.
- """
- return _bt_ctf_get_float(self._d)
-
- def get_variant(self):
- """
- 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 CTFReader.field_error() function after accessing a field.
- """
- return _bt_ctf_get_variant(self._d)
-
- def get_struct_field_count(self):
- """
- Return the number of fields contained in the structure.
- If the field does not exist or is not of the type requested,
- the value returned is undefined.
- """
- return _bt_ctf_get_struct_field_count(self._d)
-
- def get_struct_field_at(self, i):
- """
- 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 CTFReader.field_error() function after accessing a field.
- """
- return _bt_ctf_get_struct_field_index(self._d, i)
-
- def get_value(self):
- """
- Return the value associated with the field according to its type.
- Return None on error.
- """
- id = self.field_type()
- value = None
- if id == CTFTypeId.STRING:
- value = self.get_str()
- 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())
- elif id == CTFTypeId.INTEGER:
- if self.get_int_signedness() == 0:
- value = self.get_uint64()
- else:
- value = self.get_int64()
- elif id == CTFTypeId.ENUM:
- value = self.get_enum_str()
- 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())
- elif id == CTFTypeId.FLOAT:
- value = self.get_float()
- elif id == CTFTypeId.VARIANT:
- variant = CTFReader.Definition.__new__(CTFReader.Definition)
- variant._d = self.get_variant();
- value = variant.get_value()
- elif id == CTFTypeId.STRUCT:
- value = {}
- for i in range(self.get_struct_field_count()):
- member = CTFReader.Definition.__new__(CTFReader.Definition)
- member._d = self.get_struct_field_at(i);
- value[member.field_name()] = member.get_value()
-
- 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 the scope of a field or None on error."""
- return self._s
-
- class EventDecl(object):
- """Event declaration class. Do not instantiate."""
+ def __str__(self):
+ return repr(self.value)
- def __init__(self):
- raise NotImplementedError("CTFReader.EventDecl cannot be instantiated")
+class EventDeclaration(object):
+ """Event declaration class. Do not instantiate."""
- def __repr__(self):
- return "Babeltrace EventDecl: name {0}".format(self.get_name())
+ def __init__(self):
+ raise NotImplementedError("EventDeclaration cannot be instantiated")
- def get_name(self):
- """Return the name of the event or None on error"""
- return _bt_ctf_get_decl_event_name(self._d)
+ @property
+ def name(self):
+ """Return the name of the event or None on error"""
+ return _bt_ctf_get_decl_event_name(self._ed)
- def get_decl_fields(self, scope):
- """
- Return a list of CTFReader.FieldDecl
- Return None on error.
- """
- ptr_list = _by_python_field_decl_listcaller(self._d, scope)
+ @property
+ def fields(self):
+ """
+ Generator returning all FieldDeclarations of an event, going through
+ each scope in the following order:
+ 1) EVENT_FIELDS
+ 2) EVENT_CONTEXT
+ 3) STREAM_EVENT_CONTEXT
+ 4) STREAM_EVENT_HEADER
+ 5) STREAM_PACKET_CONTEXT
+ 6) TRACE_PACKET_HEADER
+ """
+ for scope in _scopes:
+ for declaration in self.fields_scope(scope):
+ yield declaration
- if ptr_list is None:
- return None
+ def fields_scope(self, scope):
+ """
+ Generator returning FieldDeclarations of the current event in scope.
+ """
+ ret = _by_python_field_decl_listcaller(self._ed, scope)
- decl_list = []
- i = 0
- while True:
- tmp = CTFReader.FieldDecl.__new__(CTFReader.FieldDecl)
- tmp._d = _bt_python_field_decl_one_from_list(
- ptr_list, i)
+ if not isinstance(ret, list):
+ return
- if tmp._d is None:
- #Last item of list is None
- break
+ list_ptr, count = ret
+ for i in range(count):
+ field_declaration_ptr = _bt_python_field_decl_one_from_list(list_ptr, i)
+ if field_declaration_ptr is not None:
+ declaration_ptr = _bt_ctf_get_decl_from_field_decl(field_declaration_ptr)
+ field_declaration = _create_field_declaration(declaration_ptr, _bt_ctf_get_decl_field_name(field_declaration_ptr), scope)
+ yield field_declaration
+
+class FieldDeclaration(object):
+ """Field declaration class. Do not instantiate."""
+ def __init__(self):
+ raise NotImplementedError("FieldDeclaration cannot be instantiated")
- decl_list.append(tmp)
- i += 1
- return decl_list
+ def __repr__(self):
+ return "({0}) {1} {2}".format(CTFScope.scope_name(self.scope), CTFTypeId.type_name(self.type), self.name)
+ @property
+ def name(self):
+ """Return the name of a FieldDeclaration or None on error."""
+ return self._name
- class FieldDecl(object):
- """Field declaration class. Do not instantiate."""
+ @property
+ def type(self):
+ """
+ Return the FieldDeclaration's type. One of the entries in class
+ CTFTypeId.
+ """
+ return _bt_ctf_field_type(self._fd)
- def __init__(self):
- raise NotImplementedError("CTFReader.FieldDecl cannot be instantiated")
+ @property
+ def scope(self):
+ """
+ Return the FieldDeclaration's scope.
+ """
+ return self._s
- def __repr__(self):
- return "Babeltrace FieldDecl: name {0}".format(self.get_name())
+class IntegerFieldDeclaration(FieldDeclaration):
+ """Do not instantiate."""
+ def __init__(self):
+ raise NotImplementedError("IntegerFieldDeclaration cannot be instantiated")
- def get_name(self):
- """Return the name of a FieldDecl or None on error"""
- return _bt_ctf_get_decl_field_name(self._d)
+ @property
+ def signedness(self):
+ """
+ Return the signedness of an integer:
+ 0 if unsigned; 1 if signed; -1 on error.
+ """
+ return _bt_ctf_get_int_signedness(self._fd)
+ @property
+ def base(self):
+ """Return the base of an int or a negative value on error."""
+ return _bt_ctf_get_int_base(self._fd)
- @staticmethod
- def field_error():
+ @property
+ def byte_order(self):
"""
- Return the last error code encountered while
- accessing a field and reset the error flag.
- Return 0 if no error, a negative value otherwise.
+ Return the byte order. One of class ByteOrder's entries.
"""
- return _bt_ctf_field_get_error()
+ ret = _bt_ctf_get_int_byte_order(self._fd)
+ if ret == 1234:
+ return ByteOrder.BYTE_ORDER_LITTLE_ENDIAN
+ elif ret == 4321:
+ return ByteOrder.BYTE_ORDER_BIG_ENDIAN
+ else:
+ return ByteOrder.BYTE_ORDER_UNKNOWN
- @staticmethod
- def get_event_decl_list(trace_handle, context):
+ @property
+ def length(self):
"""
- Return a list of CTFReader.EventDecl
- Return None on error.
+ Return the size, in bits, of an int or a negative
+ value on error.
"""
- try:
- handle_id = trace_handle._id
- except AttributeError:
- raise TypeError("in get_event_decl_list, "
- "argument 1 must be a TraceHandle instance")
- try:
- ptr_list, count = _bt_python_event_decl_listcaller(handle_id, context._c)
- except AttributeError:
- raise TypeError("in get_event_decl_list, "
- "argument 2 must be a Context instance")
+ return _bt_ctf_get_int_len(self._fd)
- if ptr_list is None:
- return None
+ @property
+ def encoding(self):
+ """
+ Return the encoding. One of class CTFStringEncoding's entries.
+ Return a negative value on error.
+ """
+ return _bt_ctf_get_encoding(self._fd)
- decl_list = []
- for i in range(count):
- tmp = CTFReader.EventDecl.__new__(CTFReader.EventDecl)
- tmp._d = _bt_python_decl_one_from_list(ptr_list, i)
- decl_list.append(tmp)
+class EnumerationFieldDeclaration(FieldDeclaration):
+ """Do not instantiate."""
+ def __init__(self):
+ raise NotImplementedError("EnumerationFieldDeclaration cannot be instantiated")
- return decl_list
+class ArrayFieldDeclaration(FieldDeclaration):
+ """Do not instantiate."""
+ def __init__(self):
+ raise NotImplementedError("ArrayFieldDeclaration cannot be instantiated")
-%}
+ @property
+ def length(self):
+ """
+ Return the length of an array or a negative
+ value on error.
+ """
+ return _bt_ctf_get_array_len(self._fd)
+ @property
+ def element_declaration(self):
+ """
+ Return element declaration.
+ """
+ field_decl_ptr = _bt_python_get_array_element_declaration(self._fd)
+ return _create_field_declaration(field_decl_ptr, "", self.scope)
+class SequenceFieldDeclaration(FieldDeclaration):
+ """Do not instantiate."""
+ def __init__(self):
+ raise NotImplementedError("SequenceFieldDeclaration cannot be instantiated")
-// =================================================================
-// NEW FUNCTIONS
-// File and list-related
-// python-complements.h
-// =================================================================
+ @property
+ def element_declaration(self):
+ """
+ Return element declaration.
+ """
+ field_decl_ptr = _bt_python_get_sequence_element_declaration(self._fd)
+ return _create_field_declaration(field_decl_ptr, "", self.scope)
-%pythoncode %{
+class FloatFieldDeclaration(FieldDeclaration):
+ """Do not instantiate."""
+ def __init__(self):
+ raise NotImplementedError("FloatFieldDeclaration cannot be instantiated")
-class File(object):
- """
- Open a file for babeltrace.
+class StructureFieldDeclaration(FieldDeclaration):
+ """Do not instantiate."""
+ def __init__(self):
+ raise NotImplementedError("StructureFieldDeclaration cannot be instantiated")
- file_path is a string containing the path or None to use the
- standard output in writing mode.
+class StringFieldDeclaration(FieldDeclaration):
+ """Do not instantiate."""
+ def __init__(self):
+ raise NotImplementedError("StringFieldDeclaration cannot be instantiated")
- The mode can be 'r', 'w' or 'a' for reading (default), writing or
- appending. The file will be created if it doesn't exist when
- opened for writing or appending; it will be truncated when opened
- for writing. Add a 'b' to the mode for binary files. Add a '+'
- to the mode to allow simultaneous reading and writing.
+class VariantFieldDeclaration(FieldDeclaration):
+ """Do not instantiate."""
+ def __init__(self):
+ raise NotImplementedError("VariantFieldDeclaration cannot be instantiated")
+
+def field_error():
+ """
+ Return the last error code encountered while
+ accessing a field and reset the error flag.
+ Return 0 if no error, a negative value otherwise.
"""
+ return _bt_ctf_field_get_error()
- def __new__(cls, file_path, mode='r'):
- # __new__ is used to control the return value
- # as the File class should return None
- # if _bt_file_open returns NULL
+def _create_field_declaration(declaration_ptr, name, scope):
+ """
+ Private field declaration factory.
+ """
+ if declaration_ptr is None:
+ raise ValueError("declaration_ptr must be valid")
+ if not scope in _scopes:
+ raise ValueError("Invalid scope provided")
+
+ type = _bt_ctf_field_type(declaration_ptr)
+ declaration = None
+ if type == CTFTypeId.INTEGER:
+ declaration = IntegerFieldDeclaration.__new__(IntegerFieldDeclaration)
+ elif type == CTFTypeId.ENUM:
+ declaration = EnumerationFieldDeclaration.__new__(EnumerationFieldDeclaration)
+ elif type == CTFTypeId.ARRAY:
+ declaration = ArrayFieldDeclaration.__new__(ArrayFieldDeclaration)
+ elif type == CTFTypeId.SEQUENCE:
+ declaration = SequenceFieldDeclaration.__new__(SequenceFieldDeclaration)
+ elif type == CTFTypeId.FLOAT:
+ declaration = FloatFieldDeclaration.__new__(FloatFieldDeclaration)
+ elif type == CTFTypeId.STRUCT:
+ declaration = StructureFieldDeclaration.__new__(StructureFieldDeclaration)
+ elif type == CTFTypeId.STRING:
+ declaration = StringFieldDeclaration.__new__(StringFieldDeclaration)
+ elif type == CTFTypeId.VARIANT:
+ declaration = VariantFieldDeclaration.__new__(VariantFieldDeclaration)
+ else:
+ return declaration
+
+ declaration._fd = declaration_ptr
+ declaration._s = scope
+ declaration._name = name
+ return declaration
+
+class _Definition(object):
+ def __init__(self, definition_ptr, scope):
+ self._d = definition_ptr
+ self._s = scope
+ if not scope in _scopes:
+ ValueError("Invalid scope provided")
+
+ @property
+ def name(self):
+ """Return the name of a field or None on error."""
+ return _bt_ctf_field_name(self._d)
+
+ @property
+ def type(self):
+ """Return the type of a field or -1 if unknown."""
+ return _bt_ctf_field_type(_bt_ctf_get_decl_from_def(self._d))
+
+ @property
+ def declaration(self):
+ """Return the associated Definition object."""
+ return _create_field_declaration(_bt_ctf_get_decl_from_def(self._d), self.name, self.scope)
+
+ def _get_enum_str(self):
+ """
+ Return the string matching the current enumeration.
+ Return None on error.
+ """
+ return _bt_ctf_get_enum_str(self._d)
- # Type check
- if file_path is not None and type(file_path) is not str:
- raise TypeError("in method __init__, argument 2 of type 'str'")
- if type(mode) is not str:
- raise TypeError("in method __init__, argument 3 of type 'str'")
+ def _get_array_element_at(self, index):
+ """
+ Return the array's element at position index.
+ Return None on error
+ """
+ array_ptr = _bt_python_get_array_from_def(self._d)
+ if array_ptr is None:
+ return None
- # Opening file
- file_ptr = _bt_file_open(file_path, mode)
- if file_ptr is None:
+ definition_ptr = _bt_array_index(array_ptr, index)
+ if definition_ptr is None:
return None
+ return _Definition(definition_ptr, self.scope)
- # Class instantiation
- file_inst = super(File, cls).__new__(cls)
- file_inst._file = file_ptr
- return file_inst
+ def _get_sequence_len(self):
+ """
+ Return the len of a sequence or a negative
+ value on error.
+ """
+ seq = _bt_python_get_sequence_from_def(self._d)
+ return _bt_sequence_len(seq)
- def __init__(self, file_path, mode='r'):
- self._opened = True
- self._use_stdout = False
+ def _get_sequence_element_at(self, index):
+ """
+ Return the sequence's element at position index,
+ otherwise return None
+ """
+ seq = _bt_python_get_sequence_from_def(self._d)
+ if seq is not None:
+ definition_ptr = _bt_sequence_index(seq, index)
+ if definition_ptr is not None:
+ return _Definition(definition_ptr, self.scope)
+ return None
- if file_path is None:
- # use stdout
- file_path = "stdout"
- mode = 'w'
- self._use_stdout = True
+ def _get_uint64(self):
+ """
+ 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 field_error() function after accessing a field.
+ """
+ return _bt_ctf_get_uint64(self._d)
- self._file_path = file_path
- self._mode = mode
+ def _get_int64(self):
+ """
+ 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 field_error() function after accessing a field.
+ """
+ return _bt_ctf_get_int64(self._d)
- def __del__(self):
- self.close()
+ def _get_char_array(self):
+ """
+ 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 occurred,
+ use the field_error() function after accessing a field.
+ """
+ return _bt_ctf_get_char_array(self._d)
+
+ def _get_str(self):
+ """
+ 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 occurred,
+ use the field_error() function after accessing a field.
+ """
+ return _bt_ctf_get_string(self._d)
+
+ def _get_float(self):
+ """
+ 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 occurred,
+ use the field_error() function after accessing a field.
+ """
+ return _bt_ctf_get_float(self._d)
+
+ def _get_variant(self):
+ """
+ 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 occurred,
+ use the field_error() function after accessing a field.
+ """
+ return _bt_ctf_get_variant(self._d)
+
+ def _get_struct_field_count(self):
+ """
+ Return the number of fields contained in the structure.
+ If the field does not exist or is not of the type requested,
+ the value returned is undefined.
+ """
+ return _bt_ctf_get_struct_field_count(self._d)
+
+ def _get_struct_field_at(self, i):
+ """
+ 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 occurred,
+ use the field_error() function after accessing a field.
+ """
+ return _bt_ctf_get_struct_field_index(self._d, i)
+
+ @property
+ def value(self):
+ """
+ Return the value associated with the field according to its type.
+ Return None on error.
+ """
+ id = self.type
+ value = None
+
+ if id == CTFTypeId.STRING:
+ value = self._get_str()
+ elif id == CTFTypeId.ARRAY:
+ element_decl = self.declaration.element_declaration
+ if ((element_decl.type == CTFTypeId.INTEGER
+ and element_decl.length == 8)
+ and (element_decl.encoding == CTFStringEncoding.ASCII or element_decl.encoding == CTFStringEncoding.UTF8)):
+ value = _bt_python_get_array_string(self._d)
+ else:
+ value = []
+ for i in range(self.declaration.length):
+ element = self._get_array_element_at(i)
+ value.append(element.value)
+ elif id == CTFTypeId.INTEGER:
+ if self.declaration.signedness == 0:
+ value = self._get_uint64()
+ else:
+ value = self._get_int64()
+ elif id == CTFTypeId.ENUM:
+ value = self._get_enum_str()
+ elif id == CTFTypeId.SEQUENCE:
+ element_decl = self.declaration.element_declaration
+ if ((element_decl.type == CTFTypeId.INTEGER
+ and element_decl.length == 8)
+ and (element_decl.encoding == CTFStringEncoding.ASCII or element_decl.encoding == CTFStringEncoding.UTF8)):
+ value = _bt_python_get_sequence_string(self._d)
+ else:
+ seq_len = self._get_sequence_len()
+ value = []
+ for i in range(seq_len):
+ evDef = self._get_sequence_element_at(i)
+ value.append(evDef.value)
+ elif id == CTFTypeId.FLOAT:
+ value = self._get_float()
+ elif id == CTFTypeId.VARIANT:
+ variant = _Definition.__new__(_Definition)
+ variant._d = self._get_variant();
+ value = variant.value
+ elif id == CTFTypeId.STRUCT:
+ value = {}
+ for i in range(self._get_struct_field_count()):
+ member = _Definition(self._get_struct_field_at(i), self.scope)
+ value[member.name] = member.value
+
+ if field_error():
+ raise FieldError("Error occurred while accessing field {} of type {}".format(self.name, CTFTypeId.type_name(id)))
+ return value
+
+ @property
+ def scope(self):
+ """Return the scope of a field or None on error."""
+ return self._s
- def __repr__(self):
- if self._opened:
- stat = 'opened'
- else:
- stat = 'closed'
- return "{0} babeltrace File; file_path('{1}'), mode('{2}')".format(
- stat, self._file_path, self._mode)
-
- def close(self):
- """Close the file. Is also called using del."""
- if self._opened and not self._use_stdout:
- _bt_file_close(self._file)
- self._opened = False
%}
+
// =================================================================
// CTF Writer
// =================================================================
%pythoncode %{
class CTFWriter:
- class ByteOrder:
- BYTE_ORDER_NATIVE = 0
- BYTE_ORDER_LITTLE_ENDIAN = 1
- BYTE_ORDER_BIG_ENDIAN = 2
- BYTE_ORDER_NETWORK = 3
class Clock:
def __init__(self, name):
def __del__(self):
_bt_ctf_clock_put(self._c)
+ """
+ Get the clock's description.
+ """
+ @property
+ def description(self):
+ raise NotImplementedError("Getter not implemented.")
+
"""
Set the clock's description. The description appears in the clock's TSDL
meta-data.
"""
- def set_description(self, desc):
- ret = _bt_ctf_clock_set_description(self._c, desc)
+ @description.setter
+ def description(self, desc):
+ ret = _bt_ctf_clock_set_description(self._c, str(desc))
if ret < 0:
raise ValueError("Invalid clock description.")
+ """
+ Get the clock's frequency (Hz).
+ """
+ @property
+ def frequency(self):
+ raise NotImplementedError("Getter not implemented.")
+
"""
Set the clock's frequency (Hz).
"""
- def set_frequency(self, freq):
+ @frequency.setter
+ def frequency(self, freq):
ret = _bt_ctf_clock_set_frequency(self._c, freq)
if ret < 0:
raise ValueError("Invalid frequency value.")
+ """
+ Get the clock's precision (in clock ticks).
+ """
+ @property
+ def precision(self):
+ raise NotImplementedError("Getter not implemented.")
+
"""
Set the clock's precision (in clock ticks).
"""
- def set_precision(self, precision):
+ @precision.setter
+ def precision(self, precision):
ret = _bt_ctf_clock_set_precision(self._c, precision)
+ """
+ Get the clock's offset in seconds from POSIX.1 Epoch.
+ """
+ @property
+ def offset_seconds(self):
+ raise NotImplementedError("Getter not implemented.")
+
"""
Set the clock's offset in seconds from POSIX.1 Epoch.
"""
- def set_offset_seconds(self, offset_s):
+ @offset_seconds.setter
+ def offset_seconds(self, offset_s):
ret = _bt_ctf_clock_set_offset_s(self._c, offset_s)
if ret < 0:
raise ValueError("Invalid offset value.")
+ """
+ Get the clock's offset in ticks from POSIX.1 Epoch + offset in seconds.
+ """
+ @property
+ def offset(self):
+ raise NotImplementedError("Getter not implemented.")
+
"""
Set the clock's offset in ticks from POSIX.1 Epoch + offset in seconds.
"""
- def set_offset_seconds(self, offset):
+ @offset.setter
+ def offset(self, offset):
ret = _bt_ctf_clock_set_offset(self._c, offset)
if ret < 0:
raise ValueError("Invalid offset value.")
+ """
+ Get a clock's absolute attribute. A clock is absolute if the clock
+ is a global reference across the trace's other clocks.
+ """
+ @property
+ def absolute(self):
+ raise NotImplementedError("Getter not implemented.")
+
"""
Set a clock's absolute attribute. A clock is absolute if the clock
is a global reference across the trace's other clocks.
"""
- def set_is_absolute(self, is_absolute):
- ret = _bt_ctf_clock_set_is_absolute(self._c, is_absolute)
+ @absolute.setter
+ def absolute(self, is_absolute):
+ ret = _bt_ctf_clock_set_is_absolute(self._c, int(is_absolute))
if ret < 0:
raise ValueError("Could not set the clock's absolute attribute.")
+ """
+ Get the current time in nanoseconds since the clock's origin (offset and
+ offset_s attributes).
+ """
+ @property
+ def time(self):
+ raise NotImplementedError("Getter not implemented.")
+
"""
Set the current time in nanoseconds since the clock's origin (offset and
offset_s attributes). The clock's value will be sampled as events are
appended to a stream.
"""
- def set_time(self, time):
+ @time.setter
+ def time(self, time):
ret = _bt_ctf_clock_set_time(self._c, time)
if ret < 0:
raise ValueError("Invalid time value.")
- class FieldType:
+ class FieldDeclaration:
"""
- FieldType should not be instantiated directly. Please instantiate
- one of the concrete FieldType classes.
+ FieldDeclaration should not be instantiated directly. Please instantiate
+ one of the concrete FieldDeclaration classes.
"""
class IntegerBase:
# These values are based on the bt_ctf_integer_base enum
def __init__(self):
if self._ft is None:
- raise ValueError("FieldType creation failed.")
+ raise ValueError("FieldDeclaration creation failed.")
def __del__(self):
_bt_ctf_field_type_put(self._ft)
+ """
+ Get the field type's alignment.
+ """
+ @property
+ def alignment(self):
+ raise NotImplementedError("Getter not implemented.")
+
"""
Set the field type's alignment. Defaults to 1 (bit-aligned). However,
some types, such as structures and string, may impose other alignment
constraints.
"""
- def set_alignment(self, alignment):
+ @alignment.setter
+ def alignment(self, alignment):
ret = _bt_ctf_field_type_set_alignment(self._ft, alignment)
if ret < 0:
raise ValueError("Invalid alignment value.")
+ """
+ Get the field type's byte order. One of the ByteOrder's constant.
+ """
+ @property
+ def byte_order(self):
+ raise NotImplementedError("Getter not implemented.")
+
"""
Set the field type's byte order. Use constants defined in the ByteOrder
class.
"""
- def set_byte_order(self, byte_order):
+ @byte_order.setter
+ def byte_order(self, byte_order):
ret = _bt_ctf_field_type_set_byte_order(self._ft, byte_order)
if ret < 0:
raise ValueError("Could not set byte order value.")
- class FieldTypeInteger(FieldType):
+ class IntegerFieldDeclaration(FieldDeclaration):
"""
Create a new integer field type of the given size.
"""
super().__init__()
"""
- Set an integer type's signedness attribute.
+ Get an integer's signedness attribute.
+ """
+ @property
+ def signed(self):
+ raise NotImplementedError("Getter not implemented.")
+
+ """
+ Set an integer's signedness attribute.
"""
- def set_signed(self, signed):
+ @signed.setter
+ def signed(self, signed):
ret = _bt_ctf_field_type_integer_set_signed(self._ft, signed)
if ret < 0:
raise ValueError("Could not set signed attribute.")
"""
- Set the integer type's base used to pretty-print the resulting trace.
+ Get the integer's base used to pretty-print the resulting trace.
+ """
+ @property
+ def base(self):
+ raise NotImplementedError("Getter not implemented.")
+
+ """
+ Set the integer's base used to pretty-print the resulting trace.
The base must be a constant of the IntegerBase class.
"""
- def set_base(self, base):
+ @base.setter
+ def base(self, base):
ret = _bt_ctf_field_type_integer_set_base(self._ft, base)
if ret < 0:
raise ValueError("Could not set base value.")
+ """
+ Get the integer's encoding (one of the constants of the
+ CTFStringEncoding class).
+ """
+ @property
+ def encoding(self):
+ raise NotImplementedError("Getter not implemented.")
+
"""
An integer encoding may be set to signal that the integer must be printed
as a text character. Must be a constant from the CTFStringEncoding class.
"""
- def set_encoding(self, encoding):
+ @encoding.setter
+ def encoding(self, encoding):
ret = _bt_ctf_field_type_integer_set_encoding(self._ft, encoding)
if ret < 0:
raise ValueError("Could not set integer encoding.")
- class FieldTypeEnumeration(FieldType):
+ class EnumerationFieldDeclaration(FieldDeclaration):
"""
Create a new enumeration field type with the given underlying type.
"""
def __init__(self, integer_type):
- if integer_type is None or not isinstance(integer_type, CTFWriter.FieldTypeInteger):
+ if integer_type is None or not isinstance(integer_type, CTFWriter.IntegerFieldDeclaration):
raise TypeError("Invalid integer container.")
self._ft = _bt_ctf_field_type_enumeration_create(integer_type._ft)
Add a mapping to the enumeration. The range's values are inclusive.
"""
def add_mapping(self, name, range_start, range_end):
- ret = _bt_ctf_field_type_enumeration_add_mapping(self._ft, name, range_start, range_end)
+ ret = _bt_ctf_field_type_enumeration_add_mapping(self._ft, str(name), range_start, range_end)
if ret < 0:
raise ValueError("Could not add mapping to enumeration type.")
- class FieldTypeFloatingPoint(FieldType):
+ class FloatFieldDeclaration(FieldDeclaration):
FLT_EXP_DIG = 8
DBL_EXP_DIG = 11
FLT_MANT_DIG = 24
super().__init__()
"""
- Set the number of exponent digits to use to store the floatingpoint field.
+ Get the number of exponent digits to use to store the floating point field.
+ """
+ @property
+ def exponent_digits(self):
+ raise NotImplementedError("Getter not implemented.")
+
+ """
+ Set the number of exponent digits to use to store the floating point field.
The only values currently supported are FLT_EXP_DIG and DBL_EXP_DIG which
are defined as constants of this class.
"""
- def set_exponent_digits(self, exponent_digits):
+ @exponent_digits.setter
+ def exponent_digits(self, exponent_digits):
ret = _bt_ctf_field_type_floating_point_set_exponent_digits(self._ft, exponent_digits)
if ret < 0:
raise ValueError("Could not set exponent digit count.")
"""
- Set the numberof mantissa digits to use to store the floatingpoint field.
+ Get the number of mantissa digits to use to store the floating point field.
+ """
+ @property
+ def mantissa_digits(self):
+ raise NotImplementedError("Getter not implemented.")
+
+ """
+ Set the number of mantissa digits to use to store the floating point field.
The only values currently supported are FLT_MANT_DIG and DBL_MANT_DIG which
are defined as constants of this class.
"""
- def set_mantissa_digits(self, mantissa_digits):
+ @mantissa_digits.setter
+ def mantissa_digits(self, mantissa_digits):
ret = _bt_ctf_field_type_floating_point_set_mantissa_digits(self._ft, mantissa_digits)
if ret < 0:
raise ValueError("Could not set mantissa digit count.")
- class FieldTypeStructure(FieldType):
+ class StructureFieldDeclaration(FieldDeclaration):
"""
Create a new structure field type.
"""
Add a field of type "field_type" to the structure.
"""
def add_field(self, field_type, field_name):
- ret = _bt_ctf_field_type_structure_add_field(self._ft, field_type._ft, field_name)
+ ret = _bt_ctf_field_type_structure_add_field(self._ft, field_type._ft, str(field_name))
if ret < 0:
raise ValueError("Could not add field to structure.")
- class FieldTypeVariant(FieldType):
+ class VariantFieldDeclaration(FieldDeclaration):
"""
Create a new variant field type.
"""
def __init__(self, enum_tag, tag_name):
- if enum_tag is None or not isinstance(enum_tag, CTFWriter.FieldTypeEnumeration):
- raise TypeError("Invalid tag type; must be of type FieldTypeEnumeration.")
+ if enum_tag is None or not isinstance(enum_tag, CTFWriter.EnumerationFieldDeclaration):
+ raise TypeError("Invalid tag type; must be of type EnumerationFieldDeclaration.")
- self._ft = _bt_ctf_field_type_variant_create(enum_tag._ft, tag_name)
+ self._ft = _bt_ctf_field_type_variant_create(enum_tag._ft, str(tag_name))
super().__init__()
"""
Add a field of type "field_type" to the variant.
"""
def add_field(self, field_type, field_name):
- ret = _bt_ctf_field_type_variant_add_field(self._ft, field_type._ft, field_name)
+ ret = _bt_ctf_field_type_variant_add_field(self._ft, field_type._ft, str(field_name))
if ret < 0:
raise ValueError("Could not add field to variant.")
- class FieldTypeArray(FieldType):
+ class ArrayFieldDeclaration(FieldDeclaration):
"""
Create a new array field type.
"""
self._ft = _bt_ctf_field_type_array_create(element_type._ft, length)
super().__init__()
- class FieldTypeSequence(FieldType):
+ class SequenceFieldDeclaration(FieldDeclaration):
"""
Create a new sequence field type.
"""
def __init__(self, element_type, length_field_name):
- self._ft = _bt_ctf_field_type_sequence_create(element_type._ft, length_field_name)
+ self._ft = _bt_ctf_field_type_sequence_create(element_type._ft, str(length_field_name))
super().__init__()
- class FieldTypeString(FieldType):
+ class StringFieldDeclaration(FieldDeclaration):
"""
Create a new string field type.
"""
self._ft = _bt_ctf_field_type_string_create()
super().__init__()
+ """
+ Get a string type's encoding (a constant from the CTFStringEncoding class).
+ """
+ @property
+ def encoding(self):
+ raise NotImplementedError("Getter not implemented.")
+
"""
Set a string type's encoding. Must be a constant from the CTFStringEncoding class.
"""
- def set_encoding(self, encoding):
+ @encoding.setter
+ def encoding(self, encoding):
ret = _bt_ctf_field_type_string_set_encoding(self._ft, encoding)
if ret < 0:
raise ValueError("Could not set string encoding.")
"""
@staticmethod
def create_field(self, field_type):
- if field_type is None or not isinstance(field_type, CTFWriter.FieldType):
- raise TypeError("Invalid field_type. Type must be a FieldType-derived class.")
-
- if isinstance(field_type, CTFWriter.FieldTypeInteger):
- return CTFWriter.FieldInteger(field_type)
- elif isinstance(field_type, CTFWriter.FieldTypeEnumeration):
- return CTFWriter.FieldEnumeration(field_type)
- elif isinstance(field_type, CTFWriter.FieldTypeFloatingPoint):
- return CTFWriter.FieldFloatingPoint(field_type)
- elif isinstance(field_type, CTFWriter.FieldTypeStructure):
- return CTFWriter.FieldStructure(field_type)
- elif isinstance(field_type, CTFWriter.FieldTypeVariant):
- return CTFWriter.FieldVariant(field_type)
- elif isinstance(field_type, CTFWriter.FieldTypeArray):
- return CTFWriter.FieldArray(field_type)
- elif isinstance(field_type, CTFWriter.FieldTypeSequence):
- return CTFWriter.FieldSequence(field_type)
- elif isinstance(field_type, CTFWriter.FieldTypeString):
- return CTFWriter.FieldString(field_type)
+ if field_type is None or not isinstance(field_type, CTFWriter.FieldDeclaration):
+ raise TypeError("Invalid field_type. Type must be a FieldDeclaration-derived class.")
+
+ if isinstance(field_type, CTFWriter.IntegerFieldDeclaration):
+ return CTFWriter.IntegerField(field_type)
+ elif isinstance(field_type, CTFWriter.EnumerationFieldDeclaration):
+ return CTFWriter.EnumerationField(field_type)
+ elif isinstance(field_type, CTFWriter.FloatFieldDeclaration):
+ return CTFWriter.FloatFieldingPoint(field_type)
+ elif isinstance(field_type, CTFWriter.StructureFieldDeclaration):
+ return CTFWriter.StructureField(field_type)
+ elif isinstance(field_type, CTFWriter.VariantFieldDeclaration):
+ return CTFWriter.VariantField(field_type)
+ elif isinstance(field_type, CTFWriter.ArrayFieldDeclaration):
+ return CTFWriter.ArrayField(field_type)
+ elif isinstance(field_type, CTFWriter.SequenceFieldDeclaration):
+ return CTFWriter.SequenceField(field_type)
+ elif isinstance(field_type, CTFWriter.StringFieldDeclaration):
+ return CTFWriter.StringField(field_type)
class Field:
"""
Base class, do not instantiate.
"""
def __init__(self, field_type):
- if not isinstance(field_type, CTFWriter.FieldType):
+ if not isinstance(field_type, CTFWriter.FieldDeclaration):
raise TypeError("Invalid field_type argument.")
self._f = _bt_ctf_field_create(field_type._ft)
@staticmethod
def _create_field_from_native_instance(native_field_instance):
type_dict = {
- CTFTypeId.INTEGER : CTFWriter.FieldInteger,
- CTFTypeId.FLOAT : CTFWriter.FieldFloatingPoint,
- CTFTypeId.ENUM : CTFWriter.FieldEnumeration,
- CTFTypeId.STRING : CTFWriter.FieldString,
- CTFTypeId.STRUCT : CTFWriter.FieldStructure,
- CTFTypeId.VARIANT : CTFWriter.FieldVariant,
- CTFTypeId.ARRAY : CTFWriter.FieldArray,
- CTFTypeId.SEQUENCE : CTFWriter.FieldSequence
+ CTFTypeId.INTEGER : CTFWriter.IntegerField,
+ CTFTypeId.FLOAT : CTFWriter.FloatFieldingPoint,
+ CTFTypeId.ENUM : CTFWriter.EnumerationField,
+ CTFTypeId.STRING : CTFWriter.StringField,
+ CTFTypeId.STRUCT : CTFWriter.StructureField,
+ CTFTypeId.VARIANT : CTFWriter.VariantField,
+ CTFTypeId.ARRAY : CTFWriter.ArrayField,
+ CTFTypeId.SEQUENCE : CTFWriter.SequenceField
}
field_type = _bt_python_get_field_type(native_field_instance)
field.__class__ = type_dict[field_type]
return field
- class FieldInteger(Field):
+ class IntegerField(Field):
+ """
+ Get an integer field's value.
+ """
+ @property
+ def value(self):
+ raise NotImplementedError("Getter not implemented.")
+
"""
Set an integer field's value.
"""
- def set_value(self, value):
+ @value.setter
+ def value(self, value):
signedness = _bt_python_field_integer_get_signedness(self._f)
if signedness < 0:
raise TypeError("Invalid integer instance.")
if ret < 0:
raise ValueError("Could not set integer field value.")
- class FieldEnumeration(Field):
+ class EnumerationField(Field):
"""
Return the enumeration's underlying container field (an integer field).
"""
- def get_container(self):
- container = CTFWriter.FieldInteger.__new__(CTFWriter.FieldInteger)
+ @property
+ def container(self):
+ container = CTFWriter.IntegerField.__new__(CTFWriter.IntegerField)
container._f = _bt_ctf_field_enumeration_get_container(self._f)
if container._f is None:
raise TypeError("Invalid enumeration field type.")
return container
- class FieldFloatingPoint(Field):
+ class FloatFieldingPoint(Field):
+ """
+ Get a floating point field's value.
+ """
+ @property
+ def value(self):
+ raise NotImplementedError("Getter not implemented.")
+
"""
Set a floating point field's value.
"""
- def set_value(self, value):
- ret = _bt_ctf_field_floating_point_set_value(self._f, value)
+ @value.setter
+ def value(self, value):
+ ret = _bt_ctf_field_floating_point_set_value(self._f, float(value))
if ret < 0:
raise ValueError("Could not set floating point field value.")
- class FieldStructure(Field):
+ class StructureField(Field):
"""
Get the structure's field corresponding to the provided field name.
"""
- def get_field(self, field_name):
- native_instance = _bt_ctf_field_structure_get_field(self._f, field_name)
+ def field(self, field_name):
+ native_instance = _bt_ctf_field_structure_get_field(self._f, str(field_name))
if native_instance is None:
raise ValueError("Invalid field_name provided.")
return CTFWriter.Field._create_field_from_native_instance(native_instance)
- class FieldVariant(Field):
+ class VariantField(Field):
"""
Return the variant's selected field. The "tag" field is the selector enum field.
"""
- def get_field(self, tag):
+ def field(self, tag):
native_instance = _bt_ctf_field_variant_get_field(self._f, tag._f)
if native_instance is None:
raise ValueError("Invalid tag provided.")
return CTFWriter.Field._create_field_from_native_instance(native_instance)
- class FieldArray(Field):
+ class ArrayField(Field):
"""
Return the array's field at position "index".
"""
- def get_field(self, index):
+ def field(self, index):
native_instance = _bt_ctf_field_array_get_field(self._f, index)
if native_instance is None:
raise IndexError("Invalid index provided.")
return CTFWriter.Field._create_field_from_native_instance(native_instance)
- class FieldSequence(Field):
+ class SequenceField(Field):
+ """
+ Get the sequence's length field (IntegerField).
+ """
+ @property
+ def length(self):
+ raise NotImplementedError("Getter not implemented.")
+
"""
Set the sequence's length field (IntegerField).
"""
- def set_length(self, length):
- if not isinstance(length, CTFWriter.FieldInteger):
+ @length.setter
+ def length(self, length_field):
+ if not isinstance(length_field, CTFWriter.IntegerField):
raise TypeError("Invalid length field.")
- ret = _bt_ctf_field_sequence_set_length(self._f, length._f)
+ ret = _bt_ctf_field_sequence_set_length(self._f, length_field._f)
if ret < 0:
raise ValueError("Could not set sequence length.")
"""
Return the sequence's field at position "index".
"""
- def get_field(self, index):
+ def field(self, index):
native_instance = _bt_ctf_field_sequence_get_field(self._f, index)
if native_instance is None:
raise ValueError("Could not get sequence element at index.")
return CTFWriter.Field._create_field_from_native_instance(native_instance)
- class FieldString(Field):
+ class StringField(Field):
+ """
+ Get a string field's value.
+ """
+ @property
+ def value(self):
+ raise NotImplementedError("Getter not implemented.")
+
"""
Set a string field's value.
"""
- def set_value(self, value):
- ret = _bt_ctf_field_string_set_value(self._f, value)
+ @value.setter
+ def value(self, value):
+ ret = _bt_ctf_field_string_set_value(self._f, str(value))
if ret < 0:
raise ValueError("Could not set string field value.")
Add a field of type "field_type" to the event class.
"""
def add_field(self, field_type, field_name):
- ret = _bt_ctf_event_class_add_field(self._ec, field_type._ft, field_name)
+ ret = _bt_ctf_event_class_add_field(self._ec, field_type._ft, str(field_name))
if ret < 0:
raise ValueError("Could not add field to event class.")
"""
Set a manually created field as an event's payload.
"""
- def set_payload(self, field_name, value):
+ def set_payload(self, field_name, value_field):
if not isinstance(value, CTFWriter.Field):
raise TypeError("Invalid value type.")
- ret = _bt_ctf_event_set_payload(self._e, field_name, value._f)
+ ret = _bt_ctf_event_set_payload(self._e, str(field_name), value_field._f)
if ret < 0:
raise ValueError("Could not set event field payload.")
"""
- Set a manually created field as an event's payload.
+ Get a field from event.
"""
- def get_payload(self, field_name):
- native_instance = _bt_ctf_event_get_payload(self._e, field_name)
+ def payload(self, field_name):
+ native_instance = _bt_ctf_event_get_payload(self._e, str(field_name))
if native_instance is None:
raise ValueError("Could not get event payload.")
return CTFWriter.Field._create_field_from_native_instance(native_instance)
def __del__(self):
_bt_ctf_stream_class_put(self._sc)
+ """
+ Get a stream class' clock.
+ """
+ @property
+ def clock(self):
+ raise NotImplementedError("Getter not implemented.")
+
"""
Assign a clock to a stream class.
"""
- def set_clock(self, clock):
+ @clock.setter
+ def clock(self, clock):
if not isinstance(clock, CTFWriter.Clock):
raise TypeError("Invalid clock type.")
"""
Add an event class to a stream class. New events can be added even after a
- stream has beem instanciated and events have been appended. However, a stream
+ stream has been instantiated and events have been appended. However, a stream
will not accept events of a class that has not been registered beforehand.
"""
def add_event_class(self, event_class):
Increase the current packet's discarded event count.
"""
def append_discarded_events(self, event_count):
- ret = _bt_ctf_stream_append_discarded_events(self._s, event_count)
- if ret < 0:
- raise ValueError("Could not append discarded events.")
+ _bt_ctf_stream_append_discarded_events(self._s, event_count)
"""
Append "event" to the stream's current packet. The stream's associated clock
Add an environment field to the trace.
"""
def add_environment_field(self, name, value):
- ret = _bt_ctf_writer_add_environment_field(self._w, name, value)
+ ret = _bt_ctf_writer_add_environment_field(self._w, str(name), str(value))
if ret < 0:
raise ValueError("Could not add environment field to trace.")
"""
Get the trace's TSDL meta-data.
"""
- def get_metadata(self):
+ @property
+ def metadata(self):
return _bt_ctf_writer_get_metadata_string(self._w)
"""
def flush_metadata(self):
_bt_ctf_writer_flush_metadata(self._w)
+ """
+ Get the trace's byte order. Must be a constant from the ByteOrder
+ class.
+ """
+ @property
+ def byte_order(self):
+ raise NotImplementedError("Getter not implemented.")
+
"""
Set the trace's byte order. Must be a constant from the ByteOrder
class. Defaults to BYTE_ORDER_NATIVE, the host machine's endianness.
"""
- def set_byte_order(self, byte_order):
+ @byte_order.setter
+ def byte_order(self, byte_order):
ret = _bt_ctf_writer_set_byte_order(self._w, byte_order)
if ret < 0:
raise ValueError("Could not set trace's byte order.")
projects / babeltrace.git / blobdiff