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

# The MIT License (MIT)
#
# Copyright (c) 2017 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 bt2.stream
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_by_index(self._trace._ptr,
                                                               self._at)
        assert(sc_ptr)
        id = native_bt.ctf_stream_class_get_id(sc_ptr)
        native_bt.put(sc_ptr)
        assert(id >= 0)
        self._at += 1
        return id


class _TraceStreams(collections.abc.Sequence):
    def __init__(self, trace):
        self._trace = trace

    def __len__(self):
        count = native_bt.ctf_trace_get_stream_count(self._trace._ptr)
        assert(count >= 0)
        return count

    def __getitem__(self, index):
        utils._check_uint64(index)

        if index >= len(self):
            raise IndexError

        stream_ptr = native_bt.ctf_trace_get_stream_by_index(self._trace._ptr,
                                                             index)
        assert(stream_ptr)
        return bt2.stream._create_from_ptr(stream_ptr)


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_by_index(trace_ptr, self._at)
        assert(cc_ptr)
        name = native_bt.ctf_clock_class_get_name(cc_ptr)
        native_bt.put(cc_ptr)
        assert(name is not None)
        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)
        assert(count >= 0)
        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_by_index(trace_ptr,
                                                                             self._at)
        assert(entry_name is not None)
        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)
        assert(count >= 0)
        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)
        assert(count >= 0)
        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_native_byte_order(self._ptr)
        assert(bo >= 0)
        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_native_byte_order(self._ptr, native_byte_order)
        utils._handle_ret(ret, "cannot set trace class object's native byte order")

    @property
    def is_static(self):
        is_static = native_bt.ctf_trace_is_static(self._ptr)
        return is_static > 0

    def set_is_static(self):
        ret = native_bt.ctf_trace_set_is_static(self._ptr)
        utils._handle_ret(ret, "cannot set trace object as static")

    @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 streams(self):
        return _TraceStreams(self)

    @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)
81447b5b PP	2	#
811644b8	3	# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
81447b5b PP	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
811644b8	27	import bt2.stream
81447b5b PP	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
811644b8 PP	41	sc_ptr = native_bt.ctf_trace_get_stream_class_by_index(self._trace._ptr,
	42	self._at)
	43	assert(sc_ptr)
81447b5b PP	44	id = native_bt.ctf_stream_class_get_id(sc_ptr)
81447b5b PP	45	native_bt.put(sc_ptr)
811644b8	46	assert(id >= 0)
81447b5b PP	47	self._at += 1
	48	return id
	49
	50
811644b8 PP	51	class _TraceStreams(collections.abc.Sequence):
	52	def __init__(self, trace):
	53	self._trace = trace
	54
	55	def __len__(self):
	56	count = native_bt.ctf_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.ctf_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
81447b5b PP	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
811644b8 PP	82	cc_ptr = native_bt.ctf_trace_get_clock_class_by_index(trace_ptr, self._at)
811644b8 PP	83	assert(cc_ptr)
81447b5b PP	84	name = native_bt.ctf_clock_class_get_name(cc_ptr)
81447b5b PP	85	native_bt.put(cc_ptr)
811644b8	86	assert(name is not None)
81447b5b PP	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.ctf_trace_get_clock_class_by_name(self._trace._ptr,
	98	key)
	99
	100	if cc_ptr is None:
	101	raise KeyError(key)
	102
	103	return bt2.ClockClass._create_from_ptr(cc_ptr)
	104
	105	def __len__(self):
	106	count = native_bt.ctf_trace_get_clock_class_count(self._trace._ptr)
811644b8	107	assert(count >= 0)
81447b5b PP	108	return count
	109
	110	def __iter__(self):
	111	return _TraceClockClassesIterator(self)
	112
	113
	114	class _TraceEnvIterator(collections.abc.Iterator):
	115	def __init__(self, trace_env):
	116	self._trace_env = trace_env
	117	self._at = 0
	118
	119	def __next__(self):
	120	if self._at == len(self._trace_env):
	121	raise StopIteration
	122
	123	trace_ptr = self._trace_env._trace._ptr
