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