src/bindings/python/bt2/bt2/message.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 clock_snapshot as bt2_clock_snapshot
   7 from bt2 import packet as bt2_packet
   8 from bt2 import stream as bt2_stream
   9 from bt2 import event as bt2_event
  10
  11
  12 def _create_from_ptr(ptr):
  13     msg_type = native_bt.message_get_type(ptr)
  14     return _MESSAGE_TYPE_TO_CLS[msg_type]._create_from_ptr(ptr)
  15
  16
  17 class _MessageConst(object._SharedObject):
  18     @staticmethod
  19     def _get_ref(ptr):
  20         native_bt.message_get_ref(ptr)
  21
  22     @staticmethod
  23     def _put_ref(ptr):
  24         native_bt.message_put_ref(ptr)
  25
  26     @staticmethod
  27     def _check_has_default_clock_class(clock_class):
  28         if clock_class is None:
  29             raise ValueError(
  30                 "cannot get default clock snapshot: stream class has no default clock class"
  31             )
  32
  33
  34 class _Message(_MessageConst):
  35     pass
  36
  37
  38 class _MessageWithDefaultClockSnapshot:
  39     def _get_default_clock_snapshot(self, borrow_clock_snapshot_ptr):
  40         snapshot_ptr = borrow_clock_snapshot_ptr(self._ptr)
  41
  42         return bt2_clock_snapshot._ClockSnapshotConst._create_from_ptr_and_get_ref(
  43             snapshot_ptr, self._ptr, self._get_ref, self._put_ref
  44         )
  45
  46
  47 class _EventMessageConst(_MessageConst, _MessageWithDefaultClockSnapshot):
  48     _borrow_default_clock_snapshot = staticmethod(
  49         native_bt.message_event_borrow_default_clock_snapshot_const
  50     )
  51     _borrow_event = staticmethod(native_bt.message_event_borrow_event_const)
  52     _event_pycls = property(lambda _: bt2_event._EventConst)
  53
  54     @property
  55     def default_clock_snapshot(self):
  56         self._check_has_default_clock_class(self.event.stream.cls.default_clock_class)
  57         return self._get_default_clock_snapshot(self._borrow_default_clock_snapshot)
  58
  59     @property
  60     def event(self):
  61         event_ptr = self._borrow_event(self._ptr)
  62         assert event_ptr is not None
  63         return self._event_pycls._create_from_ptr_and_get_ref(
  64             event_ptr, self._ptr, self._get_ref, self._put_ref
  65         )
  66
  67
  68 class _EventMessage(_EventMessageConst, _Message):
  69     _borrow_event = staticmethod(native_bt.message_event_borrow_event)
  70     _stream_pycls = property(lambda _: bt2_stream._Stream)
  71     _event_pycls = property(lambda _: bt2_event._Event)
  72
  73
  74 class _PacketMessageConst(_MessageConst, _MessageWithDefaultClockSnapshot):
  75     _packet_pycls = bt2_packet._PacketConst
  76
  77     @property
  78     def default_clock_snapshot(self):
  79         self._check_has_default_clock_class(self.packet.stream.cls.default_clock_class)
  80         return self._get_default_clock_snapshot(self._borrow_default_clock_snapshot_ptr)
  81
  82     @property
  83     def packet(self):
  84         packet_ptr = self._borrow_packet(self._ptr)
  85         assert packet_ptr is not None
  86         return self._packet_pycls._create_from_ptr_and_get_ref(packet_ptr)
  87
  88
  89 class _PacketMessage(_PacketMessageConst, _Message):
  90     _packet_pycls = bt2_packet._Packet
  91
  92
  93 class _PacketBeginningMessageConst(_PacketMessageConst):
  94     _borrow_packet = staticmethod(
  95         native_bt.message_packet_beginning_borrow_packet_const
  96     )
  97     _borrow_default_clock_snapshot_ptr = staticmethod(
  98         native_bt.message_packet_beginning_borrow_default_clock_snapshot_const
  99     )
 100
 101
 102 class _PacketBeginningMessage(_PacketMessage):
 103     _borrow_packet = staticmethod(native_bt.message_packet_beginning_borrow_packet)
 104
 105
 106 class _PacketEndMessageConst(_PacketMessageConst):
 107     _borrow_packet = staticmethod(native_bt.message_packet_end_borrow_packet_const)
 108     _borrow_default_clock_snapshot_ptr = staticmethod(
 109         native_bt.message_packet_end_borrow_default_clock_snapshot_const
 110     )
 111
 112
 113 class _PacketEndMessage(_PacketMessage):
 114     _borrow_packet = staticmethod(native_bt.message_packet_end_borrow_packet)
 115
 116
 117 class _StreamMessageConst(_MessageConst, _MessageWithDefaultClockSnapshot):
 118     _stream_pycls = property(lambda _: bt2_stream._StreamConst)
 119
 120     @property
 121     def stream(self):
 122         stream_ptr = self._borrow_stream_ptr(self._ptr)
 123         assert stream_ptr
 124         return self._stream_pycls._create_from_ptr_and_get_ref(stream_ptr)
 125
 126     @property
 127     def default_clock_snapshot(self):
 128         self._check_has_default_clock_class(self.stream.cls.default_clock_class)
 129
 130         status, snapshot_ptr = self._borrow_default_clock_snapshot_ptr(self._ptr)
 131
 132         if status == native_bt.MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_UNKNOWN:
 133             return bt2_clock_snapshot._UnknownClockSnapshot()
 134
 135         return bt2_clock_snapshot._ClockSnapshotConst._create_from_ptr_and_get_ref(
 136             snapshot_ptr, self._ptr, self._get_ref, self._put_ref
 137         )
 138
 139
 140 class _StreamMessage(_StreamMessageConst, _Message):
 141     def _default_clock_snapshot(self, raw_value):
 142         utils._check_uint64(raw_value)
 143         self._set_default_clock_snapshot(self._ptr, raw_value)
 144
 145     _default_clock_snapshot = property(
 146         fget=_StreamMessageConst.default_clock_snapshot.fget,
 147         fset=_default_clock_snapshot,
 148     )
 149     _stream_pycls = property(lambda _: bt2_stream._Stream)
 150
 151
 152 class _StreamBeginningMessageConst(_StreamMessageConst):
 153     _borrow_stream_ptr = staticmethod(
 154         native_bt.message_stream_beginning_borrow_stream_const
 155     )
 156     _borrow_default_clock_snapshot_ptr = staticmethod(
 157         native_bt.message_stream_beginning_borrow_default_clock_snapshot_const
 158     )
 159
 160
 161 class _StreamBeginningMessage(_StreamMessage):
 162     _borrow_stream_ptr = staticmethod(native_bt.message_stream_beginning_borrow_stream)
 163     _set_default_clock_snapshot = staticmethod(
 164         native_bt.message_stream_beginning_set_default_clock_snapshot
 165     )
 166
 167
 168 class _StreamEndMessageConst(_StreamMessageConst):
 169     _borrow_stream_ptr = staticmethod(native_bt.message_stream_end_borrow_stream_const)
 170     _borrow_default_clock_snapshot_ptr = staticmethod(
 171         native_bt.message_stream_end_borrow_default_clock_snapshot_const
 172     )
 173
 174
 175 class _StreamEndMessage(_StreamMessage):
 176     _borrow_stream_ptr = staticmethod(native_bt.message_stream_end_borrow_stream)
 177     _set_default_clock_snapshot = staticmethod(
 178         native_bt.message_stream_end_set_default_clock_snapshot
 179     )
 180
 181
 182 class _MessageIteratorInactivityMessageConst(
 183     _MessageConst, _MessageWithDefaultClockSnapshot
 184 ):
 185     _borrow_clock_snapshot_ptr = staticmethod(
 186         native_bt.message_message_iterator_inactivity_borrow_clock_snapshot_const
 187     )
 188
 189     @property
 190     def clock_snapshot(self):
 191         # This kind of message always has a clock class: no
 192         # need to call self._check_has_default_clock_class() here.
 193         return self._get_default_clock_snapshot(self._borrow_clock_snapshot_ptr)
 194
 195
 196 class _MessageIteratorInactivityMessage(
 197     _MessageIteratorInactivityMessageConst, _Message
 198 ):
 199     pass
 200
 201
 202 class _DiscardedMessageConst(_MessageConst, _MessageWithDefaultClockSnapshot):
 203     _stream_pycls = property(lambda _: bt2_stream._StreamConst)
 204
 205     @property
 206     def stream(self):
 207         stream_ptr = self._borrow_stream_ptr(self._ptr)
 208         assert stream_ptr
 209         return self._stream_pycls._create_from_ptr_and_get_ref(stream_ptr)
 210
 211     @property
 212     def count(self):
 213         avail, count = self._get_count(self._ptr)
 214         if avail is native_bt.PROPERTY_AVAILABILITY_AVAILABLE:
 215             return count
 216
 217     def _check_has_default_clock_snapshots(self):
 218         if not self._has_default_clock_snapshots:
 219             raise ValueError(
 220                 "cannot get default clock snapshot: such a message has no clock snapshots for this stream class"
 221             )
 222
 223     @property
 224     def beginning_default_clock_snapshot(self):
 225         self._check_has_default_clock_snapshots()
 226         return self._get_default_clock_snapshot(
 227             self._borrow_beginning_clock_snapshot_ptr
 228         )
 229
 230     @property
 231     def end_default_clock_snapshot(self):
 232         self._check_has_default_clock_snapshots()
 233         return self._get_default_clock_snapshot(self._borrow_end_clock_snapshot_ptr)
 234
 235
 236 class _DiscardedMessage(_DiscardedMessageConst, _Message):
 237     _stream_pycls = property(lambda _: bt2_stream._Stream)
 238
 239     def _set_count(self, count):
 240         utils._check_uint64(count)
 241
 242         if count == 0:
 243             raise ValueError("discarded {} count is 0".format(self._item_name))
 244
 245         self._set_count(self._ptr, count)
 246
 247     _count = property(fget=_DiscardedMessageConst.count.fget, fset=_set_count)
 248
 249
 250 class _DiscardedEventsMessageConst(_DiscardedMessageConst):
 251     _borrow_stream_ptr = staticmethod(
 252         native_bt.message_discarded_events_borrow_stream_const
 253     )
 254     _get_count = staticmethod(native_bt.message_discarded_events_get_count)
 255     _borrow_beginning_clock_snapshot_ptr = staticmethod(
 256         native_bt.message_discarded_events_borrow_beginning_default_clock_snapshot_const
 257     )
 258     _borrow_end_clock_snapshot_ptr = staticmethod(
 259         native_bt.message_discarded_events_borrow_end_default_clock_snapshot_const
 260     )
 261
 262     @property
 263     def _has_default_clock_snapshots(self):
 264         return self.stream.cls.discarded_events_have_default_clock_snapshots
 265
 266
 267 class _DiscardedEventsMessage(_DiscardedEventsMessageConst, _DiscardedMessage):
 268     _borrow_stream_ptr = staticmethod(native_bt.message_discarded_events_borrow_stream)
 269     _set_count = staticmethod(native_bt.message_discarded_events_set_count)
 270     _item_name = "event"
 271
 272
 273 class _DiscardedPacketsMessageConst(_DiscardedMessageConst):
 274     _borrow_stream_ptr = staticmethod(
 275         native_bt.message_discarded_packets_borrow_stream_const
 276     )
 277     _get_count = staticmethod(native_bt.message_discarded_packets_get_count)
 278     _borrow_beginning_clock_snapshot_ptr = staticmethod(
 279         native_bt.message_discarded_packets_borrow_beginning_default_clock_snapshot_const
 280     )
 281     _borrow_end_clock_snapshot_ptr = staticmethod(
 282         native_bt.message_discarded_packets_borrow_end_default_clock_snapshot_const
 283     )
 284
 285     @property
 286     def _has_default_clock_snapshots(self):
 287         return self.stream.cls.discarded_packets_have_default_clock_snapshots
 288
 289
 290 class _DiscardedPacketsMessage(_DiscardedPacketsMessageConst, _DiscardedMessage):
 291     _borrow_stream_ptr = staticmethod(native_bt.message_discarded_packets_borrow_stream)
 292     _set_count = staticmethod(native_bt.message_discarded_packets_set_count)
 293     _item_name = "packet"
 294
 295
 296 _MESSAGE_TYPE_TO_CLS = {
 297     native_bt.MESSAGE_TYPE_EVENT: _EventMessage,
 298     native_bt.MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY: _MessageIteratorInactivityMessage,
 299     native_bt.MESSAGE_TYPE_STREAM_BEGINNING: _StreamBeginningMessage,
 300     native_bt.MESSAGE_TYPE_STREAM_END: _StreamEndMessage,
 301     native_bt.MESSAGE_TYPE_PACKET_BEGINNING: _PacketBeginningMessage,
 302     native_bt.MESSAGE_TYPE_PACKET_END: _PacketEndMessage,
 303     native_bt.MESSAGE_TYPE_DISCARDED_EVENTS: _DiscardedEventsMessage,
 304     native_bt.MESSAGE_TYPE_DISCARDED_PACKETS: _DiscardedPacketsMessage,
 305 }
 306
 307 _MESSAGE_TYPE_TO_CLS = {
 308     native_bt.MESSAGE_TYPE_EVENT: _EventMessageConst,
 309     native_bt.MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY: _MessageIteratorInactivityMessageConst,
 310     native_bt.MESSAGE_TYPE_STREAM_BEGINNING: _StreamBeginningMessageConst,
 311     native_bt.MESSAGE_TYPE_STREAM_END: _StreamEndMessageConst,
 312     native_bt.MESSAGE_TYPE_PACKET_BEGINNING: _PacketBeginningMessageConst,
 313     native_bt.MESSAGE_TYPE_PACKET_END: _PacketEndMessageConst,
 314     native_bt.MESSAGE_TYPE_DISCARDED_EVENTS: _DiscardedEventsMessageConst,
 315     native_bt.MESSAGE_TYPE_DISCARDED_PACKETS: _DiscardedPacketsMessageConst,
 316 }