[babeltrace.git] / bindings / python / bt2 / trace.py

# The MIT License (MIT)
#
# Copyright (c) 2016 Philippe Proulx <pproulx@efficios.com>
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.

from bt2 import native_bt, object, utils
import bt2.field_types
import collections.abc
import bt2.values
import copy
import bt2


class _StreamClassIterator(collections.abc.Iterator):
    def __init__(self, trace):
        self._trace = trace
        self._at = 0

    def __next__(self):
        if self._at == len(self._trace):
            raise StopIteration

        sc_ptr = native_bt.ctf_trace_get_stream_class(self._trace._ptr,
                                                      self._at)
        utils._handle_ptr(sc_ptr, "cannot get trace class object's stream class object")
        id = native_bt.ctf_stream_class_get_id(sc_ptr)
        native_bt.put(sc_ptr)

        if utils._is_m1ull(id):
            raise bt2.Error("cannot get stream class object's ID")

        self._at += 1
        return id


class _TraceClockClassesIterator(collections.abc.Iterator):
    def __init__(self, trace_clock_classes):
        self._trace_clock_classes = trace_clock_classes
        self._at = 0

    def __next__(self):
        if self._at == len(self._trace_clock_classes):
            raise StopIteration

        trace_ptr = self._trace_clock_classes._trace._ptr
        cc_ptr = native_bt.ctf_trace_get_clock_class(trace_ptr, self._at)
        utils._handle_ptr(cc_ptr, "cannot get trace class object's clock class")
        name = native_bt.ctf_clock_class_get_name(cc_ptr)
        native_bt.put(cc_ptr)
        utils._handle_ptr(name, "cannot get clock class object's name")
        self._at += 1
        return name


class _TraceClockClasses(collections.abc.Mapping):
    def __init__(self, trace):
        self._trace = trace

    def __getitem__(self, key):
        utils._check_str(key)
        cc_ptr = native_bt.ctf_trace_get_clock_class_by_name(self._trace._ptr,
                                                             key)

        if cc_ptr is None:
            raise KeyError(key)

        return bt2.ClockClass._create_from_ptr(cc_ptr)

    def __len__(self):
        count = native_bt.ctf_trace_get_clock_class_count(self._trace._ptr)
        utils._handle_ret(count, "cannot get trace class object's clock class count")
        return count

    def __iter__(self):
        return _TraceClockClassesIterator(self)


class _TraceEnvIterator(collections.abc.Iterator):
    def __init__(self, trace_env):
        self._trace_env = trace_env
        self._at = 0

    def __next__(self):
        if self._at == len(self._trace_env):
            raise StopIteration

        trace_ptr = self._trace_env._trace._ptr
        entry_name = native_bt.ctf_trace_get_environment_field_name(trace_ptr,
                                                                    self._at)
        utils._handle_ptr(entry_name, "cannot get trace class object's environment entry name")
        self._at += 1
        return entry_name


class _TraceEnv(collections.abc.MutableMapping):
    def __init__(self, trace):
        self._trace = trace

    def __getitem__(self, key):
        utils._check_str(key)
        value_ptr = native_bt.ctf_trace_get_environment_field_value_by_name(self._trace._ptr,
                                                                            key)

        if value_ptr is None:
            raise KeyError(key)

        return bt2.values._create_from_ptr(value_ptr)

    def __setitem__(self, key, value):
        utils._check_str(key)
        value = bt2.create_value(value)
        ret = native_bt.ctf_trace_set_environment_field(self._trace._ptr,
                                                        key, value._ptr)
        utils._handle_ret(ret, "cannot set trace class object's environment entry")

    def __delitem__(self, key):
        raise NotImplementedError

    def __len__(self):
        count = native_bt.ctf_trace_get_environment_field_count(self._trace._ptr)
        utils._handle_ret(count, "cannot get trace class object's environment entry count")
        return count

    def __iter__(self):
        return _TraceEnvIterator(self)


