bt2: Add remaining trace-ir `*Const` classes and adapt tests
[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 _stream_class_pycls = bt2_stream_class._StreamClassConst
51 _create_value_from_ptr_and_get_ref = staticmethod(
52 bt2_value._create_from_const_ptr_and_get_ref
53 )
54
55 @property
56 def user_attributes(self):
57 ptr = native_bt.trace_class_borrow_user_attributes(self._ptr)
58 assert ptr is not None
59 return self._create_value_from_ptr_and_get_ref(ptr)
60
61 # Number of stream classes in this trace class.
62
63 def __len__(self):
64 count = native_bt.trace_class_get_stream_class_count(self._ptr)
65 assert count >= 0
66 return count
67
68 # Get a stream class by stream id.
69
70 def __getitem__(self, key):
71 utils._check_uint64(key)
72
73 sc_ptr = self._borrow_stream_class_ptr_by_id(self._ptr, key)
74 if sc_ptr is None:
75 raise KeyError(key)
76
77 return self._stream_class_pycls._create_from_ptr_and_get_ref(sc_ptr)
78
79 def __iter__(self):
80 for idx in range(len(self)):
81 sc_ptr = self._borrow_stream_class_ptr_by_index(self._ptr, idx)
82 assert sc_ptr is not None
83
84 id = native_bt.stream_class_get_id(sc_ptr)
85 assert id >= 0
86
87 yield id
88
89 @property
90 def assigns_automatic_stream_class_id(self):
91 return native_bt.trace_class_assigns_automatic_stream_class_id(self._ptr)
92
93 # Add a listener to be called when the trace class is destroyed.
94
95 def add_destruction_listener(self, listener):
96
97 if not callable(listener):
98 raise TypeError("'listener' parameter is not callable")
99
100 fn = native_bt.bt2_trace_class_add_destruction_listener
101 listener_from_native = functools.partial(
102 _trace_class_destruction_listener_from_native, listener
103 )
104
105 status, listener_id = fn(self._ptr, listener_from_native)
106 utils._handle_func_status(
107 status, 'cannot add destruction listener to trace class object'
108 )
109
110 return utils._ListenerHandle(listener_id, self)
111
112
113 class _TraceClass(_TraceClassConst):
114 _borrow_stream_class_ptr_by_index = staticmethod(
115 native_bt.trace_class_borrow_stream_class_by_index
116 )
117 _borrow_stream_class_ptr_by_id = staticmethod(
118 native_bt.trace_class_borrow_stream_class_by_id
119 )
120 _stream_class_pycls = bt2_stream_class._StreamClass
121 _create_value_from_ptr_and_get_ref = staticmethod(
122 bt2_value._create_from_ptr_and_get_ref
123 )
124
125 # Instantiate a trace of this class.
126
127 def __call__(self, name=None, user_attributes=None, uuid=None, environment=None):
128 trace_ptr = native_bt.trace_create(self._ptr)
129
130 if trace_ptr is None:
131 raise bt2._MemoryError('cannot create trace class object')
132
133 trace = bt2_trace._Trace._create_from_ptr(trace_ptr)
134
135 if name is not None:
136 trace._name = name
137
138 if user_attributes is not None:
139 trace._user_attributes = user_attributes
140
141 if uuid is not None:
142 trace._uuid = uuid
143
144 if environment is not None:
145 for key, value in environment.items():
146 trace.environment[key] = value
147
148 return trace
149
150 def create_stream_class(
151 self,
152 id=None,
153 name=None,
154 user_attributes=None,
155 packet_context_field_class=None,
156 event_common_context_field_class=None,
157 default_clock_class=None,
158 assigns_automatic_event_class_id=True,
159 assigns_automatic_stream_id=True,
160 supports_packets=False,
161 packets_have_beginning_default_clock_snapshot=False,
162 packets_have_end_default_clock_snapshot=False,
163 supports_discarded_events=False,
164 discarded_events_have_default_clock_snapshots=False,
165 supports_discarded_packets=False,
166 discarded_packets_have_default_clock_snapshots=False,
167 ):
168
169 if self.assigns_automatic_stream_class_id:
170 if id is not None:
171 raise ValueError(
172 'id provided, but trace class assigns automatic stream class ids'
173 )
174
175 sc_ptr = native_bt.stream_class_create(self._ptr)
176 else:
177 if id is None:
178 raise ValueError(
179 'id not provided, but trace class does not assign automatic stream class ids'
180 )
181
182 utils._check_uint64(id)
183 sc_ptr = native_bt.stream_class_create_with_id(self._ptr, id)
184
185 sc = bt2_stream_class._StreamClass._create_from_ptr(sc_ptr)
186
187 if name is not None:
188 sc._name = name
189
190 if user_attributes is not None:
191 sc._user_attributes = user_attributes
192
193 if event_common_context_field_class is not None:
194 sc._event_common_context_field_class = event_common_context_field_class
195
196 if default_clock_class is not None:
197 sc._default_clock_class = default_clock_class
198
199 # call after `sc._default_clock_class` because, if
200 # `packets_have_beginning_default_clock_snapshot` or
201 # `packets_have_end_default_clock_snapshot` is true, then this
202 # stream class needs a default clock class already.
203 sc._set_supports_packets(
204 supports_packets,
205 packets_have_beginning_default_clock_snapshot,
206 packets_have_end_default_clock_snapshot,
207 )
208
209 # call after sc._set_supports_packets() because, if
210 # `packet_context_field_class` is not `None`, then this stream
211 # class needs to support packets already.
212 if packet_context_field_class is not None:
213 sc._packet_context_field_class = packet_context_field_class
214
215 sc._assigns_automatic_event_class_id = assigns_automatic_event_class_id
216 sc._assigns_automatic_stream_id = assigns_automatic_stream_id
217 sc._set_supports_discarded_events(
218 supports_discarded_events, discarded_events_have_default_clock_snapshots
219 )
220 sc._set_supports_discarded_packets(
221 supports_discarded_packets, discarded_packets_have_default_clock_snapshots
222 )
223 return sc
224
225 def _user_attributes(self, user_attributes):
226 value = bt2_value.create_value(user_attributes)
227 utils._check_type(value, bt2_value.MapValue)
228 native_bt.trace_class_set_user_attributes(self._ptr, value._ptr)
229
230 _user_attributes = property(fset=_user_attributes)
231
232 def _assigns_automatic_stream_class_id(self, auto_id):
233 utils._check_bool(auto_id)
234 return native_bt.trace_class_set_assigns_automatic_stream_class_id(
235 self._ptr, auto_id
236 )
237
238 _assigns_automatic_stream_class_id = property(
239 fset=_assigns_automatic_stream_class_id
240 )
241
242 # Field class creation methods.
243
244 def _check_field_class_create_status(self, ptr, type_name):
245 if ptr is None:
246 raise bt2._MemoryError('cannot create {} field class'.format(type_name))
247
248 @staticmethod
249 def _set_field_class_user_attrs(fc, user_attributes):
250 if user_attributes is not None:
251 fc._user_attributes = user_attributes
252
253 def create_bool_field_class(self, user_attributes=None):
254 field_class_ptr = native_bt.field_class_bool_create(self._ptr)
255 self._check_field_class_create_status(field_class_ptr, 'boolean')
256 fc = bt2_field_class._BoolFieldClass._create_from_ptr(field_class_ptr)
257 self._set_field_class_user_attrs(fc, user_attributes)
258 return fc
259
260 def create_bit_array_field_class(self, length, user_attributes=None):
261 utils._check_uint64(length)
262
263 if length < 1 or length > 64:
264 raise ValueError(
265 'invalid length {}: expecting a value in the [1, 64] range'.format(
266 length
267 )
268 )
269
270 field_class_ptr = native_bt.field_class_bit_array_create(self._ptr, length)
271 self._check_field_class_create_status(field_class_ptr, 'bit array')
272 fc = bt2_field_class._BitArrayFieldClass._create_from_ptr(field_class_ptr)
273 self._set_field_class_user_attrs(fc, user_attributes)
274 return fc
275
276 def _create_integer_field_class(
277 self,
278 create_func,
279 py_cls,
280 type_name,
281 field_value_range,
282 preferred_display_base,
283 user_attributes,
284 ):
285 field_class_ptr = create_func(self._ptr)
286 self._check_field_class_create_status(field_class_ptr, type_name)
287
288 field_class = py_cls._create_from_ptr(field_class_ptr)
289
290 if field_value_range is not None:
291 field_class._field_value_range = field_value_range
292
293 if preferred_display_base is not None:
294 field_class._preferred_display_base = preferred_display_base
295
296 self._set_field_class_user_attrs(field_class, user_attributes)
297 return field_class
298
299 def create_signed_integer_field_class(
300 self, field_value_range=None, preferred_display_base=None, user_attributes=None
301 ):
302 return self._create_integer_field_class(
303 native_bt.field_class_integer_signed_create,
304 bt2_field_class._SignedIntegerFieldClass,
305 'signed integer',
306 field_value_range,
307 preferred_display_base,
308 user_attributes,
309 )
310
311 def create_unsigned_integer_field_class(
312 self, field_value_range=None, preferred_display_base=None, user_attributes=None
313 ):
314 return self._create_integer_field_class(
315 native_bt.field_class_integer_unsigned_create,
316 bt2_field_class._UnsignedIntegerFieldClass,
317 'unsigned integer',
318 field_value_range,
319 preferred_display_base,
320 user_attributes,
321 )
322
323 def create_signed_enumeration_field_class(
324 self, field_value_range=None, preferred_display_base=None, user_attributes=None
325 ):
326 return self._create_integer_field_class(
327 native_bt.field_class_enumeration_signed_create,
328 bt2_field_class._SignedEnumerationFieldClass,
329 'signed enumeration',
330 field_value_range,
331 preferred_display_base,
332 user_attributes,
333 )
334
335 def create_unsigned_enumeration_field_class(
336 self, field_value_range=None, preferred_display_base=None, user_attributes=None
337 ):
338 return self._create_integer_field_class(
339 native_bt.field_class_enumeration_unsigned_create,
340 bt2_field_class._UnsignedEnumerationFieldClass,
341 'unsigned enumeration',
342 field_value_range,
343 preferred_display_base,
344 user_attributes,
345 )
346
347 def create_real_field_class(self, is_single_precision=False, user_attributes=None):
348 field_class_ptr = native_bt.field_class_real_create(self._ptr)
349 self._check_field_class_create_status(field_class_ptr, 'real')
350
351 field_class = bt2_field_class._RealFieldClass._create_from_ptr(field_class_ptr)
352
353 field_class._is_single_precision = is_single_precision
354 self._set_field_class_user_attrs(field_class, user_attributes)
355
356 return field_class
357
358 def create_structure_field_class(self, user_attributes=None):
359 field_class_ptr = native_bt.field_class_structure_create(self._ptr)
360 self._check_field_class_create_status(field_class_ptr, 'structure')
361 fc = bt2_field_class._StructureFieldClass._create_from_ptr(field_class_ptr)
362 self._set_field_class_user_attrs(fc, user_attributes)
363 return fc
364
365 def create_string_field_class(self, user_attributes=None):
366 field_class_ptr = native_bt.field_class_string_create(self._ptr)
367 self._check_field_class_create_status(field_class_ptr, 'string')
368 fc = bt2_field_class._StringFieldClass._create_from_ptr(field_class_ptr)
369 self._set_field_class_user_attrs(fc, user_attributes)
370 return fc
371
372 def create_static_array_field_class(self, elem_fc, length, user_attributes=None):
373 utils._check_type(elem_fc, bt2_field_class._FieldClass)
374 utils._check_uint64(length)
375 ptr = native_bt.field_class_array_static_create(self._ptr, elem_fc._ptr, length)
376 self._check_field_class_create_status(ptr, 'static array')
377 fc = bt2_field_class._StaticArrayFieldClass._create_from_ptr_and_get_ref(ptr)
378 self._set_field_class_user_attrs(fc, user_attributes)
379 return fc
380
381 def create_dynamic_array_field_class(
382 self, elem_fc, length_fc=None, user_attributes=None
383 ):
384 utils._check_type(elem_fc, bt2_field_class._FieldClass)
385 length_fc_ptr = None
386
387 if length_fc is not None:
388 utils._check_type(length_fc, bt2_field_class._UnsignedIntegerFieldClass)
389 length_fc_ptr = length_fc._ptr
390
391 ptr = native_bt.field_class_array_dynamic_create(
392 self._ptr, elem_fc._ptr, length_fc_ptr
393 )
394 self._check_field_class_create_status(ptr, 'dynamic array')
395 fc = bt2_field_class._DynamicArrayFieldClass._create_from_ptr(ptr)
396 self._set_field_class_user_attrs(fc, user_attributes)
397 return fc
398
399 def create_option_field_class(
400 self, content_fc, selector_fc=None, user_attributes=None
401 ):
402 utils._check_type(content_fc, bt2_field_class._FieldClass)
403
404 selector_fc_ptr = None
405
406 if selector_fc is not None:
407 utils._check_type(selector_fc, bt2_field_class._BoolFieldClass)
408 selector_fc_ptr = selector_fc._ptr
409
410 ptr = native_bt.field_class_option_create(
411 self._ptr, content_fc._ptr, selector_fc_ptr
412 )
413 self._check_field_class_create_status(ptr, 'option')
414 fc = bt2_field_class._create_field_class_from_ptr_and_get_ref(ptr)
415 self._set_field_class_user_attrs(fc, user_attributes)
416 return fc
417
418 def create_variant_field_class(self, selector_fc=None, user_attributes=None):
419 selector_fc_ptr = None
420
421 if selector_fc is not None:
422 utils._check_type(selector_fc, bt2_field_class._IntegerFieldClass)
423 selector_fc_ptr = selector_fc._ptr
424
425 ptr = native_bt.field_class_variant_create(self._ptr, selector_fc_ptr)
426 self._check_field_class_create_status(ptr, 'variant')
427 fc = bt2_field_class._create_field_class_from_ptr_and_get_ref(ptr)
428 self._set_field_class_user_attrs(fc, user_attributes)
429 return fc
This page took 0.043075 seconds and 5 git commands to generate.