811644b8 PP	124	entry_name = native_bt.ctf_trace_get_environment_field_name_by_index(trace_ptr,
	125	self._at)
	126	assert(entry_name is not None)
81447b5b PP	127	self._at += 1
	128	return entry_name
	129
	130
	131	class _TraceEnv(collections.abc.MutableMapping):
	132	def __init__(self, trace):
	133	self._trace = trace
	134
	135	def __getitem__(self, key):
	136	utils._check_str(key)
	137	value_ptr = native_bt.ctf_trace_get_environment_field_value_by_name(self._trace._ptr,
	138	key)
	139
	140	if value_ptr is None:
	141	raise KeyError(key)
	142
	143	return bt2.values._create_from_ptr(value_ptr)
	144
	145	def __setitem__(self, key, value):
	146	utils._check_str(key)
	147	value = bt2.create_value(value)
	148	ret = native_bt.ctf_trace_set_environment_field(self._trace._ptr,
	149	key, value._ptr)
	150	utils._handle_ret(ret, "cannot set trace class object's environment entry")
	151
	152	def __delitem__(self, key):
	153	raise NotImplementedError
	154
	155	def __len__(self):
	156	count = native_bt.ctf_trace_get_environment_field_count(self._trace._ptr)
811644b8	157	assert(count >= 0)
81447b5b PP	158	return count
	159
	160	def __iter__(self):
	161	return _TraceEnvIterator(self)
	162
	163
	164	class Trace(object._Object, collections.abc.Mapping):
	165	def __init__(self, name=None, native_byte_order=None, env=None,
	166	packet_header_field_type=None, clock_classes=None,
	167	stream_classes=None):
	168	ptr = native_bt.ctf_trace_create()
	169
	170	if ptr is None:
	171	raise bt2.CreationError('cannot create trace class object')
	172
	173	super().__init__(ptr)
	174
	175	if name is not None:
	176	self.name = name
	177
	178	if native_byte_order is not None:
	179	self.native_byte_order = native_byte_order
	180
	181	if packet_header_field_type is not None:
	182	self.packet_header_field_type = packet_header_field_type
	183
	184	if env is not None:
	185	for key, value in env.items():
	186	self.env[key] = value
	187
	188	if clock_classes is not None:
	189	for clock_class in clock_classes:
	190	self.add_clock_class(clock_class)
	191
	192	if stream_classes is not None:
	193	for stream_class in stream_classes:
	194	self.add_stream_class(stream_class)
	195
	196	def __getitem__(self, key):
	197	utils._check_int64(key)
	198	sc_ptr = native_bt.ctf_trace_get_stream_class_by_id(self._ptr, key)
	199
	200	if sc_ptr is None:
	201	raise KeyError(key)
	202
	203	return bt2.StreamClass._create_from_ptr(sc_ptr)
	204
	205	def __len__(self):
	206	count = native_bt.ctf_trace_get_stream_class_count(self._ptr)
811644b8	207	assert(count >= 0)
81447b5b PP	208	return count
	209
	210	def __iter__(self):
	211	return _StreamClassIterator(self)
	212
	213	def add_stream_class(self, stream_class):
	214	utils._check_type(stream_class, bt2.StreamClass)
	215	ret = native_bt.ctf_trace_add_stream_class(self._ptr, stream_class._ptr)
	216	utils._handle_ret(ret, "cannot add stream class object to trace class object")
	217
	218	@property
	219	def name(self):
	220	return native_bt.ctf_trace_get_name(self._ptr)
	221
	222	@name.setter
	223	def name(self, name):
	224	utils._check_str(name)
	225	ret = native_bt.ctf_trace_set_name(self._ptr, name)
	226	utils._handle_ret(ret, "cannot set trace class object's name")
	227
	228	@property
	229	def native_byte_order(self):
811644b8 PP	230	bo = native_bt.ctf_trace_get_native_byte_order(self._ptr)
811644b8 PP	231	assert(bo >= 0)
81447b5b PP	232	return bo
	233
	234	@native_byte_order.setter
	235	def native_byte_order(self, native_byte_order):
	236	utils._check_int(native_byte_order)
