1 # The MIT License (MIT)
3 # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
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:
12 # The above copyright notice and this permission notice shall be included in
13 # all copies or substantial portions of the Software.
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
23 from bt2
import native_bt
, object, utils
24 import bt2
.field_class
25 import collections
.abc
32 class _StreamClassIterator(collections
.abc
.Iterator
):
33 def __init__(self
, trace
):
38 if self
._at
== len(self
._trace
):
41 sc_ptr
= native_bt
.trace_get_stream_class_by_index(self
._trace
._ptr
,
44 id = native_bt
.stream_class_get_id(sc_ptr
)
51 class _TraceStreams(collections
.abc
.Sequence
):
52 def __init__(self
, trace
):
56 count
= native_bt
.trace_get_stream_count(self
._trace
._ptr
)
60 def __getitem__(self
, index
):
61 utils
._check
_uint
64(index
)
63 if index
>= len(self
):
66 stream_ptr
= native_bt
.trace_get_stream_by_index(self
._trace
._ptr
,
69 return bt2
.stream
._create
_from
_ptr
(stream_ptr
)
72 class _TraceClockClassesIterator(collections
.abc
.Iterator
):
73 def __init__(self
, trace_clock_classes
):
74 self
._trace
_clock
_classes
= trace_clock_classes
78 if self
._at
== len(self
._trace
_clock
_classes
):
81 trace_ptr
= self
._trace
_clock
_classes
._trace
._ptr
82 cc_ptr
= native_bt
.trace_get_clock_class_by_index(trace_ptr
, self
._at
)
84 name
= native_bt
.clock_class_get_name(cc_ptr
)
86 assert(name
is not None)
91 class _TraceClockClasses(collections
.abc
.Mapping
):
92 def __init__(self
, trace
):
95 def __getitem__(self
, key
):
97 cc_ptr
= native_bt
.trace_get_clock_class_by_name(self
._trace
._ptr
, key
)
102 return bt2
.ClockClass
._create
_from
_ptr
(cc_ptr
)
105 count
= native_bt
.trace_get_clock_class_count(self
._trace
._ptr
)
110 return _TraceClockClassesIterator(self
)
113 class _TraceEnvIterator(collections
.abc
.Iterator
):
114 def __init__(self
, trace_env
):
115 self
._trace
_env
= trace_env
119 if self
._at
== len(self
._trace
_env
):
122 trace_ptr
= self
._trace
_env
._trace
._ptr
123 entry_name
= native_bt
.trace_get_environment_field_name_by_index(trace_ptr
,
125 assert(entry_name
is not None)
130 class _TraceEnv(collections
.abc
.MutableMapping
):
131 def __init__(self
, trace
):
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
,
139 if value_ptr
is None:
142 return bt2
.value
._create
_from
_ptr
(value_ptr
)
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
,
149 utils
._handle
_ret
(ret
, "cannot set trace class object's environment entry")
151 def __delitem__(self
, key
):
152 raise NotImplementedError
155 count
= native_bt
.trace_get_environment_field_count(self
._trace
._ptr
)
160 return _TraceEnvIterator(self
)
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()
170 raise bt2
.CreationError('cannot create trace class object')
172 super().__init
__(ptr
)
177 if native_byte_order
is not None:
178 self
.native_byte_order
= native_byte_order
180 if packet_header_field_class
is not None:
181 self
.packet_header_field_class
= packet_header_field_class
184 for key
, value
in env
.items():
185 self
.env
[key
] = value
187 if clock_classes
is not None:
188 for clock_class
in clock_classes
:
189 self
.add_clock_class(clock_class
)
191 if stream_classes
is not None:
192 for stream_class
in stream_classes
:
193 self
.add_stream_class(stream_class
)
195 def __getitem__(self
, key
):
196 utils
._check
_int
64(key
)
197 sc_ptr
= native_bt
.trace_get_stream_class_by_id(self
._ptr
, key
)
202 return bt2
.StreamClass
._create
_from
_ptr
(sc_ptr
)
205 count
= native_bt
.trace_get_stream_class_count(self
._ptr
)
210 return _StreamClassIterator(self
)
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")
219 return native_bt
.trace_get_name(self
._ptr
)
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")
228 def native_byte_order(self
):
229 bo
= native_bt
.trace_get_native_byte_order(self
._ptr
)
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")
241 is_static
= native_bt
.trace_is_static(self
._ptr
)
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")
250 return _TraceEnv(self
)
253 def clock_classes(self
):
254 return _TraceClockClasses(self
)
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")
263 return _TraceStreams(self
)
266 def packet_header_field_class(self
):
267 fc_ptr
= native_bt
.trace_get_packet_header_type(self
._ptr
)
272 return bt2
.field_class
._create
_from
_ptr
(fc_ptr
)
274 @packet_header_field_class.setter
275 def packet_header_field_class(self
, packet_header_field_class
):
276 packet_header_field_class_ptr
= None
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
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")
286 def __eq__(self
, other
):
287 if type(other
) is not type(self
):
288 # not comparing apples to apples
291 if self
.addr
== other
.addr
:
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()}
305 self
.native_byte_order
,
306 self
.packet_header_field_class
,
309 other_stream_classes
,
313 other
.native_byte_order
,
314 other
.packet_header_field_class
,
316 return self_props
== other_props
318 def _copy(self
, gen_copy_func
, sc_copy_func
):
321 if self
.name
is not None:
324 cpy
.packet_header_field_class
= gen_copy_func(self
.packet_header_field_class
)
326 for key
, val
in self
.env
.items():
327 cpy
.env
[key
] = gen_copy_func(val
)
329 for clock_class
in self
.clock_classes
.values():
330 cpy
.add_clock_class(gen_copy_func(clock_class
))
332 for stream_class
in self
.values():
333 cpy
.add_stream_class(sc_copy_func(stream_class
))
338 return self
._copy
(lambda obj
: obj
, copy
.copy
)
340 def __deepcopy__(self
, memo
):
341 cpy
= self
._copy
(copy
.deepcopy
, copy
.deepcopy
)