src/bindings/python/bt2/bt2/stream_class.py

   1 # SPDX-License-Identifier: MIT
   2 #
   3 # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
   4
   5 import collections.abc
   6
   7 from bt2 import utils as bt2_utils
   8 from bt2 import value as bt2_value
   9 from bt2 import object as bt2_object
  10 from bt2 import native_bt
  11 from bt2 import clock_class as bt2_clock_class
  12 from bt2 import event_class as bt2_event_class
  13 from bt2 import field_class as bt2_field_class
  14
  15
  16 def _bt2_trace_class():
  17     from bt2 import trace_class as bt2_trace_class
  18
  19     return bt2_trace_class
  20
  21
  22 class _StreamClassConst(bt2_object._SharedObject, collections.abc.Mapping):
  23     @staticmethod
  24     def _get_ref(ptr):
  25         native_bt.stream_class_get_ref(ptr)
  26
  27     @staticmethod
  28     def _put_ref(ptr):
  29         native_bt.stream_class_put_ref(ptr)
  30
  31     _borrow_event_class_ptr_by_id = staticmethod(
  32         native_bt.stream_class_borrow_event_class_by_id_const
  33     )
  34     _borrow_event_class_ptr_by_index = staticmethod(
  35         native_bt.stream_class_borrow_event_class_by_index_const
  36     )
  37     _borrow_trace_class_ptr = staticmethod(
  38         native_bt.stream_class_borrow_trace_class_const
  39     )
  40     _borrow_packet_context_field_class_ptr = staticmethod(
  41         native_bt.stream_class_borrow_packet_context_field_class_const
  42     )
  43     _borrow_event_common_context_field_class_ptr = staticmethod(
  44         native_bt.stream_class_borrow_event_common_context_field_class_const
  45     )
  46     _borrow_default_clock_class_ptr = staticmethod(
  47         native_bt.stream_class_borrow_default_clock_class_const
  48     )
  49     _borrow_user_attributes_ptr = staticmethod(
  50         native_bt.stream_class_borrow_user_attributes_const
  51     )
  52
  53     _event_class_cls = property(lambda _: bt2_event_class._EventClassConst)
  54     _trace_class_cls = property(lambda _: _bt2_trace_class()._TraceClassConst)
  55     _clock_class_cls = property(lambda _: bt2_clock_class._ClockClassConst)
  56
  57     def __getitem__(self, key):
  58         bt2_utils._check_int64(key)
  59         ec_ptr = self._borrow_event_class_ptr_by_id(self._ptr, key)
  60
  61         if ec_ptr is None:
  62             raise KeyError(key)
  63
  64         return self._event_class_cls._create_from_ptr_and_get_ref(ec_ptr)
  65
  66     def __len__(self):
  67         count = native_bt.stream_class_get_event_class_count(self._ptr)
  68         assert count >= 0
  69         return count
  70
  71     def __iter__(self):
  72         for idx in range(len(self)):
  73             ec_ptr = self._borrow_event_class_ptr_by_index(self._ptr, idx)
  74             assert ec_ptr is not None
  75
  76             id = native_bt.event_class_get_id(ec_ptr)
  77             assert id >= 0
  78
  79             yield id
  80
  81     @property
  82     def trace_class(self):
  83         tc_ptr = self._borrow_trace_class_ptr(self._ptr)
  84
  85         if tc_ptr is not None:
  86             return self._trace_class_cls._create_from_ptr_and_get_ref(tc_ptr)
  87
  88     @property
  89     def user_attributes(self):
  90         ptr = self._borrow_user_attributes_ptr(self._ptr)
  91         assert ptr is not None
  92         return bt2_value._create_from_ptr_and_get_ref(ptr)
  93
  94     @property
  95     def name(self):
  96         return native_bt.stream_class_get_name(self._ptr)
  97
  98     @property
  99     def assigns_automatic_event_class_id(self):
 100         return native_bt.stream_class_assigns_automatic_event_class_id(self._ptr)
 101
 102     @property
 103     def assigns_automatic_stream_id(self):
 104         return native_bt.stream_class_assigns_automatic_stream_id(self._ptr)
 105
 106     @property
 107     def supports_packets(self):
 108         return native_bt.stream_class_supports_packets(self._ptr)
 109
 110     @property
 111     def packets_have_beginning_default_clock_snapshot(self):
 112         return native_bt.stream_class_packets_have_beginning_default_clock_snapshot(
 113             self._ptr
 114         )
 115
 116     @property
 117     def packets_have_end_default_clock_snapshot(self):
 118         return native_bt.stream_class_packets_have_end_default_clock_snapshot(self._ptr)
 119
 120     @property
 121     def supports_discarded_events(self):
 122         return native_bt.stream_class_supports_discarded_events(self._ptr)
 123
 124     @property
 125     def discarded_events_have_default_clock_snapshots(self):
 126         return native_bt.stream_class_discarded_events_have_default_clock_snapshots(
 127             self._ptr
 128         )
 129
 130     @property
 131     def supports_discarded_packets(self):
 132         return native_bt.stream_class_supports_discarded_packets(self._ptr)
 133
 134     @property
 135     def discarded_packets_have_default_clock_snapshots(self):
 136         return native_bt.stream_class_discarded_packets_have_default_clock_snapshots(
 137             self._ptr
 138         )
 139
 140     @property
 141     def id(self):
 142         id = native_bt.stream_class_get_id(self._ptr)
 143
 144         if id < 0:
 145             return
 146
 147         return id
 148
 149     @property
 150     def packet_context_field_class(self):
 151         fc_ptr = self._borrow_packet_context_field_class_ptr(self._ptr)
 152
 153         if fc_ptr is None:
 154             return
 155
 156         return bt2_field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
 157
 158     @property
 159     def event_common_context_field_class(self):
 160         fc_ptr = self._borrow_event_common_context_field_class_ptr(self._ptr)
 161
 162         if fc_ptr is None:
 163             return
 164
 165         return bt2_field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
 166
 167     @property
 168     def default_clock_class(self):
 169         cc_ptr = self._borrow_default_clock_class_ptr(self._ptr)
 170         if cc_ptr is None:
 171             return
 172
 173         return self._clock_class_cls._create_from_ptr_and_get_ref(cc_ptr)
 174
 175
 176 class _StreamClass(_StreamClassConst):
 177     @staticmethod
 178     def _get_ref(ptr):
 179         native_bt.stream_class_get_ref(ptr)
 180
 181     @staticmethod
 182     def _put_ref(ptr):
 183         native_bt.stream_class_put_ref(ptr)
 184
 185     _borrow_event_class_ptr_by_id = staticmethod(
 186         native_bt.stream_class_borrow_event_class_by_id
 187     )
 188     _borrow_event_class_ptr_by_index = staticmethod(
 189         native_bt.stream_class_borrow_event_class_by_index
 190     )
 191     _borrow_trace_class_ptr = staticmethod(native_bt.stream_class_borrow_trace_class)
 192     _borrow_packet_context_field_class_ptr = staticmethod(
 193         native_bt.stream_class_borrow_packet_context_field_class
 194     )
 195     _borrow_event_common_context_field_class_ptr = staticmethod(
 196         native_bt.stream_class_borrow_event_common_context_field_class
 197     )
 198     _borrow_default_clock_class_ptr = staticmethod(
 199         native_bt.stream_class_borrow_default_clock_class
 200     )
 201     _borrow_user_attributes_ptr = staticmethod(
 202         native_bt.stream_class_borrow_user_attributes
 203     )
 204
 205     _event_class_cls = property(lambda s: bt2_event_class._EventClass)
 206     _trace_class_cls = property(lambda s: _bt2_trace_class()._TraceClass)
 207     _clock_class_cls = property(lambda s: bt2_clock_class._ClockClass)
 208
 209     def create_event_class(
 210         self,
 211         id=None,
 212         name=None,
 213         user_attributes=None,
 214         log_level=None,
 215         emf_uri=None,
 216         specific_context_field_class=None,
 217         payload_field_class=None,
 218     ):
 219         # Validate parameters before we create the object.
 220         bt2_event_class._EventClass._validate_create_params(
 221             name,
 222             user_attributes,
 223             log_level,
 224             emf_uri,
 225             specific_context_field_class,
 226             payload_field_class,
 227         )
 228
 229         if self.assigns_automatic_event_class_id:
 230             if id is not None:
 231                 raise ValueError(
 232                     "id provided, but stream class assigns automatic event class ids"
 233                 )
 234
 235             ec_ptr = native_bt.event_class_create(self._ptr)
 236         else:
 237             if id is None:
 238                 raise ValueError(
 239                     "id not provided, but stream class does not assign automatic event class ids"
 240                 )
 241
 242             bt2_utils._check_uint64(id)
 243             ec_ptr = native_bt.event_class_create_with_id(self._ptr, id)
 244
 245         event_class = bt2_event_class._EventClass._create_from_ptr(ec_ptr)
 246
 247         if name is not None:
 248             event_class._name = name
 249
 250         if user_attributes is not None:
 251             event_class._user_attributes = user_attributes
 252
 253         if log_level is not None:
 254             event_class._log_level = log_level
 255
 256         if emf_uri is not None:
 257             event_class._emf_uri = emf_uri
 258
 259         if specific_context_field_class is not None:
 260             event_class._specific_context_field_class = specific_context_field_class
 261
 262         if payload_field_class is not None:
 263             event_class._payload_field_class = payload_field_class
 264
 265         return event_class
 266
 267     def _user_attributes(self, user_attributes):
 268         value = bt2_value.create_value(user_attributes)
 269         native_bt.stream_class_set_user_attributes(self._ptr, value._ptr)
 270
 271     _user_attributes = property(fset=_user_attributes)
 272
 273     def _name(self, name):
 274         status = native_bt.stream_class_set_name(self._ptr, name)
 275         bt2_utils._handle_func_status(status, "cannot set stream class object's name")
 276
 277     _name = property(fset=_name)
 278
 279     def _assigns_automatic_event_class_id(self, auto_id):
 280         native_bt.stream_class_set_assigns_automatic_event_class_id(self._ptr, auto_id)
 281
 282     _assigns_automatic_event_class_id = property(fset=_assigns_automatic_event_class_id)
 283
 284     def _assigns_automatic_stream_id(self, auto_id):
 285         native_bt.stream_class_set_assigns_automatic_stream_id(self._ptr, auto_id)
 286
 287     _assigns_automatic_stream_id = property(fset=_assigns_automatic_stream_id)
 288
 289     def _set_supports_packets(self, supports, with_begin_cs=False, with_end_cs=False):
 290         native_bt.stream_class_set_supports_packets(
 291             self._ptr, supports, with_begin_cs, with_end_cs
 292         )
 293
 294     def _set_supports_discarded_events(self, supports, with_cs=False):
 295         native_bt.stream_class_set_supports_discarded_events(
 296             self._ptr, supports, with_cs
 297         )
 298
 299     _supports_discarded_events = property(fset=_set_supports_discarded_events)
 300
 301     def _set_supports_discarded_packets(self, supports, with_cs):
 302         native_bt.stream_class_set_supports_discarded_packets(
 303             self._ptr, supports, with_cs
 304         )
 305
 306     _supports_discarded_packets = property(fset=_set_supports_discarded_packets)
 307
 308     def _packet_context_field_class(self, packet_context_field_class):
 309         status = native_bt.stream_class_set_packet_context_field_class(
 310             self._ptr, packet_context_field_class._ptr
 311         )
 312         bt2_utils._handle_func_status(
 313             status, "cannot set stream class' packet context field class"
 314         )
 315
 316     _packet_context_field_class = property(fset=_packet_context_field_class)
 317
 318     def _event_common_context_field_class(self, event_common_context_field_class):
 319         set_context_fn = native_bt.stream_class_set_event_common_context_field_class
 320         status = set_context_fn(self._ptr, event_common_context_field_class._ptr)
 321         bt2_utils._handle_func_status(
 322             status, "cannot set stream class' event context field type"
 323         )
 324
 325     _event_common_context_field_class = property(fset=_event_common_context_field_class)
 326
 327     def _default_clock_class(self, clock_class):
 328         native_bt.stream_class_set_default_clock_class(self._ptr, clock_class._ptr)
 329
 330     _default_clock_class = property(fset=_default_clock_class)
 331
 332     @classmethod
 333     def _validate_create_params(
 334         cls,
 335         name,
 336         user_attributes,
 337         packet_context_field_class,
 338         event_common_context_field_class,
 339         default_clock_class,
 340         assigns_automatic_event_class_id,
 341         assigns_automatic_stream_id,
 342         supports_packets,
 343         packets_have_beginning_default_clock_snapshot,
 344         packets_have_end_default_clock_snapshot,
 345         supports_discarded_events,
 346         discarded_events_have_default_clock_snapshots,
 347         supports_discarded_packets,
 348         discarded_packets_have_default_clock_snapshots,
 349     ):
 350         # Name
 351         if name is not None:
 352             bt2_utils._check_str(name)
 353
 354         # User attributes
 355         if user_attributes is not None:
 356             value = bt2_value.create_value(user_attributes)
 357             bt2_utils._check_type(value, bt2_value.MapValue)
 358
 359         # Packet context field class
 360         if packet_context_field_class is not None:
 361             if not supports_packets:
 362                 raise ValueError(
 363                     "cannot have a packet context field class without supporting packets"
 364                 )
 365
 366             bt2_utils._check_type(
 367                 packet_context_field_class, bt2_field_class._StructureFieldClass
 368             )
 369
 370         # Event common context field class
 371         if event_common_context_field_class is not None:
 372             bt2_utils._check_type(
 373                 event_common_context_field_class, bt2_field_class._StructureFieldClass
 374             )
 375
 376         # Default clock class
 377         if default_clock_class is not None:
 378             bt2_utils._check_type(default_clock_class, bt2_clock_class._ClockClass)
 379
 380         # Assigns automatic event class id
 381         bt2_utils._check_bool(assigns_automatic_event_class_id)
 382
 383         # Assigns automatic stream id
 384         bt2_utils._check_bool(assigns_automatic_stream_id)
 385
 386         # Packets
 387         bt2_utils._check_bool(supports_packets)
 388         bt2_utils._check_bool(packets_have_beginning_default_clock_snapshot)
 389         bt2_utils._check_bool(packets_have_end_default_clock_snapshot)
 390
 391         if not supports_packets:
 392             if packets_have_beginning_default_clock_snapshot:
 393                 raise ValueError(
 394                     "cannot not support packets, but have packet beginning default clock snapshot"
 395                 )
 396             if packets_have_end_default_clock_snapshot:
 397                 raise ValueError(
 398                     "cannot not support packets, but have packet end default clock snapshots"
 399                 )
 400
 401         # Discarded events
 402         bt2_utils._check_bool(supports_discarded_events)
 403         bt2_utils._check_bool(discarded_events_have_default_clock_snapshots)
 404
 405         if discarded_events_have_default_clock_snapshots:
 406             if not supports_discarded_events:
 407                 raise ValueError(
 408                     "cannot not support discarded events, but have default clock snapshots for discarded event messages"
 409                 )
 410
 411             if default_clock_class is None:
 412                 raise ValueError(
 413                     "cannot have no default clock class, but have default clock snapshots for discarded event messages"
 414                 )
 415
 416         # Discarded packets
 417         bt2_utils._check_bool(supports_discarded_packets)
 418         bt2_utils._check_bool(discarded_packets_have_default_clock_snapshots)
 419
 420         if supports_discarded_packets and not supports_packets:
 421             raise ValueError(
 422                 "cannot support discarded packets, but not support packets"
 423             )
 424
 425         if discarded_packets_have_default_clock_snapshots:
 426             if not supports_discarded_packets:
 427                 raise ValueError(
 428                     "cannot not support discarded packets, but have default clock snapshots for discarded packet messages"
 429                 )
 430
 431             if default_clock_class is None:
 432                 raise ValueError(
 433                     "cannot have no default clock class, but have default clock snapshots for discarded packet messages"
 434                 )