src/bindings/python/bt2/bt2/message_iterator.py

   1 # SPDX-License-Identifier: MIT
   2 #
   3 # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
   4
   5 from bt2 import native_bt, object, utils
   6 from bt2 import message as bt2_message
   7 import collections.abc
   8 from bt2 import stream as bt2_stream
   9 from bt2 import event_class as bt2_event_class
  10 from bt2 import packet as bt2_packet
  11 from bt2 import port as bt2_port
  12 from bt2 import clock_class as bt2_clock_class
  13 import bt2
  14
  15
  16 class _MessageIterator(collections.abc.Iterator):
  17     def __next__(self):
  18         raise NotImplementedError
  19
  20
  21 class _UserComponentInputPortMessageIterator(object._SharedObject, _MessageIterator):
  22     _get_ref = staticmethod(native_bt.message_iterator_get_ref)
  23     _put_ref = staticmethod(native_bt.message_iterator_put_ref)
  24
  25     def __init__(self, ptr):
  26         self._current_msgs = []
  27         self._at = 0
  28         super().__init__(ptr)
  29
  30     def __next__(self):
  31         if len(self._current_msgs) == self._at:
  32             status, msgs = native_bt.bt2_self_component_port_input_get_msg_range(
  33                 self._ptr
  34             )
  35             utils._handle_func_status(
  36                 status, 'unexpected error: cannot advance the message iterator'
  37             )
  38             self._current_msgs = msgs
  39             self._at = 0
  40
  41         msg_ptr = self._current_msgs[self._at]
  42         self._at += 1
  43
  44         return bt2_message._create_from_ptr(msg_ptr)
  45
  46     def can_seek_beginning(self):
  47         (status, res) = native_bt.message_iterator_can_seek_beginning(self._ptr)
  48         utils._handle_func_status(
  49             status,
  50             'cannot check whether or not message iterator can seek its beginning',
  51         )
  52         return res != 0
  53
  54     def seek_beginning(self):
  55         # Forget about buffered messages, they won't be valid after seeking.
  56         self._current_msgs.clear()
  57         self._at = 0
  58
  59         status = native_bt.message_iterator_seek_beginning(self._ptr)
  60         utils._handle_func_status(status, 'cannot seek message iterator beginning')
  61
  62     def can_seek_ns_from_origin(self, ns_from_origin):
  63         utils._check_int64(ns_from_origin)
  64         (status, res) = native_bt.message_iterator_can_seek_ns_from_origin(
  65             self._ptr, ns_from_origin
  66         )
  67         utils._handle_func_status(
  68             status,
  69             'cannot check whether or not message iterator can seek given ns from origin',
  70         )
  71         return res != 0
  72
  73     def seek_ns_from_origin(self, ns_from_origin):
  74         utils._check_int64(ns_from_origin)
  75
  76         # Forget about buffered messages, they won't be valid after seeking.
  77         self._current_msgs.clear()
  78         self._at = 0
  79
  80         status = native_bt.message_iterator_seek_ns_from_origin(
  81             self._ptr, ns_from_origin
  82         )
  83         utils._handle_func_status(
  84             status, 'message iterator cannot seek given ns from origin'
  85         )
  86
  87     @property
  88     def can_seek_forward(self):
  89         return native_bt.message_iterator_can_seek_forward(self._ptr)
  90
  91
  92 class _MessageIteratorConfiguration:
  93     def __init__(self, ptr):
  94         self._ptr = ptr
  95
  96     def can_seek_forward(self, value):
  97         utils._check_bool(value)
  98         native_bt.self_message_iterator_configuration_set_can_seek_forward(
  99             self._ptr, value
 100         )
 101
 102     can_seek_forward = property(fset=can_seek_forward)
 103
 104
 105 # This is extended by the user to implement component classes in Python.  It
 106 # is created for a given output port when an input port message iterator is
 107 # created on the input port on the other side of the connection.  It is also
 108 # created when an output port message iterator is created on this output port.
 109 #
 110 # Its purpose is to feed the messages that should go out through this output
 111 # port.
 112 class _UserMessageIterator(_MessageIterator):
 113     def __new__(cls, ptr):
 114         # User iterator objects are always created by the native side,
 115         # that is, never instantiated directly by Python code.
 116         #
 117         # The native code calls this, then manually calls
 118         # self.__init__() without the `ptr` argument. The user has
 119         # access to self.component during this call, thanks to this
 120         # self._bt_ptr argument being set.
 121         #
 122         # self._bt_ptr is NOT owned by this object here, so there's nothing
 123         # to do in __del__().
 124         self = super().__new__(cls)
 125         self._bt_ptr = ptr
 126         return self
 127
 128     def _bt_init_from_native(self, config_ptr, self_output_port_ptr):
 129         self_output_port = bt2_port._create_self_from_ptr_and_get_ref(
 130             self_output_port_ptr, native_bt.PORT_TYPE_OUTPUT
 131         )
 132         config = _MessageIteratorConfiguration(config_ptr)
 133         self.__init__(config, self_output_port)
 134
 135     def __init__(self, config, self_output_port):
 136         pass
 137
 138     @property
 139     def _component(self):
 140         return native_bt.bt2_get_user_component_from_user_msg_iter(self._bt_ptr)
 141
 142     @property
 143     def _port(self):
 144         port_ptr = native_bt.self_message_iterator_borrow_port(self._bt_ptr)
 145         assert port_ptr is not None
 146         return bt2_port._create_self_from_ptr_and_get_ref(
 147             port_ptr, native_bt.PORT_TYPE_OUTPUT
 148         )
 149
 150     @property
 151     def addr(self):
 152         return int(self._bt_ptr)
 153
 154     @property
 155     def _is_interrupted(self):
 156         return bool(native_bt.self_message_iterator_is_interrupted(self._bt_ptr))
 157
 158     def _user_finalize(self):
 159         pass
 160
 161     def __next__(self):
 162         raise bt2.Stop
 163
 164     def _bt_next_from_native(self):
 165         # this can raise anything: it's catched by the native part
 166         try:
 167             msg = next(self)
 168         except StopIteration:
 169             raise bt2.Stop
 170         except Exception:
 171             raise
 172
 173         utils._check_type(msg, bt2_message._MessageConst)
 174
 175         # The reference we return will be given to the message array.
 176         # However, the `msg` Python object may stay alive, if the user has kept
 177         # a reference to it.  Acquire a new reference to account for that.
 178         msg._get_ref(msg._ptr)
 179         return int(msg._ptr)
 180
 181     def _bt_can_seek_beginning_from_native(self):
 182         # Here, we mimic the behavior of the C API:
 183         #
 184         # - If the iterator has a _user_can_seek_beginning method,
 185         #   read it and use that result.
 186         # - Otherwise, the presence or absence of a `_user_seek_beginning`
 187         #   method indicates whether the iterator can seek beginning.
 188         if hasattr(self, '_user_can_seek_beginning'):
 189             can_seek_beginning = self._user_can_seek_beginning()
 190             utils._check_bool(can_seek_beginning)
 191             return can_seek_beginning
 192         else:
 193             return hasattr(self, '_user_seek_beginning')
 194
 195     def _bt_seek_beginning_from_native(self):
 196         self._user_seek_beginning()
 197
 198     def _bt_can_seek_ns_from_origin_from_native(self, ns_from_origin):
 199         # Return whether the iterator can seek ns from origin using the
 200         # user-implemented seek_ns_from_origin method.  We mimic the behavior
 201         # of the C API:
 202         #
 203         # - If the iterator has a _user_can_seek_ns_from_origin method,
 204         #   call it and use its return value.
 205         # - Otherwise, if there is a `_user_seek_ns_from_origin` method,
 206         #   we presume it's possible.
 207
 208         if hasattr(self, '_user_can_seek_ns_from_origin'):
 209             can_seek_ns_from_origin = self._user_can_seek_ns_from_origin(ns_from_origin)
 210             utils._check_bool(can_seek_ns_from_origin)
 211             return can_seek_ns_from_origin
 212         else:
 213             return hasattr(self, '_user_seek_ns_from_origin')
 214
 215     def _bt_seek_ns_from_origin_from_native(self, ns_from_origin):
 216         self._user_seek_ns_from_origin(ns_from_origin)
 217
 218     def _create_message_iterator(self, input_port):
 219         utils._check_type(input_port, bt2_port._UserComponentInputPort)
 220
 221         if not input_port.is_connected:
 222             raise ValueError('input port is not connected')
 223
 224         (
 225             status,
 226             msg_iter_ptr,
 227         ) = native_bt.bt2_message_iterator_create_from_message_iterator(
 228             self._bt_ptr, input_port._ptr
 229         )
 230         utils._handle_func_status(status, 'cannot create message iterator object')
 231         assert msg_iter_ptr is not None
 232
 233         return _UserComponentInputPortMessageIterator(msg_iter_ptr)
 234
 235     def _create_event_message(self, event_class, parent, default_clock_snapshot=None):
 236         utils._check_type(event_class, bt2_event_class._EventClass)
 237
 238         if event_class.stream_class.supports_packets:
 239             utils._check_type(parent, bt2_packet._Packet)
 240         else:
 241             utils._check_type(parent, bt2_stream._Stream)
 242
 243         if default_clock_snapshot is not None:
 244             if event_class.stream_class.default_clock_class is None:
 245                 raise ValueError(
 246                     'event messages in this stream must not have a default clock snapshot'
 247                 )
 248
 249             utils._check_uint64(default_clock_snapshot)
 250
 251             if event_class.stream_class.supports_packets:
 252                 ptr = native_bt.message_event_create_with_packet_and_default_clock_snapshot(
 253                     self._bt_ptr, event_class._ptr, parent._ptr, default_clock_snapshot
 254                 )
 255             else:
 256                 ptr = native_bt.message_event_create_with_default_clock_snapshot(
 257                     self._bt_ptr, event_class._ptr, parent._ptr, default_clock_snapshot
 258                 )
 259         else:
 260             if event_class.stream_class.default_clock_class is not None:
 261                 raise ValueError(
 262                     'event messages in this stream must have a default clock snapshot'
 263                 )
 264
 265             if event_class.stream_class.supports_packets:
 266                 ptr = native_bt.message_event_create_with_packet(
 267                     self._bt_ptr, event_class._ptr, parent._ptr
 268                 )
 269             else:
 270                 ptr = native_bt.message_event_create(
 271                     self._bt_ptr, event_class._ptr, parent._ptr
 272                 )
 273
 274         if ptr is None:
 275             raise bt2._MemoryError('cannot create event message object')
 276
 277         return bt2_message._EventMessage(ptr)
 278
 279     def _create_message_iterator_inactivity_message(self, clock_class, clock_snapshot):
 280         utils._check_type(clock_class, bt2_clock_class._ClockClass)
 281         ptr = native_bt.message_message_iterator_inactivity_create(
 282             self._bt_ptr, clock_class._ptr, clock_snapshot
 283         )
 284
 285         if ptr is None:
 286             raise bt2._MemoryError('cannot create inactivity message object')
 287
 288         return bt2_message._MessageIteratorInactivityMessage(ptr)
 289
 290     def _create_stream_beginning_message(self, stream, default_clock_snapshot=None):
 291         utils._check_type(stream, bt2_stream._Stream)
 292
 293         ptr = native_bt.message_stream_beginning_create(self._bt_ptr, stream._ptr)
 294         if ptr is None:
 295             raise bt2._MemoryError('cannot create stream beginning message object')
 296
 297         msg = bt2_message._StreamBeginningMessage(ptr)
 298
 299         if default_clock_snapshot is not None:
 300             msg._default_clock_snapshot = default_clock_snapshot
 301
 302         return msg
 303
 304     def _create_stream_end_message(self, stream, default_clock_snapshot=None):
 305         utils._check_type(stream, bt2_stream._Stream)
 306
 307         ptr = native_bt.message_stream_end_create(self._bt_ptr, stream._ptr)
 308         if ptr is None:
 309             raise bt2._MemoryError('cannot create stream end message object')
 310
 311         msg = bt2_message._StreamEndMessage(ptr)
 312
 313         if default_clock_snapshot is not None:
 314             msg._default_clock_snapshot = default_clock_snapshot
 315
 316         return msg
 317
 318     def _create_packet_beginning_message(self, packet, default_clock_snapshot=None):
 319         utils._check_type(packet, bt2_packet._Packet)
 320
 321         if packet.stream.cls.packets_have_beginning_default_clock_snapshot:
 322             if default_clock_snapshot is None:
 323                 raise ValueError(
 324                     "packet beginning messages in this stream must have a default clock snapshot"
 325                 )
 326
 327             utils._check_uint64(default_clock_snapshot)
 328             ptr = native_bt.message_packet_beginning_create_with_default_clock_snapshot(
 329                 self._bt_ptr, packet._ptr, default_clock_snapshot
 330             )
 331         else:
 332             if default_clock_snapshot is not None:
 333                 raise ValueError(
 334                     "packet beginning messages in this stream must not have a default clock snapshot"
 335                 )
 336
 337             ptr = native_bt.message_packet_beginning_create(self._bt_ptr, packet._ptr)
 338
 339         if ptr is None:
 340             raise bt2._MemoryError('cannot create packet beginning message object')
 341
 342         return bt2_message._PacketBeginningMessage(ptr)
 343
 344     def _create_packet_end_message(self, packet, default_clock_snapshot=None):
 345         utils._check_type(packet, bt2_packet._Packet)
 346
 347         if packet.stream.cls.packets_have_end_default_clock_snapshot:
 348             if default_clock_snapshot is None:
 349                 raise ValueError(
 350                     "packet end messages in this stream must have a default clock snapshot"
 351                 )
 352
 353             utils._check_uint64(default_clock_snapshot)
 354             ptr = native_bt.message_packet_end_create_with_default_clock_snapshot(
 355                 self._bt_ptr, packet._ptr, default_clock_snapshot
 356             )
 357         else:
 358             if default_clock_snapshot is not None:
 359                 raise ValueError(
 360                     "packet end messages in this stream must not have a default clock snapshot"
 361                 )
 362
 363             ptr = native_bt.message_packet_end_create(self._bt_ptr, packet._ptr)
 364
 365         if ptr is None:
 366             raise bt2._MemoryError('cannot create packet end message object')
 367
 368         return bt2_message._PacketEndMessage(ptr)
 369
 370     def _create_discarded_events_message(
 371         self, stream, count=None, beg_clock_snapshot=None, end_clock_snapshot=None
 372     ):
 373         utils._check_type(stream, bt2_stream._Stream)
 374
 375         if not stream.cls.supports_discarded_events:
 376             raise ValueError('stream class does not support discarded events')
 377
 378         if stream.cls.discarded_events_have_default_clock_snapshots:
 379             if beg_clock_snapshot is None or end_clock_snapshot is None:
 380                 raise ValueError(
 381                     'discarded events have default clock snapshots for this stream class'
 382                 )
 383
 384             utils._check_uint64(beg_clock_snapshot)
 385             utils._check_uint64(end_clock_snapshot)
 386
 387             if beg_clock_snapshot > end_clock_snapshot:
 388                 raise ValueError(
 389                     'beginning default clock snapshot value ({}) is greater than end default clock snapshot value ({})'.format(
 390                         beg_clock_snapshot, end_clock_snapshot
 391                     )
 392                 )
 393
 394             ptr = native_bt.message_discarded_events_create_with_default_clock_snapshots(
 395                 self._bt_ptr, stream._ptr, beg_clock_snapshot, end_clock_snapshot
 396             )
 397         else:
 398             if beg_clock_snapshot is not None or end_clock_snapshot is not None:
 399                 raise ValueError(
 400                     'discarded events have no default clock snapshots for this stream class'
 401                 )
 402
 403             ptr = native_bt.message_discarded_events_create(self._bt_ptr, stream._ptr)
 404
 405         if ptr is None:
 406             raise bt2._MemoryError('cannot discarded events message object')
 407
 408         msg = bt2_message._DiscardedEventsMessage(ptr)
 409
 410         if count is not None:
 411             msg._count = count
 412
 413         return msg
 414
 415     def _create_discarded_packets_message(
 416         self, stream, count=None, beg_clock_snapshot=None, end_clock_snapshot=None
 417     ):
 418         utils._check_type(stream, bt2_stream._Stream)
 419
 420         if not stream.cls.supports_discarded_packets:
 421             raise ValueError('stream class does not support discarded packets')
 422
 423         if stream.cls.discarded_packets_have_default_clock_snapshots:
 424             if beg_clock_snapshot is None or end_clock_snapshot is None:
 425                 raise ValueError(
 426                     'discarded packets have default clock snapshots for this stream class'
 427                 )
 428
 429             utils._check_uint64(beg_clock_snapshot)
 430             utils._check_uint64(end_clock_snapshot)
 431
 432             if beg_clock_snapshot > end_clock_snapshot:
 433                 raise ValueError(
 434                     'beginning default clock snapshot value ({}) is greater than end default clock snapshot value ({})'.format(
 435                         beg_clock_snapshot, end_clock_snapshot
 436                     )
 437                 )
 438
 439             ptr = native_bt.message_discarded_packets_create_with_default_clock_snapshots(
 440                 self._bt_ptr, stream._ptr, beg_clock_snapshot, end_clock_snapshot
 441             )
 442         else:
 443             if beg_clock_snapshot is not None or end_clock_snapshot is not None:
 444                 raise ValueError(
 445                     'discarded packets have no default clock snapshots for this stream class'
 446                 )
 447
 448             ptr = native_bt.message_discarded_packets_create(self._bt_ptr, stream._ptr)
 449
 450         if ptr is None:
 451             raise bt2._MemoryError('cannot discarded packets message object')
 452
 453         msg = bt2_message._DiscardedPacketsMessage(ptr)
 454
 455         if count is not None:
 456             msg._count = count
 457
 458         return msg