class Trace(object._Object, collections.abc.Mapping):
    def __init__(self, name=None, native_byte_order=None, env=None,
                 packet_header_field_type=None, clock_classes=None,
                 stream_classes=None):
        ptr = native_bt.ctf_trace_create()

        if ptr is None:
            raise bt2.CreationError('cannot create trace class object')

        super().__init__(ptr)

        if name is not None:
            self.name = name

        if native_byte_order is not None:
            self.native_byte_order = native_byte_order

        if packet_header_field_type is not None:
            self.packet_header_field_type = packet_header_field_type

        if env is not None:
            for key, value in env.items():
                self.env[key] = value

        if clock_classes is not None:
            for clock_class in clock_classes:
                self.add_clock_class(clock_class)

        if stream_classes is not None:
            for stream_class in stream_classes:
                self.add_stream_class(stream_class)

    def __getitem__(self, key):
        utils._check_int64(key)
        sc_ptr = native_bt.ctf_trace_get_stream_class_by_id(self._ptr, key)

        if sc_ptr is None:
            raise KeyError(key)

        return bt2.StreamClass._create_from_ptr(sc_ptr)

    def __len__(self):
        count = native_bt.ctf_trace_get_stream_class_count(self._ptr)
        utils._handle_ret(count, "cannot get trace class object's stream class count")
        return count

    def __iter__(self):
        return _StreamClassIterator(self)

    def add_stream_class(self, stream_class):
        utils._check_type(stream_class, bt2.StreamClass)
        ret = native_bt.ctf_trace_add_stream_class(self._ptr, stream_class._ptr)
        utils._handle_ret(ret, "cannot add stream class object to trace class object")

    @property
    def name(self):
        return native_bt.ctf_trace_get_name(self._ptr)

    @name.setter
    def name(self, name):
        utils._check_str(name)
        ret = native_bt.ctf_trace_set_name(self._ptr, name)
        utils._handle_ret(ret, "cannot set trace class object's name")

    @property
    def native_byte_order(self):
        bo = native_bt.ctf_trace_get_byte_order(self._ptr)
        utils._handle_ret(bo, "cannot get trace class object's native byte order")
        return bo

    @native_byte_order.setter
    def native_byte_order(self, native_byte_order):
        utils._check_int(native_byte_order)
        ret = native_bt.ctf_trace_set_byte_order(self._ptr, native_byte_order)
        utils._handle_ret(ret, "cannot set trace class object's native byte order")

    @property
    def env(self):
        return _TraceEnv(self)

    @property
    def clock_classes(self):
        return _TraceClockClasses(self)

    def add_clock_class(self, clock_class):
        utils._check_type(clock_class, bt2.ClockClass)
        ret = native_bt.ctf_trace_add_clock_class(self._ptr, clock_class._ptr)
        utils._handle_ret(ret, "cannot add clock class object to trace class object")

    @property
    def packet_header_field_type(self):
        ft_ptr = native_bt.ctf_trace_get_packet_header_type(self._ptr)

        if ft_ptr is None:
            return

        return bt2.field_types._create_from_ptr(ft_ptr)

    @packet_header_field_type.setter
    def packet_header_field_type(self, packet_header_field_type):
        packet_header_field_type_ptr = None

        if packet_header_field_type is not None:
            utils._check_type(packet_header_field_type, bt2.field_types._FieldType)
            packet_header_field_type_ptr = packet_header_field_type._ptr

        ret = native_bt.ctf_trace_set_packet_header_type(self._ptr,
                                                         packet_header_field_type_ptr)
        utils._handle_ret(ret, "cannot set trace class object's packet header field type")

    def __eq__(self, other):
        if type(other) is not type(self):
            # not comparing apples to apples
            return False

        if self.addr == other.addr:
            return True

        self_stream_classes = list(self.values())
        self_clock_classes = list(self.clock_classes.values())
        self_env = {key: val for key, val in self.env.items()}
        other_stream_classes = list(other.values())
        other_clock_classes = list(other.clock_classes.values())
        other_env = {key: val for key, val in other.env.items()}
        self_props = (
            self_stream_classes,
            self_clock_classes,
            self_env,
            self.name,
            self.native_byte_order,
            self.packet_header_field_type,
        )
        other_props = (
            other_stream_classes,
            other_clock_classes,
            other_env,
            other.name,
            other.native_byte_order,
            other.packet_header_field_type,
        )
        return self_props == other_props

    def _copy(self, gen_copy_func, sc_copy_func):
        cpy = Trace()

        if self.name is not None:
            cpy.name = self.name

        cpy.packet_header_field_type = gen_copy_func(self.packet_header_field_type)

        for key, val in self.env.items():
            cpy.env[key] = gen_copy_func(val)

        for clock_class in self.clock_classes.values():
            cpy.add_clock_class(gen_copy_func(clock_class))

        for stream_class in self.values():
            cpy.add_stream_class(sc_copy_func(stream_class))

        return cpy

    def __copy__(self):
        return self._copy(lambda obj: obj, copy.copy)

    def __deepcopy__(self, memo):
        cpy = self._copy(copy.deepcopy, copy.deepcopy)
        memo[id(self)] = cpy
        return cpy
