Fix: string format warnings on 32 bits
[babeltrace.git] / bindings / python / bt2 / bt2 / trace.py
1 # The MIT License (MIT)
2 #
3 # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
4 #
5 # Permission is hereby granted, free of charge, to any person obtaining a copy
6 # of this software and associated documentation files (the "Software"), to deal
7 # in the Software without restriction, including without limitation the rights
8 # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9 # copies of the Software, and to permit persons to whom the Software is
10 # furnished to do so, subject to the following conditions:
11 #
12 # The above copyright notice and this permission notice shall be included in
13 # all copies or substantial portions of the Software.
14 #
15 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16 # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17 # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18 # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
21 # THE SOFTWARE.
22
23 from bt2 import native_bt, object, utils
24 import bt2.field_class
25 import collections.abc
26 import bt2.value
27 import bt2.stream
28 import copy
29 import bt2
30
31
32 class _StreamClassIterator(collections.abc.Iterator):
33 def __init__(self, trace):
34 self._trace = trace
35 self._at = 0
36
37 def __next__(self):
38 if self._at == len(self._trace):
39 raise StopIteration
40
41 sc_ptr = native_bt.trace_get_stream_class_by_index(self._trace._ptr,
42 self._at)
43 assert(sc_ptr)
44 id = native_bt.stream_class_get_id(sc_ptr)
45 native_bt.put(sc_ptr)
46 assert(id >= 0)
47 self._at += 1
48 return id
49
50
51 class _TraceStreams(collections.abc.Sequence):
52 def __init__(self, trace):
53 self._trace = trace
54
55 def __len__(self):
56 count = native_bt.trace_get_stream_count(self._trace._ptr)
57 assert(count >= 0)
58 return count
59
60 def __getitem__(self, index):
61 utils._check_uint64(index)
62
63 if index >= len(self):
64 raise IndexError
65
66 stream_ptr = native_bt.trace_get_stream_by_index(self._trace._ptr,
67 index)
68 assert(stream_ptr)
69 return bt2.stream._create_from_ptr(stream_ptr)
70
71
72 class _TraceClockClassesIterator(collections.abc.Iterator):
73 def __init__(self, trace_clock_classes):
74 self._trace_clock_classes = trace_clock_classes
75 self._at = 0
76
77 def __next__(self):
78 if self._at == len(self._trace_clock_classes):
79 raise StopIteration
80
81 trace_ptr = self._trace_clock_classes._trace._ptr
82 cc_ptr = native_bt.trace_get_clock_class_by_index(trace_ptr, self._at)
83 assert(cc_ptr)
84 name = native_bt.clock_class_get_name(cc_ptr)
85 native_bt.put(cc_ptr)
86 assert(name is not None)
87 self._at += 1
88 return name
89
90
91 class _TraceClockClasses(collections.abc.Mapping):
92 def __init__(self, trace):
93 self._trace = trace
94
95 def __getitem__(self, key):
96 utils._check_str(key)
97 cc_ptr = native_bt.trace_get_clock_class_by_name(self._trace._ptr, key)
98
99 if cc_ptr is None:
100 raise KeyError(key)
101
102 return bt2.ClockClass._create_from_ptr(cc_ptr)
103
104 def __len__(self):
105 count = native_bt.trace_get_clock_class_count(self._trace._ptr)
106 assert(count >= 0)
107 return count
108
109 def __iter__(self):
110 return _TraceClockClassesIterator(self)
111
112
113 class _TraceEnvIterator(collections.abc.Iterator):
114 def __init__(self, trace_env):
115 self._trace_env = trace_env
116 self._at = 0
117
118 def __next__(self):
119 if self._at == len(self._trace_env):
120 raise StopIteration
121
122 trace_ptr = self._trace_env._trace._ptr
123 entry_name = native_bt.trace_get_environment_field_name_by_index(trace_ptr,
124 self._at)
125 assert(entry_name is not None)
126 self._at += 1
127 return entry_name
128
129
130 class _TraceEnv(collections.abc.MutableMapping):
131 def __init__(self, trace):
132 self._trace = trace
133
134 def __getitem__(self, key):
135 utils._check_str(key)
136 value_ptr = native_bt.trace_get_environment_field_value_by_name(self._trace._ptr,
137 key)
138
139 if value_ptr is None:
140 raise KeyError(key)
141
142 return bt2.value._create_from_ptr(value_ptr)
143
144 def __setitem__(self, key, value):
145 utils._check_str(key)
146 value = bt2.create_value(value)
147 ret = native_bt.trace_set_environment_field(self._trace._ptr,
148 key, value._ptr)
149 utils._handle_ret(ret, "cannot set trace class object's environment entry")
150
151 def __delitem__(self, key):
152 raise NotImplementedError
153
154 def __len__(self):
155 count = native_bt.trace_get_environment_field_count(self._trace._ptr)
156 assert(count >= 0)
157 return count
158
159 def __iter__(self):
160 return _TraceEnvIterator(self)
161
162
163 class Trace(object._Object, collections.abc.Mapping):
164 def __init__(self, name=None, native_byte_order=None, env=None,
165 packet_header_field_class=None, clock_classes=None,
166 stream_classes=None):
167 ptr = native_bt.trace_create()
168
169 if ptr is None:
170 raise bt2.CreationError('cannot create trace class object')
171
172 super().__init__(ptr)
173
174 if name is not None:
175 self.name = name
176
177 if native_byte_order is not None:
178 self.native_byte_order = native_byte_order
179
180 if packet_header_field_class is not None:
181 self.packet_header_field_class = packet_header_field_class
182
183 if env is not None:
184 for key, value in env.items():
185 self.env[key] = value
186
187 if clock_classes is not None:
188 for clock_class in clock_classes:
189 self.add_clock_class(clock_class)
190
191 if stream_classes is not None:
192 for stream_class in stream_classes:
193 self.add_stream_class(stream_class)
194
195 def __getitem__(self, key):
196 utils._check_int64(key)
197 sc_ptr = native_bt.trace_get_stream_class_by_id(self._ptr, key)
198
199 if sc_ptr is None:
200 raise KeyError(key)
201
202 return bt2.StreamClass._create_from_ptr(sc_ptr)
203
204 def __len__(self):
205 count = native_bt.trace_get_stream_class_count(self._ptr)
206 assert(count >= 0)
207 return count
208
209 def __iter__(self):
210 return _StreamClassIterator(self)
211
212 def add_stream_class(self, stream_class):
213 utils._check_type(stream_class, bt2.StreamClass)
214 ret = native_bt.trace_add_stream_class(self._ptr, stream_class._ptr)
215 utils._handle_ret(ret, "cannot add stream class object to trace class object")
216
217 @property
218 def name(self):
219 return native_bt.trace_get_name(self._ptr)
220
221 @name.setter
222 def name(self, name):
223 utils._check_str(name)
224 ret = native_bt.trace_set_name(self._ptr, name)
225 utils._handle_ret(ret, "cannot set trace class object's name")
226
227 @property
228 def native_byte_order(self):
229 bo = native_bt.trace_get_native_byte_order(self._ptr)
230 assert(bo >= 0)
231 return bo
232
233 @native_byte_order.setter
234 def native_byte_order(self, native_byte_order):
235 utils._check_int(native_byte_order)
236 ret = native_bt.trace_set_native_byte_order(self._ptr, native_byte_order)
237 utils._handle_ret(ret, "cannot set trace class object's native byte order")
238
239 @property
240 def is_static(self):
241 is_static = native_bt.trace_is_static(self._ptr)
242 return is_static > 0
243
244 def set_is_static(self):
245 ret = native_bt.trace_set_is_static(self._ptr)
246 utils._handle_ret(ret, "cannot set trace object as static")
247
248 @property
249 def env(self):
250 return _TraceEnv(self)
251
252 @property
253 def clock_classes(self):
254 return _TraceClockClasses(self)
255
256 def add_clock_class(self, clock_class):
257 utils._check_type(clock_class, bt2.ClockClass)
258 ret = native_bt.trace_add_clock_class(self._ptr, clock_class._ptr)
259 utils._handle_ret(ret, "cannot add clock class object to trace class object")
260
261 @property
262 def streams(self):
263 return _TraceStreams(self)
264
265 @property
266 def packet_header_field_class(self):
267 fc_ptr = native_bt.trace_get_packet_header_type(self._ptr)
268
269 if fc_ptr is None:
270 return
271
272 return bt2.field_class._create_from_ptr(fc_ptr)
273
274 @packet_header_field_class.setter
275 def packet_header_field_class(self, packet_header_field_class):
276 packet_header_field_class_ptr = None
277
278 if packet_header_field_class is not None:
279 utils._check_type(packet_header_field_class, bt2.field_class._FieldClass)
280 packet_header_field_class_ptr = packet_header_field_class._ptr
281
282 ret = native_bt.trace_set_packet_header_type(self._ptr,
283 packet_header_field_class_ptr)
284 utils._handle_ret(ret, "cannot set trace class object's packet header field class")
285
286 def __eq__(self, other):
287 if type(other) is not type(self):
288 # not comparing apples to apples
289 return False
290
291 if self.addr == other.addr:
292 return True
293
294 self_stream_classes = list(self.values())
295 self_clock_classes = list(self.clock_classes.values())
296 self_env = {key: val for key, val in self.env.items()}
297 other_stream_classes = list(other.values())
298 other_clock_classes = list(other.clock_classes.values())
299 other_env = {key: val for key, val in other.env.items()}
300 self_props = (
301 self_stream_classes,
302 self_clock_classes,
303 self_env,
304 self.name,
305 self.native_byte_order,
306 self.packet_header_field_class,
307 )
308 other_props = (
309 other_stream_classes,
310 other_clock_classes,
311 other_env,
312 other.name,
313 other.native_byte_order,
314 other.packet_header_field_class,
315 )
316 return self_props == other_props
317
318 def _copy(self, gen_copy_func, sc_copy_func):
319 cpy = Trace()
320
321 if self.name is not None:
322 cpy.name = self.name
323
324 cpy.packet_header_field_class = gen_copy_func(self.packet_header_field_class)
325
326 for key, val in self.env.items():
327 cpy.env[key] = gen_copy_func(val)
328
329 for clock_class in self.clock_classes.values():
330 cpy.add_clock_class(gen_copy_func(clock_class))
331
332 for stream_class in self.values():
333 cpy.add_stream_class(sc_copy_func(stream_class))
334
335 return cpy
336
337 def __copy__(self):
338 return self._copy(lambda obj: obj, copy.copy)
339
340 def __deepcopy__(self, memo):
341 cpy = self._copy(copy.deepcopy, copy.deepcopy)
342 memo[id(self)] = cpy
343 return cpy
This page took 0.038543 seconds and 5 git commands to generate.