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 import bt2.field_class
  25 import bt2.event_class
  26 import bt2.trace_class
  27 import collections.abc
  28 import bt2.stream
  29 import bt2
  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         log_level=None,
  67         emf_uri=None,
  68         specific_context_field_class=None,
  69         payload_field_class=None,
  70     ):
  71         if self.assigns_automatic_event_class_id:
  72             if id is not None:
  73                 raise ValueError(
  74                     'id provided, but stream class assigns automatic event class ids'
  75                 )
  76
  77             ec_ptr = native_bt.event_class_create(self._ptr)
  78         else:
  79             if id is None:
  80                 raise ValueError(
  81                     'id not provided, but stream class does not assign automatic event class ids'
  82                 )
  83
  84             utils._check_uint64(id)
  85             ec_ptr = native_bt.event_class_create_with_id(self._ptr, id)
  86
  87         event_class = bt2.event_class._EventClass._create_from_ptr(ec_ptr)
  88
  89         if name is not None:
  90             event_class._name = name
  91
  92         if log_level is not None:
  93             event_class._log_level = log_level
  94
  95         if emf_uri is not None:
  96             event_class._emf_uri = emf_uri
  97
  98         if specific_context_field_class is not None:
  99             event_class._specific_context_field_class = specific_context_field_class
 100
 101         if payload_field_class is not None:
 102             event_class._payload_field_class = payload_field_class
 103
 104         return event_class
 105
 106     @property
 107     def trace_class(self):
 108         tc_ptr = native_bt.stream_class_borrow_trace_class_const(self._ptr)
 109
 110         if tc_ptr is not None:
 111             return bt2.trace_class._TraceClass._create_from_ptr_and_get_ref(tc_ptr)
 112
 113     @property
 114     def name(self):
 115         return native_bt.stream_class_get_name(self._ptr)
 116
 117     def _name(self, name):
 118         utils._check_str(name)
 119         status = native_bt.stream_class_set_name(self._ptr, name)
 120         utils._handle_func_status(status, "cannot set stream class object's name")
 121
 122     _name = property(fset=_name)
 123
 124     @property
 125     def assigns_automatic_event_class_id(self):
 126         return native_bt.stream_class_assigns_automatic_event_class_id(self._ptr)
 127
 128     def _assigns_automatic_event_class_id(self, auto_id):
 129         utils._check_bool(auto_id)
 130         return native_bt.stream_class_set_assigns_automatic_event_class_id(
 131             self._ptr, auto_id
 132         )
 133
 134     _assigns_automatic_event_class_id = property(fset=_assigns_automatic_event_class_id)
 135
 136     @property
 137     def assigns_automatic_stream_id(self):
 138         return native_bt.stream_class_assigns_automatic_stream_id(self._ptr)
 139
 140     def _assigns_automatic_stream_id(self, auto_id):
 141         utils._check_bool(auto_id)
 142         return native_bt.stream_class_set_assigns_automatic_stream_id(
 143             self._ptr, auto_id
 144         )
 145
 146     _assigns_automatic_stream_id = property(fset=_assigns_automatic_stream_id)
 147
 148     @property
 149     def supports_packets(self):
 150         return native_bt.stream_class_supports_packets(self._ptr)
 151
 152     @property
 153     def packets_have_beginning_default_clock_snapshot(self):
 154         return native_bt.stream_class_packets_have_beginning_default_clock_snapshot(
 155             self._ptr
 156         )
 157
 158     @property
 159     def packets_have_end_default_clock_snapshot(self):
 160         return native_bt.stream_class_packets_have_end_default_clock_snapshot(self._ptr)
 161
 162     def _set_supports_packets(self, supports, with_begin_cs=False, with_end_cs=False):
 163         utils._check_bool(supports)
 164         utils._check_bool(with_begin_cs)
 165         utils._check_bool(with_end_cs)
 166
 167         if not supports and (with_begin_cs or with_end_cs):
 168             raise ValueError(
 169                 'cannot not support packets, but have default clock snapshots'
 170             )
 171
 172         if not supports and self.packet_context_field_class is not None:
 173             raise ValueError('stream class already has a packet context field class')
 174
 175         native_bt.stream_class_set_supports_packets(
 176             self._ptr, supports, with_begin_cs, with_end_cs
 177         )
 178
 179     @property
 180     def supports_discarded_events(self):
 181         return native_bt.stream_class_supports_discarded_events(self._ptr)
 182
 183     def _set_supports_discarded_events(self, supports, with_cs=False):
 184         utils._check_bool(supports)
 185         utils._check_bool(with_cs)
 186
 187         if not supports and with_cs:
 188             raise ValueError(
 189                 'cannot not support discarded events, but have default clock snapshots'
 190             )
 191
 192         native_bt.stream_class_set_supports_discarded_events(
 193             self._ptr, supports, with_cs
 194         )
 195
 196     @property
 197     def discarded_events_have_default_clock_snapshots(self):
 198         return native_bt.stream_class_discarded_events_have_default_clock_snapshots(
 199             self._ptr
 200         )
 201
 202     @property
 203     def supports_discarded_packets(self):
 204         return native_bt.stream_class_supports_discarded_packets(self._ptr)
 205
 206     def _set_supports_discarded_packets(self, supports, with_cs):
 207         utils._check_bool(supports)
 208         utils._check_bool(with_cs)
 209
 210         if supports and not self.supports_packets:
 211             raise ValueError(
 212                 'cannot support discarded packets, but not support packets'
 213             )
 214
 215         if not supports and with_cs:
 216             raise ValueError(
 217                 'cannot not support discarded packets, but have default clock snapshots'
 218             )
 219
 220         native_bt.stream_class_set_supports_discarded_packets(
 221             self._ptr, supports, with_cs
 222         )
 223
 224     @property
 225     def discarded_packets_have_default_clock_snapshots(self):
 226         return native_bt.stream_class_discarded_packets_have_default_clock_snapshots(
 227             self._ptr
 228         )
 229
 230     @property
 231     def id(self):
 232         id = native_bt.stream_class_get_id(self._ptr)
 233
 234         if id < 0:
 235             return
 236
 237         return id
 238
 239     @id.setter
 240     def id(self, id):
 241         utils._check_int64(id)
 242         status = native_bt.stream_class_set_id(self._ptr, id)
 243         utils._handle_func_status(status, "cannot set stream class object's ID")
 244
 245     @property
 246     def packet_context_field_class(self):
 247         fc_ptr = native_bt.stream_class_borrow_packet_context_field_class_const(
 248             self._ptr
 249         )
 250
 251         if fc_ptr is None:
 252             return
 253
 254         return bt2.field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
 255
 256     def _packet_context_field_class(self, packet_context_field_class):
 257         if packet_context_field_class is not None:
 258             utils._check_type(
 259                 packet_context_field_class, bt2.field_class._StructureFieldClass
 260             )
 261
 262             if not self.supports_packets:
 263                 raise ValueError('stream class does not support packets')
 264
 265             status = native_bt.stream_class_set_packet_context_field_class(
 266                 self._ptr, packet_context_field_class._ptr
 267             )
 268             utils._handle_func_status(
 269                 status, "cannot set stream class' packet context field class"
 270             )
 271
 272     _packet_context_field_class = property(fset=_packet_context_field_class)
 273
 274     @property
 275     def event_common_context_field_class(self):
 276         fc_ptr = native_bt.stream_class_borrow_event_common_context_field_class_const(
 277             self._ptr
 278         )
 279
 280         if fc_ptr is None:
 281             return
 282
 283         return bt2.field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
 284
 285     def _event_common_context_field_class(self, event_common_context_field_class):
 286         if event_common_context_field_class is not None:
 287             utils._check_type(
 288                 event_common_context_field_class, bt2.field_class._StructureFieldClass
 289             )
 290
 291             set_context_fn = native_bt.stream_class_set_event_common_context_field_class
 292             status = set_context_fn(self._ptr, event_common_context_field_class._ptr)
 293             utils._handle_func_status(
 294                 status, "cannot set stream class' event context field type"
 295             )
 296
 297     _event_common_context_field_class = property(fset=_event_common_context_field_class)
 298
 299     @property
 300     def default_clock_class(self):
 301         cc_ptr = native_bt.stream_class_borrow_default_clock_class(self._ptr)
 302         if cc_ptr is None:
 303             return
 304
 305         return bt2.clock_class._ClockClass._create_from_ptr_and_get_ref(cc_ptr)
 306
 307     def _default_clock_class(self, clock_class):
 308         utils._check_type(clock_class, bt2.clock_class._ClockClass)
 309         native_bt.stream_class_set_default_clock_class(self._ptr, clock_class._ptr)
 310
 311     _default_clock_class = property(fset=_default_clock_class)