bindings/python/bt2/bt2/ctf_writer.py

   1 # The MIT License (MIT)
   2 #
   3 # Copyright (c) 2016-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, stream, utils
  24 import uuid as uuidp
  25 import bt2.event
  26 import abc
  27 import bt2
  28
  29
  30 class CtfWriterClock(object._Object):
  31     def __init__(self, name, description=None, frequency=None, precision=None,
  32                  offset=None, is_absolute=None, uuid=None):
  33         utils._check_str(name)
  34         ptr = native_bt.ctf_clock_create(name)
  35
  36         if ptr is None:
  37             raise bt2.CreationError('cannot create CTF writer clock object')
  38
  39         super().__init__(ptr)
  40
  41         if description is not None:
  42             self.description = description
  43
  44         if frequency is not None:
  45             self.frequency = frequency
  46
  47         if precision is not None:
  48             self.precision = precision
  49
  50         if offset is not None:
  51             self.offset = offset
  52
  53         if is_absolute is not None:
  54             self.is_absolute = is_absolute
  55
  56         if uuid is not None:
  57             self.uuid = uuid
  58
  59     def __eq__(self, other):
  60         if type(self) is not type(other):
  61             # not comparing apples to apples
  62             return False
  63
  64         if self.addr == other.addr:
  65             return True
  66
  67         self_props = (
  68             self.name,
  69             self.description,
  70             self.frequency,
  71             self.precision,
  72             self.offset,
  73             self.is_absolute,
  74             self.uuid
  75         )
  76         other_props = (
  77             other.name,
  78             other.description,
  79             other.frequency,
  80             other.precision,
  81             other.offset,
  82             other.is_absolute,
  83             other.uuid
  84         )
  85         return self_props == other_props
  86
  87     def __copy__(self):
  88         return CtfWriterClock(name=self.name, description=self.description,
  89                               frequency=self.frequency,
  90                               precision=self.precision, offset=self.offset,
  91                               is_absolute=self.is_absolute, uuid=self.uuid)
  92
  93     def __deepcopy__(self, memo):
  94         cpy = self.__copy__()
  95         memo[id(self)] = cpy
  96         return cpy
  97
  98     @property
  99     def name(self):
 100         name = native_bt.ctf_clock_get_name(self._ptr)
 101         assert(name is not None)
 102         return name
 103
 104     @property
 105     def description(self):
 106         description = native_bt.ctf_clock_get_description(self._ptr)
 107         return description
 108
 109     @description.setter
 110     def description(self, description):
 111         utils._check_str(description)
 112         ret = native_bt.ctf_clock_set_description(self._ptr, description)
 113         utils._handle_ret(ret, "cannot set CTF writer clock object's description")
 114
 115     @property
 116     def frequency(self):
 117         frequency = native_bt.ctf_clock_get_frequency(self._ptr)
 118         assert(frequency >= 1)
 119         return frequency
 120
 121     @frequency.setter
 122     def frequency(self, frequency):
 123         utils._check_uint64(frequency)
 124         ret = native_bt.ctf_clock_set_frequency(self._ptr, frequency)
 125         utils._handle_ret(ret, "cannot set CTF writer clock object's frequency")
 126
 127     @property
 128     def precision(self):
 129         precision = native_bt.ctf_clock_get_precision(self._ptr)
 130         assert(precision >= 0)
 131         return precision
 132
 133     @precision.setter
 134     def precision(self, precision):
 135         utils._check_uint64(precision)
 136         ret = native_bt.ctf_clock_set_precision(self._ptr, precision)
 137         utils._handle_ret(ret, "cannot set CTF writer clock object's precision")
 138
 139     @property
 140     def offset(self):
 141         ret, offset_s = native_bt.ctf_clock_get_offset_s(self._ptr)
 142         assert(ret == 0)
 143         ret, offset_cycles = native_bt.ctf_clock_get_offset(self._ptr)
 144         assert(ret == 0)
 145         return bt2.ClockClassOffset(offset_s, offset_cycles)
 146
 147     @offset.setter
 148     def offset(self, offset):
 149         utils._check_type(offset, bt2.ClockClassOffset)
 150         ret = native_bt.ctf_clock_set_offset_s(self._ptr, offset.seconds)
 151         utils._handle_ret(ret, "cannot set CTF writer clock object's offset (seconds)")
 152         ret = native_bt.ctf_clock_set_offset(self._ptr, offset.cycles)
 153         utils._handle_ret(ret, "cannot set CTF writer clock object's offset (cycles)")
 154
 155     @property
 156     def is_absolute(self):
 157         is_absolute = native_bt.ctf_clock_get_is_absolute(self._ptr)
 158         assert(is_absolute >= 0)
 159         return is_absolute > 0
 160
 161     @is_absolute.setter
 162     def is_absolute(self, is_absolute):
 163         utils._check_bool(is_absolute)
 164         ret = native_bt.ctf_clock_set_is_absolute(self._ptr, int(is_absolute))
 165         utils._handle_ret(ret, "cannot set CTF writer clock object's absoluteness")
 166
 167     @property
 168     def uuid(self):
 169         uuid_bytes = native_bt.ctf_clock_get_uuid(self._ptr)
 170         assert(uuid_bytes is not None)
 171         return uuidp.UUID(bytes=uuid_bytes)
 172
 173     @uuid.setter
 174     def uuid(self, uuid):
 175         utils._check_type(uuid, uuidp.UUID)
 176         ret = native_bt.ctf_clock_set_uuid(self._ptr, uuid.bytes)
 177         utils._handle_ret(ret, "cannot set CTF writer clock object's UUID")
 178
 179     def _time(self, time):
 180         utils._check_int64(time)
 181         ret = native_bt.ctf_clock_set_time(self._ptr, time)
 182
 183     time = property(fset=_time)
 184
 185
 186 class _CtfWriterStream(stream._StreamBase):
 187     @property
 188     def discarded_events_count(self):
 189         ret, count = native_bt.ctf_stream_get_discarded_events_count(self._ptr)
 190         utils._handle_ret(ret, "cannot get CTF writer stream object's discarded events count")
 191         return count
 192
 193     def append_discarded_events(self, count):
 194         utils._check_uint64(count)
 195         native_bt.ctf_stream_append_discarded_events(self._ptr, count)
 196
 197     def append_event(self, event):
 198         utils._check_type(event, bt2.event._Event)
 199         ret = native_bt.ctf_stream_append_event(self._ptr, event._ptr)
 200         utils._handle_ret(ret, 'cannot append event object to CTF writer stream object')
 201
 202     def flush(self):
 203         ret = native_bt.ctf_stream_flush(self._ptr)
 204         utils._handle_ret(ret, 'cannot cannot flush CTF writer stream object')
 205
 206     @property
 207     def packet_header_field(self):
 208         field_ptr = native_bt.ctf_stream_get_packet_header(self._ptr)
 209
 210         if field_ptr is None:
 211             return
 212
 213         return fields._create_from_ptr(field_ptr)
 214
 215     @packet_header_field.setter
 216     def packet_header_field(self, packet_header_field):
 217         packet_header_field_ptr = None
 218
 219         if packet_header_field is not None:
 220             utils._check_type(packet_header_field, fields._Field)
 221             packet_header_field_ptr = packet_header_field._ptr
 222
 223         ret = native_bt.ctf_stream_set_packet_header(self._ptr,
 224                                                      packet_header_field_ptr)
 225         utils._handle_ret(ret, "cannot set CTF writer stream object's packet header field")
 226
 227     @property
 228     def packet_context_field(self):
 229         field_ptr = native_bt.ctf_stream_get_packet_context(self._ptr)
 230
 231         if field_ptr is None:
 232             return
 233
 234         return fields._create_from_ptr(field_ptr)
 235
 236     @packet_context_field.setter
 237     def packet_context_field(self, packet_context_field):
 238         packet_context_field_ptr = None
 239
 240         if packet_context_field is not None:
 241             utils._check_type(packet_context_field, fields._Field)
 242             packet_context_field_ptr = packet_context_field._ptr
 243
 244         ret = native_bt.ctf_stream_set_packet_context(self._ptr,
 245                                                       packet_context_field_ptr)
 246         utils._handle_ret(ret, "cannot set CTF writer stream object's packet context field")
 247
 248     def __eq__(self, other):
 249         if type(other) is not type(self):
 250             return False
 251
 252         if self.addr == other.addr:
 253             return True
 254
 255         if not _StreamBase.__eq__(self, other):
 256             return False
 257
 258         self_props = (
 259             self.discarded_events_count,
 260             self.packet_header_field,
 261             self.packet_context_field,
 262         )
 263         other_props = (
 264             other.discarded_events_count,
 265             other.packet_header_field,
 266             other.packet_context_field,
 267         )
 268         return self_props == other_props
 269
 270     def _copy(self, copy_func):
 271         cpy = self.stream_class(self.name)
 272         cpy.append_discarded_events(self.discarded_events_count)
 273         cpy.packet_header_field = copy_func(self.packet_header_field)
 274         cpy.packet_context_field = copy_func(self.packet_context_field)
 275         return cpy
 276
 277     def __copy__(self):
 278         return self._copy(copy.copy)
 279
 280     def __deepcopy__(self, memo):
 281         cpy = self._copy(copy.deepcopy)
 282         memo[id(self)] = cpy
 283         return cpy
 284
 285
 286 class CtfWriter(object._Object):
 287     def __init__(self, path):
 288         utils._check_str(path)
 289         ptr = native_bt.ctf_writer_create(path)
 290
 291         if ptr is None:
 292             raise bt2.CreationError('cannot create CTF writer object')
 293
 294         super().__init__(ptr)
 295
 296     @property
 297     def trace(self):
 298         trace_ptr = native_bt.ctf_writer_get_trace(self._ptr)
 299         assert(trace_ptr)
 300         return bt2.Trace._create_from_ptr(trace_ptr)
 301
 302     @property
 303     def metadata_string(self):
 304         metadata_string = native_bt.ctf_writer_get_metadata_string(self._ptr)
 305         assert(metadata_string is not None)
 306         return metadata_string
 307
 308     def flush_metadata(self):
 309         native_bt.ctf_writer_flush_metadata(self._ptr)
 310
 311     def add_clock(self, clock):
 312         utils._check_type(clock, CtfWriterClock)
 313         ret = native_bt.ctf_writer_add_clock(self._ptr, clock._ptr)
 314         utils._handle_ret(ret, 'cannot add CTF writer clock object to CTF writer object')