import bt2
from bt2 import native_bt, utils, object
import bt2.stream_class
-import uuid as uuidp
import collections.abc
import functools
-class _TraceClassEnv(collections.abc.MutableMapping):
- def __init__(self, trace_class):
- self._trace_class = trace_class
-
- def __getitem__(self, key):
- utils._check_str(key)
-
- borrow_entry_fn = native_bt.trace_class_borrow_environment_entry_value_by_name_const
- value_ptr = borrow_entry_fn(self._trace_class._ptr, key)
-
- if value_ptr is None:
- raise KeyError(key)
-
- return bt2.value._create_from_ptr_and_get_ref(value_ptr)
-
- def __setitem__(self, key, value):
- if isinstance(value, str):
- set_env_entry_fn = native_bt.trace_class_set_environment_entry_string
- elif isinstance(value, int):
- set_env_entry_fn = native_bt.trace_class_set_environment_entry_integer
- else:
- raise TypeError('expected str or int, got {}'.format(type(value)))
-
- ret = set_env_entry_fn(self._trace_class._ptr, key, value)
-
- utils._handle_ret(ret, "cannot set trace class object's environment entry")
-
- def __delitem__(self, key):
- raise NotImplementedError
-
- def __len__(self):
- count = native_bt.trace_class_get_environment_entry_count(self._trace_class._ptr)
- assert count >= 0
- return count
-
- def __iter__(self):
- trace_class_ptr = self._trace_class_env._trace_class._ptr
-
- for idx in range(len(self)):
- borrow_entry_fn = native_bt.trace_class_borrow_environment_entry_by_index_const
- entry_name, _ = borrow_entry_fn(trace_class_ptr, idx)
- assert entry_name is not None
- yield entry_name
-
-
class _StreamClassIterator(collections.abc.Iterator):
def __init__(self, trace_class):
self._trace_class = trace_class
_get_ref = staticmethod(native_bt.trace_class_get_ref)
_put_ref = staticmethod(native_bt.trace_class_put_ref)
- @property
- def uuid(self):
- uuid_bytes = native_bt.trace_class_get_uuid(self._ptr)
- if uuid_bytes is None:
- return
-
- return uuidp.UUID(bytes=uuid_bytes)
-
- def _uuid(self, uuid):
- utils._check_type(uuid, uuidp.UUID)
- native_bt.trace_class_set_uuid(self._ptr, uuid.bytes)
-
- _uuid = property(fset=_uuid)
-
# Instantiate a trace of this class.
- def __call__(self, name=None):
+ def __call__(self, name=None, uuid=None, env=None):
trace_ptr = native_bt.trace_create(self._ptr)
if trace_ptr is None:
if name is not None:
trace._name = name
+ if uuid is not None:
+ trace._uuid = uuid
+
+ if env is not None:
+ for key, value in env.items():
+ trace.env[key] = value
+
return trace
# Number of stream classes in this trace class.
yield id
- @property
- def env(self):
- return _TraceClassEnv(self)
-
def create_stream_class(self, id=None,
name=None,
packet_context_field_class=None,
default_clock_class=None,
assigns_automatic_event_class_id=True,
assigns_automatic_stream_id=True,
+ supports_packets=False,
packets_have_beginning_default_clock_snapshot=False,
packets_have_end_default_clock_snapshot=False,
supports_discarded_events=False,
if name is not None:
sc._name = name
- if packet_context_field_class is not None:
- sc._packet_context_field_class = packet_context_field_class
-
if event_common_context_field_class is not None:
sc._event_common_context_field_class = event_common_context_field_class
if default_clock_class is not None:
sc._default_clock_class = default_clock_class
+ # call after `sc._default_clock_class` because, if
+ # `packets_have_beginning_default_clock_snapshot` or
+ # `packets_have_end_default_clock_snapshot` is true, then this
+ # stream class needs a default clock class already.
+ sc._set_supports_packets(supports_packets,
+ packets_have_beginning_default_clock_snapshot,
+ packets_have_end_default_clock_snapshot)
+
+ # call after sc._set_supports_packets() because, if
+ # `packet_context_field_class` is not `None`, then this stream
+ # class needs to support packets already.
+ if packet_context_field_class is not None:
+ sc._packet_context_field_class = packet_context_field_class
+
sc._assigns_automatic_event_class_id = assigns_automatic_event_class_id
sc._assigns_automatic_stream_id = assigns_automatic_stream_id
- sc._packets_have_beginning_default_clock_snapshot = packets_have_beginning_default_clock_snapshot
- sc._packets_have_end_default_clock_snapshot = packets_have_end_default_clock_snapshot
sc._set_supports_discarded_events(supports_discarded_events,
discarded_events_have_default_clock_snapshots)
sc._set_supports_discarded_packets(supports_discarded_packets,
if not callable(listener):
raise TypeError("'listener' parameter is not callable")
- fn = native_bt.py3_trace_class_add_destruction_listener
+ fn = native_bt.bt2_trace_class_add_destruction_listener
listener_from_native = functools.partial(_trace_class_destruction_listener_from_native,
listener)