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         (
 222             status,
 223             msg_iter_ptr,
 224         ) = native_bt.bt2_message_iterator_create_from_message_iterator(
 225             self._bt_ptr, input_port._ptr
 226         )
 227         utils._handle_func_status(status, 'cannot create message iterator object')
 228         assert msg_iter_ptr is not None
 229
 230         return _UserComponentInputPortMessageIterator(msg_iter_ptr)
 231
 232     def _create_event_message(self, event_class, parent, default_clock_snapshot=None):
 233         utils._check_type(event_class, bt2_event_class._EventClass)
 234
 235         if event_class.stream_class.supports_packets:
 236             utils._check_type(parent, bt2_packet._Packet)
 237         else:
 238             utils._check_type(parent, bt2_stream._Stream)
 239
 240         if default_clock_snapshot is not None:
 241             if event_class.stream_class.default_clock_class is None:
 242                 raise ValueError(
 243                     'event messages in this stream must not have a default clock snapshot'
 244                 )
 245
 246             utils._check_uint64(default_clock_snapshot)
 247
 248             if event_class.stream_class.supports_packets:
 249                 ptr = native_bt.message_event_create_with_packet_and_default_clock_snapshot(
 250                     self._bt_ptr, event_class._ptr, parent._ptr, default_clock_snapshot
 251                 )
 252             else:
 253                 ptr = native_bt.message_event_create_with_default_clock_snapshot(
 254                     self._bt_ptr, event_class._ptr, parent._ptr, default_clock_snapshot
 255                 )
 256         else:
 257             if event_class.stream_class.default_clock_class is not None:
 258                 raise ValueError(
 259                     'event messages in this stream must have a default clock snapshot'
 260                 )
 261
 262             if event_class.stream_class.supports_packets:
 263                 ptr = native_bt.message_event_create_with_packet(
 264                     self._bt_ptr, event_class._ptr, parent._ptr
 265                 )
 266             else:
 267                 ptr = native_bt.message_event_create(
 268                     self._bt_ptr, event_class._ptr, parent._ptr
 269                 )
 270
 271         if ptr is None:
 272             raise bt2._MemoryError('cannot create event message object')
 273
 274         return bt2_message._EventMessage(ptr)
 275
 276     def _create_message_iterator_inactivity_message(self, clock_class, clock_snapshot):
 277         utils._check_type(clock_class, bt2_clock_class._ClockClass)
 278         ptr = native_bt.message_message_iterator_inactivity_create(
 279             self._bt_ptr, clock_class._ptr, clock_snapshot
 280         )
 281
 282         if ptr is None:
 283             raise bt2._MemoryError('cannot create inactivity message object')
 284
 285         return bt2_message._MessageIteratorInactivityMessage(ptr)
 286
 287     def _create_stream_beginning_message(self, stream, default_clock_snapshot=None):
 288         utils._check_type(stream, bt2_stream._Stream)
 289
 290         ptr = native_bt.message_stream_beginning_create(self._bt_ptr, stream._ptr)
 291         if ptr is None:
 292             raise bt2._MemoryError('cannot create stream beginning message object')
 293
 294         msg = bt2_message._StreamBeginningMessage(ptr)
 295
 296         if default_clock_snapshot is not None:
 297             msg._default_clock_snapshot = default_clock_snapshot
 298
 299         return msg
 300
 301     def _create_stream_end_message(self, stream, default_clock_snapshot=None):
 302         utils._check_type(stream, bt2_stream._Stream)
 303
 304         ptr = native_bt.message_stream_end_create(self._bt_ptr, stream._ptr)
 305         if ptr is None:
 306             raise bt2._MemoryError('cannot create stream end message object')
 307
 308         msg = bt2_message._StreamEndMessage(ptr)
 309
 310         if default_clock_snapshot is not None:
 311             msg._default_clock_snapshot = default_clock_snapshot
 312
 313         return msg
 314
 315     def _create_packet_beginning_message(self, packet, default_clock_snapshot=None):
 316         utils._check_type(packet, bt2_packet._Packet)
 317
 318         if packet.stream.cls.packets_have_beginning_default_clock_snapshot:
 319             if default_clock_snapshot is None:
 320                 raise ValueError(
 321                     "packet beginning messages in this stream must have a default clock snapshot"
 322                 )
 323
 324             utils._check_uint64(default_clock_snapshot)
 325             ptr = native_bt.message_packet_beginning_create_with_default_clock_snapshot(
 326                 self._bt_ptr, packet._ptr, default_clock_snapshot
 327             )
 328         else:
 329             if default_clock_snapshot is not None:
 330                 raise ValueError(
 331                     "packet beginning messages in this stream must not have a default clock snapshot"
 332                 )
 333
 334             ptr = native_bt.message_packet_beginning_create(self._bt_ptr, packet._ptr)
 335
 336         if ptr is None:
 337             raise bt2._MemoryError('cannot create packet beginning message object')
 338
 339         return bt2_message._PacketBeginningMessage(ptr)
 340
 341     def _create_packet_end_message(self, packet, default_clock_snapshot=None):
 342         utils._check_type(packet, bt2_packet._Packet)
 343
 344         if packet.stream.cls.packets_have_end_default_clock_snapshot:
 345             if default_clock_snapshot is None:
 346                 raise ValueError(
 347                     "packet end messages in this stream must have a default clock snapshot"
 348                 )
 349
 350             utils._check_uint64(default_clock_snapshot)
 351             ptr = native_bt.message_packet_end_create_with_default_clock_snapshot(
 352                 self._bt_ptr, packet._ptr, default_clock_snapshot
 353             )
 354         else:
 355             if default_clock_snapshot is not None:
 356                 raise ValueError(
 357                     "packet end messages in this stream must not have a default clock snapshot"
 358                 )
 359
 360             ptr = native_bt.message_packet_end_create(self._bt_ptr, packet._ptr)
 361
 362         if ptr is None:
 363             raise bt2._MemoryError('cannot create packet end message object')
 364
 365         return bt2_message._PacketEndMessage(ptr)
 366
 367     def _create_discarded_events_message(
 368         self, stream, count=None, beg_clock_snapshot=None, end_clock_snapshot=None
 369     ):
 370         utils._check_type(stream, bt2_stream._Stream)
 371
 372         if not stream.cls.supports_discarded_events:
 373             raise ValueError('stream class does not support discarded events')
 374
 375         if stream.cls.discarded_events_have_default_clock_snapshots:
 376             if beg_clock_snapshot is None or end_clock_snapshot is None:
 377                 raise ValueError(
 378                     'discarded events have default clock snapshots for this stream class'
 379                 )
 380
 381             utils._check_uint64(beg_clock_snapshot)
 382             utils._check_uint64(end_clock_snapshot)
 383             ptr = native_bt.message_discarded_events_create_with_default_clock_snapshots(
 384                 self._bt_ptr, stream._ptr, beg_clock_snapshot, end_clock_snapshot
 385             )
 386         else:
 387             if beg_clock_snapshot is not None or end_clock_snapshot is not None:
 388                 raise ValueError(
 389                     'discarded events have no default clock snapshots for this stream class'
 390                 )
 391
 392             ptr = native_bt.message_discarded_events_create(self._bt_ptr, stream._ptr)
 393
 394         if ptr is None:
 395             raise bt2._MemoryError('cannot discarded events message object')
 396
 397         msg = bt2_message._DiscardedEventsMessage(ptr)
 398
 399         if count is not None:
 400             msg._count = count
 401
 402         return msg
 403
 404     def _create_discarded_packets_message(
 405         self, stream, count=None, beg_clock_snapshot=None, end_clock_snapshot=None
 406     ):
 407         utils._check_type(stream, bt2_stream._Stream)
 408
 409         if not stream.cls.supports_discarded_packets:
 410             raise ValueError('stream class does not support discarded packets')
 411
 412         if stream.cls.discarded_packets_have_default_clock_snapshots:
 413             if beg_clock_snapshot is None or end_clock_snapshot is None:
 414                 raise ValueError(
 415                     'discarded packets have default clock snapshots for this stream class'
 416                 )
 417
 418             utils._check_uint64(beg_clock_snapshot)
 419             utils._check_uint64(end_clock_snapshot)
 420             ptr = native_bt.message_discarded_packets_create_with_default_clock_snapshots(
 421                 self._bt_ptr, stream._ptr, beg_clock_snapshot, end_clock_snapshot
 422             )
 423         else:
 424             if beg_clock_snapshot is not None or end_clock_snapshot is not None:
 425                 raise ValueError(
 426                     'discarded packets have no default clock snapshots for this stream class'
 427                 )
 428
 429             ptr = native_bt.message_discarded_packets_create(self._bt_ptr, stream._ptr)
 430
 431         if ptr is None:
 432             raise bt2._MemoryError('cannot discarded packets message object')
 433
 434         msg = bt2_message._DiscardedPacketsMessage(ptr)
 435
 436         if count is not None:
 437             msg._count = count
 438
 439         return msg