Commit	Line	Data
81447b5b PP	1	# The MIT License (MIT)
	2	#
	3	# Copyright (c) 2016 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_types
	25	import collections.abc
	26	import bt2.values
	27	import copy
	28	import bt2
	29
	30
	31	class _StreamClassIterator(collections.abc.Iterator):
	32	def __init__(self, trace):
	33	self._trace = trace
	34	self._at = 0
	35
	36	def __next__(self):
	37	if self._at == len(self._trace):
	38	raise StopIteration
	39
	40	sc_ptr = native_bt.ctf_trace_get_stream_class(self._trace._ptr,
	41	self._at)
	42	utils._handle_ptr(sc_ptr, "cannot get trace class object's stream class object")
	43	id = native_bt.ctf_stream_class_get_id(sc_ptr)
	44	native_bt.put(sc_ptr)
	45
	46	if utils._is_m1ull(id):
	47	raise bt2.Error("cannot get stream class object's ID")
	48
	49	self._at += 1
	50	return id
	51
	52
	53	class _TraceClockClassesIterator(collections.abc.Iterator):
	54	def __init__(self, trace_clock_classes):
	55	self._trace_clock_classes = trace_clock_classes
	56	self._at = 0
	57
	58	def __next__(self):
	59	if self._at == len(self._trace_clock_classes):
	60	raise StopIteration
	61
	62	trace_ptr = self._trace_clock_classes._trace._ptr
	63	cc_ptr = native_bt.ctf_trace_get_clock_class(trace_ptr, self._at)
	64	utils._handle_ptr(cc_ptr, "cannot get trace class object's clock class")
