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