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_types
  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.ctf_stream_class_get_event_class_by_index(self._stream_class._ptr,
  42                                                                      self._at)
  43         assert(ec_ptr)
  44         ev_id = native_bt.ctf_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._Object, collections.abc.Mapping):
  52     def __init__(self, name=None, id=None, packet_context_field_type=None,
  53                  event_header_field_type=None, event_context_field_type=None,
  54                  event_classes=None):
  55         ptr = native_bt.ctf_stream_class_create_empty(None)
  56
  57         if ptr is None:
  58             raise bt2.CreationError('cannot create stream class object')
  59
  60         super().__init__(ptr)
  61
  62         if name is not None:
  63             self.name = name
  64
  65         if id is not None:
  66             self.id = id
  67
  68         if packet_context_field_type is not None:
  69             self.packet_context_field_type = packet_context_field_type
  70
  71         if event_header_field_type is not None:
  72             self.event_header_field_type = event_header_field_type
  73
  74         if event_context_field_type is not None:
  75             self.event_context_field_type = event_context_field_type
  76
  77         if event_classes is not None:
  78             for event_class in event_classes:
  79                 self.add_event_class(event_class)
  80
  81     def __getitem__(self, key):
  82         utils._check_int64(key)
  83         ec_ptr = native_bt.ctf_stream_class_get_event_class_by_id(self._ptr,
  84                                                                   key)
  85
  86         if ec_ptr is None:
  87             raise KeyError(key)
  88
  89         return bt2.EventClass._create_from_ptr(ec_ptr)
  90
  91     def __len__(self):
  92         count = native_bt.ctf_stream_class_get_event_class_count(self._ptr)
  93         assert(count >= 0)
  94         return count
  95
  96     def __iter__(self):
  97         return _EventClassIterator(self)
  98
  99     def add_event_class(self, event_class):
 100         utils._check_type(event_class, bt2.EventClass)
 101         ret = native_bt.ctf_stream_class_add_event_class(self._ptr, event_class._ptr)
 102         utils._handle_ret(ret, "cannot add event class object to stream class object's")
 103
 104     @property
 105     def trace(self):
 106         tc_ptr = native_bt.ctf_stream_class_get_trace(self._ptr)
 107
 108         if tc_ptr is not None:
 109             return bt2.Trace._create_from_ptr(tc_ptr)
 110
 111     @property
 112     def name(self):
 113         return native_bt.ctf_stream_class_get_name(self._ptr)
 114
 115     @name.setter
 116     def name(self, name):
 117         utils._check_str(name)
 118         ret = native_bt.ctf_stream_class_set_name(self._ptr, name)
 119         utils._handle_ret(ret, "cannot set stream class object's name")
 120
 121     @property
 122     def id(self):
 123         id = native_bt.ctf_stream_class_get_id(self._ptr)
 124
 125         if id < 0:
 126             return
 127
 128         return id
 129
 130     @id.setter
 131     def id(self, id):
 132         utils._check_int64(id)
 133         ret = native_bt.ctf_stream_class_set_id(self._ptr, id)
 134         utils._handle_ret(ret, "cannot set stream class object's ID")
 135
 136     @property
 137     def clock(self):
 138         clock_ptr = native_bt.ctf_stream_class_get_clock(self._ptr)
 139
 140         if clock_ptr is None:
 141             return
 142
 143         return bt2.ctf_writer.CtfWriterClock._create_from_ptr(clock_ptr)
 144
 145     @clock.setter
 146     def clock(self, clock):
 147         utils._check_type(clock, bt2.ctf_writer.CtfWriterClock)
 148         ret = native_bt.ctf_stream_class_set_clock(self._ptr, clock._ptr)
 149         utils._handle_ret(ret, "cannot set stream class object's CTF writer clock object")
 150
 151     @property
 152     def packet_context_field_type(self):
 153         ft_ptr = native_bt.ctf_stream_class_get_packet_context_type(self._ptr)
 154
 155         if ft_ptr is None:
 156             return
 157
 158         return bt2.field_types._create_from_ptr(ft_ptr)
 159
 160     @packet_context_field_type.setter
 161     def packet_context_field_type(self, packet_context_field_type):
 162         packet_context_field_type_ptr = None
 163
 164         if packet_context_field_type is not None:
 165             utils._check_type(packet_context_field_type, bt2.field_types._FieldType)
 166             packet_context_field_type_ptr = packet_context_field_type._ptr
 167
 168         ret = native_bt.ctf_stream_class_set_packet_context_type(self._ptr,
 169                                                                  packet_context_field_type_ptr)
 170         utils._handle_ret(ret, "cannot set stream class object's packet context field type")
 171
 172     @property
 173     def event_header_field_type(self):
 174         ft_ptr = native_bt.ctf_stream_class_get_event_header_type(self._ptr)
 175
 176         if ft_ptr is None:
 177             return
 178
 179         return bt2.field_types._create_from_ptr(ft_ptr)
 180
 181     @event_header_field_type.setter
 182     def event_header_field_type(self, event_header_field_type):
 183         event_header_field_type_ptr = None
 184
 185         if event_header_field_type is not None:
 186             utils._check_type(event_header_field_type, bt2.field_types._FieldType)
 187             event_header_field_type_ptr = event_header_field_type._ptr
 188
 189         ret = native_bt.ctf_stream_class_set_event_header_type(self._ptr,
 190                                                                event_header_field_type_ptr)
 191         utils._handle_ret(ret, "cannot set stream class object's event header field type")
 192
 193     @property
 194     def event_context_field_type(self):
 195         ft_ptr = native_bt.ctf_stream_class_get_event_context_type(self._ptr)
 196
 197         if ft_ptr is None:
 198             return
 199
 200         return bt2.field_types._create_from_ptr(ft_ptr)
 201
 202     @event_context_field_type.setter
 203     def event_context_field_type(self, event_context_field_type):
 204         event_context_field_type_ptr = None
 205
 206         if event_context_field_type is not None:
 207             utils._check_type(event_context_field_type, bt2.field_types._FieldType)
 208             event_context_field_type_ptr = event_context_field_type._ptr
 209
 210         ret = native_bt.ctf_stream_class_set_event_context_type(self._ptr,
 211                                                                event_context_field_type_ptr)
 212         utils._handle_ret(ret, "cannot set stream class object's event context field type")
 213
 214     def __call__(self, name=None, id=None):
 215         if name is not None:
 216             utils._check_str(name)
 217
 218         if id is None:
 219             stream_ptr = native_bt.ctf_stream_create(self._ptr, name)
 220         else:
 221             stream_ptr = native_bt.ctf_stream_create_with_id(self._ptr, name, id)
 222
 223         if stream_ptr is None:
 224             raise bt2.CreationError('cannot create stream object')
 225
 226         return bt2.stream._create_from_ptr(stream_ptr)
 227
 228     def __eq__(self, other):
 229         if type(other) is not type(self):
 230             return False
 231
 232         if self.addr == other.addr:
 233             return True
 234
 235         self_event_classes = list(self.values())
 236         other_event_classes = list(other.values())
 237         self_props = (
 238             self_event_classes,
 239             self.name,
 240             self.id,
 241             self.packet_context_field_type,
 242             self.event_header_field_type,
 243             self.event_context_field_type,
 244             self.clock,
 245         )
 246         other_props = (
 247             other_event_classes,
 248             other.name,
 249             other.id,
 250             other.packet_context_field_type,
 251             other.event_header_field_type,
 252             other.event_context_field_type,
 253             other.clock,
 254         )
 255
 256         return self_props == other_props
 257
 258     def _copy(self, ft_copy_func, ev_copy_func):
 259         cpy = StreamClass()
 260
 261         if self.id is not None:
 262             cpy.id = self.id
 263
 264         if self.name is not None:
 265             cpy.name = self.name
 266
 267         if self.clock is not None:
 268             cpy.clock = self.clock
 269
 270         cpy.packet_context_field_type = ft_copy_func(self.packet_context_field_type)
 271         cpy.event_header_field_type = ft_copy_func(self.event_header_field_type)
 272         cpy.event_context_field_type = ft_copy_func(self.event_context_field_type)
 273
 274         for event_class in self.values():
 275             cpy.add_event_class(ev_copy_func(event_class))
 276
 277         return cpy
 278
 279     def __copy__(self):
 280         return self._copy(lambda ft: ft, copy.copy)
 281
 282     def __deepcopy__(self, memo):
 283         cpy = self._copy(copy.deepcopy, copy.deepcopy)
 284         memo[id(self)] = cpy
 285         return cpy