Move to kernel style SPDX license identifiers
[babeltrace.git] / src / bindings / python / bt2 / bt2 / stream.py
1 # SPDX-License-Identifier: MIT
2 #
3 # Copyright (c) 2016-2017 Philippe Proulx <pproulx@efficios.com>
4
5 from bt2 import native_bt, utils
6 from bt2 import object as bt2_object
7 from bt2 import packet as bt2_packet
8 from bt2 import stream_class as bt2_stream_class
9 from bt2 import value as bt2_value
10 import bt2
11
12
13 def _bt2_trace():
14 from bt2 import trace as bt2_trace
15
16 return bt2_trace
17
18
19 class _StreamConst(bt2_object._SharedObject):
20 _get_ref = staticmethod(native_bt.stream_get_ref)
21 _put_ref = staticmethod(native_bt.stream_put_ref)
22 _borrow_class_ptr = staticmethod(native_bt.stream_borrow_class_const)
23 _borrow_user_attributes_ptr = staticmethod(
24 native_bt.stream_borrow_user_attributes_const
25 )
26 _create_value_from_ptr_and_get_ref = staticmethod(
27 bt2_value._create_from_const_ptr_and_get_ref
28 )
29 _borrow_trace_ptr = staticmethod(native_bt.stream_borrow_trace_const)
30 _stream_class_pycls = bt2_stream_class._StreamClassConst
31 _trace_pycls = property(lambda _: _bt2_trace()._TraceConst)
32
33 @property
34 def cls(self):
35 stream_class_ptr = self._borrow_class_ptr(self._ptr)
36 assert stream_class_ptr is not None
37 return self._stream_class_pycls._create_from_ptr_and_get_ref(stream_class_ptr)
38
39 @property
40 def name(self):
41 return native_bt.stream_get_name(self._ptr)
42
43 @property
44 def user_attributes(self):
45 ptr = self._borrow_user_attributes_ptr(self._ptr)
46 assert ptr is not None
47 return self._create_value_from_ptr_and_get_ref(ptr)
48
49 @property
50 def id(self):
51 id = native_bt.stream_get_id(self._ptr)
52 return id if id >= 0 else None
53
54 @property
55 def trace(self):
56 trace_ptr = self._borrow_trace_ptr(self._ptr)
57 assert trace_ptr is not None
58 return self._trace_pycls._create_from_ptr_and_get_ref(trace_ptr)
59
60
61 class _Stream(_StreamConst):
62 _borrow_class_ptr = staticmethod(native_bt.stream_borrow_class)
63 _borrow_user_attributes_ptr = staticmethod(native_bt.stream_borrow_user_attributes)
64 _create_value_from_ptr_and_get_ref = staticmethod(
65 bt2_value._create_from_ptr_and_get_ref
66 )
67 _borrow_trace_ptr = staticmethod(native_bt.stream_borrow_trace)
68 _stream_class_pycls = bt2_stream_class._StreamClass
69 _trace_pycls = property(lambda _: _bt2_trace()._Trace)
70
71 def create_packet(self):
72 if not self.cls.supports_packets:
73 raise ValueError(
74 'cannot create packet: stream class does not support packets'
75 )
76
77 packet_ptr = native_bt.packet_create(self._ptr)
78
79 if packet_ptr is None:
80 raise bt2._MemoryError('cannot create packet object')
81
82 return bt2_packet._Packet._create_from_ptr(packet_ptr)
83
84 def _user_attributes(self, user_attributes):
85 value = bt2_value.create_value(user_attributes)
86 utils._check_type(value, bt2_value.MapValue)
87 native_bt.stream_set_user_attributes(self._ptr, value._ptr)
88
89 _user_attributes = property(
90 fget=_StreamConst.user_attributes.fget, fset=_user_attributes
91 )
92
93 def _name(self, name):
94 utils._check_str(name)
95 native_bt.stream_set_name(self._ptr, name)
96
97 _name = property(fget=_StreamConst.name.fget, fset=_name)
This page took 0.030872 seconds and 4 git commands to generate.