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