bindings/python/bt2/bt2/trace_class.py

   1 # The MIT License (MIT)
   2 #
   3 # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
   4 # Copyright (c) 2018 Francis Deslauriers <francis.deslauriers@efficios.com>
   5 # Copyright (c) 2019 Simon Marchi <simon.marchi@efficios.com>
   6 #
   7 # Permission is hereby granted, free of charge, to any person obtaining a copy
   8 # of this software and associated documentation files (the "Software"), to deal
   9 # in the Software without restriction, including without limitation the rights
  10 # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  11 # copies of the Software, and to permit persons to whom the Software is
  12 # furnished to do so, subject to the following conditions:
  13 #
  14 # The above copyright notice and this permission notice shall be included in
  15 # all copies or substantial portions of the Software.
  16 #
  17 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  18 # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  19 # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  20 # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  21 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  22 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  23 # THE SOFTWARE.
  24
  25 __all__ = ['TraceClass']
  26
  27 import bt2
  28 from bt2 import native_bt, utils, object
  29 import uuid as uuidp
  30 import collections.abc
  31 import functools
  32
  33
  34 class _TraceClassEnv(collections.abc.MutableMapping):
  35     def __init__(self, trace_class):
  36         self._trace_class = trace_class
  37
  38     def __getitem__(self, key):
  39         utils._check_str(key)
  40
  41         borrow_entry_fn = native_bt.trace_class_borrow_environment_entry_value_by_name_const
  42         value_ptr = borrow_entry_fn(self._trace_class._ptr, key)
  43
  44         if value_ptr is None:
  45             raise KeyError(key)
  46
  47         return bt2.value._create_from_ptr_and_get_ref(value_ptr)
  48
  49     def __setitem__(self, key, value):
  50         if isinstance(value, str):
  51             set_env_entry_fn = native_bt.trace_class_set_environment_entry_string
  52         elif isinstance(value, int):
  53             set_env_entry_fn = native_bt.trace_class_set_environment_entry_integer
  54         else:
  55             raise TypeError('expected str or int, got {}'.format(type(value)))
  56
  57         ret = set_env_entry_fn(self._trace_class._ptr, key, value)
  58
  59         utils._handle_ret(ret, "cannot set trace class object's environment entry")
  60
  61     def __delitem__(self, key):
  62         raise NotImplementedError
  63
  64     def __len__(self):
  65         count = native_bt.trace_class_get_environment_entry_count(self._trace_class._ptr)
  66         assert count >= 0
  67         return count
  68
  69     def __iter__(self):
  70         trace_class_ptr = self._trace_class_env._trace_class._ptr
  71
  72         for idx in range(len(self)):
  73             borrow_entry_fn = native_bt.trace_class_borrow_environment_entry_by_index_const
  74             entry_name, _ = borrow_entry_fn(trace_class_ptr, idx)
  75             assert entry_name is not None
  76             yield entry_name
  77
  78
  79 class _StreamClassIterator(collections.abc.Iterator):
  80     def __init__(self, trace_class):
  81         self._trace_class = trace_class
  82         self._at = 0
  83
  84     def __next__(self):
  85         if self._at == len(self._trace_class):
  86             raise StopIteration
  87
  88         borrow_stream_class_fn = native_bt.trace_class_borrow_stream_class_by_index_const
  89         sc_ptr = borrow_stream_class_fn(self._trace_class._ptr, self._at)
  90         assert sc_ptr
  91         id = native_bt.stream_class_get_id(sc_ptr)
  92         assert id >= 0
  93         self._at += 1
  94         return id
  95
  96
  97 def _trace_class_destruction_listener_from_native(user_listener, trace_class_ptr):
  98     trace_class = bt2.trace_class.TraceClass._create_from_ptr_and_get_ref(trace_class_ptr)
  99     user_listener(trace_class)
 100
 101
 102 class TraceClass(object._SharedObject, collections.abc.Mapping):
 103     _get_ref = staticmethod(native_bt.trace_class_get_ref)
 104     _put_ref = staticmethod(native_bt.trace_class_put_ref)
 105
 106     @property
 107     def uuid(self):
 108         uuid_bytes = native_bt.trace_class_get_uuid(self._ptr)
 109         if uuid_bytes is None:
 110             return
 111
 112         return uuidp.UUID(bytes=uuid_bytes)
 113
 114     def _uuid(self, uuid):
 115         utils._check_type(uuid, uuidp.UUID)
 116         native_bt.trace_class_set_uuid(self._ptr, uuid.bytes)
 117
 118     _uuid = property(fset=_uuid)
 119
 120     # Instantiate a trace of this class.
 121
 122     def __call__(self, name=None):
 123         trace_ptr = native_bt.trace_create(self._ptr)
 124
 125         if trace_ptr is None:
 126             raise bt2.CreationError('cannot create trace class object')
 127
 128         trace = bt2.trace.Trace._create_from_ptr(trace_ptr)
 129
 130         if name is not None:
 131             trace._name = name
 132
 133         return trace
 134
 135     # Number of stream classes in this trace class.
 136
 137     def __len__(self):
 138         count = native_bt.trace_class_get_stream_class_count(self._ptr)
 139         assert count >= 0
 140         return count
 141
 142     # Get a stream class by stream id.
 143
 144     def __getitem__(self, key):
 145         utils._check_uint64(key)
 146
 147         sc_ptr = native_bt.trace_class_borrow_stream_class_by_id_const(self._ptr, key)
 148         if sc_ptr is None:
 149             raise KeyError(key)
 150
 151         return bt2.StreamClass._create_from_ptr_and_get_ref(sc_ptr)
 152
 153     def __iter__(self):
 154         for idx in range(len(self)):
 155             sc_ptr = native_bt.trace_class_borrow_stream_class_by_index_const(self._ptr, idx)
 156             assert sc_ptr is not None
 157
 158             id = native_bt.stream_class_get_id(sc_ptr)
 159             assert id >= 0
 160
 161             yield id
 162
 163     @property
 164     def env(self):
 165         return _TraceClassEnv(self)
 166
 167     def create_stream_class(self, id=None,
 168                             name=None,
 169                             packet_context_field_class=None,
 170                             event_common_context_field_class=None,
 171                             default_clock_class=None,
 172                             assigns_automatic_event_class_id=True,
 173                             assigns_automatic_stream_id=True,
 174                             packets_have_beginning_default_clock_snapshot=False,
 175                             packets_have_end_default_clock_snapshot=False,
 176                             supports_discarded_events=False,
 177                             discarded_events_have_default_clock_snapshots=False,
 178                             supports_discarded_packets=False,
 179                             discarded_packets_have_default_clock_snapshots=False):
 180
 181         if self.assigns_automatic_stream_class_id:
 182             if id is not None:
 183                 raise ValueError('id provided, but trace class assigns automatic stream class ids')
 184
 185             sc_ptr = native_bt.stream_class_create(self._ptr)
 186         else:
 187             if id is None:
 188                 raise ValueError('id not provided, but trace class does not assign automatic stream class ids')
 189
 190             utils._check_uint64(id)
 191             sc_ptr = native_bt.stream_class_create_with_id(self._ptr, id)
 192
 193         sc = bt2.stream_class.StreamClass._create_from_ptr(sc_ptr)
 194
 195         if name is not None:
 196             sc._name = name
 197
 198         if packet_context_field_class is not None:
 199             sc._packet_context_field_class = packet_context_field_class
 200
 201         if event_common_context_field_class is not None:
 202             sc._event_common_context_field_class = event_common_context_field_class
 203
 204         if default_clock_class is not None:
 205             sc._default_clock_class = default_clock_class
 206
 207         sc._assigns_automatic_event_class_id = assigns_automatic_event_class_id
 208         sc._assigns_automatic_stream_id = assigns_automatic_stream_id
 209         sc._packets_have_beginning_default_clock_snapshot = packets_have_beginning_default_clock_snapshot
 210         sc._packets_have_end_default_clock_snapshot = packets_have_end_default_clock_snapshot
 211         sc._set_supports_discarded_events(supports_discarded_events,
 212                                           discarded_events_have_default_clock_snapshots)
 213         sc._set_supports_discarded_packets(supports_discarded_packets,
 214                                            discarded_packets_have_default_clock_snapshots)
 215         return sc
 216
 217     @property
 218     def assigns_automatic_stream_class_id(self):
 219         return native_bt.trace_class_assigns_automatic_stream_class_id(self._ptr)
 220
 221     def _assigns_automatic_stream_class_id(self, auto_id):
 222         utils._check_bool(auto_id)
 223         return native_bt.trace_class_set_assigns_automatic_stream_class_id(self._ptr, auto_id)
 224
 225     _assigns_automatic_stream_class_id = property(fset=_assigns_automatic_stream_class_id)
 226
 227     # Field class creation methods.
 228
 229     def _check_create_status(self, ptr, type_name):
 230         if ptr is None:
 231             raise bt2.CreationError(
 232                 'cannot create {} field class'.format(type_name))
 233
 234     def _create_integer_field_class(self, create_func, py_cls, type_name, field_value_range, preferred_display_base):
 235         field_class_ptr = create_func(self._ptr)
 236         self._check_create_status(field_class_ptr, type_name)
 237
 238         field_class = py_cls._create_from_ptr(field_class_ptr)
 239
 240         if field_value_range is not None:
 241             field_class._field_value_range = field_value_range
 242
 243         if preferred_display_base is not None:
 244             field_class._preferred_display_base = preferred_display_base
 245
 246         return field_class
 247
 248     def create_signed_integer_field_class(self, field_value_range=None, preferred_display_base=None):
 249         return self._create_integer_field_class(native_bt.field_class_signed_integer_create,
 250                                                 bt2.field_class._SignedIntegerFieldClass,
 251                                                 'signed integer', field_value_range, preferred_display_base)
 252
 253     def create_unsigned_integer_field_class(self, field_value_range=None, preferred_display_base=None):
 254         return self._create_integer_field_class(native_bt.field_class_unsigned_integer_create,
 255                                                 bt2.field_class._UnsignedIntegerFieldClass,
 256                                                 'unsigned integer', field_value_range, preferred_display_base)
 257
 258     def create_signed_enumeration_field_class(self, field_value_range=None, preferred_display_base=None):
 259         return self._create_integer_field_class(native_bt.field_class_signed_enumeration_create,
 260                                                 bt2.field_class._SignedEnumerationFieldClass,
 261                                                 'signed enumeration', field_value_range, preferred_display_base)
 262
 263     def create_unsigned_enumeration_field_class(self, field_value_range=None, preferred_display_base=None):
 264         return self._create_integer_field_class(native_bt.field_class_unsigned_enumeration_create,
 265                                                 bt2.field_class._UnsignedEnumerationFieldClass,
 266                                                 'unsigned enumeration', field_value_range, preferred_display_base)
 267
 268     def create_real_field_class(self, is_single_precision=False):
 269         field_class_ptr = native_bt.field_class_real_create(self._ptr)
 270         self._check_create_status(field_class_ptr, 'real')
 271
 272         field_class = bt2.field_class._RealFieldClass._create_from_ptr(field_class_ptr)
 273
 274         field_class._is_single_precision = is_single_precision
 275
 276         return field_class
 277
 278     def create_structure_field_class(self):
 279         field_class_ptr = native_bt.field_class_structure_create(self._ptr)
 280         self._check_create_status(field_class_ptr, 'structure')
 281
 282         return bt2.field_class._StructureFieldClass._create_from_ptr(field_class_ptr)
 283
 284     def create_string_field_class(self):
 285         field_class_ptr = native_bt.field_class_string_create(self._ptr)
 286         self._check_create_status(field_class_ptr, 'string')
 287
 288         return bt2.field_class._StringFieldClass._create_from_ptr(field_class_ptr)
 289
 290     def create_static_array_field_class(self, elem_fc, length):
 291         utils._check_type(elem_fc, bt2.field_class._FieldClass)
 292         utils._check_uint64(length)
 293         ptr = native_bt.field_class_static_array_create(self._ptr, elem_fc._ptr, length)
 294         self._check_create_status(ptr, 'static array')
 295
 296         return bt2.field_class._StaticArrayFieldClass._create_from_ptr_and_get_ref(ptr)
 297
 298     def create_dynamic_array_field_class(self, elem_fc, length_fc=None):
 299         utils._check_type(elem_fc, bt2.field_class._FieldClass)
 300         ptr = native_bt.field_class_dynamic_array_create(self._ptr, elem_fc._ptr)
 301         self._check_create_status(ptr, 'dynamic array')
 302         obj = bt2.field_class._DynamicArrayFieldClass._create_from_ptr(ptr)
 303
 304         if length_fc is not None:
 305             obj._length_field_class = length_fc
 306
 307         return obj
 308
 309     def create_variant_field_class(self, selector_fc=None):
 310         ptr = native_bt.field_class_variant_create(self._ptr)
 311         self._check_create_status(ptr, 'variant')
 312         obj = bt2.field_class._VariantFieldClass._create_from_ptr(ptr)
 313
 314         if selector_fc is not None:
 315             obj._selector_field_class = selector_fc
 316
 317         return obj
 318
 319     # Add a listener to be called when the trace class is destroyed.
 320
 321     def add_destruction_listener(self, listener):
 322
 323         if not callable(listener):
 324             raise TypeError("'listener' parameter is not callable")
 325
 326         fn = native_bt.py3_trace_class_add_destruction_listener
 327         listener_from_native = functools.partial(_trace_class_destruction_listener_from_native,
 328                                                  listener)
 329
 330         listener_id = fn(self._ptr, listener_from_native)
 331         if listener_id is None:
 332             utils._raise_bt2_error('cannot add destruction listener to trace class object')
 333
 334         return bt2._ListenerHandle(listener_id, self)