65	name = native_bt.ctf_clock_class_get_name(cc_ptr)
66	native_bt.put(cc_ptr)
67	utils._handle_ptr(name, "cannot get clock class object's name")
68	self._at += 1
69	return name
70
71
72	class _TraceClockClasses(collections.abc.Mapping):
73	def __init__(self, trace):
74	self._trace = trace
75
76	def __getitem__(self, key):
77	utils._check_str(key)
78	cc_ptr = native_bt.ctf_trace_get_clock_class_by_name(self._trace._ptr,
79	key)
80
81	if cc_ptr is None:
82	raise KeyError(key)
83
84	return bt2.ClockClass._create_from_ptr(cc_ptr)
85
86	def __len__(self):
87	count = native_bt.ctf_trace_get_clock_class_count(self._trace._ptr)
88	utils._handle_ret(count, "cannot get trace class object's clock class count")
89	return count
90
91	def __iter__(self):
92	return _TraceClockClassesIterator(self)
93
94
95	class _TraceEnvIterator(collections.abc.Iterator):
96	def __init__(self, trace_env):
97	self._trace_env = trace_env
98	self._at = 0
99
100	def __next__(self):
101	if self._at == len(self._trace_env):
102	raise StopIteration
103
104	trace_ptr = self._trace_env._trace._ptr
105	entry_name = native_bt.ctf_trace_get_environment_field_name(trace_ptr,
106	self._at)
107	utils._handle_ptr(entry_name, "cannot get trace class object's environment entry name")
108	self._at += 1
109	return entry_name
110
111
112	class _TraceEnv(collections.abc.MutableMapping):
113	def __init__(self, trace):
114	self._trace = trace
115
116	def __getitem__(self, key):
117	utils._check_str(key)
118	value_ptr = native_bt.ctf_trace_get_environment_field_value_by_name(self._trace._ptr,
119	key)
120
121	if value_ptr is None:
122	raise KeyError(key)
123
124	return bt2.values._create_from_ptr(value_ptr)
125
126	def __setitem__(self, key, value):
127	utils._check_str(key)
128	value = bt2.create_value(value)
129	ret = native_bt.ctf_trace_set_environment_field(self._trace._ptr,
130	key, value._ptr)
131	utils._handle_ret(ret, "cannot set trace class object's environment entry")
132
133	def __delitem__(self, key):
134	raise NotImplementedError
135
136	def __len__(self):
137	count = native_bt.ctf_trace_get_environment_field_count(self._trace._ptr)
138	utils._handle_ret(count, "cannot get trace class object's environment entry count")
139	return count
140
141	def __iter__(self):
142	return _TraceEnvIterator(self)
143
144
145	class Trace(object._Object, collections.abc.Mapping):
146	def __init__(self, name=None, native_byte_order=None, env=None,
147	packet_header_field_type=None, clock_classes=None,
148	stream_classes=None):
149	ptr = native_bt.ctf_trace_create()
150
151	if ptr is None:
152	raise bt2.CreationError('cannot create trace class object')
153
154	super().__init__(ptr)
155
156	if name is not None:
157	self.name = name
158
159	if native_byte_order is not None:
160	self.native_byte_order = native_byte_order
161
162	if packet_header_field_type is not None:
163	self.packet_header_field_type = packet_header_field_type
164
165	if env is not None:
166	for key, value in env.items():
167	self.env[key] = value
168
169	if clock_classes is not None:
170	for clock_class in clock_classes:
171	self.add_clock_class(clock_class)
172
173	if stream_classes is not None:
174	for stream_class in stream_classes:
175	self.add_stream_class(stream_class)
176
177	def __getitem__(self, key):
178	utils._check_int64(key)
179	sc_ptr = native_bt.ctf_trace_get_stream_class_by_id(self._ptr, key)
180
181	if sc_ptr is None:
182	raise KeyError(key)
183
184	return bt2.StreamClass._create_from_ptr(sc_ptr)
185
186	def __len__(self):
187	count = native_bt.ctf_trace_get_stream_class_count(self._ptr)
188	utils._handle_ret(count, "cannot get trace class object's stream class count")
189	return count
190
191	def __iter__(self):
192	return _StreamClassIterator(self)
193
194	def add_stream_class(self, stream_class):
195	utils._check_type(stream_class, bt2.StreamClass)
196	ret = native_bt.ctf_trace_add_stream_class(self._ptr, stream_class._ptr)
197	utils._handle_ret(ret, "cannot add stream class object to trace class object")
198
199	@property
200	def name(self):
201	return native_bt.ctf_trace_get_name(self._ptr)
202
203	@name.setter
204	def name(self, name):
205	utils._check_str(name)
206	ret = native_bt.ctf_trace_set_name(self._ptr, name)
207	utils._handle_ret(ret, "cannot set trace class object's name")
208
209	@property
210	def native_byte_order(self):
211	bo = native_bt.ctf_trace_get_byte_order(self._ptr)
212	utils._handle_ret(bo, "cannot get trace class object's native byte order")
213	return bo
214
215	@native_byte_order.setter
216	def native_byte_order(self, native_byte_order):
217	utils._check_int(native_byte_order)
218	ret = native_bt.ctf_trace_set_byte_order(self._ptr, native_byte_order)
219	utils._handle_ret(ret, "cannot set trace class object's native byte order")
220
221	@property
222	def env(self):
223	return _TraceEnv(self)
224
225	@property
226	def clock_classes(self):
227	return _TraceClockClasses(self)
228
229	def add_clock_class(self, clock_class):
230	utils._check_type(clock_class, bt2.ClockClass)
231	ret = native_bt.ctf_trace_add_clock_class(self._ptr, clock_class._ptr)
232	utils._handle_ret(ret, "cannot add clock class object to trace class object")
233
234	@property
235	def packet_header_field_type(self):
236	ft_ptr = native_bt.ctf_trace_get_packet_header_type(self._ptr)
237
238	if ft_ptr is None:
239	return
240
241	return bt2.field_types._create_from_ptr(ft_ptr)
242
243	@packet_header_field_type.setter
244	def packet_header_field_type(self, packet_header_field_type):
245	packet_header_field_type_ptr = None
246
247	if packet_header_field_type is not None:
248	utils._check_type(packet_header_field_type, bt2.field_types._FieldType)
249	packet_header_field_type_ptr = packet_header_field_type._ptr
250
251	ret = native_bt.ctf_trace_set_packet_header_type(self._ptr,
252	packet_header_field_type_ptr)
253	utils._handle_ret(ret, "cannot set trace class object's packet header field type")
254
255	def __eq__(self, other):
256	if type(other) is not type(self):
257	# not comparing apples to apples
258	return False
259
260	if self.addr == other.addr:
261	return True
262
263	self_stream_classes = list(self.values())
264	self_clock_classes = list(self.clock_classes.values())
265	self_env = {key: val for key, val in self.env.items()}
266	other_stream_classes = list(other.values())
267	other_clock_classes = list(other.clock_classes.values())
268	other_env = {key: val for key, val in other.env.items()}
269	self_props = (
270	self_stream_classes,
271	self_clock_classes,
272	self_env,
273	self.name,
274	self.native_byte_order,
275	self.packet_header_field_type,
276	)
277	other_props = (
278	other_stream_classes,
279	other_clock_classes,
280	other_env,
281	other.name,
282	other.native_byte_order,
283	other.packet_header_field_type,
284	)
285	return self_props == other_props
286
287	def _copy(self, gen_copy_func, sc_copy_func):
288	cpy = Trace()
289
290	if self.name is not None:
291	cpy.name = self.name
292
293	cpy.packet_header_field_type = gen_copy_func(self.packet_header_field_type)
294
295	for key, val in self.env.items():
296	cpy.env[key] = gen_copy_func(val)
297
298	for clock_class in self.clock_classes.values():
299	cpy.add_clock_class(gen_copy_func(clock_class))
300
301	for stream_class in self.values():
302	cpy.add_stream_class(sc_copy_func(stream_class))
303
304	return cpy
305
306	def __copy__(self):
307	return self._copy(lambda obj: obj, copy.copy)
308
309	def __deepcopy__(self, memo):
310	cpy = self._copy(copy.deepcopy, copy.deepcopy)
311	memo[id(self)] = cpy
312	return cpy