src/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 bt2.stream_class
  30 import bt2.field_class
  31 import bt2.trace
  32 import collections.abc
  33 import functools
  34
  35
  36 class _StreamClassIterator(collections.abc.Iterator):
  37     def __init__(self, trace_class):
  38         self._trace_class = trace_class
  39         self._at = 0
  40
  41     def __next__(self):
  42         if self._at == len(self._trace_class):
  43             raise StopIteration
  44
  45         borrow_stream_class_fn = (
  46             native_bt.trace_class_borrow_stream_class_by_index_const
  47         )
  48         sc_ptr = borrow_stream_class_fn(self._trace_class._ptr, self._at)
  49         assert sc_ptr
  50         id = native_bt.stream_class_get_id(sc_ptr)
  51         assert id >= 0
  52         self._at += 1
  53         return id
  54
  55
  56 def _trace_class_destruction_listener_from_native(user_listener, trace_class_ptr):
  57     trace_class = bt2.trace_class._TraceClass._create_from_ptr_and_get_ref(
  58         trace_class_ptr
  59     )
  60     user_listener(trace_class)
  61
  62
  63 class _TraceClass(object._SharedObject, collections.abc.Mapping):
  64     _get_ref = staticmethod(native_bt.trace_class_get_ref)
  65     _put_ref = staticmethod(native_bt.trace_class_put_ref)
  66
  67     # Instantiate a trace of this class.
  68
  69     def __call__(self, name=None, uuid=None, env=None):
  70         trace_ptr = native_bt.trace_create(self._ptr)
  71
  72         if trace_ptr is None:
  73             raise bt2._MemoryError('cannot create trace class object')
  74
  75         trace = bt2.trace._Trace._create_from_ptr(trace_ptr)
  76
  77         if name is not None:
  78             trace._name = name
  79
  80         if uuid is not None:
  81             trace._uuid = uuid
  82
  83         if env is not None:
  84             for key, value in env.items():
  85                 trace.env[key] = value
  86
  87         return trace
  88
  89     # Number of stream classes in this trace class.
  90
  91     def __len__(self):
  92         count = native_bt.trace_class_get_stream_class_count(self._ptr)
  93         assert count >= 0
  94         return count
  95
  96     # Get a stream class by stream id.
  97
  98     def __getitem__(self, key):
  99         utils._check_uint64(key)
 100
 101         sc_ptr = native_bt.trace_class_borrow_stream_class_by_id_const(self._ptr, key)
 102         if sc_ptr is None:
 103             raise KeyError(key)
 104
 105         return bt2.stream_class._StreamClass._create_from_ptr_and_get_ref(sc_ptr)
 106
 107     def __iter__(self):
 108         for idx in range(len(self)):
 109             sc_ptr = native_bt.trace_class_borrow_stream_class_by_index_const(
 110                 self._ptr, idx
 111             )
 112             assert sc_ptr is not None
 113
 114             id = native_bt.stream_class_get_id(sc_ptr)
 115             assert id >= 0
 116
 117             yield id
 118
 119     def create_stream_class(
 120         self,
 121         id=None,
 122         name=None,
 123         packet_context_field_class=None,
 124         event_common_context_field_class=None,
 125         default_clock_class=None,
 126         assigns_automatic_event_class_id=True,
 127         assigns_automatic_stream_id=True,
 128         supports_packets=False,
 129         packets_have_beginning_default_clock_snapshot=False,
 130         packets_have_end_default_clock_snapshot=False,
 131         supports_discarded_events=False,
 132         discarded_events_have_default_clock_snapshots=False,
 133         supports_discarded_packets=False,
 134         discarded_packets_have_default_clock_snapshots=False,
 135     ):
 136
 137         if self.assigns_automatic_stream_class_id:
 138             if id is not None:
 139                 raise ValueError(
 140                     'id provided, but trace class assigns automatic stream class ids'
 141                 )
 142
 143             sc_ptr = native_bt.stream_class_create(self._ptr)
 144         else:
 145             if id is None:
 146                 raise ValueError(
 147                     'id not provided, but trace class does not assign automatic stream class ids'
 148                 )
 149
 150             utils._check_uint64(id)
 151             sc_ptr = native_bt.stream_class_create_with_id(self._ptr, id)
 152
 153         sc = bt2.stream_class._StreamClass._create_from_ptr(sc_ptr)
 154
 155         if name is not None:
 156             sc._name = name
 157
 158         if event_common_context_field_class is not None:
 159             sc._event_common_context_field_class = event_common_context_field_class
 160
 161         if default_clock_class is not None:
 162             sc._default_clock_class = default_clock_class
 163
 164         # call after `sc._default_clock_class` because, if
 165         # `packets_have_beginning_default_clock_snapshot` or
 166         # `packets_have_end_default_clock_snapshot` is true, then this
 167         # stream class needs a default clock class already.
 168         sc._set_supports_packets(
 169             supports_packets,
 170             packets_have_beginning_default_clock_snapshot,
 171             packets_have_end_default_clock_snapshot,
 172         )
 173
 174         # call after sc._set_supports_packets() because, if
 175         # `packet_context_field_class` is not `None`, then this stream
 176         # class needs to support packets already.
 177         if packet_context_field_class is not None:
 178             sc._packet_context_field_class = packet_context_field_class
 179
 180         sc._assigns_automatic_event_class_id = assigns_automatic_event_class_id
 181         sc._assigns_automatic_stream_id = assigns_automatic_stream_id
 182         sc._set_supports_discarded_events(
 183             supports_discarded_events, discarded_events_have_default_clock_snapshots
 184         )
 185         sc._set_supports_discarded_packets(
 186             supports_discarded_packets, discarded_packets_have_default_clock_snapshots
 187         )
 188         return sc
 189
 190     @property
 191     def assigns_automatic_stream_class_id(self):
 192         return native_bt.trace_class_assigns_automatic_stream_class_id(self._ptr)
 193
 194     def _assigns_automatic_stream_class_id(self, auto_id):
 195         utils._check_bool(auto_id)
 196         return native_bt.trace_class_set_assigns_automatic_stream_class_id(
 197             self._ptr, auto_id
 198         )
 199
 200     _assigns_automatic_stream_class_id = property(
 201         fset=_assigns_automatic_stream_class_id
 202     )
 203
 204     # Field class creation methods.
 205
 206     def _check_create_status(self, ptr, type_name):
 207         if ptr is None:
 208             raise bt2._MemoryError('cannot create {} field class'.format(type_name))
 209
 210     def _create_integer_field_class(
 211         self, create_func, py_cls, type_name, field_value_range, preferred_display_base
 212     ):
 213         field_class_ptr = create_func(self._ptr)
 214         self._check_create_status(field_class_ptr, type_name)
 215
 216         field_class = py_cls._create_from_ptr(field_class_ptr)
 217
 218         if field_value_range is not None:
 219             field_class._field_value_range = field_value_range
 220
 221         if preferred_display_base is not None:
 222             field_class._preferred_display_base = preferred_display_base
 223
 224         return field_class
 225
 226     def create_signed_integer_field_class(
 227         self, field_value_range=None, preferred_display_base=None
 228     ):
 229         return self._create_integer_field_class(
 230             native_bt.field_class_integer_signed_create,
 231             bt2.field_class._SignedIntegerFieldClass,
 232             'signed integer',
 233             field_value_range,
 234             preferred_display_base,
 235         )
 236
 237     def create_unsigned_integer_field_class(
 238         self, field_value_range=None, preferred_display_base=None
 239     ):
 240         return self._create_integer_field_class(
 241             native_bt.field_class_integer_unsigned_create,
 242             bt2.field_class._UnsignedIntegerFieldClass,
 243             'unsigned integer',
 244             field_value_range,
 245             preferred_display_base,
 246         )
 247
 248     def create_signed_enumeration_field_class(
 249         self, field_value_range=None, preferred_display_base=None
 250     ):
 251         return self._create_integer_field_class(
 252             native_bt.field_class_enumeration_signed_create,
 253             bt2.field_class._SignedEnumerationFieldClass,
 254             'signed enumeration',
 255             field_value_range,
 256             preferred_display_base,
 257         )
 258
 259     def create_unsigned_enumeration_field_class(
 260         self, field_value_range=None, preferred_display_base=None
 261     ):
 262         return self._create_integer_field_class(
 263             native_bt.field_class_enumeration_unsigned_create,
 264             bt2.field_class._UnsignedEnumerationFieldClass,
 265             'unsigned enumeration',
 266             field_value_range,
 267             preferred_display_base,
 268         )
 269
 270     def create_real_field_class(self, is_single_precision=False):
 271         field_class_ptr = native_bt.field_class_real_create(self._ptr)
 272         self._check_create_status(field_class_ptr, 'real')
 273
 274         field_class = bt2.field_class._RealFieldClass._create_from_ptr(field_class_ptr)
 275
 276         field_class._is_single_precision = is_single_precision
 277
 278         return field_class
 279
 280     def create_structure_field_class(self):
 281         field_class_ptr = native_bt.field_class_structure_create(self._ptr)
 282         self._check_create_status(field_class_ptr, 'structure')
 283
 284         return bt2.field_class._StructureFieldClass._create_from_ptr(field_class_ptr)
 285
 286     def create_string_field_class(self):
 287         field_class_ptr = native_bt.field_class_string_create(self._ptr)
 288         self._check_create_status(field_class_ptr, 'string')
 289
 290         return bt2.field_class._StringFieldClass._create_from_ptr(field_class_ptr)
 291
 292     def create_static_array_field_class(self, elem_fc, length):
 293         utils._check_type(elem_fc, bt2.field_class._FieldClass)
 294         utils._check_uint64(length)
 295         ptr = native_bt.field_class_array_static_create(self._ptr, elem_fc._ptr, length)
 296         self._check_create_status(ptr, 'static array')
 297
 298         return bt2.field_class._StaticArrayFieldClass._create_from_ptr_and_get_ref(ptr)
 299
 300     def create_dynamic_array_field_class(self, elem_fc, length_fc=None):
 301         utils._check_type(elem_fc, bt2.field_class._FieldClass)
 302         length_fc_ptr = None
 303
 304         if length_fc is not None:
 305             utils._check_type(length_fc, bt2.field_class._UnsignedIntegerFieldClass)
 306             length_fc_ptr = length_fc._ptr
 307
 308         ptr = native_bt.field_class_array_dynamic_create(
 309             self._ptr, elem_fc._ptr, length_fc_ptr
 310         )
 311         self._check_create_status(ptr, 'dynamic array')
 312         return bt2.field_class._DynamicArrayFieldClass._create_from_ptr(ptr)
 313
 314     def create_variant_field_class(self, selector_fc=None):
 315         selector_fc_ptr = None
 316
 317         if selector_fc is not None:
 318             utils._check_type(selector_fc, bt2.field_class._IntegerFieldClass)
 319             selector_fc_ptr = selector_fc._ptr
 320
 321         ptr = native_bt.field_class_variant_create(self._ptr, selector_fc_ptr)
 322         self._check_create_status(ptr, 'variant')
 323         return bt2.field_class._create_field_class_from_ptr_and_get_ref(ptr)
 324
 325     # Add a listener to be called when the trace class is destroyed.
 326
 327     def add_destruction_listener(self, listener):
 328
 329         if not callable(listener):
 330             raise TypeError("'listener' parameter is not callable")
 331
 332         fn = native_bt.bt2_trace_class_add_destruction_listener
 333         listener_from_native = functools.partial(
 334             _trace_class_destruction_listener_from_native, listener
 335         )
 336
 337         status, listener_id = fn(self._ptr, listener_from_native)
 338         utils._handle_func_status(
 339             status, 'cannot add destruction listener to trace class object'
 340         )
 341
 342         return bt2.utils._ListenerHandle(listener_id, self)