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 collections.abc
  26 import bt2.ctf_writer
  27 import bt2.stream
  28 import copy
  29 import bt2
  30
  31
  32 class _EventClassIterator(collections.abc.Iterator):
  33     def __init__(self, stream_class):
  34         self._stream_class = stream_class
  35         self._at = 0
  36
  37     def __next__(self):
  38         if self._at == len(self._stream_class):
  39             raise StopIteration
  40
  41         ec_ptr = native_bt.stream_class_get_event_class_by_index(self._stream_class._ptr,
  42                                                                  self._at)
  43         assert(ec_ptr)
  44         ev_id = native_bt.event_class_get_id(ec_ptr)
  45         native_bt.put(ec_ptr)
  46         utils._handle_ret(ev_id, "cannot get event class object's ID")
  47         self._at += 1
  48         return ev_id
  49
  50
  51 class StreamClass(object._SharedObject, collections.abc.Mapping):
  52     _get_ref = staticmethod(native_bt.stream_class_get_ref)
  53     _put_ref = staticmethod(native_bt.stream_class_put_ref)
  54
  55     def __init__(self, name=None, id=None, packet_context_field_class=None,
  56                  event_header_field_class=None, event_context_field_class=None,
  57                  event_classes=None):
  58         ptr = native_bt.stream_class_create_empty(None)
  59
  60         if ptr is None:
  61             raise bt2.CreationError('cannot create stream class object')
  62
  63         super().__init__(ptr)
  64
  65         if name is not None:
  66             self.name = name
  67
  68         if id is not None:
  69             self.id = id
  70
  71         if packet_context_field_class is not None:
  72             self.packet_context_field_class = packet_context_field_class
  73
  74         if event_header_field_class is not None:
  75             self.event_header_field_class = event_header_field_class
  76
  77         if event_context_field_class is not None:
  78             self.event_context_field_class = event_context_field_class
  79
  80         if event_classes is not None:
  81             for event_class in event_classes:
  82                 self.add_event_class(event_class)
  83
  84     def __getitem__(self, key):
  85         utils._check_int64(key)
  86         ec_ptr = native_bt.stream_class_get_event_class_by_id(self._ptr,
  87                                                               key)
  88
  89         if ec_ptr is None:
  90             raise KeyError(key)
  91
  92         return bt2.EventClass._create_from_ptr(ec_ptr)
  93
  94     def __len__(self):
  95         count = native_bt.stream_class_get_event_class_count(self._ptr)
  96         assert(count >= 0)
  97         return count
  98
  99     def __iter__(self):
 100         return _EventClassIterator(self)
 101
 102     def add_event_class(self, event_class):
 103         utils._check_type(event_class, bt2.EventClass)
 104         ret = native_bt.stream_class_add_event_class(self._ptr, event_class._ptr)
 105         utils._handle_ret(ret, "cannot add event class object to stream class object's")
 106
 107     @property
 108     def trace(self):
 109         tc_ptr = native_bt.stream_class_get_trace(self._ptr)
 110
 111         if tc_ptr is not None:
 112             return bt2.Trace._create_from_ptr(tc_ptr)
 113
 114     @property
 115     def name(self):
 116         return native_bt.stream_class_get_name(self._ptr)
 117
 118     @name.setter
 119     def name(self, name):
 120         utils._check_str(name)
 121         ret = native_bt.stream_class_set_name(self._ptr, name)
 122         utils._handle_ret(ret, "cannot set stream class object's name")
 123
 124     @property
 125     def assigns_automatic_stream_id(self):
 126         return native_bt.stream_class_assigns_automatic_stream_id(self._ptr)
 127
 128     def _assigns_automatic_stream_id(self, auto_id):
 129         utils._check_bool(auto_id)
 130         return native_bt.stream_class_set_assigns_automatic_stream_id(self._ptr, auto_id)
 131
 132     _assigns_automatic_stream_id = property(fset=_assigns_automatic_stream_id)
 133
 134     @property
 135     def id(self):
 136         id = native_bt.stream_class_get_id(self._ptr)
 137
 138         if id < 0:
 139             return
 140
 141         return id
 142
 143     @id.setter
 144     def id(self, id):
 145         utils._check_int64(id)
 146         ret = native_bt.stream_class_set_id(self._ptr, id)
 147         utils._handle_ret(ret, "cannot set stream class object's ID")
 148
 149     @property
 150     def clock(self):
 151         clock_ptr = native_bt.stream_class_get_clock(self._ptr)
 152
 153         if clock_ptr is None:
 154             return
 155
 156         return bt2.ctf_writer.CtfWriterClock._create_from_ptr(clock_ptr)
 157
 158     @clock.setter
 159     def clock(self, clock):
 160         utils._check_type(clock, bt2.ctf_writer.CtfWriterClock)
 161         ret = native_bt.stream_class_set_clock(self._ptr, clock._ptr)
 162         utils._handle_ret(ret, "cannot set stream class object's CTF writer clock object")
 163
 164     @property
 165     def packet_context_field_class(self):
 166         fc_ptr = native_bt.stream_class_get_packet_context_type(self._ptr)
 167
 168         if fc_ptr is None:
 169             return
 170
 171         return bt2.field_class._create_from_ptr(fc_ptr)
 172
 173     @packet_context_field_class.setter
 174     def packet_context_field_class(self, packet_context_field_class):
 175         packet_context_field_class_ptr = None
 176
 177         if packet_context_field_class is not None:
 178             utils._check_type(packet_context_field_class, bt2.field_class._FieldClass)
 179             packet_context_field_class_ptr = packet_context_field_class._ptr
 180
 181         ret = native_bt.stream_class_set_packet_context_type(self._ptr,
 182                                                              packet_context_field_class_ptr)
 183         utils._handle_ret(ret, "cannot set stream class object's packet context field class")
 184
 185     @property
 186     def event_header_field_class(self):
 187         fc_ptr = native_bt.stream_class_get_event_header_type(self._ptr)
 188
 189         if fc_ptr is None:
 190             return
 191
 192         return bt2.field_class._create_from_ptr(fc_ptr)
 193
 194     @event_header_field_class.setter
 195     def event_header_field_class(self, event_header_field_class):
 196         event_header_field_class_ptr = None
 197
 198         if event_header_field_class is not None:
 199             utils._check_type(event_header_field_class, bt2.field_class._FieldClass)
 200             event_header_field_class_ptr = event_header_field_class._ptr
 201
 202         ret = native_bt.stream_class_set_event_header_type(self._ptr,
 203                                                            event_header_field_class_ptr)
 204         utils._handle_ret(ret, "cannot set stream class object's event header field class")
 205
 206     @property
 207     def event_context_field_class(self):
 208         fc_ptr = native_bt.stream_class_get_event_context_type(self._ptr)
 209
 210         if fc_ptr is None:
 211             return
 212
 213         return bt2.field_class._create_from_ptr(fc_ptr)
 214
 215     @event_context_field_class.setter
 216     def event_context_field_class(self, event_context_field_class):
 217         event_context_field_class_ptr = None
 218
 219         if event_context_field_class is not None:
 220             utils._check_type(event_context_field_class, bt2.field_class._FieldClass)
 221             event_context_field_class_ptr = event_context_field_class._ptr
 222
 223         ret = native_bt.stream_class_set_event_context_type(self._ptr,
 224                                                             event_context_field_class_ptr)
 225         utils._handle_ret(ret, "cannot set stream class object's event context field class")
 226
 227     def __call__(self, name=None, id=None):
 228         if name is not None:
 229             utils._check_str(name)
 230
 231         if id is None:
 232             stream_ptr = native_bt.stream_create(self._ptr, name)
 233         else:
 234             stream_ptr = native_bt.stream_create_with_id(self._ptr, name, id)
 235
 236         if stream_ptr is None:
 237             raise bt2.CreationError('cannot create stream object')
 238
 239         return bt2.stream._create_from_ptr(stream_ptr)
 240
 241     def __eq__(self, other):
 242         if type(other) is not type(self):
 243             return False
 244
 245         if self.addr == other.addr:
 246             return True
 247
 248         self_event_classes = list(self.values())
 249         other_event_classes = list(other.values())
 250         self_props = (
 251             self_event_classes,
 252             self.name,
 253             self.id,
 254             self.packet_context_field_class,
 255             self.event_header_field_class,
 256             self.event_context_field_class,
 257             self.clock,
 258         )
 259         other_props = (
 260             other_event_classes,
 261             other.name,
 262             other.id,
 263             other.packet_context_field_class,
 264             other.event_header_field_class,
 265             other.event_context_field_class,
 266             other.clock,
 267         )
 268
 269         return self_props == other_props
 270
 271     def _copy(self, fc_copy_func, ev_copy_func):
 272         cpy = StreamClass()
 273
 274         if self.id is not None:
 275             cpy.id = self.id
 276
 277         if self.name is not None:
 278             cpy.name = self.name
 279
 280         if self.clock is not None:
 281             cpy.clock = self.clock
 282
 283         cpy.packet_context_field_class = fc_copy_func(self.packet_context_field_class)
 284         cpy.event_header_field_class = fc_copy_func(self.event_header_field_class)
 285         cpy.event_context_field_class = fc_copy_func(self.event_context_field_class)
 286
 287         for event_class in self.values():
 288             cpy.add_event_class(ev_copy_func(event_class))
 289
 290         return cpy
 291
 292     def __copy__(self):
 293         return self._copy(lambda fc: fc, copy.copy)
 294
 295     def __deepcopy__(self, memo):
 296         cpy = self._copy(copy.deepcopy, copy.deepcopy)
 297         memo[id(self)] = cpy
 298         return cpy