src/bindings/python/bt2/bt2/trace_class.py

   1 # SPDX-License-Identifier: 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 import functools
   8 import collections.abc
   9
  10 from bt2 import error as bt2_error
  11 from bt2 import trace as bt2_trace
  12 from bt2 import utils as bt2_utils
  13 from bt2 import value as bt2_value
  14 from bt2 import object as bt2_object
  15 from bt2 import native_bt
  16 from bt2 import field_class as bt2_field_class
  17 from bt2 import stream_class as bt2_stream_class
  18 from bt2 import integer_range_set as bt2_integer_range_set
  19
  20
  21 def _trace_class_destruction_listener_from_native(
  22     user_listener, handle, trace_class_ptr
  23 ):
  24     trace_class = _TraceClassConst._create_from_ptr_and_get_ref(trace_class_ptr)
  25     user_listener(trace_class)
  26     handle._invalidate()
  27
  28
  29 class _TraceClassConst(bt2_object._SharedObject, collections.abc.Mapping):
  30     @staticmethod
  31     def _get_ref(ptr):
  32         native_bt.trace_class_get_ref(ptr)
  33
  34     @staticmethod
  35     def _put_ref(ptr):
  36         native_bt.trace_class_put_ref(ptr)
  37
  38     _borrow_stream_class_ptr_by_index = staticmethod(
  39         native_bt.trace_class_borrow_stream_class_by_index_const
  40     )
  41     _borrow_stream_class_ptr_by_id = staticmethod(
  42         native_bt.trace_class_borrow_stream_class_by_id_const
  43     )
  44     _borrow_user_attributes_ptr = staticmethod(
  45         native_bt.trace_class_borrow_user_attributes_const
  46     )
  47     _stream_class_pycls = bt2_stream_class._StreamClassConst
  48     _create_value_from_ptr_and_get_ref = staticmethod(
  49         bt2_value._create_from_const_ptr_and_get_ref
  50     )
  51
  52     @property
  53     def user_attributes(self):
  54         ptr = self._borrow_user_attributes_ptr(self._ptr)
  55         assert ptr is not None
  56         return self._create_value_from_ptr_and_get_ref(ptr)
  57
  58     # Number of stream classes in this trace class.
  59
  60     def __len__(self):
  61         count = native_bt.trace_class_get_stream_class_count(self._ptr)
  62         assert count >= 0
  63         return count
  64
  65     # Get a stream class by stream id.
  66
  67     def __getitem__(self, key):
  68         bt2_utils._check_uint64(key)
  69
  70         sc_ptr = self._borrow_stream_class_ptr_by_id(self._ptr, key)
  71         if sc_ptr is None:
  72             raise KeyError(key)
  73
  74         return self._stream_class_pycls._create_from_ptr_and_get_ref(sc_ptr)
  75
  76     def __iter__(self):
  77         for idx in range(len(self)):
  78             sc_ptr = self._borrow_stream_class_ptr_by_index(self._ptr, idx)
  79             assert sc_ptr is not None
  80
  81             id = native_bt.stream_class_get_id(sc_ptr)
  82             assert id >= 0
  83
  84             yield id
  85
  86     @property
  87     def assigns_automatic_stream_class_id(self):
  88         return native_bt.trace_class_assigns_automatic_stream_class_id(self._ptr)
  89
  90     # Add a listener to be called when the trace class is destroyed.
  91
  92     def add_destruction_listener(self, listener):
  93         if not callable(listener):
  94             raise TypeError("'listener' parameter is not callable")
  95
  96         handle = bt2_utils._ListenerHandle(self.addr)
  97
  98         listener_from_native = functools.partial(
  99             _trace_class_destruction_listener_from_native, listener, handle
 100         )
 101
 102         fn = native_bt.bt2_trace_class_add_destruction_listener
 103         status, listener_id = fn(self._ptr, listener_from_native)
 104         bt2_utils._handle_func_status(
 105             status, "cannot add destruction listener to trace class object"
 106         )
 107
 108         handle._set_listener_id(listener_id)
 109
 110         return handle
 111
 112     def remove_destruction_listener(self, listener_handle):
 113         bt2_utils._check_type(listener_handle, bt2_utils._ListenerHandle)
 114
 115         if listener_handle._addr != self.addr:
 116             raise ValueError(
 117                 "This trace class destruction listener does not match the trace class object."
 118             )
 119
 120         if listener_handle._listener_id is None:
 121             raise ValueError(
 122                 "This trace class destruction listener was already removed."
 123             )
 124
 125         status = native_bt.trace_class_remove_destruction_listener(
 126             self._ptr, listener_handle._listener_id
 127         )
 128         bt2_utils._handle_func_status(status)
 129         listener_handle._invalidate()
 130
 131
 132 class _TraceClass(_TraceClassConst):
 133     _borrow_stream_class_ptr_by_index = staticmethod(
 134         native_bt.trace_class_borrow_stream_class_by_index
 135     )
 136     _borrow_stream_class_ptr_by_id = staticmethod(
 137         native_bt.trace_class_borrow_stream_class_by_id
 138     )
 139     _borrow_user_attributes_ptr = staticmethod(
 140         native_bt.trace_class_borrow_user_attributes
 141     )
 142     _stream_class_pycls = bt2_stream_class._StreamClass
 143     _create_value_from_ptr_and_get_ref = staticmethod(
 144         bt2_value._create_from_ptr_and_get_ref
 145     )
 146
 147     # Instantiate a trace of this class.
 148
 149     def __call__(self, name=None, user_attributes=None, uuid=None, environment=None):
 150         trace_ptr = native_bt.trace_create(self._ptr)
 151
 152         if trace_ptr is None:
 153             raise bt2_error._MemoryError("cannot create trace class object")
 154
 155         trace = bt2_trace._Trace._create_from_ptr(trace_ptr)
 156
 157         if name is not None:
 158             trace._name = name
 159
 160         if user_attributes is not None:
 161             trace._user_attributes = user_attributes
 162
 163         if uuid is not None:
 164             trace._uuid = uuid
 165
 166         if environment is not None:
 167             for key, value in environment.items():
 168                 trace.environment[key] = value
 169
 170         return trace
 171
 172     def create_stream_class(
 173         self,
 174         id=None,
 175         name=None,
 176         user_attributes=None,
 177         packet_context_field_class=None,
 178         event_common_context_field_class=None,
 179         default_clock_class=None,
 180         assigns_automatic_event_class_id=True,
 181         assigns_automatic_stream_id=True,
 182         supports_packets=False,
 183         packets_have_beginning_default_clock_snapshot=False,
 184         packets_have_end_default_clock_snapshot=False,
 185         supports_discarded_events=False,
 186         discarded_events_have_default_clock_snapshots=False,
 187         supports_discarded_packets=False,
 188         discarded_packets_have_default_clock_snapshots=False,
 189     ):
 190         # Validate parameters before we create the object.
 191         bt2_stream_class._StreamClass._validate_create_params(
 192             name,
 193             user_attributes,
 194             packet_context_field_class,
 195             event_common_context_field_class,
 196             default_clock_class,
 197             assigns_automatic_event_class_id,
 198             assigns_automatic_stream_id,
 199             supports_packets,
 200             packets_have_beginning_default_clock_snapshot,
 201             packets_have_end_default_clock_snapshot,
 202             supports_discarded_events,
 203             discarded_events_have_default_clock_snapshots,
 204             supports_discarded_packets,
 205             discarded_packets_have_default_clock_snapshots,
 206         )
 207
 208         if self.assigns_automatic_stream_class_id:
 209             if id is not None:
 210                 raise ValueError(
 211                     "id provided, but trace class assigns automatic stream class ids"
 212                 )
 213
 214             sc_ptr = native_bt.stream_class_create(self._ptr)
 215         else:
 216             if id is None:
 217                 raise ValueError(
 218                     "id not provided, but trace class does not assign automatic stream class ids"
 219                 )
 220
 221             bt2_utils._check_uint64(id)
 222             sc_ptr = native_bt.stream_class_create_with_id(self._ptr, id)
 223
 224         sc = bt2_stream_class._StreamClass._create_from_ptr(sc_ptr)
 225
 226         if name is not None:
 227             sc._name = name
 228
 229         if user_attributes is not None:
 230             sc._user_attributes = user_attributes
 231
 232         if event_common_context_field_class is not None:
 233             sc._event_common_context_field_class = event_common_context_field_class
 234
 235         if default_clock_class is not None:
 236             sc._default_clock_class = default_clock_class
 237
 238         # call after `sc._default_clock_class` because, if
 239         # `packets_have_beginning_default_clock_snapshot` or
 240         # `packets_have_end_default_clock_snapshot` is true, then this
 241         # stream class needs a default clock class already.
 242         sc._set_supports_packets(
 243             supports_packets,
 244             packets_have_beginning_default_clock_snapshot,
 245             packets_have_end_default_clock_snapshot,
 246         )
 247
 248         # call after sc._set_supports_packets() because, if
 249         # `packet_context_field_class` is not `None`, then this stream
 250         # class needs to support packets already.
 251         if packet_context_field_class is not None:
 252             sc._packet_context_field_class = packet_context_field_class
 253
 254         sc._assigns_automatic_event_class_id = assigns_automatic_event_class_id
 255         sc._assigns_automatic_stream_id = assigns_automatic_stream_id
 256         sc._set_supports_discarded_events(
 257             supports_discarded_events, discarded_events_have_default_clock_snapshots
 258         )
 259         sc._set_supports_discarded_packets(
 260             supports_discarded_packets, discarded_packets_have_default_clock_snapshots
 261         )
 262         return sc
 263
 264     def _user_attributes(self, user_attributes):
 265         value = bt2_value.create_value(user_attributes)
 266         bt2_utils._check_type(value, bt2_value.MapValue)
 267         native_bt.trace_class_set_user_attributes(self._ptr, value._ptr)
 268
 269     _user_attributes = property(fset=_user_attributes)
 270
 271     def _assigns_automatic_stream_class_id(self, auto_id):
 272         bt2_utils._check_bool(auto_id)
 273         return native_bt.trace_class_set_assigns_automatic_stream_class_id(
 274             self._ptr, auto_id
 275         )
 276
 277     _assigns_automatic_stream_class_id = property(
 278         fset=_assigns_automatic_stream_class_id
 279     )
 280
 281     # Field class creation methods.
 282
 283     def _check_field_class_create_status(self, ptr, type_name):
 284         if ptr is None:
 285             raise bt2_error._MemoryError(
 286                 "cannot create {} field class".format(type_name)
 287             )
 288
 289     @staticmethod
 290     def _set_field_class_user_attrs(fc, user_attributes):
 291         if user_attributes is not None:
 292             fc._user_attributes = user_attributes
 293
 294     def create_bool_field_class(self, user_attributes=None):
 295         field_class_ptr = native_bt.field_class_bool_create(self._ptr)
 296         self._check_field_class_create_status(field_class_ptr, "boolean")
 297         fc = bt2_field_class._BoolFieldClass._create_from_ptr(field_class_ptr)
 298         self._set_field_class_user_attrs(fc, user_attributes)
 299         return fc
 300
 301     def create_bit_array_field_class(self, length, user_attributes=None):
 302         bt2_utils._check_uint64(length)
 303
 304         if length < 1 or length > 64:
 305             raise ValueError(
 306                 "invalid length {}: expecting a value in the [1, 64] range".format(
 307                     length
 308                 )
 309             )
 310
 311         field_class_ptr = native_bt.field_class_bit_array_create(self._ptr, length)
 312         self._check_field_class_create_status(field_class_ptr, "bit array")
 313         fc = bt2_field_class._BitArrayFieldClass._create_from_ptr(field_class_ptr)
 314         self._set_field_class_user_attrs(fc, user_attributes)
 315         return fc
 316
 317     def _create_integer_field_class(
 318         self,
 319         create_func,
 320         py_cls,
 321         type_name,
 322         field_value_range,
 323         preferred_display_base,
 324         user_attributes,
 325     ):
 326         field_class_ptr = create_func(self._ptr)
 327         self._check_field_class_create_status(field_class_ptr, type_name)
 328
 329         field_class = py_cls._create_from_ptr(field_class_ptr)
 330
 331         if field_value_range is not None:
 332             field_class._field_value_range = field_value_range
 333
 334         if preferred_display_base is not None:
 335             field_class._preferred_display_base = preferred_display_base
 336
 337         self._set_field_class_user_attrs(field_class, user_attributes)
 338         return field_class
 339
 340     def create_signed_integer_field_class(
 341         self, field_value_range=None, preferred_display_base=None, user_attributes=None
 342     ):
 343         return self._create_integer_field_class(
 344             native_bt.field_class_integer_signed_create,
 345             bt2_field_class._SignedIntegerFieldClass,
 346             "signed integer",
 347             field_value_range,
 348             preferred_display_base,
 349             user_attributes,
 350         )
 351
 352     def create_unsigned_integer_field_class(
 353         self, field_value_range=None, preferred_display_base=None, user_attributes=None
 354     ):
 355         return self._create_integer_field_class(
 356             native_bt.field_class_integer_unsigned_create,
 357             bt2_field_class._UnsignedIntegerFieldClass,
 358             "unsigned integer",
 359             field_value_range,
 360             preferred_display_base,
 361             user_attributes,
 362         )
 363
 364     def create_signed_enumeration_field_class(
 365         self, field_value_range=None, preferred_display_base=None, user_attributes=None
 366     ):
 367         return self._create_integer_field_class(
 368             native_bt.field_class_enumeration_signed_create,
 369             bt2_field_class._SignedEnumerationFieldClass,
 370             "signed enumeration",
 371             field_value_range,
 372             preferred_display_base,
 373             user_attributes,
 374         )
 375
 376     def create_unsigned_enumeration_field_class(
 377         self, field_value_range=None, preferred_display_base=None, user_attributes=None
 378     ):
 379         return self._create_integer_field_class(
 380             native_bt.field_class_enumeration_unsigned_create,
 381             bt2_field_class._UnsignedEnumerationFieldClass,
 382             "unsigned enumeration",
 383             field_value_range,
 384             preferred_display_base,
 385             user_attributes,
 386         )
 387
 388     def create_single_precision_real_field_class(self, user_attributes=None):
 389         field_class_ptr = native_bt.field_class_real_single_precision_create(self._ptr)
 390         self._check_field_class_create_status(field_class_ptr, "single-precision real")
 391
 392         field_class = bt2_field_class._SinglePrecisionRealFieldClass._create_from_ptr(
 393             field_class_ptr
 394         )
 395
 396         self._set_field_class_user_attrs(field_class, user_attributes)
 397
 398         return field_class
 399
 400     def create_double_precision_real_field_class(self, user_attributes=None):
 401         field_class_ptr = native_bt.field_class_real_double_precision_create(self._ptr)
 402         self._check_field_class_create_status(field_class_ptr, "double-precision real")
 403
 404         field_class = bt2_field_class._DoublePrecisionRealFieldClass._create_from_ptr(
 405             field_class_ptr
 406         )
 407
 408         self._set_field_class_user_attrs(field_class, user_attributes)
 409
 410         return field_class
 411
 412     def create_structure_field_class(self, user_attributes=None):
 413         field_class_ptr = native_bt.field_class_structure_create(self._ptr)
 414         self._check_field_class_create_status(field_class_ptr, "structure")
 415         fc = bt2_field_class._StructureFieldClass._create_from_ptr(field_class_ptr)
 416         self._set_field_class_user_attrs(fc, user_attributes)
 417         return fc
 418
 419     def create_string_field_class(self, user_attributes=None):
 420         field_class_ptr = native_bt.field_class_string_create(self._ptr)
 421         self._check_field_class_create_status(field_class_ptr, "string")
 422         fc = bt2_field_class._StringFieldClass._create_from_ptr(field_class_ptr)
 423         self._set_field_class_user_attrs(fc, user_attributes)
 424         return fc
 425
 426     def create_static_array_field_class(self, elem_fc, length, user_attributes=None):
 427         bt2_utils._check_type(elem_fc, bt2_field_class._FieldClass)
 428         bt2_utils._check_uint64(length)
 429         ptr = native_bt.field_class_array_static_create(self._ptr, elem_fc._ptr, length)
 430         self._check_field_class_create_status(ptr, "static array")
 431         fc = bt2_field_class._StaticArrayFieldClass._create_from_ptr(ptr)
 432         self._set_field_class_user_attrs(fc, user_attributes)
 433         return fc
 434
 435     def create_dynamic_array_field_class(
 436         self, elem_fc, length_fc=None, user_attributes=None
 437     ):
 438         bt2_utils._check_type(elem_fc, bt2_field_class._FieldClass)
 439         length_fc_ptr = None
 440
 441         if length_fc is not None:
 442             bt2_utils._check_type(length_fc, bt2_field_class._UnsignedIntegerFieldClass)
 443             length_fc_ptr = length_fc._ptr
 444
 445         ptr = native_bt.field_class_array_dynamic_create(
 446             self._ptr, elem_fc._ptr, length_fc_ptr
 447         )
 448         self._check_field_class_create_status(ptr, "dynamic array")
 449         fc = bt2_field_class._obj_type_from_field_class_ptr(ptr)._create_from_ptr(ptr)
 450         self._set_field_class_user_attrs(fc, user_attributes)
 451         return fc
 452
 453     def create_option_without_selector_field_class(
 454         self, content_fc, user_attributes=None
 455     ):
 456         bt2_utils._check_type(content_fc, bt2_field_class._FieldClass)
 457         ptr = native_bt.field_class_option_without_selector_create(
 458             self._ptr, content_fc._ptr
 459         )
 460         self._check_field_class_create_status(ptr, "option")
 461         fc = bt2_field_class._obj_type_from_field_class_ptr(ptr)._create_from_ptr(ptr)
 462         self._set_field_class_user_attrs(fc, user_attributes)
 463         return fc
 464
 465     def create_option_with_bool_selector_field_class(
 466         self, content_fc, selector_fc, selector_is_reversed=False, user_attributes=None
 467     ):
 468         bt2_utils._check_type(content_fc, bt2_field_class._FieldClass)
 469         bt2_utils._check_bool(selector_is_reversed)
 470         bt2_utils._check_type(selector_fc, bt2_field_class._BoolFieldClass)
 471         ptr = native_bt.field_class_option_with_selector_field_bool_create(
 472             self._ptr, content_fc._ptr, selector_fc._ptr
 473         )
 474         self._check_field_class_create_status(ptr, "option")
 475         fc = bt2_field_class._obj_type_from_field_class_ptr(ptr)._create_from_ptr(ptr)
 476         self._set_field_class_user_attrs(fc, user_attributes)
 477         fc._selector_is_reversed = selector_is_reversed
 478         return fc
 479
 480     def create_option_with_integer_selector_field_class(
 481         self, content_fc, selector_fc, ranges, user_attributes=None
 482     ):
 483         bt2_utils._check_type(content_fc, bt2_field_class._FieldClass)
 484         bt2_utils._check_type(selector_fc, bt2_field_class._IntegerFieldClass)
 485
 486         if len(ranges) == 0:
 487             raise ValueError("integer range set is empty")
 488
 489         if isinstance(selector_fc, bt2_field_class._UnsignedIntegerFieldClass):
 490             bt2_utils._check_type(ranges, bt2_integer_range_set.UnsignedIntegerRangeSet)
 491             ptr = native_bt.field_class_option_with_selector_field_integer_unsigned_create(
 492                 self._ptr, content_fc._ptr, selector_fc._ptr, ranges._ptr
 493             )
 494         else:
 495             bt2_utils._check_type(ranges, bt2_integer_range_set.SignedIntegerRangeSet)
 496             ptr = (
 497                 native_bt.field_class_option_with_selector_field_integer_signed_create(
 498                     self._ptr, content_fc._ptr, selector_fc._ptr, ranges._ptr
 499                 )
 500             )
 501
 502         self._check_field_class_create_status(ptr, "option")
 503         fc = bt2_field_class._obj_type_from_field_class_ptr(ptr)._create_from_ptr(ptr)
 504         self._set_field_class_user_attrs(fc, user_attributes)
 505         return fc
 506
 507     def create_variant_field_class(self, selector_fc=None, user_attributes=None):
 508         selector_fc_ptr = None
 509
 510         if selector_fc is not None:
 511             bt2_utils._check_type(selector_fc, bt2_field_class._IntegerFieldClass)
 512             selector_fc_ptr = selector_fc._ptr
 513
 514         ptr = native_bt.field_class_variant_create(self._ptr, selector_fc_ptr)
 515         self._check_field_class_create_status(ptr, "variant")
 516         fc = bt2_field_class._obj_type_from_field_class_ptr(ptr)._create_from_ptr(ptr)
 517         self._set_field_class_user_attrs(fc, user_attributes)
 518         return fc