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

   1 # The MIT License (MIT)
   2 #
   3 # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
   4 #
   5 # Permission is hereby granted, free of charge, to any person obtaining a copy
   6 # of this software and associated documentation files (the "Software"), to deal
   7 # in the Software without restriction, including without limitation the rights
   8 # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   9 # copies of the Software, and to permit persons to whom the Software is
  10 # furnished to do so, subject to the following conditions:
  11 #
  12 # The above copyright notice and this permission notice shall be included in
  13 # all copies or substantial portions of the Software.
  14 #
  15 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  16 # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  17 # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  18 # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  19 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  20 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  21 # THE SOFTWARE.
  22
  23 from bt2 import native_bt, object, utils
  24 from bt2 import field_class as bt2_field_class
  25 from bt2 import event_class as bt2_event_class
  26 from bt2 import trace_class as bt2_trace_class
  27 from bt2 import clock_class as bt2_clock_class
  28 from bt2 import value as bt2_value
  29 import collections.abc
  30
  31
  32 class _StreamClass(object._SharedObject, collections.abc.Mapping):
  33     _get_ref = staticmethod(native_bt.stream_class_get_ref)
  34     _put_ref = staticmethod(native_bt.stream_class_put_ref)
  35
  36     def __getitem__(self, key):
  37         utils._check_int64(key)
  38         ec_ptr = native_bt.stream_class_borrow_event_class_by_id(self._ptr, key)
  39
  40         if ec_ptr is None:
  41             raise KeyError(key)
  42
  43         return bt2_event_class._EventClass._create_from_ptr_and_get_ref(ec_ptr)
  44
  45     def __len__(self):
  46         count = native_bt.stream_class_get_event_class_count(self._ptr)
  47         assert count >= 0
  48         return count
  49
  50     def __iter__(self):
  51         for idx in range(len(self)):
  52             ec_ptr = native_bt.stream_class_borrow_event_class_by_index_const(
  53                 self._ptr, idx
  54             )
  55             assert ec_ptr is not None
  56
  57             id = native_bt.event_class_get_id(ec_ptr)
  58             assert id >= 0
  59
  60             yield id
  61
  62     def create_event_class(
  63         self,
  64         id=None,
  65         name=None,
  66         user_attributes=None,
  67         log_level=None,
  68         emf_uri=None,
  69         specific_context_field_class=None,
  70         payload_field_class=None,
  71     ):
  72         if self.assigns_automatic_event_class_id:
  73             if id is not None:
  74                 raise ValueError(
  75                     'id provided, but stream class assigns automatic event class ids'
  76                 )
  77
  78             ec_ptr = native_bt.event_class_create(self._ptr)
  79         else:
  80             if id is None:
  81                 raise ValueError(
  82                     'id not provided, but stream class does not assign automatic event class ids'
  83                 )
  84
  85             utils._check_uint64(id)
  86             ec_ptr = native_bt.event_class_create_with_id(self._ptr, id)
  87
  88         event_class = bt2_event_class._EventClass._create_from_ptr(ec_ptr)
  89
  90         if name is not None:
  91             event_class._name = name
  92
  93         if user_attributes is not None:
  94             event_class._user_attributes = user_attributes
  95
  96         if log_level is not None:
  97             event_class._log_level = log_level
  98
  99         if emf_uri is not None:
 100             event_class._emf_uri = emf_uri
 101
 102         if specific_context_field_class is not None:
 103             event_class._specific_context_field_class = specific_context_field_class
 104
 105         if payload_field_class is not None:
 106             event_class._payload_field_class = payload_field_class
 107
 108         return event_class
 109
 110     @property
 111     def trace_class(self):
 112         tc_ptr = native_bt.stream_class_borrow_trace_class_const(self._ptr)
 113
 114         if tc_ptr is not None:
 115             return bt2_trace_class._TraceClass._create_from_ptr_and_get_ref(tc_ptr)
 116
 117     @property
 118     def user_attributes(self):
 119         ptr = native_bt.stream_class_borrow_user_attributes(self._ptr)
 120         assert ptr is not None
 121         return bt2_value._create_from_ptr_and_get_ref(ptr)
 122
 123     def _user_attributes(self, user_attributes):
 124         value = bt2_value.create_value(user_attributes)
 125         utils._check_type(value, bt2_value.MapValue)
 126         native_bt.stream_class_set_user_attributes(self._ptr, value._ptr)
 127
 128     _user_attributes = property(fset=_user_attributes)
 129
 130     @property
 131     def name(self):
 132         return native_bt.stream_class_get_name(self._ptr)
 133
 134     def _name(self, name):
 135         utils._check_str(name)
 136         status = native_bt.stream_class_set_name(self._ptr, name)
 137         utils._handle_func_status(status, "cannot set stream class object's name")
 138
 139     _name = property(fset=_name)
 140
 141     @property
 142     def assigns_automatic_event_class_id(self):
 143         return native_bt.stream_class_assigns_automatic_event_class_id(self._ptr)
 144
 145     def _assigns_automatic_event_class_id(self, auto_id):
 146         utils._check_bool(auto_id)
 147         return native_bt.stream_class_set_assigns_automatic_event_class_id(
 148             self._ptr, auto_id
 149         )
 150
 151     _assigns_automatic_event_class_id = property(fset=_assigns_automatic_event_class_id)
 152
 153     @property
 154     def assigns_automatic_stream_id(self):
 155         return native_bt.stream_class_assigns_automatic_stream_id(self._ptr)
 156
 157     def _assigns_automatic_stream_id(self, auto_id):
 158         utils._check_bool(auto_id)
 159         return native_bt.stream_class_set_assigns_automatic_stream_id(
 160             self._ptr, auto_id
 161         )
 162
 163     _assigns_automatic_stream_id = property(fset=_assigns_automatic_stream_id)
 164
 165     @property
 166     def supports_packets(self):
 167         return native_bt.stream_class_supports_packets(self._ptr)
 168
 169     @property
 170     def packets_have_beginning_default_clock_snapshot(self):
 171         return native_bt.stream_class_packets_have_beginning_default_clock_snapshot(
 172             self._ptr
 173         )
 174
 175     @property
 176     def packets_have_end_default_clock_snapshot(self):
 177         return native_bt.stream_class_packets_have_end_default_clock_snapshot(self._ptr)
 178
 179     def _set_supports_packets(self, supports, with_begin_cs=False, with_end_cs=False):
 180         utils._check_bool(supports)
 181         utils._check_bool(with_begin_cs)
 182         utils._check_bool(with_end_cs)
 183
 184         if not supports and (with_begin_cs or with_end_cs):
 185             raise ValueError(
 186                 'cannot not support packets, but have default clock snapshots'
 187             )
 188
 189         if not supports and self.packet_context_field_class is not None:
 190             raise ValueError('stream class already has a packet context field class')
 191
 192         native_bt.stream_class_set_supports_packets(
 193             self._ptr, supports, with_begin_cs, with_end_cs
 194         )
 195
 196     @property
 197     def supports_discarded_events(self):
 198         return native_bt.stream_class_supports_discarded_events(self._ptr)
 199
 200     def _set_supports_discarded_events(self, supports, with_cs=False):
 201         utils._check_bool(supports)
 202         utils._check_bool(with_cs)
 203
 204         if not supports and with_cs:
 205             raise ValueError(
 206                 'cannot not support discarded events, but have default clock snapshots'
 207             )
 208
 209         native_bt.stream_class_set_supports_discarded_events(
 210             self._ptr, supports, with_cs
 211         )
 212
 213     @property
 214     def discarded_events_have_default_clock_snapshots(self):
 215         return native_bt.stream_class_discarded_events_have_default_clock_snapshots(
 216             self._ptr
 217         )
 218
 219     @property
 220     def supports_discarded_packets(self):
 221         return native_bt.stream_class_supports_discarded_packets(self._ptr)
 222
 223     def _set_supports_discarded_packets(self, supports, with_cs):
 224         utils._check_bool(supports)
 225         utils._check_bool(with_cs)
 226
 227         if supports and not self.supports_packets:
 228             raise ValueError(
 229                 'cannot support discarded packets, but not support packets'
 230             )
 231
 232         if not supports and with_cs:
 233             raise ValueError(
 234                 'cannot not support discarded packets, but have default clock snapshots'
 235             )
 236
 237         native_bt.stream_class_set_supports_discarded_packets(
 238             self._ptr, supports, with_cs
 239         )
 240
 241     @property
 242     def discarded_packets_have_default_clock_snapshots(self):
 243         return native_bt.stream_class_discarded_packets_have_default_clock_snapshots(
 244             self._ptr
 245         )
 246
 247     @property
 248     def id(self):
 249         id = native_bt.stream_class_get_id(self._ptr)
 250
 251         if id < 0:
 252             return
 253
 254         return id
 255
 256     @property
 257     def packet_context_field_class(self):
 258         fc_ptr = native_bt.stream_class_borrow_packet_context_field_class_const(
 259             self._ptr
 260         )
 261
 262         if fc_ptr is None:
 263             return
 264
 265         return bt2_field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
 266
 267     def _packet_context_field_class(self, packet_context_field_class):
 268         if packet_context_field_class is not None:
 269             utils._check_type(
 270                 packet_context_field_class, bt2_field_class._StructureFieldClass
 271             )
 272
 273             if not self.supports_packets:
 274                 raise ValueError('stream class does not support packets')
 275
 276             status = native_bt.stream_class_set_packet_context_field_class(
 277                 self._ptr, packet_context_field_class._ptr
 278             )
 279             utils._handle_func_status(
 280                 status, "cannot set stream class' packet context field class"
 281             )
 282
 283     _packet_context_field_class = property(fset=_packet_context_field_class)
 284
 285     @property
 286     def event_common_context_field_class(self):
 287         fc_ptr = native_bt.stream_class_borrow_event_common_context_field_class_const(
 288             self._ptr
 289         )
 290
 291         if fc_ptr is None:
 292             return
 293
 294         return bt2_field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
 295
 296     def _event_common_context_field_class(self, event_common_context_field_class):
 297         if event_common_context_field_class is not None:
 298             utils._check_type(
 299                 event_common_context_field_class, bt2_field_class._StructureFieldClass
 300             )
 301
 302             set_context_fn = native_bt.stream_class_set_event_common_context_field_class
 303             status = set_context_fn(self._ptr, event_common_context_field_class._ptr)
 304             utils._handle_func_status(
 305                 status, "cannot set stream class' event context field type"
 306             )
 307
 308     _event_common_context_field_class = property(fset=_event_common_context_field_class)
 309
 310     @property
 311     def default_clock_class(self):
 312         cc_ptr = native_bt.stream_class_borrow_default_clock_class(self._ptr)
 313         if cc_ptr is None:
 314             return
 315
 316         return bt2_clock_class._ClockClass._create_from_ptr_and_get_ref(cc_ptr)
 317
 318     def _default_clock_class(self, clock_class):
 319         utils._check_type(clock_class, bt2_clock_class._ClockClass)
 320         native_bt.stream_class_set_default_clock_class(self._ptr, clock_class._ptr)
 321
 322     _default_clock_class = property(fset=_default_clock_class)