Remove `skip-string-normalization` in Python formatter config
[babeltrace.git] / src / bindings / python / bt2 / bt2 / trace.py
... / ...
CommitLineData
1# SPDX-License-Identifier: MIT
2#
3# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
4
5from bt2 import native_bt, object, utils
6import collections.abc
7from bt2 import value as bt2_value
8from bt2 import stream as bt2_stream
9from bt2 import stream_class as bt2_stream_class
10import bt2
11import functools
12import uuid as uuidp
13
14
15def _bt2_trace_class():
16 from bt2 import trace_class as bt2_trace_class
17
18 return bt2_trace_class
19
20
21class _TraceEnvironmentConst(collections.abc.Mapping):
22 _create_value_from_ptr_and_get_ref = staticmethod(
23 bt2_value._create_from_const_ptr_and_get_ref
24 )
25
26 def __init__(self, trace):
27 self._trace = trace
28
29 def __getitem__(self, key):
30 utils._check_str(key)
31
32 borrow_entry_fn = native_bt.trace_borrow_environment_entry_value_by_name_const
33 value_ptr = borrow_entry_fn(self._trace._ptr, key)
34
35 if value_ptr is None:
36 raise KeyError(key)
37
38 return self._create_value_from_ptr_and_get_ref(value_ptr)
39
40 def __len__(self):
41 count = native_bt.trace_get_environment_entry_count(self._trace._ptr)
42 assert count >= 0
43 return count
44
45 def __iter__(self):
46 trace_ptr = self._trace._ptr
47
48 for idx in range(len(self)):
49 borrow_entry_fn = native_bt.trace_borrow_environment_entry_by_index_const
50 entry_name, _ = borrow_entry_fn(trace_ptr, idx)
51 assert entry_name is not None
52 yield entry_name
53
54
55class _TraceEnvironment(_TraceEnvironmentConst, collections.abc.MutableMapping):
56 _create_value_from_ptr_and_get_ref = staticmethod(
57 bt2_value._create_from_ptr_and_get_ref
58 )
59
60 def __setitem__(self, key, value):
61 if isinstance(value, str):
62 set_env_entry_fn = native_bt.trace_set_environment_entry_string
63 elif isinstance(value, int):
64 set_env_entry_fn = native_bt.trace_set_environment_entry_integer
65 else:
66 raise TypeError("expected str or int, got {}".format(type(value)))
67
68 status = set_env_entry_fn(self._trace._ptr, key, value)
69 utils._handle_func_status(status, "cannot set trace object's environment entry")
70
71 def __delitem__(self, key):
72 raise NotImplementedError
73
74
75class _TraceConst(object._SharedObject, collections.abc.Mapping):
76 _get_ref = staticmethod(native_bt.trace_get_ref)
77 _put_ref = staticmethod(native_bt.trace_put_ref)
78 _borrow_stream_ptr_by_id = staticmethod(native_bt.trace_borrow_stream_by_id_const)
79 _borrow_stream_ptr_by_index = staticmethod(
80 native_bt.trace_borrow_stream_by_index_const
81 )
82 _borrow_class_ptr = staticmethod(native_bt.trace_borrow_class_const)
83 _borrow_user_attributes_ptr = staticmethod(
84 native_bt.trace_borrow_user_attributes_const
85 )
86 _create_value_from_ptr_and_get_ref = staticmethod(
87 bt2_value._create_from_const_ptr_and_get_ref
88 )
89 _stream_pycls = property(lambda _: bt2_stream._StreamConst)
90 _trace_class_pycls = property(lambda _: _bt2_trace_class()._TraceClassConst)
91 _trace_env_pycls = property(lambda _: _TraceEnvironmentConst)
92
93 def __len__(self):
94 count = native_bt.trace_get_stream_count(self._ptr)
95 assert count >= 0
96 return count
97
98 def __getitem__(self, id):
99 utils._check_uint64(id)
100
101 stream_ptr = self._borrow_stream_ptr_by_id(self._ptr, id)
102
103 if stream_ptr is None:
104 raise KeyError(id)
105
106 return self._stream_pycls._create_from_ptr_and_get_ref(stream_ptr)
107
108 def __iter__(self):
109 for idx in range(len(self)):
110 stream_ptr = self._borrow_stream_ptr_by_index(self._ptr, idx)
111 assert stream_ptr is not None
112
113 id = native_bt.stream_get_id(stream_ptr)
114 assert id >= 0
115
116 yield id
117
118 @property
119 def cls(self):
120 trace_class_ptr = self._borrow_class_ptr(self._ptr)
121 assert trace_class_ptr is not None
122 return self._trace_class_pycls._create_from_ptr_and_get_ref(trace_class_ptr)
123
124 @property
125 def user_attributes(self):
126 ptr = self._borrow_user_attributes_ptr(self._ptr)
127 assert ptr is not None
128 return self._create_value_from_ptr_and_get_ref(ptr)
129
130 @property
131 def name(self):
132 return native_bt.trace_get_name(self._ptr)
133
134 @property
135 def uuid(self):
136 uuid_bytes = native_bt.trace_get_uuid(self._ptr)
137 if uuid_bytes is None:
138 return
139
140 return uuidp.UUID(bytes=uuid_bytes)
141
142 @property
143 def environment(self):
144 return self._trace_env_pycls(self)
145
146 def add_destruction_listener(self, listener):
147 """Add a listener to be called when the trace is destroyed."""
148 if not callable(listener):
149 raise TypeError("'listener' parameter is not callable")
150
151 handle = utils._ListenerHandle(self.addr)
152
153 fn = native_bt.bt2_trace_add_destruction_listener
154 listener_from_native = functools.partial(
155 _trace_destruction_listener_from_native, listener, handle
156 )
157
158 status, listener_id = fn(self._ptr, listener_from_native)
159 utils._handle_func_status(
160 status, "cannot add destruction listener to trace object"
161 )
162
163 handle._set_listener_id(listener_id)
164
165 return handle
166
167 def remove_destruction_listener(self, listener_handle):
168 utils._check_type(listener_handle, utils._ListenerHandle)
169
170 if listener_handle._addr != self.addr:
171 raise ValueError(
172 "This trace destruction listener does not match the trace object."
173 )
174
175 if listener_handle._listener_id is None:
176 raise ValueError("This trace destruction listener was already removed.")
177
178 status = native_bt.trace_remove_destruction_listener(
179 self._ptr, listener_handle._listener_id
180 )
181 utils._handle_func_status(status)
182 listener_handle._invalidate()
183
184
185class _Trace(_TraceConst):
186 _borrow_stream_ptr_by_id = staticmethod(native_bt.trace_borrow_stream_by_id)
187 _borrow_stream_ptr_by_index = staticmethod(native_bt.trace_borrow_stream_by_index)
188 _borrow_class_ptr = staticmethod(native_bt.trace_borrow_class)
189 _borrow_user_attributes_ptr = staticmethod(native_bt.trace_borrow_user_attributes)
190 _create_value_from_ptr_and_get_ref = staticmethod(
191 bt2_value._create_from_ptr_and_get_ref
192 )
193 _stream_pycls = property(lambda _: bt2_stream._Stream)
194 _trace_class_pycls = property(lambda _: _bt2_trace_class()._TraceClass)
195 _trace_env_pycls = property(lambda _: _TraceEnvironment)
196
197 def _name(self, name):
198 utils._check_str(name)
199 status = native_bt.trace_set_name(self._ptr, name)
200 utils._handle_func_status(status, "cannot set trace class object's name")
201
202 _name = property(fset=_name)
203
204 def _user_attributes(self, user_attributes):
205 value = bt2_value.create_value(user_attributes)
206 utils._check_type(value, bt2_value.MapValue)
207 native_bt.trace_set_user_attributes(self._ptr, value._ptr)
208
209 _user_attributes = property(fset=_user_attributes)
210
211 def _uuid(self, uuid):
212 utils._check_type(uuid, uuidp.UUID)
213 native_bt.trace_set_uuid(self._ptr, uuid.bytes)
214
215 _uuid = property(fset=_uuid)
216
217 def create_stream(self, stream_class, id=None, name=None, user_attributes=None):
218 utils._check_type(stream_class, bt2_stream_class._StreamClass)
219
220 if stream_class.assigns_automatic_stream_id:
221 if id is not None:
222 raise ValueError(
223 "id provided, but stream class assigns automatic stream ids"
224 )
225
226 stream_ptr = native_bt.stream_create(stream_class._ptr, self._ptr)
227 else:
228 if id is None:
229 raise ValueError(
230 "id not provided, but stream class does not assign automatic stream ids"
231 )
232
233 utils._check_uint64(id)
234 stream_ptr = native_bt.stream_create_with_id(
235 stream_class._ptr, self._ptr, id
236 )
237
238 if stream_ptr is None:
239 raise bt2._MemoryError("cannot create stream object")
240
241 stream = bt2_stream._Stream._create_from_ptr(stream_ptr)
242
243 if name is not None:
244 stream._name = name
245
246 if user_attributes is not None:
247 stream._user_attributes = user_attributes
248
249 return stream
250
251
252def _trace_destruction_listener_from_native(user_listener, handle, trace_ptr):
253 trace = _TraceConst._create_from_ptr_and_get_ref(trace_ptr)
254 user_listener(trace)
255 handle._invalidate()
This page took 0.024088 seconds and 4 git commands to generate.