811644b8	237	ret = native_bt.ctf_trace_set_native_byte_order(self._ptr, native_byte_order)
81447b5b PP	238	utils._handle_ret(ret, "cannot set trace class object's native byte order")
81447b5b PP	239
811644b8 PP	240	@property
	241	def is_static(self):
	242	is_static = native_bt.ctf_trace_is_static(self._ptr)
	243	return is_static > 0
	244
	245	def set_is_static(self):
	246	ret = native_bt.ctf_trace_set_is_static(self._ptr)
	247	utils._handle_ret(ret, "cannot set trace object as static")
	248
81447b5b PP	249	@property
	250	def env(self):
	251	return _TraceEnv(self)
	252
	253	@property
	254	def clock_classes(self):
	255	return _TraceClockClasses(self)
	256
	257	def add_clock_class(self, clock_class):
	258	utils._check_type(clock_class, bt2.ClockClass)
	259	ret = native_bt.ctf_trace_add_clock_class(self._ptr, clock_class._ptr)
	260	utils._handle_ret(ret, "cannot add clock class object to trace class object")
	261
811644b8 PP	262	@property
	263	def streams(self):
	264	return _TraceStreams(self)
	265
81447b5b PP	266	@property
	267	def packet_header_field_type(self):
	268	ft_ptr = native_bt.ctf_trace_get_packet_header_type(self._ptr)
	269
	270	if ft_ptr is None:
	271	return
	272
	273	return bt2.field_types._create_from_ptr(ft_ptr)
	274
	275	@packet_header_field_type.setter
	276	def packet_header_field_type(self, packet_header_field_type):
	277	packet_header_field_type_ptr = None
	278
	279	if packet_header_field_type is not None:
	280	utils._check_type(packet_header_field_type, bt2.field_types._FieldType)
	281	packet_header_field_type_ptr = packet_header_field_type._ptr
	282
	283	ret = native_bt.ctf_trace_set_packet_header_type(self._ptr,
	284	packet_header_field_type_ptr)
	285	utils._handle_ret(ret, "cannot set trace class object's packet header field type")
	286
	287	def __eq__(self, other):
	288	if type(other) is not type(self):
	289	# not comparing apples to apples
	290	return False
	291
	292	if self.addr == other.addr:
	293	return True
	294
	295	self_stream_classes = list(self.values())
	296	self_clock_classes = list(self.clock_classes.values())
	297	self_env = {key: val for key, val in self.env.items()}
	298	other_stream_classes = list(other.values())
	299	other_clock_classes = list(other.clock_classes.values())
	300	other_env = {key: val for key, val in other.env.items()}
	301	self_props = (
	302	self_stream_classes,
	303	self_clock_classes,
	304	self_env,
	305	self.name,
	306	self.native_byte_order,
	307	self.packet_header_field_type,
	308	)
	309	other_props = (
	310	other_stream_classes,
	311	other_clock_classes,
	312	other_env,
	313	other.name,
	314	other.native_byte_order,
	315	other.packet_header_field_type,
	316	)
	317	return self_props == other_props
	318
	319	def _copy(self, gen_copy_func, sc_copy_func):
	320	cpy = Trace()
	321
	322	if self.name is not None:
	323	cpy.name = self.name
	324
	325	cpy.packet_header_field_type = gen_copy_func(self.packet_header_field_type)
	326
	327	for key, val in self.env.items():
	328	cpy.env[key] = gen_copy_func(val)
	329
330	for clock_class in self.clock_classes.values():
331	cpy.add_clock_class(gen_copy_func(clock_class))
332
333	for stream_class in self.values():
334	cpy.add_stream_class(sc_copy_func(stream_class))
335
336	return cpy
337
338	def __copy__(self):
339	return self._copy(lambda obj: obj, copy.copy)
340
341	def __deepcopy__(self, memo):
342	cpy = self._copy(copy.deepcopy, copy.deepcopy)
343	memo[id(self)] = cpy
344	return cpy