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 from bt2 import error as bt2_error
  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_utils.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_utils.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_error._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_error._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_error._MemoryError(
 304                 "cannot create stream beginning message object"
 305             )
 306
 307         msg = bt2_message._StreamBeginningMessage(ptr)
 308
 309         if default_clock_snapshot is not None:
 310             msg._default_clock_snapshot = default_clock_snapshot
 311
 312         return msg
 313
 314     def _create_stream_end_message(self, stream, default_clock_snapshot=None):
 315         bt2_utils._check_type(stream, bt2_stream._Stream)
 316
 317         ptr = native_bt.message_stream_end_create(self._bt_ptr, stream._ptr)
 318         if ptr is None:
 319             raise bt2_error._MemoryError("cannot create stream end message object")
 320
 321         msg = bt2_message._StreamEndMessage(ptr)
 322
 323         if default_clock_snapshot is not None:
 324             msg._default_clock_snapshot = default_clock_snapshot
 325
 326         return msg
 327
 328     def _create_packet_beginning_message(self, packet, default_clock_snapshot=None):
 329         bt2_utils._check_type(packet, bt2_packet._Packet)
 330
 331         if packet.stream.cls.packets_have_beginning_default_clock_snapshot:
 332             if default_clock_snapshot is None:
 333                 raise ValueError(
 334                     "packet beginning messages in this stream must have a default clock snapshot"
 335                 )
 336
 337             bt2_utils._check_uint64(default_clock_snapshot)
 338             ptr = native_bt.message_packet_beginning_create_with_default_clock_snapshot(
 339                 self._bt_ptr, packet._ptr, default_clock_snapshot
 340             )
 341         else:
 342             if default_clock_snapshot is not None:
 343                 raise ValueError(
 344                     "packet beginning messages in this stream must not have a default clock snapshot"
 345                 )
 346
 347             ptr = native_bt.message_packet_beginning_create(self._bt_ptr, packet._ptr)
 348
 349         if ptr is None:
 350             raise bt2_error._MemoryError(
 351                 "cannot create packet beginning message object"
 352             )
 353
 354         return bt2_message._PacketBeginningMessage(ptr)
 355
 356     def _create_packet_end_message(self, packet, default_clock_snapshot=None):
 357         bt2_utils._check_type(packet, bt2_packet._Packet)
 358
 359         if packet.stream.cls.packets_have_end_default_clock_snapshot:
 360             if default_clock_snapshot is None:
 361                 raise ValueError(
 362                     "packet end messages in this stream must have a default clock snapshot"
 363                 )
 364
 365             bt2_utils._check_uint64(default_clock_snapshot)
 366             ptr = native_bt.message_packet_end_create_with_default_clock_snapshot(
 367                 self._bt_ptr, packet._ptr, default_clock_snapshot
 368             )
 369         else:
 370             if default_clock_snapshot is not None:
 371                 raise ValueError(
 372                     "packet end messages in this stream must not have a default clock snapshot"
 373                 )
 374
 375             ptr = native_bt.message_packet_end_create(self._bt_ptr, packet._ptr)
 376
 377         if ptr is None:
 378             raise bt2_error._MemoryError("cannot create packet end message object")
 379
 380         return bt2_message._PacketEndMessage(ptr)
 381
 382     def _create_discarded_events_message(
 383         self, stream, count=None, beg_clock_snapshot=None, end_clock_snapshot=None
 384     ):
 385         bt2_utils._check_type(stream, bt2_stream._Stream)
 386
 387         if not stream.cls.supports_discarded_events:
 388             raise ValueError("stream class does not support discarded events")
 389
 390         if stream.cls.discarded_events_have_default_clock_snapshots:
 391             if beg_clock_snapshot is None or end_clock_snapshot is None:
 392                 raise ValueError(
 393                     "discarded events have default clock snapshots for this stream class"
 394                 )
 395
 396             bt2_utils._check_uint64(beg_clock_snapshot)
 397             bt2_utils._check_uint64(end_clock_snapshot)
 398
 399             if beg_clock_snapshot > end_clock_snapshot:
 400                 raise ValueError(
 401                     "beginning default clock snapshot value ({}) is greater than end default clock snapshot value ({})".format(
 402                         beg_clock_snapshot, end_clock_snapshot
 403                     )
 404                 )
 405
 406             ptr = (
 407                 native_bt.message_discarded_events_create_with_default_clock_snapshots(
 408                     self._bt_ptr, stream._ptr, beg_clock_snapshot, end_clock_snapshot
 409                 )
 410             )
 411         else:
 412             if beg_clock_snapshot is not None or end_clock_snapshot is not None:
 413                 raise ValueError(
 414                     "discarded events have no default clock snapshots for this stream class"
 415                 )
 416
 417             ptr = native_bt.message_discarded_events_create(self._bt_ptr, stream._ptr)
 418
 419         if ptr is None:
 420             raise bt2_error._MemoryError("cannot discarded events message object")
 421
 422         msg = bt2_message._DiscardedEventsMessage(ptr)
 423
 424         if count is not None:
 425             msg._count = count
 426
 427         return msg
 428
 429     def _create_discarded_packets_message(
 430         self, stream, count=None, beg_clock_snapshot=None, end_clock_snapshot=None
 431     ):
 432         bt2_utils._check_type(stream, bt2_stream._Stream)
 433
 434         if not stream.cls.supports_discarded_packets:
 435             raise ValueError("stream class does not support discarded packets")
 436
 437         if stream.cls.discarded_packets_have_default_clock_snapshots:
 438             if beg_clock_snapshot is None or end_clock_snapshot is None:
 439                 raise ValueError(
 440                     "discarded packets have default clock snapshots for this stream class"
 441                 )
 442
 443             bt2_utils._check_uint64(beg_clock_snapshot)
 444             bt2_utils._check_uint64(end_clock_snapshot)
 445
 446             if beg_clock_snapshot > end_clock_snapshot:
 447                 raise ValueError(
 448                     "beginning default clock snapshot value ({}) is greater than end default clock snapshot value ({})".format(
 449                         beg_clock_snapshot, end_clock_snapshot
 450                     )
 451                 )
 452
 453             ptr = (
 454                 native_bt.message_discarded_packets_create_with_default_clock_snapshots(
 455                     self._bt_ptr, stream._ptr, beg_clock_snapshot, end_clock_snapshot
 456                 )
 457             )
 458         else:
 459             if beg_clock_snapshot is not None or end_clock_snapshot is not None:
 460                 raise ValueError(
 461                     "discarded packets have no default clock snapshots for this stream class"
 462                 )
 463
 464             ptr = native_bt.message_discarded_packets_create(self._bt_ptr, stream._ptr)
 465
 466         if ptr is None:
 467             raise bt2_error._MemoryError("cannot discarded packets message object")
 468
 469         msg = bt2_message._DiscardedPacketsMessage(ptr)
 470
 471         if count is not None:
 472             msg._count = count
 473
 474         return msg