# THE SOFTWARE.
from bt2 import native_bt, object, utils
-import bt2.clock_snapshot
-import bt2.packet
-import bt2.stream
-import bt2.event
+from bt2 import clock_snapshot as bt2_clock_snapshot
+from bt2 import packet as bt2_packet
+from bt2 import stream as bt2_stream
+from bt2 import event as bt2_event
import bt2
def _create_from_ptr(ptr):
msg_type = native_bt.message_get_type(ptr)
-
- if msg_type not in _MESSAGE_TYPE_TO_CLS:
- raise bt2.Error('unknown message type: {}'.format(msg_type))
-
return _MESSAGE_TYPE_TO_CLS[msg_type]._create_from_ptr(ptr)
@staticmethod
def _check_has_default_clock_class(clock_class):
if clock_class is None:
- raise bt2.NonexistentClockSnapshot(
+ raise ValueError(
'cannot get default clock snapshot: stream class has no default clock class'
)
def _get_default_clock_snapshot(self, borrow_clock_snapshot_ptr):
snapshot_ptr = borrow_clock_snapshot_ptr(self._ptr)
- return bt2.clock_snapshot._ClockSnapshot._create_from_ptr_and_get_ref(
+ return bt2_clock_snapshot._ClockSnapshot._create_from_ptr_and_get_ref(
snapshot_ptr, self._ptr, self._get_ref, self._put_ref
)
def event(self):
event_ptr = native_bt.message_event_borrow_event(self._ptr)
assert event_ptr is not None
- return bt2.event._Event._create_from_ptr_and_get_ref(
+ return bt2_event._Event._create_from_ptr_and_get_ref(
event_ptr, self._ptr, self._get_ref, self._put_ref
)
def packet(self):
packet_ptr = self._borrow_packet_ptr(self._ptr)
assert packet_ptr is not None
- return bt2.packet._Packet._create_from_ptr_and_get_ref(packet_ptr)
+ return bt2_packet._Packet._create_from_ptr_and_get_ref(packet_ptr)
class _PacketBeginningMessage(_PacketMessage):
def stream(self):
stream_ptr = self._borrow_stream_ptr(self._ptr)
assert stream_ptr
- return bt2.stream._Stream._create_from_ptr_and_get_ref(stream_ptr)
+ return bt2_stream._Stream._create_from_ptr_and_get_ref(stream_ptr)
@property
def default_clock_snapshot(self):
status, snapshot_ptr = self._borrow_default_clock_snapshot_ptr(self._ptr)
if status == native_bt.MESSAGE_STREAM_CLOCK_SNAPSHOT_STATE_UNKNOWN:
- return bt2.clock_snapshot._UnknownClockSnapshot()
+ return bt2_clock_snapshot._UnknownClockSnapshot()
- return bt2.clock_snapshot._ClockSnapshot._create_from_ptr_and_get_ref(
+ return bt2_clock_snapshot._ClockSnapshot._create_from_ptr_and_get_ref(
snapshot_ptr, self._ptr, self._get_ref, self._put_ref
)
def stream(self):
stream_ptr = self._borrow_stream_ptr(self._ptr)
assert stream_ptr
- return bt2.stream._Stream._create_from_ptr_and_get_ref(stream_ptr)
+ return bt2_stream._Stream._create_from_ptr_and_get_ref(stream_ptr)
@property
def count(self):
def _check_has_default_clock_snapshots(self):
if not self._has_default_clock_snapshots:
- raise bt2.NonexistentClockSnapshot(
+ raise ValueError(
'cannot get default clock snapshot: such a message has no clock snapshots for this stream class'
)