bindings/python/bt2/trace.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.values
  27 import bt2.stream
  28 import copy
  29 import bt2
  30
  31
  32 class _StreamClassIterator(collections.abc.Iterator):
  33     def __init__(self, trace):
  34         self._trace = trace
  35         self._at = 0
  36
  37     def __next__(self):
  38         if self._at == len(self._trace):
  39             raise StopIteration
  40
  41         sc_ptr = native_bt.ctf_trace_get_stream_class_by_index(self._trace._ptr,
  42                                                                self._at)
  43         assert(sc_ptr)
  44         id = native_bt.ctf_stream_class_get_id(sc_ptr)
  45         native_bt.put(sc_ptr)
  46         assert(id >= 0)
  47         self._at += 1
  48         return id
  49
  50
  51 class _TraceStreams(collections.abc.Sequence):
  52     def __init__(self, trace):
  53         self._trace = trace
  54
  55     def __len__(self):
  56         count = native_bt.ctf_trace_get_stream_count(self._trace._ptr)
  57         assert(count >= 0)
  58         return count
  59
  60     def __getitem__(self, index):
  61         utils._check_uint64(index)
  62
  63         if index >= len(self):
  64             raise IndexError
  65
  66         stream_ptr = native_bt.ctf_trace_get_stream_by_index(self._trace._ptr,
  67                                                              index)
  68         assert(stream_ptr)
  69         return bt2.stream._create_from_ptr(stream_ptr)
  70
  71
  72 class _TraceClockClassesIterator(collections.abc.Iterator):
  73     def __init__(self, trace_clock_classes):
  74         self._trace_clock_classes = trace_clock_classes
  75         self._at = 0
  76
  77     def __next__(self):
  78         if self._at == len(self._trace_clock_classes):
  79             raise StopIteration
  80
  81         trace_ptr = self._trace_clock_classes._trace._ptr
  82         cc_ptr = native_bt.ctf_trace_get_clock_class_by_index(trace_ptr, self._at)
  83         assert(cc_ptr)
  84         name = native_bt.ctf_clock_class_get_name(cc_ptr)
  85         native_bt.put(cc_ptr)
  86         assert(name is not None)
  87         self._at += 1
  88         return name
  89
  90
  91 class _TraceClockClasses(collections.abc.Mapping):
  92     def __init__(self, trace):
  93         self._trace = trace
  94
  95     def __getitem__(self, key):
  96         utils._check_str(key)
  97         cc_ptr = native_bt.ctf_trace_get_clock_class_by_name(self._trace._ptr,
  98                                                              key)
  99
 100         if cc_ptr is None:
 101             raise KeyError(key)
 102
 103         return bt2.ClockClass._create_from_ptr(cc_ptr)
 104
 105     def __len__(self):
 106         count = native_bt.ctf_trace_get_clock_class_count(self._trace._ptr)
 107         assert(count >= 0)
 108         return count
 109
 110     def __iter__(self):
 111         return _TraceClockClassesIterator(self)
 112
 113
 114 class _TraceEnvIterator(collections.abc.Iterator):
 115     def __init__(self, trace_env):
 116         self._trace_env = trace_env
 117         self._at = 0
 118
 119     def __next__(self):
 120         if self._at == len(self._trace_env):
 121             raise StopIteration
 122
 123         trace_ptr = self._trace_env._trace._ptr
 124         entry_name = native_bt.ctf_trace_get_environment_field_name_by_index(trace_ptr,
 125                                                                              self._at)
 126         assert(entry_name is not None)
 127         self._at += 1
 128         return entry_name
 129
 130
 131 class _TraceEnv(collections.abc.MutableMapping):
 132     def __init__(self, trace):
 133         self._trace = trace
 134
 135     def __getitem__(self, key):
 136         utils._check_str(key)
 137         value_ptr = native_bt.ctf_trace_get_environment_field_value_by_name(self._trace._ptr,
 138                                                                             key)
 139
 140         if value_ptr is None:
 141             raise KeyError(key)
 142
 143         return bt2.values._create_from_ptr(value_ptr)
 144
 145     def __setitem__(self, key, value):
 146         utils._check_str(key)
 147         value = bt2.create_value(value)
 148         ret = native_bt.ctf_trace_set_environment_field(self._trace._ptr,
 149                                                         key, value._ptr)
 150         utils._handle_ret(ret, "cannot set trace class object's environment entry")
 151
 152     def __delitem__(self, key):
 153         raise NotImplementedError
 154
 155     def __len__(self):
 156         count = native_bt.ctf_trace_get_environment_field_count(self._trace._ptr)
 157         assert(count >= 0)
 158         return count
 159
 160     def __iter__(self):
 161         return _TraceEnvIterator(self)
 162
 163
 164 class Trace(object._Object, collections.abc.Mapping):
 165     def __init__(self, name=None, native_byte_order=None, env=None,
 166                  packet_header_field_type=None, clock_classes=None,
 167                  stream_classes=None):
 168         ptr = native_bt.ctf_trace_create()
 169
 170         if ptr is None:
 171             raise bt2.CreationError('cannot create trace class object')
 172
 173         super().__init__(ptr)
 174
 175         if name is not None:
 176             self.name = name
 177
 178         if native_byte_order is not None:
 179             self.native_byte_order = native_byte_order
 180
 181         if packet_header_field_type is not None:
 182             self.packet_header_field_type = packet_header_field_type
 183
 184         if env is not None:
 185             for key, value in env.items():
 186                 self.env[key] = value
 187
 188         if clock_classes is not None:
 189             for clock_class in clock_classes:
 190                 self.add_clock_class(clock_class)
 191
 192         if stream_classes is not None:
 193             for stream_class in stream_classes:
 194                 self.add_stream_class(stream_class)
 195
 196     def __getitem__(self, key):
 197         utils._check_int64(key)
 198         sc_ptr = native_bt.ctf_trace_get_stream_class_by_id(self._ptr, key)
 199
 200         if sc_ptr is None:
 201             raise KeyError(key)
 202
 203         return bt2.StreamClass._create_from_ptr(sc_ptr)
 204
 205     def __len__(self):
 206         count = native_bt.ctf_trace_get_stream_class_count(self._ptr)
 207         assert(count >= 0)
 208         return count
 209
 210     def __iter__(self):
 211         return _StreamClassIterator(self)
 212
 213     def add_stream_class(self, stream_class):
 214         utils._check_type(stream_class, bt2.StreamClass)
 215         ret = native_bt.ctf_trace_add_stream_class(self._ptr, stream_class._ptr)
 216         utils._handle_ret(ret, "cannot add stream class object to trace class object")
 217
 218     @property
 219     def name(self):
 220         return native_bt.ctf_trace_get_name(self._ptr)
 221
 222     @name.setter
 223     def name(self, name):
 224         utils._check_str(name)
 225         ret = native_bt.ctf_trace_set_name(self._ptr, name)
 226         utils._handle_ret(ret, "cannot set trace class object's name")
 227
 228     @property
 229     def native_byte_order(self):
 230         bo = native_bt.ctf_trace_get_native_byte_order(self._ptr)
 231         assert(bo >= 0)
 232         return bo
 233
 234     @native_byte_order.setter
 235     def native_byte_order(self, native_byte_order):
 236         utils._check_int(native_byte_order)
 237         ret = native_bt.ctf_trace_set_native_byte_order(self._ptr, native_byte_order)
 238         utils._handle_ret(ret, "cannot set trace class object's native byte order")
 239
 240     @property
 241     def is_static(self):
 242         is_static = native_bt.ctf_trace_is_static(self._ptr)
 243         return is_static > 0
 244
 245     def set_is_static(self):
 246         ret = native_bt.ctf_trace_set_is_static(self._ptr)
 247         utils._handle_ret(ret, "cannot set trace object as static")
 248
 249     @property
 250     def env(self):
 251         return _TraceEnv(self)
 252
 253     @property
 254     def clock_classes(self):
 255         return _TraceClockClasses(self)
 256
 257     def add_clock_class(self, clock_class):
 258         utils._check_type(clock_class, bt2.ClockClass)
 259         ret = native_bt.ctf_trace_add_clock_class(self._ptr, clock_class._ptr)
 260         utils._handle_ret(ret, "cannot add clock class object to trace class object")
 261
 262     @property
 263     def streams(self):
 264         return _TraceStreams(self)
 265
 266     @property
 267     def packet_header_field_type(self):
 268         ft_ptr = native_bt.ctf_trace_get_packet_header_type(self._ptr)
 269
 270         if ft_ptr is None:
 271             return
 272
 273         return bt2.field_types._create_from_ptr(ft_ptr)
 274
 275     @packet_header_field_type.setter
 276     def packet_header_field_type(self, packet_header_field_type):
 277         packet_header_field_type_ptr = None
 278
 279         if packet_header_field_type is not None:
 280             utils._check_type(packet_header_field_type, bt2.field_types._FieldType)
 281             packet_header_field_type_ptr = packet_header_field_type._ptr
 282
 283         ret = native_bt.ctf_trace_set_packet_header_type(self._ptr,
 284                                                          packet_header_field_type_ptr)
 285         utils._handle_ret(ret, "cannot set trace class object's packet header field type")
 286
 287     def __eq__(self, other):
 288         if type(other) is not type(self):
 289             # not comparing apples to apples
 290             return False
 291
 292         if self.addr == other.addr:
 293             return True
 294
 295         self_stream_classes = list(self.values())
 296         self_clock_classes = list(self.clock_classes.values())
 297         self_env = {key: val for key, val in self.env.items()}
 298         other_stream_classes = list(other.values())
 299         other_clock_classes = list(other.clock_classes.values())
 300         other_env = {key: val for key, val in other.env.items()}
 301         self_props = (
 302             self_stream_classes,
 303             self_clock_classes,
 304             self_env,
 305             self.name,
 306             self.native_byte_order,
 307             self.packet_header_field_type,
 308         )
 309         other_props = (
 310             other_stream_classes,
 311             other_clock_classes,
 312             other_env,
 313             other.name,
 314             other.native_byte_order,
 315             other.packet_header_field_type,
 316         )
 317         return self_props == other_props
 318
 319     def _copy(self, gen_copy_func, sc_copy_func):
 320         cpy = Trace()
 321
 322         if self.name is not None:
 323             cpy.name = self.name
 324
 325         cpy.packet_header_field_type = gen_copy_func(self.packet_header_field_type)
 326
 327         for key, val in self.env.items():
 328             cpy.env[key] = gen_copy_func(val)
 329
 330         for clock_class in self.clock_classes.values():
 331             cpy.add_clock_class(gen_copy_func(clock_class))
 332
 333         for stream_class in self.values():
 334             cpy.add_stream_class(sc_copy_func(stream_class))
 335
 336         return cpy
 337
 338     def __copy__(self):
 339         return self._copy(lambda obj: obj, copy.copy)
 340
 341     def __deepcopy__(self, memo):
 342         cpy = self._copy(copy.deepcopy, copy.deepcopy)
 343         memo[id(self)] = cpy
 344         return cpy