Signed-off-by: Philippe Proulx <eeppeliteloop@gmail.com>
Signed-off-by: Jérémie Galarneau <jeremie.galarneau@efficios.com>
"""
The TraceCollection is the object that contains all currently opened traces.
"""
"""
The TraceCollection is the object that contains all currently opened traces.
"""
Return: the corresponding TraceHandle on success or None on error.
"""
Return: the corresponding TraceHandle on success or None on error.
"""
- ret = nbt._bt_context_add_trace(
- self._tc,
- path,
- format_str,
- None,
- None,
- None)
+
+ ret = nbt._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
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):
return th
def add_traces_recursive(self, path, format_str):
noTrace = True
error = False
for fullpath, dirs, files in os.walk(path):
if "metadata" in files:
trace_handle = self.add_trace(fullpath, format_str)
noTrace = True
error = False
for fullpath, dirs, files in os.walk(path):
if "metadata" in files:
trace_handle = self.add_trace(fullpath, format_str)
if trace_handle is None:
error = True
continue
if trace_handle is None:
error = True
continue
if noTrace and error:
return None
if noTrace and error:
return None
return trace_handles
def remove_trace(self, trace_handle):
return trace_handles
def remove_trace(self, trace_handle):
Remove a trace from the TraceCollection.
Effectively closing the trace.
"""
Remove a trace from the TraceCollection.
Effectively closing the trace.
"""
try:
nbt._bt_context_remove_trace(self._tc, trace_handle._id)
except AttributeError:
try:
nbt._bt_context_remove_trace(self._tc, trace_handle._id)
except AttributeError:
- raise TypeError("in remove_trace, "
- "argument 2 must be a TraceHandle instance")
+ raise TypeError("in remove_trace, argument 2 must be a TraceHandle instance")
@property
def events(self):
@property
def events(self):
event after the the generator has gone out of scope may result in a
crash or data corruption.
"""
event after the the generator has gone out of scope may result in a
crash or data corruption.
"""
begin_pos_ptr = nbt._bt_iter_pos()
end_pos_ptr = nbt._bt_iter_pos()
begin_pos_ptr.type = nbt.SEEK_BEGIN
begin_pos_ptr = nbt._bt_iter_pos()
end_pos_ptr = nbt._bt_iter_pos()
begin_pos_ptr.type = nbt.SEEK_BEGIN
Generator function to iterate over the events of open in the current
TraceCollection from timestamp_begin to timestamp_end.
"""
Generator function to iterate over the events of open in the current
TraceCollection from timestamp_begin to timestamp_end.
"""
begin_pos_ptr = nbt._bt_iter_pos()
end_pos_ptr = nbt._bt_iter_pos()
begin_pos_ptr.type = end_pos_ptr.type = nbt.SEEK_TIME
begin_pos_ptr = nbt._bt_iter_pos()
end_pos_ptr = nbt._bt_iter_pos()
begin_pos_ptr.type = end_pos_ptr.type = nbt.SEEK_TIME
def timestamp_begin(self):
pos_ptr = nbt._bt_iter_pos()
pos_ptr.type = nbt.SEEK_BEGIN
def timestamp_begin(self):
pos_ptr = nbt._bt_iter_pos()
pos_ptr.type = nbt.SEEK_BEGIN
return self._timestamp_at_pos(pos_ptr)
@property
def timestamp_end(self):
pos_ptr = nbt._bt_iter_pos()
pos_ptr.type = nbt.SEEK_LAST
return self._timestamp_at_pos(pos_ptr)
@property
def timestamp_end(self):
pos_ptr = nbt._bt_iter_pos()
pos_ptr.type = nbt.SEEK_LAST
return self._timestamp_at_pos(pos_ptr)
def _timestamp_at_pos(self, pos_ptr):
ctf_it_ptr = nbt._bt_ctf_iter_create(self._tc, pos_ptr, pos_ptr)
return self._timestamp_at_pos(pos_ptr)
def _timestamp_at_pos(self, pos_ptr):
ctf_it_ptr = nbt._bt_ctf_iter_create(self._tc, pos_ptr, pos_ptr)
- raise NotImplementedError(
- "Creation of multiple iterators is unsupported.")
+ raise NotImplementedError("Creation of multiple iterators is unsupported.")
+
ev_ptr = nbt._bt_ctf_iter_read_event(ctf_it_ptr)
nbt._bt_ctf_iter_destroy(ctf_it_ptr)
ev_ptr = nbt._bt_ctf_iter_read_event(ctf_it_ptr)
nbt._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 = nbt._bt_ctf_iter_create(self._tc, begin_pos_ptr, end_pos_ptr)
if ev_ptr is None:
return None
def _events(self, begin_pos_ptr, end_pos_ptr):
ctf_it_ptr = nbt._bt_ctf_iter_create(self._tc, begin_pos_ptr, end_pos_ptr)
- raise NotImplementedError(
- "Creation of multiple iterators is unsupported.")
+ raise NotImplementedError("Creation of multiple iterators is unsupported.")
while True:
ev_ptr = nbt._bt_ctf_iter_read_event(ctf_it_ptr)
while True:
ev_ptr = nbt._bt_ctf_iter_read_event(ctf_it_ptr)
if ev_ptr is None:
break
ev = Event.__new__(Event)
ev._e = ev_ptr
if ev_ptr is None:
break
ev = Event.__new__(Event)
ev._e = ev_ptr
try:
yield ev
except GeneratorExit:
break
ret = nbt._bt_iter_next(nbt._bt_ctf_get_iter(ctf_it_ptr))
try:
yield ev
except GeneratorExit:
break
ret = nbt._bt_iter_next(nbt._bt_ctf_get_iter(ctf_it_ptr))
babeltrace_file must be a File instance opened in write mode.
"""
babeltrace_file must be a File instance opened in write mode.
"""
try:
if babeltrace_file._file is not None:
nbt._bt_print_format_list(babeltrace_file._file)
except AttributeError:
try:
if babeltrace_file._file is not None:
nbt._bt_print_format_list(babeltrace_file._file)
except AttributeError:
- raise TypeError("in print_format_list, "
- "argument 1 must be a File instance")
+ raise TypeError("in print_format_list, argument 1 must be a File instance")
# Based on enum bt_clock_type in clock-type.h
# Based on enum bt_clock_type in clock-type.h
"""
The TraceHandle allows the user to manipulate a trace file directly.
It is a unique identifier representing a trace file.
"""
The TraceHandle allows the user to manipulate a trace file directly.
It is a unique identifier representing a trace file.
@property
def id(self):
"""Return the TraceHandle id."""
@property
def id(self):
"""Return the TraceHandle id."""
return self._id
@property
def path(self):
"""Return the path of a TraceHandle."""
return self._id
@property
def path(self):
"""Return the path of a TraceHandle."""
- return nbt._bt_trace_handle_get_path(self._trace_collection._tc, self._id)
+
+ return nbt._bt_trace_handle_get_path(self._trace_collection._tc,
+ self._id)
@property
def timestamp_begin(self):
"""Return the creation time of the buffers of a trace."""
@property
def timestamp_begin(self):
"""Return the creation time of the buffers of a trace."""
- return nbt._bt_trace_handle_get_timestamp_begin(
- self._trace_collection._tc, self._id, ClockType.CLOCK_REAL)
+
+ return nbt._bt_trace_handle_get_timestamp_begin(self._trace_collection._tc,
+ self._id,
+ ClockType.CLOCK_REAL)
@property
def timestamp_end(self):
"""Return the destruction timestamp of the buffers of a trace."""
@property
def timestamp_end(self):
"""Return the destruction timestamp of the buffers of a trace."""
- return nbt._bt_trace_handle_get_timestamp_end(
- self._trace_collection._tc, self._id, ClockType.CLOCK_REAL)
+
+ return nbt._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.
"""
@property
def events(self):
"""
Generator returning all events (EventDeclaration) in a trace.
"""
- ret = nbt._bt_python_event_decl_listcaller(
- self.id,
- self._trace_collection._tc)
+
+ ret = nbt._bt_python_event_decl_listcaller(self.id,
+ self._trace_collection._tc)
if not isinstance(ret, list):
return
ptr_list, count = ret
if not isinstance(ret, list):
return
ptr_list, count = ret
for i in range(count):
tmp = EventDeclaration.__new__(EventDeclaration)
tmp._ed = nbt._bt_python_decl_one_from_list(ptr_list, i)
for i in range(count):
tmp = EventDeclaration.__new__(EventDeclaration)
tmp._ed = nbt._bt_python_decl_one_from_list(ptr_list, i)
getattr(
CTFTypeId,
attr)) and not attr.startswith("__")]
getattr(
CTFTypeId,
attr)) and not attr.startswith("__")]
for attr in constants:
if getattr(CTFTypeId, attr) == id:
name = attr
break
for attr in constants:
if getattr(CTFTypeId, attr) == id:
name = attr
break
getattr(
CTFScope,
attr)) and not attr.startswith("__")]
getattr(
CTFScope,
attr)) and not attr.startswith("__")]
for attr in constants:
if getattr(CTFScope, attr) == scope:
name = attr
break
for attr in constants:
if getattr(CTFScope, attr) == scope:
name = attr
break
return name
# Priority of the scopes when searching for event fields
return name
# 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]
+_scopes = [
+ CTFScope.EVENT_FIELDS,
+ CTFScope.EVENT_CONTEXT,
+ CTFScope.STREAM_EVENT_CONTEXT,
+ CTFScope.STREAM_EVENT_HEADER,
+ CTFScope.STREAM_PACKET_CONTEXT,
+ CTFScope.TRACE_PACKET_HEADER
+]
class Event(collections.Mapping):
class Event(collections.Mapping):
"""
This class represents an event from the trace.
It is obtained using the TraceCollection generator functions.
"""
This class represents an event from the trace.
It is obtained using the TraceCollection generator functions.
@property
def name(self):
"""Return the name of the event or None on error."""
@property
def name(self):
"""Return the name of the event or None on error."""
return nbt._bt_ctf_event_name(self._e)
@property
return nbt._bt_ctf_event_name(self._e)
@property
Return the timestamp of the event as written in
the packet (in cycles) or -1ULL on error.
"""
Return the timestamp of the event as written in
the packet (in cycles) or -1ULL on error.
"""
return nbt._bt_ctf_get_cycles(self._e)
@property
return nbt._bt_ctf_get_cycles(self._e)
@property
Return the timestamp of the event offset with the
system clock source or -1ULL on error.
"""
Return the timestamp of the event offset with the
system clock source or -1ULL on error.
"""
return nbt._bt_ctf_get_timestamp(self._e)
@property
return nbt._bt_ctf_get_timestamp(self._e)
@property
timestamp. Note that the datetime class' precision
is limited to microseconds.
"""
timestamp. Note that the datetime class' precision
is limited to microseconds.
"""
return datetime.fromtimestamp(self.timestamp / 1E9)
def field_with_scope(self, field_name, scope):
return datetime.fromtimestamp(self.timestamp / 1E9)
def field_with_scope(self, field_name, scope):
Get field_name's value in scope.
None is returned if no field matches field_name.
"""
Get field_name's value in scope.
None is returned if no field matches field_name.
"""
- if not scope in _scopes:
+
+ if scope not in _scopes:
raise ValueError("Invalid scope provided")
raise ValueError("Invalid scope provided")
field = self._field_with_scope(field_name, scope)
field = self._field_with_scope(field_name, scope)
if field is not None:
return field.value
if field is not None:
return field.value
return None
def field_list_with_scope(self, scope):
"""Return a list of field names in scope."""
return None
def field_list_with_scope(self, scope):
"""Return a list of field names in scope."""
- if not scope in _scopes:
+
+ if scope not in _scopes:
raise ValueError("Invalid scope provided")
raise ValueError("Invalid scope provided")
for field in self._field_list_with_scope(scope):
field_names.append(field.name)
for field in self._field_list_with_scope(scope):
field_names.append(field.name)
return field_names
@property
return field_names
@property
Get the TraceHandle associated with this event
Return None on error
"""
Get the TraceHandle associated with this event
Return None on error
"""
ret = nbt._bt_ctf_event_get_handle_id(self._e)
ret = nbt._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()
if ret < 0:
return None
th = TraceHandle.__new__(TraceHandle)
th._id = ret
th._trace_collection = self.get_trace_collection()
Get the TraceCollection associated with this event.
Return None on error.
"""
Get the TraceCollection associated with this event.
Return None on error.
"""
trace_collection = TraceCollection()
trace_collection._tc = nbt._bt_ctf_event_get_context(self._e)
trace_collection = TraceCollection()
trace_collection._tc = nbt._bt_ctf_event_get_context(self._e)
if trace_collection._tc is None:
return None
else:
if trace_collection._tc is None:
return None
else:
Use field_with_scope() to explicitly access fields in a given
scope.
"""
Use field_with_scope() to explicitly access fields in a given
scope.
"""
field = self._field(field_name)
field = self._field(field_name)
if field is not None:
return field.value
if field is not None:
return field.value
raise KeyError(field_name)
def __iter__(self):
raise KeyError(field_name)
def __iter__(self):
def __len__(self):
count = 0
def __len__(self):
count = 0
for scope in _scopes:
scope_ptr = nbt._bt_ctf_get_top_level_scope(self._e, scope)
ret = nbt._bt_python_field_listcaller(self._e, scope_ptr)
for scope in _scopes:
scope_ptr = nbt._bt_ctf_get_top_level_scope(self._e, scope)
ret = nbt._bt_python_field_listcaller(self._e, scope_ptr)
if isinstance(ret, list):
count += ret[1]
if isinstance(ret, list):
count += ret[1]
return count
def __contains__(self, field_name):
return count
def __contains__(self, field_name):
def keys(self):
"""Return a list of field names."""
def keys(self):
"""Return a list of field names."""
for scope in _scopes:
for name in self.field_list_with_scope(scope):
field_names.add(name)
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)
return list(field_names)
def get(self, field_name, default=None):
field = self._field(field_name)
if field is None:
return default
if field is None:
return default
return field.value
def items(self):
return field.value
def items(self):
def _field_with_scope(self, field_name, scope):
scope_ptr = nbt._bt_ctf_get_top_level_scope(self._e, scope)
def _field_with_scope(self, field_name, scope):
scope_ptr = nbt._bt_ctf_get_top_level_scope(self._e, scope)
if scope_ptr is None:
return None
definition_ptr = nbt._bt_ctf_get_field(self._e, scope_ptr, field_name)
if scope_ptr is None:
return None
definition_ptr = nbt._bt_ctf_get_field(self._e, scope_ptr, field_name)
if definition_ptr is None:
return None
field = _Definition(definition_ptr, scope)
if definition_ptr is None:
return None
field = _Definition(definition_ptr, scope)
return field
def _field(self, field_name):
field = None
return field
def _field(self, field_name):
field = None
for scope in _scopes:
field = self._field_with_scope(field_name, scope)
for scope in _scopes:
field = self._field_with_scope(field_name, scope)
if field is not None:
break
if field is not None:
break
return field
def _field_list_with_scope(self, scope):
return field
def _field_list_with_scope(self, scope):
count = 0
list_ptr = None
ret = nbt._bt_python_field_listcaller(self._e, scope_ptr)
count = 0
list_ptr = None
ret = nbt._bt_python_field_listcaller(self._e, scope_ptr)
if isinstance(ret, list):
list_ptr, count = ret
for i in range(count):
definition_ptr = nbt._bt_python_field_one_from_list(list_ptr, i)
if isinstance(ret, list):
list_ptr, count = ret
for i in range(count):
definition_ptr = nbt._bt_python_field_one_from_list(list_ptr, i)
if definition_ptr is not None:
definition = _Definition(definition_ptr, scope)
fields.append(definition)
if definition_ptr is not None:
definition = _Definition(definition_ptr, scope)
fields.append(definition)
return fields
class FieldError(Exception):
return fields
class FieldError(Exception):
def __init__(self, value):
self.value = value
def __init__(self, value):
self.value = value
"""Event declaration class. Do not instantiate."""
"""Event declaration class. Do not instantiate."""
MAX_UINT64 = 0xFFFFFFFFFFFFFFFF
def __init__(self):
MAX_UINT64 = 0xFFFFFFFFFFFFFFFF
def __init__(self):
@property
def name(self):
"""Return the name of the event or None on error"""
@property
def name(self):
"""Return the name of the event or None on error"""
return nbt._bt_ctf_get_decl_event_name(self._ed)
@property
def id(self):
"""Return the event-ID of the event or -1 on error"""
return nbt._bt_ctf_get_decl_event_name(self._ed)
@property
def id(self):
"""Return the event-ID of the event or -1 on error"""
id = nbt._bt_ctf_get_decl_event_id(self._ed)
id = nbt._bt_ctf_get_decl_event_id(self._ed)
if id == self.MAX_UINT64:
id = -1
if id == self.MAX_UINT64:
id = -1
5) STREAM_PACKET_CONTEXT
6) TRACE_PACKET_HEADER
"""
5) STREAM_PACKET_CONTEXT
6) TRACE_PACKET_HEADER
"""
for scope in _scopes:
for declaration in self.fields_scope(scope):
yield declaration
for scope in _scopes:
for declaration in self.fields_scope(scope):
yield declaration
return
list_ptr, count = ret
return
list_ptr, count = ret
- field_declaration_ptr = nbt._bt_python_field_decl_one_from_list(
- list_ptr,
- i)
- if field_declaration_ptr is not None:
- declaration_ptr = nbt._bt_ctf_get_decl_from_field_decl(
- field_declaration_ptr)
- field_declaration = _create_field_declaration(
- declaration_ptr,
- nbt._bt_ctf_get_decl_field_name(field_declaration_ptr),
- scope)
+ field_decl_ptr = nbt._bt_python_field_decl_one_from_list(list_ptr, i)
+
+ if field_decl_ptr is not None:
+ decl_ptr = nbt._bt_ctf_get_decl_from_field_decl(field_decl_ptr)
+ name = nbt._bt_ctf_get_decl_field_name(field_decl_ptr)
+ field_declaration = _create_field_declaration(decl_ptr, name,
+ scope)
yield field_declaration
class FieldDeclaration:
yield field_declaration
class FieldDeclaration:
"""Field declaration class. Do not instantiate."""
def __init__(self):
raise NotImplementedError("FieldDeclaration cannot be instantiated")
def __repr__(self):
"""Field declaration class. Do not instantiate."""
def __init__(self):
raise NotImplementedError("FieldDeclaration cannot be instantiated")
def __repr__(self):
- return "({0}) {1} {2}".format(
- CTFScope.scope_name(self.scope), CTFTypeId.type_name(self.type), self.name)
+ 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."""
@property
def name(self):
"""Return the name of a FieldDeclaration or None on error."""
return self._name
@property
return self._name
@property
Return the FieldDeclaration's type. One of the entries in class
CTFTypeId.
"""
Return the FieldDeclaration's type. One of the entries in class
CTFTypeId.
"""
return nbt._bt_ctf_field_type(self._fd)
@property
return nbt._bt_ctf_field_type(self._fd)
@property
"""
Return the FieldDeclaration's scope.
"""
"""
Return the FieldDeclaration's scope.
"""
return self._s
class IntegerFieldDeclaration(FieldDeclaration):
return self._s
class IntegerFieldDeclaration(FieldDeclaration):
"""Do not instantiate."""
def __init__(self):
"""Do not instantiate."""
def __init__(self):
- raise NotImplementedError(
- "IntegerFieldDeclaration cannot be instantiated")
+ raise NotImplementedError("IntegerFieldDeclaration cannot be instantiated")
@property
def signedness(self):
@property
def signedness(self):
Return the signedness of an integer:
0 if unsigned; 1 if signed; -1 on error.
"""
Return the signedness of an integer:
0 if unsigned; 1 if signed; -1 on error.
"""
return nbt._bt_ctf_get_int_signedness(self._fd)
@property
def base(self):
"""Return the base of an int or a negative value on error."""
return nbt._bt_ctf_get_int_signedness(self._fd)
@property
def base(self):
"""Return the base of an int or a negative value on error."""
return nbt._bt_ctf_get_int_base(self._fd)
@property
return nbt._bt_ctf_get_int_base(self._fd)
@property
"""
Return the byte order. One of class ByteOrder's entries.
"""
"""
Return the byte order. One of class ByteOrder's entries.
"""
ret = nbt._bt_ctf_get_int_byte_order(self._fd)
ret = nbt._bt_ctf_get_int_byte_order(self._fd)
if ret == 1234:
return ByteOrder.BYTE_ORDER_LITTLE_ENDIAN
elif ret == 4321:
if ret == 1234:
return ByteOrder.BYTE_ORDER_LITTLE_ENDIAN
elif ret == 4321:
Return the size, in bits, of an int or a negative
value on error.
"""
Return the size, in bits, of an int or a negative
value on error.
"""
return nbt._bt_ctf_get_int_len(self._fd)
@property
return nbt._bt_ctf_get_int_len(self._fd)
@property
Return the encoding. One of class CTFStringEncoding's entries.
Return a negative value on error.
"""
Return the encoding. One of class CTFStringEncoding's entries.
Return a negative value on error.
"""
return nbt._bt_ctf_get_encoding(self._fd)
class EnumerationFieldDeclaration(FieldDeclaration):
return nbt._bt_ctf_get_encoding(self._fd)
class EnumerationFieldDeclaration(FieldDeclaration):
"""Do not instantiate."""
def __init__(self):
"""Do not instantiate."""
def __init__(self):
- raise NotImplementedError(
- "EnumerationFieldDeclaration cannot be instantiated")
+ raise NotImplementedError("EnumerationFieldDeclaration cannot be instantiated")
class ArrayFieldDeclaration(FieldDeclaration):
class ArrayFieldDeclaration(FieldDeclaration):
"""Do not instantiate."""
def __init__(self):
"""Do not instantiate."""
def __init__(self):
- raise NotImplementedError(
- "ArrayFieldDeclaration cannot be instantiated")
+ raise NotImplementedError("ArrayFieldDeclaration cannot be instantiated")
@property
def length(self):
@property
def length(self):
Return the length of an array or a negative
value on error.
"""
Return the length of an array or a negative
value on error.
"""
return nbt._bt_ctf_get_array_len(self._fd)
@property
return nbt._bt_ctf_get_array_len(self._fd)
@property
"""
Return element declaration.
"""
"""
Return element declaration.
"""
field_decl_ptr = nbt._bt_python_get_array_element_declaration(self._fd)
field_decl_ptr = nbt._bt_python_get_array_element_declaration(self._fd)
return _create_field_declaration(field_decl_ptr, "", self.scope)
class SequenceFieldDeclaration(FieldDeclaration):
return _create_field_declaration(field_decl_ptr, "", self.scope)
class SequenceFieldDeclaration(FieldDeclaration):
"""Do not instantiate."""
def __init__(self):
"""Do not instantiate."""
def __init__(self):
- raise NotImplementedError(
- "SequenceFieldDeclaration cannot be instantiated")
+ raise NotImplementedError("SequenceFieldDeclaration cannot be instantiated")
@property
def element_declaration(self):
"""
Return element declaration.
"""
@property
def element_declaration(self):
"""
Return element declaration.
"""
field_decl_ptr = nbt._bt_python_get_sequence_element_declaration(self._fd)
field_decl_ptr = nbt._bt_python_get_sequence_element_declaration(self._fd)
return _create_field_declaration(field_decl_ptr, "", self.scope)
class FloatFieldDeclaration(FieldDeclaration):
return _create_field_declaration(field_decl_ptr, "", self.scope)
class FloatFieldDeclaration(FieldDeclaration):
"""Do not instantiate."""
def __init__(self):
"""Do not instantiate."""
def __init__(self):
- raise NotImplementedError(
- "FloatFieldDeclaration cannot be instantiated")
+ raise NotImplementedError("FloatFieldDeclaration cannot be instantiated")
class StructureFieldDeclaration(FieldDeclaration):
class StructureFieldDeclaration(FieldDeclaration):
"""Do not instantiate."""
def __init__(self):
"""Do not instantiate."""
def __init__(self):
- raise NotImplementedError(
- "StructureFieldDeclaration cannot be instantiated")
+ raise NotImplementedError("StructureFieldDeclaration cannot be instantiated")
class StringFieldDeclaration(FieldDeclaration):
class StringFieldDeclaration(FieldDeclaration):
"""Do not instantiate."""
def __init__(self):
"""Do not instantiate."""
def __init__(self):
- raise NotImplementedError(
- "StringFieldDeclaration cannot be instantiated")
+ raise NotImplementedError("StringFieldDeclaration cannot be instantiated")
class VariantFieldDeclaration(FieldDeclaration):
class VariantFieldDeclaration(FieldDeclaration):
"""Do not instantiate."""
def __init__(self):
"""Do not instantiate."""
def __init__(self):
- raise NotImplementedError(
- "VariantFieldDeclaration cannot be instantiated")
+ raise NotImplementedError("VariantFieldDeclaration cannot be instantiated")
accessing a field and reset the error flag.
Return 0 if no error, a negative value otherwise.
"""
accessing a field and reset the error flag.
Return 0 if no error, a negative value otherwise.
"""
return nbt._bt_ctf_field_get_error()
return nbt._bt_ctf_field_get_error()
"""
Private field declaration factory.
"""
"""
Private field declaration factory.
"""
if declaration_ptr is None:
raise ValueError("declaration_ptr must be valid")
if declaration_ptr is None:
raise ValueError("declaration_ptr must be valid")
- if not scope in _scopes:
+ if scope not in _scopes:
raise ValueError("Invalid scope provided")
type = nbt._bt_ctf_field_type(declaration_ptr)
declaration = None
raise ValueError("Invalid scope provided")
type = nbt._bt_ctf_field_type(declaration_ptr)
declaration = None
if type == CTFTypeId.INTEGER:
declaration = IntegerFieldDeclaration.__new__(IntegerFieldDeclaration)
elif type == CTFTypeId.ENUM:
if type == CTFTypeId.INTEGER:
declaration = IntegerFieldDeclaration.__new__(IntegerFieldDeclaration)
elif type == CTFTypeId.ENUM:
- declaration = EnumerationFieldDeclaration.__new__(
- EnumerationFieldDeclaration)
+ declaration = EnumerationFieldDeclaration.__new__(EnumerationFieldDeclaration)
elif type == CTFTypeId.ARRAY:
declaration = ArrayFieldDeclaration.__new__(ArrayFieldDeclaration)
elif type == CTFTypeId.SEQUENCE:
elif type == CTFTypeId.ARRAY:
declaration = ArrayFieldDeclaration.__new__(ArrayFieldDeclaration)
elif type == CTFTypeId.SEQUENCE:
- declaration = SequenceFieldDeclaration.__new__(
- SequenceFieldDeclaration)
+ declaration = SequenceFieldDeclaration.__new__(SequenceFieldDeclaration)
elif type == CTFTypeId.FLOAT:
declaration = FloatFieldDeclaration.__new__(FloatFieldDeclaration)
elif type == CTFTypeId.STRUCT:
elif type == CTFTypeId.FLOAT:
declaration = FloatFieldDeclaration.__new__(FloatFieldDeclaration)
elif type == CTFTypeId.STRUCT:
- declaration = StructureFieldDeclaration.__new__(
- StructureFieldDeclaration)
+ declaration = StructureFieldDeclaration.__new__(StructureFieldDeclaration)
elif type == CTFTypeId.STRING:
declaration = StringFieldDeclaration.__new__(StringFieldDeclaration)
elif type == CTFTypeId.VARIANT:
elif type == CTFTypeId.STRING:
declaration = StringFieldDeclaration.__new__(StringFieldDeclaration)
elif type == CTFTypeId.VARIANT:
declaration._fd = declaration_ptr
declaration._s = scope
declaration._name = name
declaration._fd = declaration_ptr
declaration._s = scope
declaration._name = name
return declaration
class _Definition:
return declaration
class _Definition:
def __init__(self, definition_ptr, scope):
self._d = definition_ptr
self._s = scope
def __init__(self, definition_ptr, scope):
self._d = definition_ptr
self._s = scope
- if not scope in _scopes:
+
+ if scope not in _scopes:
ValueError("Invalid scope provided")
@property
def name(self):
"""Return the name of a field or None on error."""
ValueError("Invalid scope provided")
@property
def name(self):
"""Return the name of a field or None on error."""
return nbt._bt_ctf_field_name(self._d)
@property
def type(self):
"""Return the type of a field or -1 if unknown."""
return nbt._bt_ctf_field_name(self._d)
@property
def type(self):
"""Return the type of a field or -1 if unknown."""
return nbt._bt_ctf_field_type(nbt._bt_ctf_get_decl_from_def(self._d))
@property
def declaration(self):
"""Return the associated Definition object."""
return nbt._bt_ctf_field_type(nbt._bt_ctf_get_decl_from_def(self._d))
@property
def declaration(self):
"""Return the associated Definition object."""
return _create_field_declaration(
nbt._bt_ctf_get_decl_from_def(self._d), self.name, self.scope)
return _create_field_declaration(
nbt._bt_ctf_get_decl_from_def(self._d), self.name, self.scope)
Return the string matching the current enumeration.
Return None on error.
"""
Return the string matching the current enumeration.
Return None on error.
"""
return nbt._bt_ctf_get_enum_str(self._d)
def _get_array_element_at(self, index):
return nbt._bt_ctf_get_enum_str(self._d)
def _get_array_element_at(self, index):
Return the array's element at position index.
Return None on error
"""
Return the array's element at position index.
Return None on error
"""
array_ptr = nbt._bt_python_get_array_from_def(self._d)
array_ptr = nbt._bt_python_get_array_from_def(self._d)
if array_ptr is None:
return None
definition_ptr = nbt._bt_array_index(array_ptr, index)
if array_ptr is None:
return None
definition_ptr = nbt._bt_array_index(array_ptr, index)
if definition_ptr is None:
return None
if definition_ptr is None:
return None
return _Definition(definition_ptr, self.scope)
def _get_sequence_len(self):
return _Definition(definition_ptr, self.scope)
def _get_sequence_len(self):
Return the len of a sequence or a negative
value on error.
"""
Return the len of a sequence or a negative
value on error.
"""
seq = nbt._bt_python_get_sequence_from_def(self._d)
seq = nbt._bt_python_get_sequence_from_def(self._d)
return nbt._bt_sequence_len(seq)
def _get_sequence_element_at(self, index):
return nbt._bt_sequence_len(seq)
def _get_sequence_element_at(self, index):
Return the sequence's element at position index,
otherwise return None
"""
Return the sequence's element at position index,
otherwise return None
"""
seq = nbt._bt_python_get_sequence_from_def(self._d)
seq = nbt._bt_python_get_sequence_from_def(self._d)
if seq is not None:
definition_ptr = nbt._bt_sequence_index(seq, index)
if seq is not None:
definition_ptr = nbt._bt_sequence_index(seq, index)
if definition_ptr is not None:
return _Definition(definition_ptr, self.scope)
if definition_ptr is not None:
return _Definition(definition_ptr, self.scope)
return None
def _get_uint64(self):
return None
def _get_uint64(self):
the value returned is undefined. To check if an error occured,
use the field_error() function after accessing a field.
"""
the value returned is undefined. To check if an error occured,
use the field_error() function after accessing a field.
"""
return nbt._bt_ctf_get_uint64(self._d)
def _get_int64(self):
return nbt._bt_ctf_get_uint64(self._d)
def _get_int64(self):
the value returned is undefined. To check if an error occured,
use the field_error() function after accessing a field.
"""
the value returned is undefined. To check if an error occured,
use the field_error() function after accessing a field.
"""
return nbt._bt_ctf_get_int64(self._d)
def _get_char_array(self):
return nbt._bt_ctf_get_int64(self._d)
def _get_char_array(self):
the value returned is undefined. To check if an error occurred,
use the field_error() function after accessing a field.
"""
the value returned is undefined. To check if an error occurred,
use the field_error() function after accessing a field.
"""
return nbt._bt_ctf_get_char_array(self._d)
def _get_str(self):
return nbt._bt_ctf_get_char_array(self._d)
def _get_str(self):
the value returned is undefined. To check if an error occurred,
use the field_error() function after accessing a field.
"""
the value returned is undefined. To check if an error occurred,
use the field_error() function after accessing a field.
"""
return nbt._bt_ctf_get_string(self._d)
def _get_float(self):
return nbt._bt_ctf_get_string(self._d)
def _get_float(self):
the value returned is undefined. To check if an error occurred,
use the field_error() function after accessing a field.
"""
the value returned is undefined. To check if an error occurred,
use the field_error() function after accessing a field.
"""
return nbt._bt_ctf_get_float(self._d)
def _get_variant(self):
return nbt._bt_ctf_get_float(self._d)
def _get_variant(self):
the value returned is undefined. To check if an error occurred,
use the field_error() function after accessing a field.
"""
the value returned is undefined. To check if an error occurred,
use the field_error() function after accessing a field.
"""
return nbt._bt_ctf_get_variant(self._d)
def _get_struct_field_count(self):
return nbt._bt_ctf_get_variant(self._d)
def _get_struct_field_count(self):
If the field does not exist or is not of the type requested,
the value returned is undefined.
"""
If the field does not exist or is not of the type requested,
the value returned is undefined.
"""
return nbt._bt_ctf_get_struct_field_count(self._d)
def _get_struct_field_at(self, i):
return nbt._bt_ctf_get_struct_field_count(self._d)
def _get_struct_field_at(self, i):
the value returned is undefined. To check if an error occurred,
use the field_error() function after accessing a field.
"""
the value returned is undefined. To check if an error occurred,
use the field_error() function after accessing a field.
"""
return nbt._bt_ctf_get_struct_field_index(self._d, i)
@property
return nbt._bt_ctf_get_struct_field_index(self._d, i)
@property
Return the value associated with the field according to its type.
Return None on error.
"""
Return the value associated with the field according to its type.
Return None on error.
"""
id = self.type
value = None
id = self.type
value = None
value = self._get_str()
elif id == CTFTypeId.ARRAY:
element_decl = self.declaration.element_declaration
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 = nbt._bt_python_get_array_string(self._d)
else:
value = []
if ((element_decl.type == CTFTypeId.INTEGER
and element_decl.length == 8)
and (element_decl.encoding == CTFStringEncoding.ASCII or element_decl.encoding == CTFStringEncoding.UTF8)):
value = nbt._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)
for i in range(self.declaration.length):
element = self._get_array_element_at(i)
value.append(element.value)
value = self._get_enum_str()
elif id == CTFTypeId.SEQUENCE:
element_decl = self.declaration.element_declaration
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)):
if ((element_decl.type == CTFTypeId.INTEGER
and element_decl.length == 8)
and (element_decl.encoding == CTFStringEncoding.ASCII or element_decl.encoding == CTFStringEncoding.UTF8)):
else:
seq_len = self._get_sequence_len()
value = []
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)
for i in range(seq_len):
evDef = self._get_sequence_element_at(i)
value.append(evDef.value)
value = variant.value
elif id == CTFTypeId.STRUCT:
value = {}
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
for i in range(self._get_struct_field_count()):
member = _Definition(self._get_struct_field_at(i), self.scope)
value[member.name] = member.value
"Error occurred while accessing field {} of type {}".format(
self.name,
CTFTypeId.type_name(id)))
"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 value
@property
def scope(self):
"""Return the scope of a field or None on error."""
# Used to compare to -1ULL in error checks
_MAX_UINT64 = 0xFFFFFFFFFFFFFFFF
# Used to compare to -1ULL in error checks
_MAX_UINT64 = 0xFFFFFFFFFFFFFFFF
- """
- Enumeration mapping class. start and end values are inclusive.
- """
class EnumerationMapping:
class EnumerationMapping:
+ """
+ Enumeration mapping class. start and end values are inclusive.
+ """
def __init__(self, name, start, end):
self.name = name
def __init__(self, name, start, end):
self.name = name
self.end = end
class Clock:
self.end = end
class Clock:
def __init__(self, name):
self._c = nbt._bt_ctf_clock_create(name)
def __init__(self, name):
self._c = nbt._bt_ctf_clock_create(name)
if self._c is None:
raise ValueError("Invalid clock name.")
def __del__(self):
nbt._bt_ctf_clock_put(self._c)
if self._c is None:
raise ValueError("Invalid clock name.")
def __del__(self):
nbt._bt_ctf_clock_put(self._c)
- """
- Get the clock's name.
- """
@property
def name(self):
@property
def name(self):
+ """
+ Get the clock's name.
+ """
+
name = nbt._bt_ctf_clock_get_name(self._c)
name = nbt._bt_ctf_clock_get_name(self._c)
if name is None:
raise ValueError("Invalid clock instance.")
if name is None:
raise ValueError("Invalid clock instance.")
- """
- Get the clock's description. None if unset.
- """
@property
def description(self):
@property
def description(self):
+ """
+ Get the clock's description. None if unset.
+ """
+
return nbt._bt_ctf_clock_get_description(self._c)
return nbt._bt_ctf_clock_get_description(self._c)
- """
- Set the clock's description. The description appears in the clock's TSDL
- meta-data.
- """
@description.setter
def description(self, desc):
@description.setter
def description(self, desc):
+ """
+ Set the clock's description. The description appears in the clock's TSDL
+ meta-data.
+ """
+
ret = nbt._bt_ctf_clock_set_description(self._c, str(desc))
ret = nbt._bt_ctf_clock_set_description(self._c, str(desc))
if ret < 0:
raise ValueError("Invalid clock description.")
if ret < 0:
raise ValueError("Invalid clock description.")
- """
- Get the clock's frequency (Hz).
- """
@property
def frequency(self):
@property
def frequency(self):
+ """
+ Get the clock's frequency (Hz).
+ """
+
freq = nbt._bt_ctf_clock_get_frequency(self._c)
freq = nbt._bt_ctf_clock_get_frequency(self._c)
if freq == CTFWriter._MAX_UINT64:
raise ValueError("Invalid clock instance")
if freq == CTFWriter._MAX_UINT64:
raise ValueError("Invalid clock instance")
- """
- Set the clock's frequency (Hz).
- """
@frequency.setter
def frequency(self, freq):
@frequency.setter
def frequency(self, freq):
+ """
+ Set the clock's frequency (Hz).
+ """
+
ret = nbt._bt_ctf_clock_set_frequency(self._c, freq)
ret = nbt._bt_ctf_clock_set_frequency(self._c, freq)
if ret < 0:
raise ValueError("Invalid frequency value.")
if ret < 0:
raise ValueError("Invalid frequency value.")
- """
- Get the clock's precision (in clock ticks).
- """
@property
def precision(self):
@property
def precision(self):
+ """
+ Get the clock's precision (in clock ticks).
+ """
+
precision = nbt._bt_ctf_clock_get_precision(self._c)
precision = nbt._bt_ctf_clock_get_precision(self._c)
if precision == CTFWriter._MAX_UINT64:
raise ValueError("Invalid clock instance")
if precision == CTFWriter._MAX_UINT64:
raise ValueError("Invalid clock instance")
- """
- Set the clock's precision (in clock ticks).
- """
@precision.setter
def precision(self, precision):
@precision.setter
def precision(self, precision):
+ """
+ Set the clock's precision (in clock ticks).
+ """
+
ret = nbt._bt_ctf_clock_set_precision(self._c, precision)
ret = nbt._bt_ctf_clock_set_precision(self._c, precision)
- """
- Get the clock's offset in seconds from POSIX.1 Epoch.
- """
@property
def offset_seconds(self):
@property
def offset_seconds(self):
+ """
+ Get the clock's offset in seconds from POSIX.1 Epoch.
+ """
+
offset_s = nbt._bt_ctf_clock_get_offset_s(self._c)
offset_s = nbt._bt_ctf_clock_get_offset_s(self._c)
if offset_s == CTFWriter._MAX_UINT64:
raise ValueError("Invalid clock instance")
if offset_s == CTFWriter._MAX_UINT64:
raise ValueError("Invalid clock instance")
- """
- Set the clock's offset in seconds from POSIX.1 Epoch.
- """
@offset_seconds.setter
def offset_seconds(self, offset_s):
@offset_seconds.setter
def offset_seconds(self, offset_s):
+ """
+ Set the clock's offset in seconds from POSIX.1 Epoch.
+ """
+
ret = nbt._bt_ctf_clock_set_offset_s(self._c, offset_s)
ret = nbt._bt_ctf_clock_set_offset_s(self._c, offset_s)
if ret < 0:
raise ValueError("Invalid offset value.")
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):
@property
def offset(self):
+ """
+ Get the clock's offset in ticks from POSIX.1 Epoch + offset in seconds.
+ """
+
offset = nbt._bt_ctf_clock_get_offset(self._c)
offset = nbt._bt_ctf_clock_get_offset(self._c)
if offset == CTFWriter._MAX_UINT64:
raise ValueError("Invalid clock instance")
if offset == CTFWriter._MAX_UINT64:
raise ValueError("Invalid clock instance")
- """
- Set the clock's offset in ticks from POSIX.1 Epoch + offset in seconds.
- """
@offset.setter
def offset(self, offset):
@offset.setter
def offset(self, offset):
+ """
+ Set the clock's offset in ticks from POSIX.1 Epoch + offset in seconds.
+ """
+
ret = nbt._bt_ctf_clock_set_offset(self._c, offset)
ret = nbt._bt_ctf_clock_set_offset(self._c, offset)
if ret < 0:
raise ValueError("Invalid offset value.")
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):
@property
def absolute(self):
+ """
+ Get a clock's absolute attribute. A clock is absolute if the clock
+ is a global reference across the trace's other clocks.
+ """
+
is_absolute = nbt._bt_ctf_clock_get_is_absolute(self._c)
is_absolute = nbt._bt_ctf_clock_get_is_absolute(self._c)
if is_absolute == -1:
raise ValueError("Invalid clock instance")
if is_absolute == -1:
raise ValueError("Invalid clock instance")
return False if is_absolute == 0 else True
return False if is_absolute == 0 else True
- """
- Set a clock's absolute attribute. A clock is absolute if the clock
- is a global reference across the trace's other clocks.
- """
@absolute.setter
def absolute(self, is_absolute):
@absolute.setter
def absolute(self, is_absolute):
+ """
+ Set a clock's absolute attribute. A clock is absolute if the clock
+ is a global reference across the trace's other clocks.
+ """
+
ret = nbt._bt_ctf_clock_set_is_absolute(self._c, int(is_absolute))
ret = nbt._bt_ctf_clock_set_is_absolute(self._c, int(is_absolute))
- raise ValueError(
- "Could not set the clock's absolute attribute.")
+ raise ValueError("Could not set the clock's absolute attribute.")
- """
- Get a clock's UUID (an object of type UUID).
- """
@property
def uuid(self):
@property
def uuid(self):
+ """
+ Get a clock's UUID (an object of type UUID).
+ """
+
for i in range(16):
ret, value = nbt._bt_python_ctf_clock_get_uuid_index(self._c, i)
for i in range(16):
ret, value = nbt._bt_python_ctf_clock_get_uuid_index(self._c, i)
if ret < 0:
raise ValueError("Invalid clock instance")
if ret < 0:
raise ValueError("Invalid clock instance")
return UUID(bytes=bytes(uuid_list))
return UUID(bytes=bytes(uuid_list))
- """
- Set a clock's UUID (an object of type UUID).
- """
@uuid.setter
def uuid(self, uuid):
@uuid.setter
def uuid(self, uuid):
+ """
+ Set a clock's UUID (an object of type UUID).
+ """
+
if len(uuid_bytes) != 16:
if len(uuid_bytes) != 16:
- raise ValueError(
- "Invalid UUID provided. UUID length must be 16 bytes")
+ raise ValueError("Invalid UUID provided. UUID length must be 16 bytes")
+
for i in range(len(uuid_bytes)):
for i in range(len(uuid_bytes)):
- ret = nbt._bt_python_ctf_clock_set_uuid_index(
- self._c,
- i,
- uuid_bytes[i])
+ ret = nbt._bt_python_ctf_clock_set_uuid_index(self._c, i,
+ uuid_bytes[i])
+
if ret < 0:
raise ValueError("Invalid clock instance")
if ret < 0:
raise ValueError("Invalid clock instance")
- """
- Get the current time in nanoseconds since the clock's origin (offset and
- offset_s attributes).
- """
@property
def time(self):
@property
def time(self):
+ """
+ Get the current time in nanoseconds since the clock's origin (offset and
+ offset_s attributes).
+ """
+
time = nbt._bt_ctf_clock_get_time(self._c)
time = nbt._bt_ctf_clock_get_time(self._c)
if time == CTFWriter._MAX_UINT64:
raise ValueError("Invalid clock instance")
if time == CTFWriter._MAX_UINT64:
raise ValueError("Invalid clock instance")
- """
- 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.
- """
@time.setter
def time(self, time):
@time.setter
def time(self, time):
+ """
+ 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.
+ """
+
ret = nbt._bt_ctf_clock_set_time(self._c, time)
ret = nbt._bt_ctf_clock_set_time(self._c, time)
if ret < 0:
raise ValueError("Invalid time value.")
class FieldDeclaration:
if ret < 0:
raise ValueError("Invalid time value.")
class FieldDeclaration:
"""
FieldDeclaration should not be instantiated directly. Instantiate
one of the concrete FieldDeclaration classes.
"""
"""
FieldDeclaration should not be instantiated directly. Instantiate
one of the concrete FieldDeclaration classes.
"""
class IntegerBase:
# These values are based on the bt_ctf_integer_base enum
# declared in event-types.h.
class IntegerBase:
# These values are based on the bt_ctf_integer_base enum
# declared in event-types.h.
CTFTypeId.SEQUENCE: CTFWriter.SequenceFieldDeclaration
}
CTFTypeId.SEQUENCE: CTFWriter.SequenceFieldDeclaration
}
- field_type_id = nbt._bt_ctf_field_type_get_type_id(
- native_field_declaration)
+ field_type_id = nbt._bt_ctf_field_type_get_type_id(native_field_declaration)
+
if field_type_id == CTFTypeId.UNKNOWN:
raise TypeError("Invalid field instance")
declaration = CTFWriter.Field.__new__(CTFWriter.Field)
declaration._ft = native_field_declaration
declaration.__class__ = type_dict[field_type_id]
if field_type_id == CTFTypeId.UNKNOWN:
raise TypeError("Invalid field instance")
declaration = CTFWriter.Field.__new__(CTFWriter.Field)
declaration._ft = native_field_declaration
declaration.__class__ = type_dict[field_type_id]
- """
- Get the field declaration's alignment. Returns -1 on error.
- """
@property
def alignment(self):
@property
def alignment(self):
+ """
+ Get the field declaration's alignment. Returns -1 on error.
+ """
+
return nbt._bt_ctf_field_type_get_alignment(self._ft)
return nbt._bt_ctf_field_type_get_alignment(self._ft)
- """
- Set the field declaration's alignment. Defaults to 1 (bit-aligned). However,
- some types, such as structures and string, may impose other alignment
- constraints.
- """
@alignment.setter
def alignment(self, alignment):
@alignment.setter
def alignment(self, alignment):
+ """
+ Set the field declaration's alignment. Defaults to 1 (bit-aligned). However,
+ some types, such as structures and string, may impose other alignment
+ constraints.
+ """
+
ret = nbt._bt_ctf_field_type_set_alignment(self._ft, alignment)
ret = nbt._bt_ctf_field_type_set_alignment(self._ft, alignment)
if ret < 0:
raise ValueError("Invalid alignment value.")
if ret < 0:
raise ValueError("Invalid alignment value.")
- """
- Get the field declaration's byte order. One of the ByteOrder's constant.
- """
@property
def byte_order(self):
@property
def byte_order(self):
+ """
+ Get the field declaration's byte order. One of the ByteOrder's constant.
+ """
+
return nbt._bt_ctf_field_type_get_byte_order(self._ft)
return nbt._bt_ctf_field_type_get_byte_order(self._ft)
- """
- Set the field declaration's byte order. Use constants defined in the ByteOrder
- class.
- """
@byte_order.setter
def byte_order(self, byte_order):
@byte_order.setter
def byte_order(self, byte_order):
+ """
+ Set the field declaration's byte order. Use constants defined in the ByteOrder
+ class.
+ """
+
ret = nbt._bt_ctf_field_type_set_byte_order(self._ft, byte_order)
ret = nbt._bt_ctf_field_type_set_byte_order(self._ft, byte_order)
if ret < 0:
raise ValueError("Could not set byte order value.")
class IntegerFieldDeclaration(FieldDeclaration):
if ret < 0:
raise ValueError("Could not set byte order value.")
class IntegerFieldDeclaration(FieldDeclaration):
-
- """
- Create a new integer field declaration of the given size.
- """
-
def __init__(self, size):
def __init__(self, size):
+ """
+ Create a new integer field declaration of the given size.
+ """
self._ft = nbt._bt_ctf_field_type_integer_create(size)
super().__init__()
self._ft = nbt._bt_ctf_field_type_integer_create(size)
super().__init__()
- """
- Get an integer's size.
- """
@property
def size(self):
@property
def size(self):
+ """
+ Get an integer's size.
+ """
+
ret = nbt._bt_ctf_field_type_integer_get_size(self._ft)
ret = nbt._bt_ctf_field_type_integer_get_size(self._ft)
if ret < 0:
raise ValueError("Could not get Integer's size attribute.")
else:
return ret
if ret < 0:
raise ValueError("Could not get Integer's size attribute.")
else:
return ret
- """
- Get an integer's signedness attribute.
- """
@property
def signed(self):
@property
def signed(self):
+ """
+ Get an integer's signedness attribute.
+ """
+
ret = nbt._bt_ctf_field_type_integer_get_signed(self._ft)
ret = nbt._bt_ctf_field_type_integer_get_signed(self._ft)
if ret < 0:
raise ValueError("Could not get Integer's signed attribute.")
elif ret > 0:
if ret < 0:
raise ValueError("Could not get Integer's signed attribute.")
elif ret > 0:
- """
- Set an integer's signedness attribute.
- """
@signed.setter
def signed(self, signed):
@signed.setter
def signed(self, signed):
+ """
+ Set an integer's signedness attribute.
+ """
+
ret = nbt._bt_ctf_field_type_integer_set_signed(self._ft, signed)
ret = nbt._bt_ctf_field_type_integer_set_signed(self._ft, signed)
if ret < 0:
raise ValueError("Could not set Integer's signed attribute.")
if ret < 0:
raise ValueError("Could not set Integer's signed attribute.")
- """
- Get the integer's base used to pretty-print the resulting trace.
- Returns a constant from the FieldDeclaration.IntegerBase class.
- """
@property
def base(self):
@property
def base(self):
+ """
+ Get the integer's base used to pretty-print the resulting trace.
+ Returns a constant from the FieldDeclaration.IntegerBase class.
+ """
+
return nbt._bt_ctf_field_type_integer_get_base(self._ft)
return nbt._bt_ctf_field_type_integer_get_base(self._ft)
- """
- Set the integer's base used to pretty-print the resulting trace.
- The base must be a constant of the FieldDeclarationIntegerBase class.
- """
@base.setter
def base(self, base):
@base.setter
def base(self, base):
+ """
+ Set the integer's base used to pretty-print the resulting trace.
+ The base must be a constant of the FieldDeclarationIntegerBase class.
+ """
+
ret = nbt._bt_ctf_field_type_integer_set_base(self._ft, base)
ret = nbt._bt_ctf_field_type_integer_set_base(self._ft, base)
if ret < 0:
raise ValueError("Could not set Integer's base.")
if ret < 0:
raise ValueError("Could not set Integer's base.")
- """
- Get the integer's encoding (one of the constants of the
- CTFStringEncoding class).
- Returns a constant from the CTFStringEncoding class.
- """
@property
def encoding(self):
@property
def encoding(self):
+ """
+ Get the integer's encoding (one of the constants of the
+ CTFStringEncoding class).
+ Returns a constant from the CTFStringEncoding class.
+ """
+
return nbt._bt_ctf_field_type_integer_get_encoding(self._ft)
return nbt._bt_ctf_field_type_integer_get_encoding(self._ft)
- """
- 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.
- """
@encoding.setter
def encoding(self, encoding):
@encoding.setter
def encoding(self, encoding):
+ """
+ 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.
+ """
+
ret = nbt._bt_ctf_field_type_integer_set_encoding(self._ft, encoding)
ret = nbt._bt_ctf_field_type_integer_set_encoding(self._ft, encoding)
if ret < 0:
raise ValueError("Could not set Integer's encoding.")
class EnumerationFieldDeclaration(FieldDeclaration):
if ret < 0:
raise ValueError("Could not set Integer's encoding.")
class EnumerationFieldDeclaration(FieldDeclaration):
-
- """
- Create a new enumeration field declaration with the given underlying container type.
- """
-
def __init__(self, integer_type):
def __init__(self, integer_type):
- if integer_type is None or not isinstance(
- integer_type, CTFWriter.IntegerFieldDeclaration):
+ """
+ Create a new enumeration field declaration with the given underlying container type.
+ """
+ isinst = isinstance(integer_type, CTFWriter.IntegerFieldDeclaration)
+
+ if integer_type is None or not isinst:
raise TypeError("Invalid integer container.")
self._ft = nbt._bt_ctf_field_type_enumeration_create(integer_type._ft)
super().__init__()
raise TypeError("Invalid integer container.")
self._ft = nbt._bt_ctf_field_type_enumeration_create(integer_type._ft)
super().__init__()
- """
- Get the enumeration's underlying container type.
- """
@property
def container(self):
@property
def container(self):
+ """
+ Get the enumeration's underlying container type.
+ """
+
ret = nbt._bt_ctf_field_type_enumeration_get_container_type(self._ft)
ret = nbt._bt_ctf_field_type_enumeration_get_container_type(self._ft)
if ret is None:
raise TypeError("Invalid enumeration declaration")
if ret is None:
raise TypeError("Invalid enumeration declaration")
- return CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(
- ret)
- """
- Add a mapping to the enumeration. The range's values are inclusive.
- """
+ return CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(ret)
def add_mapping(self, name, range_start, range_end):
def add_mapping(self, name, range_start, range_end):
+ """
+ Add a mapping to the enumeration. The range's values are inclusive.
+ """
+
if range_start < 0 or range_end < 0:
if range_start < 0 or range_end < 0:
- ret = nbt._bt_ctf_field_type_enumeration_add_mapping(
- self._ft,
- str(name),
- range_start,
- range_end)
+ ret = nbt._bt_ctf_field_type_enumeration_add_mapping(self._ft,
+ str(name),
+ range_start,
+ range_end)
- ret = nbt._bt_ctf_field_type_enumeration_add_mapping_unsigned(
- self._ft,
- str(name),
- range_start,
- range_end)
+ ret = nbt._bt_ctf_field_type_enumeration_add_mapping_unsigned(self._ft,
+ str(name),
+ range_start,
+ range_end)
- raise ValueError(
- "Could not add mapping to enumeration declaration.")
+ raise ValueError("Could not add mapping to enumeration declaration.")
- """
- Generator returning instances of EnumerationMapping.
- """
@property
def mappings(self):
@property
def mappings(self):
+ """
+ Generator returning instances of EnumerationMapping.
+ """
+
signed = self.container.signed
count = nbt._bt_ctf_field_type_enumeration_get_mapping_count(self._ft)
signed = self.container.signed
count = nbt._bt_ctf_field_type_enumeration_get_mapping_count(self._ft)
for i in range(count):
if signed:
for i in range(count):
if signed:
- ret = nbt._bt_python_ctf_field_type_enumeration_get_mapping(
- self._ft,
- i)
+ ret = nbt._bt_python_ctf_field_type_enumeration_get_mapping(self._ft, i)
- ret = nbt._bt_python_ctf_field_type_enumeration_get_mapping_unsigned(
- self._ft,
- i)
+ ret = nbt._bt_python_ctf_field_type_enumeration_get_mapping_unsigned(self._ft, i)
- raise TypeError(
- "Could not get Enumeration mapping at index {}".format(i))
+ msg = "Could not get Enumeration mapping at index {}".format(i)
+ raise TypeError(msg)
+
name, range_start, range_end = ret
yield CTFWriter.EnumerationMapping(name, range_start, range_end)
name, range_start, range_end = ret
yield CTFWriter.EnumerationMapping(name, range_start, range_end)
- """
- Get a mapping by name (EnumerationMapping).
- """
-
def get_mapping_by_name(self, name):
def get_mapping_by_name(self, name):
- index = nbt._bt_ctf_field_type_enumeration_get_mapping_index_by_name(
- self._ft, name)
+ """
+ Get a mapping by name (EnumerationMapping).
+ """
+
+ index = nbt._bt_ctf_field_type_enumeration_get_mapping_index_by_name(self._ft, name)
+
if index < 0:
return None
if self.container.signed:
if index < 0:
return None
if self.container.signed:
- ret = nbt._bt_python_ctf_field_type_enumeration_get_mapping(
- self._ft,
- index)
+ ret = nbt._bt_python_ctf_field_type_enumeration_get_mapping(self._ft, index)
- ret = nbt._bt_python_ctf_field_type_enumeration_get_mapping_unsigned(
- self._ft,
- index)
+ ret = nbt._bt_python_ctf_field_type_enumeration_get_mapping_unsigned(self._ft, index)
- raise TypeError(
- "Could not get Enumeration mapping at index {}".format(i))
+ msg = "Could not get Enumeration mapping at index {}".format(i)
+ raise TypeError(msg)
+
name, range_start, range_end = ret
name, range_start, range_end = ret
- return CTFWriter.EnumerationMapping(name, range_start, range_end)
- """
- Get a mapping by value (EnumerationMapping).
- """
+ return CTFWriter.EnumerationMapping(name, range_start, range_end)
def get_mapping_by_value(self, value):
def get_mapping_by_value(self, value):
+ """
+ Get a mapping by value (EnumerationMapping).
+ """
+
- index = nbt._bt_ctf_field_type_enumeration_get_mapping_index_by_value(
- self._ft, value)
+ index = nbt._bt_ctf_field_type_enumeration_get_mapping_index_by_value(self._ft, value)
- index = nbt._bt_ctf_field_type_enumeration_get_mapping_index_by_unsigned_value(
- self._ft, value)
+ index = nbt._bt_ctf_field_type_enumeration_get_mapping_index_by_unsigned_value(self._ft, value)
if index < 0:
return None
if self.container.signed:
if index < 0:
return None
if self.container.signed:
- ret = nbt._bt_python_ctf_field_type_enumeration_get_mapping(
- self._ft,
- index)
+ ret = nbt._bt_python_ctf_field_type_enumeration_get_mapping(self._ft, index)
- ret = nbt._bt_python_ctf_field_type_enumeration_get_mapping_unsigned(
- self._ft,
- index)
+ ret = nbt._bt_python_ctf_field_type_enumeration_get_mapping_unsigned(self._ft, index)
- raise TypeError(
- "Could not get Enumeration mapping at index {}".format(i))
+ msg = "Could not get Enumeration mapping at index {}".format(i)
+ raise TypeError(msg)
+
name, range_start, range_end = ret
name, range_start, range_end = ret
return CTFWriter.EnumerationMapping(name, range_start, range_end)
class FloatFieldDeclaration(FieldDeclaration):
return CTFWriter.EnumerationMapping(name, range_start, range_end)
class FloatFieldDeclaration(FieldDeclaration):
FLT_MANT_DIG = 24
DBL_MANT_DIG = 53
FLT_MANT_DIG = 24
DBL_MANT_DIG = 53
- """
- Create a new floating point field declaration.
- """
-
+ """
+ Create a new floating point field declaration.
+ """
+
self._ft = nbt._bt_ctf_field_type_floating_point_create()
super().__init__()
self._ft = nbt._bt_ctf_field_type_floating_point_create()
super().__init__()
- """
- Get the number of exponent digits used to store the floating point field.
- """
@property
def exponent_digits(self):
@property
def exponent_digits(self):
- ret = nbt._bt_ctf_field_type_floating_point_get_exponent_digits(
- self._ft)
+ """
+ Get the number of exponent digits used to store the floating point field.
+ """
+
+ ret = nbt._bt_ctf_field_type_floating_point_get_exponent_digits(self._ft)
+
if ret < 0:
raise TypeError(
"Could not get Floating point exponent digit count")
if ret < 0:
raise TypeError(
"Could not get Floating point exponent digit count")
- """
- 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.
- """
@exponent_digits.setter
def exponent_digits(self, exponent_digits):
@exponent_digits.setter
def exponent_digits(self, exponent_digits):
- ret = nbt._bt_ctf_field_type_floating_point_set_exponent_digits(
- self._ft,
- exponent_digits)
+ """
+ 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.
+ """
+
+ ret = nbt._bt_ctf_field_type_floating_point_set_exponent_digits(self._ft,
+ exponent_digits)
+
if ret < 0:
raise ValueError("Could not set exponent digit count.")
if ret < 0:
raise ValueError("Could not set exponent digit count.")
- """
- Get the number of mantissa digits used to store the floating point field.
- """
@property
def mantissa_digits(self):
@property
def mantissa_digits(self):
- ret = nbt._bt_ctf_field_type_floating_point_get_mantissa_digits(
- self._ft)
+ """
+ Get the number of mantissa digits used to store the floating point field.
+ """
+
+ ret = nbt._bt_ctf_field_type_floating_point_get_mantissa_digits(self._ft)
+
- raise TypeError(
- "Could not get Floating point mantissa digit count")
+ raise TypeError("Could not get Floating point mantissa digit count")
+
- """
- 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.
- """
@mantissa_digits.setter
def mantissa_digits(self, mantissa_digits):
@mantissa_digits.setter
def mantissa_digits(self, mantissa_digits):
- ret = nbt._bt_ctf_field_type_floating_point_set_mantissa_digits(
- self._ft,
- mantissa_digits)
+ """
+ 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.
+ """
+
+ ret = nbt._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 StructureFieldDeclaration(FieldDeclaration):
if ret < 0:
raise ValueError("Could not set mantissa digit count.")
class StructureFieldDeclaration(FieldDeclaration):
-
- """
- Create a new structure field declaration.
- """
-
+ """
+ Create a new structure field declaration.
+ """
+
self._ft = nbt._bt_ctf_field_type_structure_create()
super().__init__()
self._ft = nbt._bt_ctf_field_type_structure_create()
super().__init__()
- """
- Add a field of type "field_type" to the structure.
- """
-
def add_field(self, field_type, field_name):
def add_field(self, field_type, field_name):
- ret = nbt._bt_ctf_field_type_structure_add_field(
- self._ft,
- field_type._ft,
- str(field_name))
+ """
+ Add a field of type "field_type" to the structure.
+ """
+
+ ret = nbt._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.")
if ret < 0:
raise ValueError("Could not add field to structure.")
- """
- Generator returning the structure's field as tuples of (field name, field declaration).
- """
@property
def fields(self):
@property
def fields(self):
+ """
+ Generator returning the structure's field as tuples of (field name, field declaration).
+ """
+
count = nbt._bt_ctf_field_type_structure_get_field_count(self._ft)
count = nbt._bt_ctf_field_type_structure_get_field_count(self._ft)
if count < 0:
raise TypeError("Could not get Structure field count")
for i in range(count):
if count < 0:
raise TypeError("Could not get Structure field count")
for i in range(count):
- field_name = nbt._bt_python_ctf_field_type_structure_get_field_name(
- self._ft,
- i)
+ field_name = nbt._bt_python_ctf_field_type_structure_get_field_name(self._ft, i)
+
- raise TypeError(
- "Could not get Structure field name at index {}".format(i))
+ msg = "Could not get Structure field name at index {}".format(i)
+ raise TypeError(msg)
+
+ field_type_native = nbt._bt_python_ctf_field_type_structure_get_field_type(self._ft, i)
- field_type_native = nbt._bt_python_ctf_field_type_structure_get_field_type(
- self._ft,
- i)
if field_type_native is None:
if field_type_native is None:
- raise TypeError(
- "Could not get Structure field type at index {}".format(i))
+ msg = "Could not get Structure field type at index {}".format(i)
+ raise TypeError(msg)
- field_type = CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(
- field_type_native)
+ field_type = CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(field_type_native)
yield (field_name, field_type)
yield (field_name, field_type)
- """
- Get a field declaration by name (FieldDeclaration).
- """
-
def get_field_by_name(self, name):
def get_field_by_name(self, name):
- field_type_native = nbt._bt_ctf_field_type_structure_get_field_type_by_name(
- self._ft,
- name)
- if field_type_native is None:
- raise TypeError(
- "Could not find Structure field with name {}".format(name))
+ """
+ Get a field declaration by name (FieldDeclaration).
+ """
- return CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(
- field_type_native)
+ field_type_native = nbt._bt_ctf_field_type_structure_get_field_type_by_name(self._ft, name)
- class VariantFieldDeclaration(FieldDeclaration):
+ if field_type_native is None:
+ msg = "Could not find Structure field with name {}".format(name)
+ raise TypeError(msg)
- """
- Create a new variant field declaration.
- """
+ return CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(field_type_native)
+ class VariantFieldDeclaration(FieldDeclaration):
def __init__(self, enum_tag, tag_name):
def __init__(self, enum_tag, tag_name):
- if enum_tag is None or not isinstance(
- enum_tag, CTFWriter.EnumerationFieldDeclaration):
- raise TypeError(
- "Invalid tag type; must be of type EnumerationFieldDeclaration.")
+ """
+ Create a new variant field declaration.
+ """
+
+ isinst = isinstance(enum_tag, CTFWriter.EnumerationFieldDeclaration)
+ if enum_tag is None or not isinst:
+ raise TypeError("Invalid tag type; must be of type EnumerationFieldDeclaration.")
- self._ft = nbt._bt_ctf_field_type_variant_create(
- enum_tag._ft,
- str(tag_name))
+ self._ft = nbt._bt_ctf_field_type_variant_create(enum_tag._ft,
+ str(tag_name))
- """
- Get the variant's tag name.
- """
@property
def tag_name(self):
@property
def tag_name(self):
+ """
+ Get the variant's tag name.
+ """
+
ret = nbt._bt_ctf_field_type_variant_get_tag_name(self._ft)
ret = nbt._bt_ctf_field_type_variant_get_tag_name(self._ft)
if ret is None:
raise TypeError("Could not get Variant tag name")
if ret is None:
raise TypeError("Could not get Variant tag name")
- """
- Get the variant's tag type.
- """
@property
def tag_type(self):
@property
def tag_type(self):
+ """
+ Get the variant's tag type.
+ """
+
ret = nbt._bt_ctf_field_type_variant_get_tag_type(self._ft)
ret = nbt._bt_ctf_field_type_variant_get_tag_type(self._ft)
if ret is None:
raise TypeError("Could not get Variant tag type")
if ret is None:
raise TypeError("Could not get Variant tag type")
- return CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(
- ret)
- """
- Add a field of type "field_type" to the variant.
- """
+ return CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(ret)
def add_field(self, field_type, field_name):
def add_field(self, field_type, field_name):
- ret = nbt._bt_ctf_field_type_variant_add_field(
- self._ft,
- field_type._ft,
- str(field_name))
+ """
+ Add a field of type "field_type" to the variant.
+ """
+
+ ret = nbt._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.")
if ret < 0:
raise ValueError("Could not add field to variant.")
- """
- Generator returning the variant's field as tuples of (field name, field declaration).
- """
@property
def fields(self):
@property
def fields(self):
+ """
+ Generator returning the variant's field as tuples of (field name, field declaration).
+ """
+
count = nbt._bt_ctf_field_type_variant_get_field_count(self._ft)
count = nbt._bt_ctf_field_type_variant_get_field_count(self._ft)
if count < 0:
raise TypeError("Could not get Variant field count")
for i in range(count):
if count < 0:
raise TypeError("Could not get Variant field count")
for i in range(count):
- field_name = nbt._bt_python_ctf_field_type_variant_get_field_name(
- self._ft,
- i)
+ field_name = nbt._bt_python_ctf_field_type_variant_get_field_name(self._ft, i)
+
- raise TypeError(
- "Could not get Variant field name at index {}".format(i))
+ msg = "Could not get Variant field name at index {}".format(i)
+ raise TypeError(msg)
+
+ field_type_native = nbt._bt_python_ctf_field_type_variant_get_field_type(self._ft, i)
- field_type_native = nbt._bt_python_ctf_field_type_variant_get_field_type(
- self._ft,
- i)
if field_type_native is None:
if field_type_native is None:
- raise TypeError(
- "Could not get Variant field type at index {}".format(i))
+ msg = "Could not get Variant field type at index {}".format(i)
+ raise TypeError(msg)
- field_type = CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(
- field_type_native)
+ field_type = CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(field_type_native)
yield (field_name, field_type)
yield (field_name, field_type)
- """
- Get a field declaration by name (FieldDeclaration).
- """
-
def get_field_by_name(self, name):
def get_field_by_name(self, name):
- field_type_native = nbt._bt_ctf_field_type_variant_get_field_type_by_name(
- self._ft,
- name)
- if field_type_native is None:
- raise TypeError(
- "Could not find Variant field with name {}".format(name))
+ """
+ Get a field declaration by name (FieldDeclaration).
+ """
- return CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(
- field_type_native)
+ field_type_native = nbt._bt_ctf_field_type_variant_get_field_type_by_name(self._ft,
+ name)
- """
- Get a field declaration from tag (EnumerationField).
- """
+ if field_type_native is None:
+ msg = "Could not find Variant field with name {}".format(name)
+ raise TypeError(msg)
+
+ return CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(field_type_native)
def get_field_from_tag(self, tag):
def get_field_from_tag(self, tag):
- field_type_native = nbt._bt_ctf_field_type_variant_get_field_type_from_tag(
- self._ft,
- tag._f)
- if field_type_native is None:
- raise TypeError(
- "Could not find Variant field with tag value {}".format(
- tag.value))
+ """
+ Get a field declaration from tag (EnumerationField).
+ """
- return CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(
- field_type_native)
+ field_type_native = nbt._bt_ctf_field_type_variant_get_field_type_from_tag(self._ft, tag._f)
- class ArrayFieldDeclaration(FieldDeclaration):
+ if field_type_native is None:
+ msg = "Could not find Variant field with tag value {}".format(tag.value)
+ raise TypeError(msg)
- """
- Create a new array field declaration.
- """
+ return CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(field_type_native)
+ class ArrayFieldDeclaration(FieldDeclaration):
def __init__(self, element_type, length):
def __init__(self, element_type, length):
- self._ft = nbt._bt_ctf_field_type_array_create(
- element_type._ft,
- length)
+ """
+ Create a new array field declaration.
+ """
+
+ self._ft = nbt._bt_ctf_field_type_array_create(element_type._ft,
+ length)
- """
- Get the array's element type.
- """
@property
def element_type(self):
@property
def element_type(self):
+ """
+ Get the array's element type.
+ """
+
ret = nbt._bt_ctf_field_type_array_get_element_type(self._ft)
ret = nbt._bt_ctf_field_type_array_get_element_type(self._ft)
if ret is None:
raise TypeError("Could not get Array element type")
if ret is None:
raise TypeError("Could not get Array element type")
- return CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(
- ret)
- """
- Get the array's length.
- """
+ return CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(ret)
+
@property
def length(self):
@property
def length(self):
+ """
+ Get the array's length.
+ """
+
ret = nbt._bt_ctf_field_type_array_get_length(self._ft)
ret = nbt._bt_ctf_field_type_array_get_length(self._ft)
if ret < 0:
raise TypeError("Could not get Array length")
if ret < 0:
raise TypeError("Could not get Array length")
return ret
class SequenceFieldDeclaration(FieldDeclaration):
return ret
class SequenceFieldDeclaration(FieldDeclaration):
-
- """
- Create a new sequence field declaration.
- """
-
def __init__(self, element_type, length_field_name):
def __init__(self, element_type, length_field_name):
- self._ft = nbt._bt_ctf_field_type_sequence_create(
- element_type._ft,
- str(length_field_name))
+ """
+ Create a new sequence field declaration.
+ """
+
+ self._ft = nbt._bt_ctf_field_type_sequence_create(element_type._ft,
+ str(length_field_name))
- """
- Get the sequence's element type.
- """
@property
def element_type(self):
@property
def element_type(self):
+ """
+ Get the sequence's element type.
+ """
+
ret = nbt._bt_ctf_field_type_sequence_get_element_type(self._ft)
ret = nbt._bt_ctf_field_type_sequence_get_element_type(self._ft)
if ret is None:
raise TypeError("Could not get Sequence element type")
if ret is None:
raise TypeError("Could not get Sequence element type")
- return CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(
- ret)
- """
- Get the sequence's length field name.
- """
+ return CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(ret)
+
@property
def length_field_name(self):
@property
def length_field_name(self):
+ """
+ Get the sequence's length field name.
+ """
+
ret = nbt._bt_ctf_field_type_sequence_get_length_field_name(self._ft)
ret = nbt._bt_ctf_field_type_sequence_get_length_field_name(self._ft)
if ret is None:
raise TypeError("Could not get Sequence length field name")
if ret is None:
raise TypeError("Could not get Sequence length field name")
return ret
class StringFieldDeclaration(FieldDeclaration):
return ret
class StringFieldDeclaration(FieldDeclaration):
-
- """
- Create a new string field declaration.
- """
-
+ """
+ Create a new string field declaration.
+ """
+
self._ft = nbt._bt_ctf_field_type_string_create()
super().__init__()
self._ft = nbt._bt_ctf_field_type_string_create()
super().__init__()
- """
- Get a string declaration's encoding (a constant from the CTFStringEncoding class).
- """
@property
def encoding(self):
@property
def encoding(self):
+ """
+ Get a string declaration's encoding (a constant from the CTFStringEncoding class).
+ """
+
return nbt._bt_ctf_field_type_string_get_encoding(self._ft)
return nbt._bt_ctf_field_type_string_get_encoding(self._ft)
- """
- Set a string declaration's encoding. Must be a constant from the CTFStringEncoding class.
- """
@encoding.setter
def encoding(self, encoding):
@encoding.setter
def encoding(self, encoding):
+ """
+ Set a string declaration's encoding. Must be a constant from the CTFStringEncoding class.
+ """
+
ret = nbt._bt_ctf_field_type_string_set_encoding(self._ft, encoding)
if ret < 0:
raise ValueError("Could not set string encoding.")
ret = nbt._bt_ctf_field_type_string_set_encoding(self._ft, encoding)
if ret < 0:
raise ValueError("Could not set string encoding.")
- """
- Create an instance of a field.
- """
@staticmethod
def create_field(field_type):
@staticmethod
def create_field(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.")
+ """
+ Create an instance of a field.
+ """
+ isinst = isinstance(field_type, CTFWriter.FieldDeclaration)
+
+ if field_type is None or not isinst:
+ raise TypeError("Invalid field_type. Type must be a FieldDeclaration-derived class.")
if isinstance(field_type, CTFWriter.IntegerFieldDeclaration):
return CTFWriter.IntegerField(field_type)
if isinstance(field_type, CTFWriter.IntegerFieldDeclaration):
return CTFWriter.IntegerField(field_type)
return CTFWriter.StringField(field_type)
class Field:
return CTFWriter.StringField(field_type)
class Field:
"""
Base class, do not instantiate.
"""
"""
Base class, do not instantiate.
"""
raise TypeError("Invalid field_type argument.")
self._f = nbt._bt_ctf_field_create(field_type._ft)
raise TypeError("Invalid field_type argument.")
self._f = nbt._bt_ctf_field_create(field_type._ft)
if self._f is None:
raise ValueError("Field creation failed.")
if self._f is None:
raise ValueError("Field creation failed.")
}
field_type = nbt._bt_python_get_field_type(native_field_instance)
}
field_type = nbt._bt_python_get_field_type(native_field_instance)
if field_type == CTFTypeId.UNKNOWN:
raise TypeError("Invalid field instance")
field = CTFWriter.Field.__new__(CTFWriter.Field)
field._f = native_field_instance
field.__class__ = type_dict[field_type]
if field_type == CTFTypeId.UNKNOWN:
raise TypeError("Invalid field instance")
field = CTFWriter.Field.__new__(CTFWriter.Field)
field._f = native_field_instance
field.__class__ = type_dict[field_type]
return field
@property
def declaration(self):
native_field_type = nbt._bt_ctf_field_get_type(self._f)
return field
@property
def declaration(self):
native_field_type = nbt._bt_ctf_field_get_type(self._f)
if native_field_type is None:
raise TypeError("Invalid field instance")
return CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(
native_field_type)
class IntegerField(Field):
if native_field_type is None:
raise TypeError("Invalid field instance")
return CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(
native_field_type)
class IntegerField(Field):
-
- """
- Get an integer field's value.
- """
@property
def value(self):
@property
def value(self):
+ """
+ Get an integer field's value.
+ """
+
signedness = nbt._bt_python_field_integer_get_signedness(self._f)
signedness = nbt._bt_python_field_integer_get_signedness(self._f)
if signedness < 0:
raise TypeError("Invalid integer instance.")
if signedness < 0:
raise TypeError("Invalid integer instance.")
if ret < 0:
raise ValueError("Could not get integer field value.")
if ret < 0:
raise ValueError("Could not get integer field value.")
- """
- Set an integer field's value.
- """
@value.setter
def value(self, value):
@value.setter
def value(self, value):
+ """
+ Set an integer field's value.
+ """
+
if not isinstance(value, int):
raise TypeError("IntegerField's value must be an int")
if not isinstance(value, int):
raise TypeError("IntegerField's value must be an int")
raise ValueError("Could not set integer field value.")
class EnumerationField(Field):
raise ValueError("Could not set integer field value.")
class EnumerationField(Field):
-
- """
- Return the enumeration's underlying container field (an integer field).
- """
@property
def container(self):
@property
def container(self):
+ """
+ Return the enumeration's underlying container field (an integer field).
+ """
+
container = CTFWriter.IntegerField.__new__(CTFWriter.IntegerField)
container._f = nbt._bt_ctf_field_enumeration_get_container(self._f)
container = CTFWriter.IntegerField.__new__(CTFWriter.IntegerField)
container._f = nbt._bt_ctf_field_enumeration_get_container(self._f)
if container._f is None:
raise TypeError("Invalid enumeration field type.")
if container._f is None:
raise TypeError("Invalid enumeration field type.")
- """
- Get the enumeration field's mapping name.
- """
@property
def value(self):
@property
def value(self):
+ """
+ Get the enumeration field's mapping name.
+ """
+
value = nbt._bt_ctf_field_enumeration_get_mapping_name(self._f)
value = nbt._bt_ctf_field_enumeration_get_mapping_name(self._f)
if value is None:
raise ValueError("Could not get enumeration's mapping name.")
if value is None:
raise ValueError("Could not get enumeration's mapping name.")
- """
- Set the enumeration field's value. Must be an integer as mapping names
- may be ambiguous.
- """
@value.setter
def value(self, value):
@value.setter
def value(self, value):
+ """
+ Set the enumeration field's value. Must be an integer as mapping names
+ may be ambiguous.
+ """
+
if not isinstance(value, int):
raise TypeError("EnumerationField value must be an int")
if not isinstance(value, int):
raise TypeError("EnumerationField value must be an int")
self.container.value = value
class FloatFieldingPoint(Field):
self.container.value = value
class FloatFieldingPoint(Field):
-
- """
- Get a floating point field's value.
- """
@property
def value(self):
@property
def value(self):
+ """
+ Get a floating point field's value.
+ """
+
ret, value = nbt._bt_ctf_field_floating_point_get_value(self._f)
ret, value = nbt._bt_ctf_field_floating_point_get_value(self._f)
if ret < 0:
raise ValueError("Could not get floating point field value.")
if ret < 0:
raise ValueError("Could not get floating point field value.")
- """
- Set a floating point field's value.
- """
@value.setter
def value(self, value):
@value.setter
def value(self, value):
+ """
+ Set a floating point field's value.
+ """
+
if not isinstance(value, int) and not isinstance(value, float):
raise TypeError("Value must be either a float or an int")
ret = nbt._bt_ctf_field_floating_point_set_value(self._f, float(value))
if not isinstance(value, int) and not isinstance(value, float):
raise TypeError("Value must be either a float or an int")
ret = nbt._bt_ctf_field_floating_point_set_value(self._f, float(value))
if ret < 0:
raise ValueError("Could not set floating point field value.")
class StructureField(Field):
if ret < 0:
raise ValueError("Could not set floating point field value.")
class StructureField(Field):
+ def field(self, field_name):
+ """
+ Get the structure's field corresponding to the provided field name.
+ """
- """
- Get the structure's field corresponding to the provided field name.
- """
+ native_instance = nbt._bt_ctf_field_structure_get_field(self._f,
+ str(field_name))
- def field(self, field_name):
- native_instance = nbt._bt_ctf_field_structure_get_field(
- self._f,
- str(field_name))
if native_instance is None:
raise ValueError("Invalid field_name provided.")
if native_instance is None:
raise ValueError("Invalid field_name provided.")
- return CTFWriter.Field._create_field_from_native_instance(
- native_instance)
-
- class VariantField(Field):
- """
- Return the variant's selected field. The "tag" field is the selector enum field.
- """
+ return CTFWriter.Field._create_field_from_native_instance(native_instance)
+ class VariantField(Field):
+ """
+ Return the variant's selected field. The "tag" field is the selector enum field.
+ """
+
native_instance = nbt._bt_ctf_field_variant_get_field(self._f, tag._f)
native_instance = nbt._bt_ctf_field_variant_get_field(self._f, tag._f)
if native_instance is None:
raise ValueError("Invalid tag provided.")
if native_instance is None:
raise ValueError("Invalid tag provided.")
- return CTFWriter.Field._create_field_from_native_instance(
- native_instance)
- class ArrayField(Field):
-
- """
- Return the array's field at position "index".
- """
+ return CTFWriter.Field._create_field_from_native_instance(native_instance)
+ class ArrayField(Field):
+ """
+ Return the array's field at position "index".
+ """
+
native_instance = nbt._bt_ctf_field_array_get_field(self._f, index)
native_instance = nbt._bt_ctf_field_array_get_field(self._f, index)
if native_instance is None:
raise IndexError("Invalid index provided.")
if native_instance is None:
raise IndexError("Invalid index provided.")
- return CTFWriter.Field._create_field_from_native_instance(
- native_instance)
- class SequenceField(Field):
+ return CTFWriter.Field._create_field_from_native_instance(native_instance)
- """
- Get the sequence's length field (IntegerField).
- """
+ class SequenceField(Field):
@property
def length(self):
@property
def length(self):
+ """
+ Get the sequence's length field (IntegerField).
+ """
+
native_instance = nbt._bt_ctf_field_sequence_get_length(self._f)
native_instance = nbt._bt_ctf_field_sequence_get_length(self._f)
if native_instance is None:
length = -1
if native_instance is None:
length = -1
- return CTFWriter.Field._create_field_from_native_instance(
- native_instance)
- """
- Set the sequence's length field (IntegerField).
- """
+ return CTFWriter.Field._create_field_from_native_instance(native_instance)
+
@length.setter
def length(self, length_field):
@length.setter
def length(self, length_field):
+ """
+ Set the sequence's length field (IntegerField).
+ """
+
if not isinstance(length_field, CTFWriter.IntegerField):
raise TypeError("Invalid length field.")
if not isinstance(length_field, CTFWriter.IntegerField):
raise TypeError("Invalid length field.")
if length_field.declaration.signed:
raise TypeError("Sequence field length must be unsigned")
if length_field.declaration.signed:
raise TypeError("Sequence field length must be unsigned")
ret = nbt._bt_ctf_field_sequence_set_length(self._f, length_field._f)
ret = nbt._bt_ctf_field_sequence_set_length(self._f, length_field._f)
if ret < 0:
raise ValueError("Could not set sequence length.")
if ret < 0:
raise ValueError("Could not set sequence length.")
- """
- Return the sequence's field at position "index".
- """
-
+ """
+ Return the sequence's field at position "index".
+ """
+
native_instance = nbt._bt_ctf_field_sequence_get_field(self._f, index)
native_instance = nbt._bt_ctf_field_sequence_get_field(self._f, index)
if native_instance is None:
raise ValueError("Could not get sequence element at 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 StringField(Field):
+ return CTFWriter.Field._create_field_from_native_instance(native_instance)
- """
- Get a string field's value.
- """
+ class StringField(Field):
@property
def value(self):
@property
def value(self):
+ """
+ Get a string field's value.
+ """
+
return nbt._bt_ctf_field_string_get_value(self._f)
return nbt._bt_ctf_field_string_get_value(self._f)
- """
- Set a string field's value.
- """
@value.setter
def value(self, value):
@value.setter
def value(self, value):
+ """
+ Set a string field's value.
+ """
+
ret = nbt._bt_ctf_field_string_set_value(self._f, str(value))
ret = nbt._bt_ctf_field_string_set_value(self._f, str(value))
if ret < 0:
raise ValueError("Could not set string field value.")
class EventClass:
if ret < 0:
raise ValueError("Could not set string field value.")
class EventClass:
-
- """
- Create a new event class of the given name.
- """
-
def __init__(self, name):
def __init__(self, name):
+ """
+ Create a new event class of the given name.
+ """
+
self._ec = nbt._bt_ctf_event_class_create(name)
self._ec = nbt._bt_ctf_event_class_create(name)
if self._ec is None:
raise ValueError("Event class creation failed.")
def __del__(self):
nbt._bt_ctf_event_class_put(self._ec)
if self._ec is None:
raise ValueError("Event class creation failed.")
def __del__(self):
nbt._bt_ctf_event_class_put(self._ec)
- """
- Add a field of type "field_type" to the event class.
- """
-
def add_field(self, field_type, field_name):
def add_field(self, field_type, field_name):
- ret = nbt._bt_ctf_event_class_add_field(
- self._ec,
- field_type._ft,
- str(field_name))
+ """
+ Add a field of type "field_type" to the event class.
+ """
+
+ ret = nbt._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.")
if ret < 0:
raise ValueError("Could not add field to event class.")
- """
- Get the event class' name.
- """
@property
def name(self):
@property
def name(self):
+ """
+ Get the event class' name.
+ """
+
name = nbt._bt_ctf_event_class_get_name(self._ec)
name = nbt._bt_ctf_event_class_get_name(self._ec)
if name is None:
raise TypeError("Could not get EventClass name")
if name is None:
raise TypeError("Could not get EventClass name")
- """
- Get the event class' id. Returns a negative value if unset.
- """
+ """
+ Get the event class' id. Returns a negative value if unset.
+ """
+
id = nbt._bt_ctf_event_class_get_id(self._ec)
id = nbt._bt_ctf_event_class_get_id(self._ec)
if id < 0:
raise TypeError("Could not get EventClass id")
if id < 0:
raise TypeError("Could not get EventClass id")
- """
- Set the event class' id. Throws a TypeError if the event class
- is already registered to a stream class.
- """
@id.setter
def id(self, id):
@id.setter
def id(self, id):
+ """
+ Set the event class' id. Throws a TypeError if the event class
+ is already registered to a stream class.
+ """
+
ret = nbt._bt_ctf_event_class_set_id(self._ec, id)
ret = nbt._bt_ctf_event_class_set_id(self._ec, id)
- raise TypeError(
- "Can't change an Event Class's id after it has been assigned to a stream class")
+ raise TypeError("Can't change an Event Class's id after it has been assigned to a stream class")
- """
- Get the event class' stream class. Returns None if unset.
- """
@property
def stream_class(self):
@property
def stream_class(self):
- stream_class_native = nbt._bt_ctf_event_class_get_stream_class(
- self._ec)
+ """
+ Get the event class' stream class. Returns None if unset.
+ """
+ stream_class_native = nbt._bt_ctf_event_class_get_stream_class(self._ec)
+
if stream_class_native is None:
return None
if stream_class_native is None:
return None
stream_class = CTFWriter.StreamClass.__new__(CTFWriter.StreamClass)
stream_class._sc = stream_class_native
stream_class = CTFWriter.StreamClass.__new__(CTFWriter.StreamClass)
stream_class._sc = stream_class_native
- """
- Generator returning the event class' fields as tuples of (field name, field declaration).
- """
@property
def fields(self):
@property
def fields(self):
+ """
+ Generator returning the event class' fields as tuples of (field name, field declaration).
+ """
+
count = nbt._bt_ctf_event_class_get_field_count(self._ec)
count = nbt._bt_ctf_event_class_get_field_count(self._ec)
if count < 0:
raise TypeError("Could not get EventClass' field count")
for i in range(count):
if count < 0:
raise TypeError("Could not get EventClass' field count")
for i in range(count):
- field_name = nbt._bt_python_ctf_event_class_get_field_name(
- self._ec,
- i)
+ field_name = nbt._bt_python_ctf_event_class_get_field_name(self._ec, i)
+
- raise TypeError(
- "Could not get EventClass' field name at index {}".format(i))
+ msg = "Could not get EventClass' field name at index {}".format(i)
+ raise TypeError(msg)
+
+ field_type_native = nbt._bt_python_ctf_event_class_get_field_type(self._ec, i)
- field_type_native = nbt._bt_python_ctf_event_class_get_field_type(
- self._ec,
- i)
if field_type_native is None:
if field_type_native is None:
- raise TypeError(
- "Could not get EventClass' field type at index {}".format(i))
+ msg = "Could not get EventClass' field type at index {}".format(i)
+ raise TypeError(msg)
- field_type = CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(
- field_type_native)
+ field_type = CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(field_type_native)
yield (field_name, field_type)
yield (field_name, field_type)
- """
- Get a field declaration by name (FieldDeclaration).
- """
-
def get_field_by_name(self, name):
def get_field_by_name(self, name):
- field_type_native = nbt._bt_ctf_event_class_get_field_by_name(
- self._ec,
- name)
- if field_type_native is None:
- raise TypeError(
- "Could not find EventClass field with name {}".format(name))
- return CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(
- field_type_native)
+ """
+ Get a field declaration by name (FieldDeclaration).
+ """
+ field_type_native = nbt._bt_ctf_event_class_get_field_by_name(self._ec, name)
- """
- Create a new event of the given event class.
- """
+ if field_type_native is None:
+ msg = "Could not find EventClass field with name {}".format(name)
+ raise TypeError(msg)
+ return CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(field_type_native)
+
+ class Event:
def __init__(self, event_class):
def __init__(self, event_class):
+ """
+ Create a new event of the given event class.
+ """
+
if not isinstance(event_class, CTFWriter.EventClass):
raise TypeError("Invalid event_class argument.")
self._e = nbt._bt_ctf_event_create(event_class._ec)
if not isinstance(event_class, CTFWriter.EventClass):
raise TypeError("Invalid event_class argument.")
self._e = nbt._bt_ctf_event_create(event_class._ec)
if self._e is None:
raise ValueError("Event creation failed.")
def __del__(self):
nbt._bt_ctf_event_put(self._e)
if self._e is None:
raise ValueError("Event creation failed.")
def __del__(self):
nbt._bt_ctf_event_put(self._e)
- """
- Get the event's class.
- """
@property
def event_class(self):
@property
def event_class(self):
+ """
+ Get the event's class.
+ """
+
event_class_native = nbt._bt_ctf_event_get_class(self._e)
event_class_native = nbt._bt_ctf_event_get_class(self._e)
if event_class_native is None:
return None
if event_class_native is None:
return None
event_class = CTFWriter.EventClass.__new__(CTFWriter.EventClass)
event_class._ec = event_class_native
event_class = CTFWriter.EventClass.__new__(CTFWriter.EventClass)
event_class._ec = event_class_native
- """
- Get a clock from event. Returns None if the event's class
- is not registered to a stream class.
- """
+ """
+ Get a clock from event. Returns None if the event's class
+ is not registered to a stream class.
+ """
+
clock_instance = nbt._bt_ctf_event_get_clock(self._e)
clock_instance = nbt._bt_ctf_event_get_clock(self._e)
if clock_instance is None:
return None
if clock_instance is None:
return None
clock = CTFWriter.Clock.__new__(CTFWriter.Clock)
clock._c = clock_instance
clock = CTFWriter.Clock.__new__(CTFWriter.Clock)
clock._c = clock_instance
- """
- Get a field from event.
- """
def payload(self, field_name):
def payload(self, field_name):
- native_instance = nbt._bt_ctf_event_get_payload(
- self._e,
- str(field_name))
+ """
+ Get a field from event.
+ """
+
+ native_instance = nbt._bt_ctf_event_get_payload(self._e,
+ str(field_name))
+
if native_instance is None:
raise ValueError("Could not get event payload.")
if native_instance is None:
raise ValueError("Could not get event payload.")
- return CTFWriter.Field._create_field_from_native_instance(
- native_instance)
- """
- Set a manually created field as an event's payload.
- """
+ return CTFWriter.Field._create_field_from_native_instance(native_instance)
def set_payload(self, field_name, value_field):
def set_payload(self, field_name, value_field):
+ """
+ Set a manually created field as an event's payload.
+ """
+
if not isinstance(value, CTFWriter.Field):
raise TypeError("Invalid value type.")
if not isinstance(value, CTFWriter.Field):
raise TypeError("Invalid value type.")
- ret = nbt._bt_ctf_event_set_payload(
- self._e,
- str(field_name),
- value_field._f)
+
+ ret = nbt._bt_ctf_event_set_payload(self._e, str(field_name),
+ value_field._f)
+
if ret < 0:
raise ValueError("Could not set event field payload.")
class StreamClass:
if ret < 0:
raise ValueError("Could not set event field payload.")
class StreamClass:
-
- """
- Create a new stream class of the given name.
- """
-
def __init__(self, name):
def __init__(self, name):
+ """
+ Create a new stream class of the given name.
+ """
+
self._sc = nbt._bt_ctf_stream_class_create(name)
self._sc = nbt._bt_ctf_stream_class_create(name)
if self._sc is None:
raise ValueError("Stream class creation failed.")
def __del__(self):
nbt._bt_ctf_stream_class_put(self._sc)
if self._sc is None:
raise ValueError("Stream class creation failed.")
def __del__(self):
nbt._bt_ctf_stream_class_put(self._sc)
- """
- Get a stream class' name.
- """
@property
def name(self):
@property
def name(self):
+ """
+ Get a stream class' name.
+ """
+
name = nbt._bt_ctf_stream_class_get_name(self._sc)
name = nbt._bt_ctf_stream_class_get_name(self._sc)
if name is None:
raise TypeError("Could not get StreamClass name")
if name is None:
raise TypeError("Could not get StreamClass name")
- """
- Get a stream class' clock.
- """
@property
def clock(self):
@property
def clock(self):
+ """
+ Get a stream class' clock.
+ """
+
clock_instance = nbt._bt_ctf_stream_class_get_clock(self._sc)
clock_instance = nbt._bt_ctf_stream_class_get_clock(self._sc)
if clock_instance is None:
return None
if clock_instance is None:
return None
clock = CTFWriter.Clock.__new__(CTFWriter.Clock)
clock._c = clock_instance
clock = CTFWriter.Clock.__new__(CTFWriter.Clock)
clock._c = clock_instance
- """
- Assign a clock to a stream class.
- """
@clock.setter
def clock(self, clock):
@clock.setter
def clock(self, clock):
+ """
+ Assign a clock to a stream class.
+ """
+
if not isinstance(clock, CTFWriter.Clock):
raise TypeError("Invalid clock type.")
ret = nbt._bt_ctf_stream_class_set_clock(self._sc, clock._c)
if not isinstance(clock, CTFWriter.Clock):
raise TypeError("Invalid clock type.")
ret = nbt._bt_ctf_stream_class_set_clock(self._sc, clock._c)
if ret < 0:
raise ValueError("Could not set stream class clock.")
if ret < 0:
raise ValueError("Could not set stream class clock.")
- """
- Get a stream class' id.
- """
+ """
+ Get a stream class' id.
+ """
+
ret = nbt._bt_ctf_stream_class_get_id(self._sc)
ret = nbt._bt_ctf_stream_class_get_id(self._sc)
if ret < 0:
raise TypeError("Could not get StreamClass id")
if ret < 0:
raise TypeError("Could not get StreamClass id")
- """
- Assign an id to a stream class.
- """
@id.setter
def id(self, id):
@id.setter
def id(self, id):
+ """
+ Assign an id to a stream class.
+ """
+
ret = nbt._bt_ctf_stream_class_set_id(self._sc, id)
ret = nbt._bt_ctf_stream_class_set_id(self._sc, id)
if ret < 0:
raise TypeError("Could not set stream class id.")
if ret < 0:
raise TypeError("Could not set stream class id.")
- """
- Generator returning the stream class' event classes.
- """
@property
def event_classes(self):
@property
def event_classes(self):
+ """
+ Generator returning the stream class' event classes.
+ """
+
count = nbt._bt_ctf_stream_class_get_event_class_count(self._sc)
count = nbt._bt_ctf_stream_class_get_event_class_count(self._sc)
if count < 0:
raise TypeError("Could not get StreamClass' event class count")
for i in range(count):
if count < 0:
raise TypeError("Could not get StreamClass' event class count")
for i in range(count):
- event_class_native = nbt._bt_ctf_stream_class_get_event_class(
- self._sc,
- i)
+ event_class_native = nbt._bt_ctf_stream_class_get_event_class(self._sc, i)
+
if event_class_native is None:
if event_class_native is None:
- raise TypeError(
- "Could not get StreamClass' event class at index {}".format(i))
+ msg = "Could not get StreamClass' event class at index {}".format(i)
+ raise TypeError(msg)
- event_class = CTFWriter.EventClass.__new__(
- CTFWriter.EventClass)
+ event_class = CTFWriter.EventClass.__new__(CTFWriter.EventClass)
event_class._ec = event_class_native
yield event_class
event_class._ec = event_class_native
yield event_class
- """
- Add an event class to a stream class. New events can be added even after a
- stream has been instantiated and events have been appended. However, a stream
- will not accept events of a class that has not been added to the stream
- class beforehand.
- """
-
def add_event_class(self, event_class):
def add_event_class(self, event_class):
+ """
+ Add an event class to a stream class. New events can be added even after a
+ stream has been instantiated and events have been appended. However, a stream
+ will not accept events of a class that has not been added to the stream
+ class beforehand.
+ """
+
if not isinstance(event_class, CTFWriter.EventClass):
raise TypeError("Invalid event_class type.")
if not isinstance(event_class, CTFWriter.EventClass):
raise TypeError("Invalid event_class type.")
- ret = nbt._bt_ctf_stream_class_add_event_class(
- self._sc,
- event_class._ec)
+ ret = nbt._bt_ctf_stream_class_add_event_class(self._sc,
+ event_class._ec)
+
if ret < 0:
raise ValueError("Could not add event class.")
if ret < 0:
raise ValueError("Could not add event class.")
- """
- Get the StreamClass' packet context type (StructureFieldDeclaration)
- """
@property
def packet_context_type(self):
@property
def packet_context_type(self):
- field_type_native = nbt._bt_ctf_stream_class_get_packet_context_type(
- self._sc)
+ """
+ Get the StreamClass' packet context type (StructureFieldDeclaration)
+ """
+
+ field_type_native = nbt._bt_ctf_stream_class_get_packet_context_type(self._sc)
+
if field_type_native is None:
raise ValueError("Invalid StreamClass")
if field_type_native is None:
raise ValueError("Invalid StreamClass")
- field_type = CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(
- field_type_native)
+
+ field_type = CTFWriter.FieldDeclaration._create_field_declaration_from_native_instance(field_type_native)
+
- """
- Set a StreamClass' packet context type. Must be of type
- StructureFieldDeclaration.
- """
@packet_context_type.setter
def packet_context_type(self, field_type):
@packet_context_type.setter
def packet_context_type(self, field_type):
+ """
+ Set a StreamClass' packet context type. Must be of type
+ StructureFieldDeclaration.
+ """
+
if not isinstance(field_type, CTFWriter.StructureFieldDeclaration):
if not isinstance(field_type, CTFWriter.StructureFieldDeclaration):
- raise TypeError(
- "field_type argument must be of type StructureFieldDeclaration.")
- ret = nbt._bt_ctf_stream_class_set_packet_context_type(
- self._sc,
- field_type._ft)
+ raise TypeError("field_type argument must be of type StructureFieldDeclaration.")
+
+ ret = nbt._bt_ctf_stream_class_set_packet_context_type(self._sc,
+ field_type._ft)
+
if ret < 0:
raise ValueError("Failed to set packet context type.")
class Stream:
if ret < 0:
raise ValueError("Failed to set packet context type.")
class Stream:
-
- """
- Create a stream of the given class.
- """
-
def __init__(self, stream_class):
def __init__(self, stream_class):
+ """
+ Create a stream of the given class.
+ """
+
if not isinstance(stream_class, CTFWriter.StreamClass):
if not isinstance(stream_class, CTFWriter.StreamClass):
- raise TypeError(
- "Invalid stream_class argument must be of type StreamClass.")
+ raise TypeError("Invalid stream_class argument must be of type StreamClass.")
self._s = nbt._bt_ctf_stream_create(stream_class._sc)
self._s = nbt._bt_ctf_stream_create(stream_class._sc)
if self._s is None:
raise ValueError("Stream creation failed.")
def __del__(self):
nbt._bt_ctf_stream_put(self._s)
if self._s is None:
raise ValueError("Stream creation failed.")
def __del__(self):
nbt._bt_ctf_stream_put(self._s)
- """
- Get a stream's discarded event count.
- """
@property
def discarded_events(self):
@property
def discarded_events(self):
+ """
+ Get a stream's discarded event count.
+ """
+
ret, count = nbt._bt_ctf_stream_get_discarded_events_count(self._s)
ret, count = nbt._bt_ctf_stream_get_discarded_events_count(self._s)
- raise ValueError(
- "Could not get the stream's discarded events count")
- return count
+ raise ValueError("Could not get the stream's discarded events count")
- """
- Increase the current packet's discarded event count.
- """
def append_discarded_events(self, event_count):
def append_discarded_events(self, event_count):
- nbt._bt_ctf_stream_append_discarded_events(self._s, event_count)
+ """
+ Increase the current packet's discarded event count.
+ """
- """
- Append "event" to the stream's current packet. The stream's associated clock
- will be sampled during this call. The event shall not be modified after
- being appended to a stream.
- """
+ nbt._bt_ctf_stream_append_discarded_events(self._s, event_count)
def append_event(self, event):
def append_event(self, event):
+ """
+ Append "event" to the stream's current packet. The stream's associated clock
+ will be sampled during this call. The event shall not be modified after
+ being appended to a stream.
+ """
+
ret = nbt._bt_ctf_stream_append_event(self._s, event._e)
ret = nbt._bt_ctf_stream_append_event(self._s, event._e)
if ret < 0:
raise ValueError("Could not append event to stream.")
if ret < 0:
raise ValueError("Could not append event to stream.")
- """
- Get a Stream's packet context field (a StructureField).
- """
@property
def packet_context(self):
@property
def packet_context(self):
+ """
+ Get a Stream's packet context field (a StructureField).
+ """
+
native_field = nbt._bt_ctf_stream_get_packet_context(self._s)
native_field = nbt._bt_ctf_stream_get_packet_context(self._s)
if native_field is None:
raise ValueError("Invalid Stream.")
if native_field is None:
raise ValueError("Invalid Stream.")
- return CTFWriter.Field._create_field_from_native_instance(
- native_field)
- """
- Set a Stream's packet context field (must be a StructureField).
- """
+ return CTFWriter.Field._create_field_from_native_instance(native_field)
+
@packet_context.setter
def packet_context(self, field):
@packet_context.setter
def packet_context(self, field):
+ """
+ Set a Stream's packet context field (must be a StructureField).
+ """
+
if not isinstance(field, CTFWriter.StructureField):
if not isinstance(field, CTFWriter.StructureField):
- raise TypeError(
- "Argument field must be of type StructureField")
+ raise TypeError("Argument field must be of type StructureField")
+
ret = nbt._bt_ctf_stream_set_packet_context(self._s, field._f)
ret = nbt._bt_ctf_stream_set_packet_context(self._s, field._f)
if ret < 0:
raise ValueError("Invalid packet context field.")
if ret < 0:
raise ValueError("Invalid packet context field.")
- """
- The stream's current packet's events will be flushed to disk. Events
- subsequently appended to the stream will be added to a new packet.
- """
-
+ """
+ The stream's current packet's events will be flushed to disk. Events
+ subsequently appended to the stream will be added to a new packet.
+ """
+
ret = nbt._bt_ctf_stream_flush(self._s)
ret = nbt._bt_ctf_stream_flush(self._s)
if ret < 0:
raise ValueError("Could not flush stream.")
class Writer:
if ret < 0:
raise ValueError("Could not flush stream.")
class Writer:
-
- """
- Create a new writer that will produce a trace in the given path.
- """
-
def __init__(self, path):
def __init__(self, path):
+ """
+ Create a new writer that will produce a trace in the given path.
+ """
+
self._w = nbt._bt_ctf_writer_create(path)
self._w = nbt._bt_ctf_writer_create(path)
if self._w is None:
raise ValueError("Writer creation failed.")
def __del__(self):
nbt._bt_ctf_writer_put(self._w)
if self._w is None:
raise ValueError("Writer creation failed.")
def __del__(self):
nbt._bt_ctf_writer_put(self._w)
- """
- Create a new stream instance and register it to the writer.
- """
-
def create_stream(self, stream_class):
def create_stream(self, stream_class):
+ """
+ Create a new stream instance and register it to the writer.
+ """
+
if not isinstance(stream_class, CTFWriter.StreamClass):
raise TypeError("Invalid stream_class type.")
stream = CTFWriter.Stream.__new__(CTFWriter.Stream)
stream._s = nbt._bt_ctf_writer_create_stream(self._w, stream_class._sc)
if not isinstance(stream_class, CTFWriter.StreamClass):
raise TypeError("Invalid stream_class type.")
stream = CTFWriter.Stream.__new__(CTFWriter.Stream)
stream._s = nbt._bt_ctf_writer_create_stream(self._w, stream_class._sc)
- """
- Add an environment field to the trace.
- """
def add_environment_field(self, name, value):
def add_environment_field(self, name, value):
- ret = nbt._bt_ctf_writer_add_environment_field(
- self._w,
- str(name),
- str(value))
+ """
+ Add an environment field to the trace.
+ """
+
+ ret = nbt._bt_ctf_writer_add_environment_field(self._w, str(name),
+ str(value))
+
if ret < 0:
raise ValueError("Could not add environment field to trace.")
if ret < 0:
raise ValueError("Could not add environment field to trace.")
- """
- Add a clock to the trace. Clocks assigned to stream classes must be
- registered to the writer.
- """
-
def add_clock(self, clock):
def add_clock(self, clock):
+ """
+ Add a clock to the trace. Clocks assigned to stream classes must be
+ registered to the writer.
+ """
+
ret = nbt._bt_ctf_writer_add_clock(self._w, clock._c)
ret = nbt._bt_ctf_writer_add_clock(self._w, clock._c)
if ret < 0:
raise ValueError("Could not add clock to Writer.")
if ret < 0:
raise ValueError("Could not add clock to Writer.")
- """
- Get the trace's TSDL meta-data.
- """
@property
def metadata(self):
@property
def metadata(self):
- return nbt._bt_ctf_writer_get_metadata_string(self._w)
+ """
+ Get the trace's TSDL meta-data.
+ """
- """
- Flush the trace's metadata to the metadata file.
- """
+ return nbt._bt_ctf_writer_get_metadata_string(self._w)
def flush_metadata(self):
def flush_metadata(self):
+ """
+ Flush the trace's metadata to the metadata file.
+ """
+
nbt._bt_ctf_writer_flush_metadata(self._w)
nbt._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):
@property
def byte_order(self):
+ """
+ Get the trace's byte order. Must be a constant from the ByteOrder
+ class.
+ """
+
raise NotImplementedError("Getter not implemented.")
raise NotImplementedError("Getter not implemented.")
- """
- Set the trace's byte order. Must be a constant from the ByteOrder
- class. Defaults to the host machine's endianness
- """
@byte_order.setter
def byte_order(self, byte_order):
@byte_order.setter
def byte_order(self, byte_order):
+ """
+ Set the trace's byte order. Must be a constant from the ByteOrder
+ class. Defaults to the host machine's endianness
+ """
+
ret = nbt._bt_ctf_writer_set_byte_order(self._w, byte_order)
ret = nbt._bt_ctf_writer_set_byte_order(self._w, byte_order)
if ret < 0:
raise ValueError("Could not set trace's byte order.")
if ret < 0:
raise ValueError("Could not set trace's byte order.")
projects / babeltrace.git / commitdiff
