bt2: template borrow_user_attributes method in _TraceClassConst, _StreamClassConst...
[babeltrace.git] / src / bindings / python / bt2 / bt2 / trace_class.py
1 # The MIT License (MIT)
2 #
3 # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
4 # Copyright (c) 2018 Francis Deslauriers <francis.deslauriers@efficios.com>
5 # Copyright (c) 2019 Simon Marchi <simon.marchi@efficios.com>
6 #
7 # Permission is hereby granted, free of charge, to any person obtaining a copy
8 # of this software and associated documentation files (the "Software"), to deal
9 # in the Software without restriction, including without limitation the rights
10 # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
11 # copies of the Software, and to permit persons to whom the Software is
12 # furnished to do so, subject to the following conditions:
13 #
14 # The above copyright notice and this permission notice shall be included in
15 # all copies or substantial portions of the Software.
16 #
17 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
18 # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
19 # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
20 # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
21 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
22 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
23 # THE SOFTWARE.
24
25 from bt2 import native_bt, utils, object
26 from bt2 import stream_class as bt2_stream_class
27 from bt2 import field_class as bt2_field_class
28 from bt2 import trace as bt2_trace
29 from bt2 import trace_class as bt2_trace_class
30 from bt2 import value as bt2_value
31 import collections.abc
32 import functools
33 import bt2
34
35
36 def _trace_class_destruction_listener_from_native(user_listener, trace_class_ptr):
37 trace_class = _TraceClass._create_from_ptr_and_get_ref(trace_class_ptr)
38 user_listener(trace_class)
39
40
41 class _TraceClassConst(object._SharedObject, collections.abc.Mapping):
42 _get_ref = staticmethod(native_bt.trace_class_get_ref)
43 _put_ref = staticmethod(native_bt.trace_class_put_ref)
44 _borrow_stream_class_ptr_by_index = staticmethod(
45 native_bt.trace_class_borrow_stream_class_by_index_const
46 )
47 _borrow_stream_class_ptr_by_id = staticmethod(
48 native_bt.trace_class_borrow_stream_class_by_id_const
49 )
50 _borrow_user_attributes_ptr = staticmethod(
51 native_bt.trace_class_borrow_user_attributes_const
52 )
53 _stream_class_pycls = bt2_stream_class._StreamClassConst
54 _create_value_from_ptr_and_get_ref = staticmethod(
55 bt2_value._create_from_const_ptr_and_get_ref
56 )
57
58 @property
59 def user_attributes(self):
60 ptr = self._borrow_user_attributes_ptr(self._ptr)
61 assert ptr is not None
62 return self._create_value_from_ptr_and_get_ref(ptr)
63
64 # Number of stream classes in this trace class.
65
66 def __len__(self):
67 count = native_bt.trace_class_get_stream_class_count(self._ptr)
68 assert count >= 0
69 return count
70
71 # Get a stream class by stream id.
72
73 def __getitem__(self, key):
74 utils._check_uint64(key)
75
76 sc_ptr = self._borrow_stream_class_ptr_by_id(self._ptr, key)
77 if sc_ptr is None:
78 raise KeyError(key)
79
80 return self._stream_class_pycls._create_from_ptr_and_get_ref(sc_ptr)
81
82 def __iter__(self):
83 for idx in range(len(self)):
84 sc_ptr = self._borrow_stream_class_ptr_by_index(self._ptr, idx)
85 assert sc_ptr is not None
86
87 id = native_bt.stream_class_get_id(sc_ptr)
88 assert id >= 0
89
90 yield id
91
92 @property
93 def assigns_automatic_stream_class_id(self):
94 return native_bt.trace_class_assigns_automatic_stream_class_id(self._ptr)
95
96 # Add a listener to be called when the trace class is destroyed.
97
98 def add_destruction_listener(self, listener):
99
100 if not callable(listener):
101 raise TypeError("'listener' parameter is not callable")
102
103 fn = native_bt.bt2_trace_class_add_destruction_listener
104 listener_from_native = functools.partial(
105 _trace_class_destruction_listener_from_native, listener
106 )
107
108 status, listener_id = fn(self._ptr, listener_from_native)
109 utils._handle_func_status(
110 status, 'cannot add destruction listener to trace class object'
111 )
112
113 return utils._ListenerHandle(listener_id, self)
114
115
116 class _TraceClass(_TraceClassConst):
117 _borrow_stream_class_ptr_by_index = staticmethod(
118 native_bt.trace_class_borrow_stream_class_by_index
119 )
120 _borrow_stream_class_ptr_by_id = staticmethod(
121 native_bt.trace_class_borrow_stream_class_by_id
122 )
123 _borrow_user_attributes_ptr = staticmethod(
124 native_bt.trace_class_borrow_user_attributes
125 )
126 _stream_class_pycls = bt2_stream_class._StreamClass
127 _create_value_from_ptr_and_get_ref = staticmethod(
128 bt2_value._create_from_ptr_and_get_ref
129 )
130
131 # Instantiate a trace of this class.
132
133 def __call__(self, name=None, user_attributes=None, uuid=None, environment=None):
134 trace_ptr = native_bt.trace_create(self._ptr)
135
136 if trace_ptr is None:
137 raise bt2._MemoryError('cannot create trace class object')
138
139 trace = bt2_trace._Trace._create_from_ptr(trace_ptr)
140
141 if name is not None:
142 trace._name = name
143
144 if user_attributes is not None:
145 trace._user_attributes = user_attributes
146
147 if uuid is not None:
148 trace._uuid = uuid
149
150 if environment is not None:
151 for key, value in environment.items():
152 trace.environment[key] = value
153
154 return trace
155
156 def create_stream_class(
157 self,
158 id=None,
159 name=None,
160 user_attributes=None,
161 packet_context_field_class=None,
162 event_common_context_field_class=None,
163 default_clock_class=None,
164 assigns_automatic_event_class_id=True,
165 assigns_automatic_stream_id=True,
166 supports_packets=False,
167 packets_have_beginning_default_clock_snapshot=False,
168 packets_have_end_default_clock_snapshot=False,
169 supports_discarded_events=False,
170 discarded_events_have_default_clock_snapshots=False,
171 supports_discarded_packets=False,
172 discarded_packets_have_default_clock_snapshots=False,
173 ):
174
175 if self.assigns_automatic_stream_class_id:
176 if id is not None:
177 raise ValueError(
178 'id provided, but trace class assigns automatic stream class ids'
179 )
180
181 sc_ptr = native_bt.stream_class_create(self._ptr)
182 else:
183 if id is None:
184 raise ValueError(
185 'id not provided, but trace class does not assign automatic stream class ids'
186 )
187
188 utils._check_uint64(id)
189 sc_ptr = native_bt.stream_class_create_with_id(self._ptr, id)
190
191 sc = bt2_stream_class._StreamClass._create_from_ptr(sc_ptr)
192
193 if name is not None:
194 sc._name = name
195
196 if user_attributes is not None:
197 sc._user_attributes = user_attributes
198
199 if event_common_context_field_class is not None:
200 sc._event_common_context_field_class = event_common_context_field_class
201
202 if default_clock_class is not None:
203 sc._default_clock_class = default_clock_class
204
205 # call after `sc._default_clock_class` because, if
206 # `packets_have_beginning_default_clock_snapshot` or
207 # `packets_have_end_default_clock_snapshot` is true, then this
208 # stream class needs a default clock class already.
209 sc._set_supports_packets(
210 supports_packets,
211 packets_have_beginning_default_clock_snapshot,
212 packets_have_end_default_clock_snapshot,
213 )
214
215 # call after sc._set_supports_packets() because, if
216 # `packet_context_field_class` is not `None`, then this stream
217 # class needs to support packets already.
218 if packet_context_field_class is not None:
219 sc._packet_context_field_class = packet_context_field_class
220
221 sc._assigns_automatic_event_class_id = assigns_automatic_event_class_id
222 sc._assigns_automatic_stream_id = assigns_automatic_stream_id
223 sc._set_supports_discarded_events(
224 supports_discarded_events, discarded_events_have_default_clock_snapshots
225 )
226 sc._set_supports_discarded_packets(
227 supports_discarded_packets, discarded_packets_have_default_clock_snapshots
228 )
229 return sc
230
231 def _user_attributes(self, user_attributes):
232 value = bt2_value.create_value(user_attributes)
233 utils._check_type(value, bt2_value.MapValue)
234 native_bt.trace_class_set_user_attributes(self._ptr, value._ptr)
235
236 _user_attributes = property(fset=_user_attributes)
237
238 def _assigns_automatic_stream_class_id(self, auto_id):
239 utils._check_bool(auto_id)
240 return native_bt.trace_class_set_assigns_automatic_stream_class_id(
241 self._ptr, auto_id
242 )
243
244 _assigns_automatic_stream_class_id = property(
245 fset=_assigns_automatic_stream_class_id
246 )
247
248 # Field class creation methods.
249
250 def _check_field_class_create_status(self, ptr, type_name):
251 if ptr is None:
252 raise bt2._MemoryError('cannot create {} field class'.format(type_name))
253
254 @staticmethod
255 def _set_field_class_user_attrs(fc, user_attributes):
256 if user_attributes is not None:
257 fc._user_attributes = user_attributes
258
259 def create_bool_field_class(self, user_attributes=None):
260 field_class_ptr = native_bt.field_class_bool_create(self._ptr)
261 self._check_field_class_create_status(field_class_ptr, 'boolean')
262 fc = bt2_field_class._BoolFieldClass._create_from_ptr(field_class_ptr)
263 self._set_field_class_user_attrs(fc, user_attributes)
264 return fc
265
266 def create_bit_array_field_class(self, length, user_attributes=None):
267 utils._check_uint64(length)
268
269 if length < 1 or length > 64:
270 raise ValueError(
271 'invalid length {}: expecting a value in the [1, 64] range'.format(
272 length
273 )
274 )
275
276 field_class_ptr = native_bt.field_class_bit_array_create(self._ptr, length)
277 self._check_field_class_create_status(field_class_ptr, 'bit array')
278 fc = bt2_field_class._BitArrayFieldClass._create_from_ptr(field_class_ptr)
279 self._set_field_class_user_attrs(fc, user_attributes)
280 return fc
281
282 def _create_integer_field_class(
283 self,
284 create_func,
285 py_cls,
286 type_name,
287 field_value_range,
288 preferred_display_base,
289 user_attributes,
290 ):
291 field_class_ptr = create_func(self._ptr)
292 self._check_field_class_create_status(field_class_ptr, type_name)
293
294 field_class = py_cls._create_from_ptr(field_class_ptr)
295
296 if field_value_range is not None:
297 field_class._field_value_range = field_value_range
298
299 if preferred_display_base is not None:
300 field_class._preferred_display_base = preferred_display_base
301
302 self._set_field_class_user_attrs(field_class, user_attributes)
303 return field_class
304
305 def create_signed_integer_field_class(
306 self, field_value_range=None, preferred_display_base=None, user_attributes=None
307 ):
308 return self._create_integer_field_class(
309 native_bt.field_class_integer_signed_create,
310 bt2_field_class._SignedIntegerFieldClass,
311 'signed integer',
312 field_value_range,
313 preferred_display_base,
314 user_attributes,
315 )
316
317 def create_unsigned_integer_field_class(
318 self, field_value_range=None, preferred_display_base=None, user_attributes=None
319 ):
320 return self._create_integer_field_class(
321 native_bt.field_class_integer_unsigned_create,
322 bt2_field_class._UnsignedIntegerFieldClass,
323 'unsigned integer',
324 field_value_range,
325 preferred_display_base,
326 user_attributes,
327 )
328
329 def create_signed_enumeration_field_class(
330 self, field_value_range=None, preferred_display_base=None, user_attributes=None
331 ):
332 return self._create_integer_field_class(
333 native_bt.field_class_enumeration_signed_create,
334 bt2_field_class._SignedEnumerationFieldClass,
335 'signed enumeration',
336 field_value_range,
337 preferred_display_base,
338 user_attributes,
339 )
340
341 def create_unsigned_enumeration_field_class(
342 self, field_value_range=None, preferred_display_base=None, user_attributes=None
343 ):
344 return self._create_integer_field_class(
345 native_bt.field_class_enumeration_unsigned_create,
346 bt2_field_class._UnsignedEnumerationFieldClass,
347 'unsigned enumeration',
348 field_value_range,
349 preferred_display_base,
350 user_attributes,
351 )
352
353 def create_real_field_class(self, is_single_precision=False, user_attributes=None):
354 field_class_ptr = native_bt.field_class_real_create(self._ptr)
355 self._check_field_class_create_status(field_class_ptr, 'real')
356
357 field_class = bt2_field_class._RealFieldClass._create_from_ptr(field_class_ptr)
358
359 field_class._is_single_precision = is_single_precision
360 self._set_field_class_user_attrs(field_class, user_attributes)
361
362 return field_class
363
364 def create_structure_field_class(self, user_attributes=None):
365 field_class_ptr = native_bt.field_class_structure_create(self._ptr)
366 self._check_field_class_create_status(field_class_ptr, 'structure')
367 fc = bt2_field_class._StructureFieldClass._create_from_ptr(field_class_ptr)
368 self._set_field_class_user_attrs(fc, user_attributes)
369 return fc
370
371 def create_string_field_class(self, user_attributes=None):
372 field_class_ptr = native_bt.field_class_string_create(self._ptr)
373 self._check_field_class_create_status(field_class_ptr, 'string')
374 fc = bt2_field_class._StringFieldClass._create_from_ptr(field_class_ptr)
375 self._set_field_class_user_attrs(fc, user_attributes)
376 return fc
377
378 def create_static_array_field_class(self, elem_fc, length, user_attributes=None):
379 utils._check_type(elem_fc, bt2_field_class._FieldClass)
380 utils._check_uint64(length)
381 ptr = native_bt.field_class_array_static_create(self._ptr, elem_fc._ptr, length)
382 self._check_field_class_create_status(ptr, 'static array')
383 fc = bt2_field_class._StaticArrayFieldClass._create_from_ptr_and_get_ref(ptr)
384 self._set_field_class_user_attrs(fc, user_attributes)
385 return fc
386
387 def create_dynamic_array_field_class(
388 self, elem_fc, length_fc=None, user_attributes=None
389 ):
390 utils._check_type(elem_fc, bt2_field_class._FieldClass)
391 length_fc_ptr = None
392
393 if length_fc is not None:
394 utils._check_type(length_fc, bt2_field_class._UnsignedIntegerFieldClass)
395 length_fc_ptr = length_fc._ptr
396
397 ptr = native_bt.field_class_array_dynamic_create(
398 self._ptr, elem_fc._ptr, length_fc_ptr
399 )
400 self._check_field_class_create_status(ptr, 'dynamic array')
401 fc = bt2_field_class._DynamicArrayFieldClass._create_from_ptr(ptr)
402 self._set_field_class_user_attrs(fc, user_attributes)
403 return fc
404
405 def create_option_field_class(
406 self, content_fc, selector_fc=None, user_attributes=None
407 ):
408 utils._check_type(content_fc, bt2_field_class._FieldClass)
409
410 selector_fc_ptr = None
411
412 if selector_fc is not None:
413 utils._check_type(selector_fc, bt2_field_class._BoolFieldClass)
414 selector_fc_ptr = selector_fc._ptr
415
416 ptr = native_bt.field_class_option_create(
417 self._ptr, content_fc._ptr, selector_fc_ptr
418 )
419 self._check_field_class_create_status(ptr, 'option')
420 fc = bt2_field_class._create_field_class_from_ptr_and_get_ref(ptr)
421 self._set_field_class_user_attrs(fc, user_attributes)
422 return fc
423
424 def create_variant_field_class(self, selector_fc=None, user_attributes=None):
425 selector_fc_ptr = None
426
427 if selector_fc is not None:
428 utils._check_type(selector_fc, bt2_field_class._IntegerFieldClass)
429 selector_fc_ptr = selector_fc._ptr
430
431 ptr = native_bt.field_class_variant_create(self._ptr, selector_fc_ptr)
432 self._check_field_class_create_status(ptr, 'variant')
433 fc = bt2_field_class._create_field_class_from_ptr_and_get_ref(ptr)
434 self._set_field_class_user_attrs(fc, user_attributes)
435 return fc
This page took 0.039247 seconds and 4 git commands to generate.