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

# The MIT License (MIT)
#
# Copyright (c) 2016-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, stream, utils
import uuid as uuidp
import bt2.event
import bt2.field
import abc
import bt2


class CtfWriterClock(bt2.object._SharedObject):
    def __init__(self, name, description=None, frequency=None, precision=None,
                 offset=None, is_absolute=None, uuid=None):
        utils._check_str(name)
        ptr = native_bt.ctf_clock_create(name)

        if ptr is None:
            raise bt2.CreationError('cannot create CTF writer clock object')

        super().__init__(ptr)

        if description is not None:
            self.description = description

        if frequency is not None:
            self.frequency = frequency

        if precision is not None:
            self.precision = precision

        if offset is not None:
            self.offset = offset

        if is_absolute is not None:
            self.is_absolute = is_absolute

        if uuid is not None:
            self.uuid = uuid

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

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

        self_props = (
            self.name,
            self.description,
            self.frequency,
            self.precision,
            self.offset,
            self.is_absolute,
            self.uuid
        )
        other_props = (
            other.name,
            other.description,
            other.frequency,
            other.precision,
            other.offset,
            other.is_absolute,
            other.uuid
        )
        return self_props == other_props

    def __copy__(self):
        return CtfWriterClock(name=self.name, description=self.description,
                              frequency=self.frequency,
                              precision=self.precision, offset=self.offset,
                              is_absolute=self.is_absolute, uuid=self.uuid)

    def __deepcopy__(self, memo):
        cpy = self.__copy__()
        memo[id(self)] = cpy
        return cpy

    @property
    def name(self):
        name = native_bt.ctf_clock_get_name(self._ptr)
        assert(name is not None)
        return name

    @property
    def description(self):
        description = native_bt.ctf_clock_get_description(self._ptr)
        return description

    @description.setter
    def description(self, description):
        utils._check_str(description)
        ret = native_bt.ctf_clock_set_description(self._ptr, description)
        utils._handle_ret(ret, "cannot set CTF writer clock object's description")

    @property
    def frequency(self):
        frequency = native_bt.ctf_clock_get_frequency(self._ptr)
        assert(frequency >= 1)
        return frequency

    @frequency.setter
    def frequency(self, frequency):
        utils._check_uint64(frequency)
        ret = native_bt.ctf_clock_set_frequency(self._ptr, frequency)
        utils._handle_ret(ret, "cannot set CTF writer clock object's frequency")

    @property
    def precision(self):
        precision = native_bt.ctf_clock_get_precision(self._ptr)
        assert(precision >= 0)
        return precision

    @precision.setter
    def precision(self, precision):
        utils._check_uint64(precision)
        ret = native_bt.ctf_clock_set_precision(self._ptr, precision)
        utils._handle_ret(ret, "cannot set CTF writer clock object's precision")

    @property
    def offset(self):
        ret, offset_s = native_bt.ctf_clock_get_offset_s(self._ptr)
        assert(ret == 0)
        ret, offset_cycles = native_bt.ctf_clock_get_offset(self._ptr)
        assert(ret == 0)
        return bt2.ClockClassOffset(offset_s, offset_cycles)

    @offset.setter
    def offset(self, offset):
        utils._check_type(offset, bt2.ClockClassOffset)
        ret = native_bt.ctf_clock_set_offset_s(self._ptr, offset.seconds)
        utils._handle_ret(ret, "cannot set CTF writer clock object's offset (seconds)")
        ret = native_bt.ctf_clock_set_offset(self._ptr, offset.cycles)
        utils._handle_ret(ret, "cannot set CTF writer clock object's offset (cycles)")

    @property
    def is_absolute(self):
        is_absolute = native_bt.ctf_clock_get_is_absolute(self._ptr)
        assert(is_absolute >= 0)
        return is_absolute > 0

    @is_absolute.setter
    def is_absolute(self, is_absolute):
        utils._check_bool(is_absolute)
        ret = native_bt.ctf_clock_set_is_absolute(self._ptr, int(is_absolute))
        utils._handle_ret(ret, "cannot set CTF writer clock object's absoluteness")

    @property
    def uuid(self):
        uuid_bytes = native_bt.ctf_clock_get_uuid(self._ptr)
        assert(uuid_bytes is not None)
        return uuidp.UUID(bytes=uuid_bytes)

    @uuid.setter
    def uuid(self, uuid):
        utils._check_type(uuid, uuidp.UUID)
        ret = native_bt.ctf_clock_set_uuid(self._ptr, uuid.bytes)
        utils._handle_ret(ret, "cannot set CTF writer clock object's UUID")

    def _time(self, time):
        utils._check_int64(time)
        ret = native_bt.ctf_clock_set_time(self._ptr, time)

    time = property(fset=_time)


class _CtfWriterStream:
    @property
    def discarded_events_count(self):
        ret, count = native_bt.stream_get_discarded_events_count(self._ptr)
        utils._handle_ret(ret, "cannot get CTF writer stream object's discarded events count")
        return count

    def append_discarded_events(self, count):
        utils._check_uint64(count)
        native_bt.stream_append_discarded_events(self._ptr, count)

    def append_event(self, event):
        utils._check_type(event, bt2.event._Event)
        ret = native_bt.stream_append_event(self._ptr, event._ptr)
        utils._handle_ret(ret, 'cannot append event object to CTF writer stream object')

    def flush(self):
        ret = native_bt.stream_flush(self._ptr)
        utils._handle_ret(ret, 'cannot flush CTF writer stream object')

    @property
    def packet_header_field(self):
        field_ptr = native_bt.stream_get_packet_header(self._ptr)

        if field_ptr is None:
            return

        return bt2.field._create_from_ptr(field_ptr)

    @packet_header_field.setter
    def packet_header_field(self, packet_header_field):
        packet_header_field_ptr = None

        if packet_header_field is not None:
            utils._check_type(packet_header_field, bt2.field._Field)
            packet_header_field_ptr = packet_header_field._ptr

        ret = native_bt.stream_set_packet_header(self._ptr,
                                                 packet_header_field_ptr)
        utils._handle_ret(ret, "cannot set CTF writer stream object's packet header field")

    @property
    def packet_context_field(self):
        field_ptr = native_bt.stream_get_packet_context(self._ptr)

        if field_ptr is None:
            return

        return bt2.field._create_from_ptr(field_ptr)

    @packet_context_field.setter
    def packet_context_field(self, packet_context_field):
        packet_context_field_ptr = None

        if packet_context_field is not None:
            utils._check_type(packet_context_field, bt2.field._Field)
            packet_context_field_ptr = packet_context_field._ptr

        ret = native_bt.stream_set_packet_context(self._ptr,
                                                  packet_context_field_ptr)
        utils._handle_ret(ret, "cannot set CTF writer stream object's packet context field")

    def __eq__(self, other):
        if type(other) is not type(self):
            return False

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

        if not _StreamBase.__eq__(self, other):
            return False

        self_props = (
            self.discarded_events_count,
            self.packet_header_field,
            self.packet_context_field,
        )
        other_props = (
            other.discarded_events_count,
            other.packet_header_field,
            other.packet_context_field,
        )
        return self_props == other_props

    def _copy(self, copy_func):
        cpy = self.stream_class(self.name)
        cpy.append_discarded_events(self.discarded_events_count)
        cpy.packet_header_field = copy_func(self.packet_header_field)
        cpy.packet_context_field = copy_func(self.packet_context_field)
        return cpy

    def __copy__(self):
        return self._copy(copy.copy)

    def __deepcopy__(self, memo):
        cpy = self._copy(copy.deepcopy)
        memo[id(self)] = cpy
        return cpy


class CtfWriter(bt2.object._SharedObject):
    def __init__(self, path):
        utils._check_str(path)
        ptr = native_bt.ctf_writer_create(path)

        if ptr is None:
            raise bt2.CreationError('cannot create CTF writer object')

        super().__init__(ptr)

    @property
    def trace(self):
        trace_ptr = native_bt.ctf_writer_get_trace(self._ptr)
        assert(trace_ptr)
        return bt2.Trace._create_from_ptr(trace_ptr)

    @property
    def metadata_string(self):
        metadata_string = native_bt.ctf_writer_get_metadata_string(self._ptr)
        assert(metadata_string is not None)
        return metadata_string

    def flush_metadata(self):
        native_bt.ctf_writer_flush_metadata(self._ptr)

    def add_clock(self, clock):
        utils._check_type(clock, CtfWriterClock)
        ret = native_bt.ctf_writer_add_clock(self._ptr, clock._ptr)
        utils._handle_ret(ret, 'cannot add CTF writer clock object to CTF writer object')
Commit	Line	Data
81447b5b PP	1	# The MIT License (MIT)
	2	#
	3	# Copyright (c) 2016-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, stream, utils
	24	import uuid as uuidp
	25	import bt2.event
c4239792	26	import bt2.field
81447b5b PP	27	import abc
	28	import bt2
	29
	30
78288f58	31	class CtfWriterClock(bt2.object._SharedObject):
81447b5b PP	32	def __init__(self, name, description=None, frequency=None, precision=None,
	33	offset=None, is_absolute=None, uuid=None):
	34	utils._check_str(name)
	35	ptr = native_bt.ctf_clock_create(name)
	36
	37	if ptr is None:
	38	raise bt2.CreationError('cannot create CTF writer clock object')
	39
	40	super().__init__(ptr)
	41
	42	if description is not None:
	43	self.description = description
	44
	45	if frequency is not None:
	46	self.frequency = frequency
	47
	48	if precision is not None:
	49	self.precision = precision
	50
	51	if offset is not None:
	52	self.offset = offset
	53
	54	if is_absolute is not None:
	55	self.is_absolute = is_absolute
	56
	57	if uuid is not None:
	58	self.uuid = uuid
	59
	60	def __eq__(self, other):
	61	if type(self) is not type(other):
	62	# not comparing apples to apples
	63	return False
	64
	65	if self.addr == other.addr:
	66	return True
	67
	68	self_props = (
	69	self.name,
	70	self.description,
	71	self.frequency,
	72	self.precision,
	73	self.offset,
	74	self.is_absolute,
	75	self.uuid
	76	)
	77	other_props = (
	78	other.name,
	79	other.description,
	80	other.frequency,
	81	other.precision,
	82	other.offset,
	83	other.is_absolute,
	84	other.uuid
	85	)
	86	return self_props == other_props
	87
	88	def __copy__(self):
	89	return CtfWriterClock(name=self.name, description=self.description,
	90	frequency=self.frequency,
	91	precision=self.precision, offset=self.offset,
	92	is_absolute=self.is_absolute, uuid=self.uuid)
	93
	94	def __deepcopy__(self, memo):
	95	cpy = self.__copy__()
96	memo[id(self)] = cpy
97	return cpy
98
99	@property
100	def name(self):
101	name = native_bt.ctf_clock_get_name(self._ptr)
811644b8	102	assert(name is not None)
81447b5b PP	103	return name
	104
	105	@property
	106	def description(self):
	107	description = native_bt.ctf_clock_get_description(self._ptr)
	108	return description
	109
	110	@description.setter
	111	def description(self, description):
	112	utils._check_str(description)
	113	ret = native_bt.ctf_clock_set_description(self._ptr, description)
	114	utils._handle_ret(ret, "cannot set CTF writer clock object's description")
	115
	116	@property
	117	def frequency(self):
	118	frequency = native_bt.ctf_clock_get_frequency(self._ptr)
811644b8	119	assert(frequency >= 1)
81447b5b PP	120	return frequency
	121
	122	@frequency.setter
	123	def frequency(self, frequency):
	124	utils._check_uint64(frequency)
	125	ret = native_bt.ctf_clock_set_frequency(self._ptr, frequency)
	126	utils._handle_ret(ret, "cannot set CTF writer clock object's frequency")
	127
	128	@property
	129	def precision(self):
	130	precision = native_bt.ctf_clock_get_precision(self._ptr)
811644b8	131	assert(precision >= 0)
81447b5b PP	132	return precision
	133
	134	@precision.setter
	135	def precision(self, precision):
	136	utils._check_uint64(precision)
	137	ret = native_bt.ctf_clock_set_precision(self._ptr, precision)
	138	utils._handle_ret(ret, "cannot set CTF writer clock object's precision")
	139
	140	@property
	141	def offset(self):
	142	ret, offset_s = native_bt.ctf_clock_get_offset_s(self._ptr)
811644b8	143	assert(ret == 0)
81447b5b	144	ret, offset_cycles = native_bt.ctf_clock_get_offset(self._ptr)
811644b8	145	assert(ret == 0)
81447b5b PP	146	return bt2.ClockClassOffset(offset_s, offset_cycles)
	147
	148	@offset.setter
	149	def offset(self, offset):
	150	utils._check_type(offset, bt2.ClockClassOffset)
	151	ret = native_bt.ctf_clock_set_offset_s(self._ptr, offset.seconds)
	152	utils._handle_ret(ret, "cannot set CTF writer clock object's offset (seconds)")
	153	ret = native_bt.ctf_clock_set_offset(self._ptr, offset.cycles)
	154	utils._handle_ret(ret, "cannot set CTF writer clock object's offset (cycles)")
	155
	156	@property
	157	def is_absolute(self):
	158	is_absolute = native_bt.ctf_clock_get_is_absolute(self._ptr)
811644b8	159	assert(is_absolute >= 0)
81447b5b PP	160	return is_absolute > 0
	161
	162	@is_absolute.setter
	163	def is_absolute(self, is_absolute):
	164	utils._check_bool(is_absolute)
	165	ret = native_bt.ctf_clock_set_is_absolute(self._ptr, int(is_absolute))
	166	utils._handle_ret(ret, "cannot set CTF writer clock object's absoluteness")
	167
	168	@property
	169	def uuid(self):
	170	uuid_bytes = native_bt.ctf_clock_get_uuid(self._ptr)
811644b8	171	assert(uuid_bytes is not None)
81447b5b PP	172	return uuidp.UUID(bytes=uuid_bytes)
	173
	174	@uuid.setter
	175	def uuid(self, uuid):
	176	utils._check_type(uuid, uuidp.UUID)
	177	ret = native_bt.ctf_clock_set_uuid(self._ptr, uuid.bytes)
	178	utils._handle_ret(ret, "cannot set CTF writer clock object's UUID")
	179
	180	def _time(self, time):
	181	utils._check_int64(time)
	182	ret = native_bt.ctf_clock_set_time(self._ptr, time)
	183
	184	time = property(fset=_time)
	185
	186
af4bbfc7	187	class _CtfWriterStream:
81447b5b PP	188	@property
81447b5b PP	189	def discarded_events_count(self):
50842bdc	190	ret, count = native_bt.stream_get_discarded_events_count(self._ptr)
81447b5b PP	191	utils._handle_ret(ret, "cannot get CTF writer stream object's discarded events count")
	192	return count
	193
	194	def append_discarded_events(self, count):
	195	utils._check_uint64(count)
50842bdc	196	native_bt.stream_append_discarded_events(self._ptr, count)
81447b5b PP	197
	198	def append_event(self, event):
	199	utils._check_type(event, bt2.event._Event)
50842bdc	200	ret = native_bt.stream_append_event(self._ptr, event._ptr)
81447b5b PP	201	utils._handle_ret(ret, 'cannot append event object to CTF writer stream object')
	202
	203	def flush(self):
50842bdc	204	ret = native_bt.stream_flush(self._ptr)
6b475588	205	utils._handle_ret(ret, 'cannot flush CTF writer stream object')
81447b5b PP	206
	207	@property
	208	def packet_header_field(self):
50842bdc	209	field_ptr = native_bt.stream_get_packet_header(self._ptr)
81447b5b PP	210
	211	if field_ptr is None:
	212	return
	213
c4239792	214	return bt2.field._create_from_ptr(field_ptr)
81447b5b PP	215
	216	@packet_header_field.setter
	217	def packet_header_field(self, packet_header_field):
	218	packet_header_field_ptr = None
	219
	220	if packet_header_field is not None:
c4239792	221	utils._check_type(packet_header_field, bt2.field._Field)
81447b5b PP	222	packet_header_field_ptr = packet_header_field._ptr
81447b5b PP	223
50842bdc PP	224	ret = native_bt.stream_set_packet_header(self._ptr,
50842bdc PP	225	packet_header_field_ptr)
81447b5b PP	226	utils._handle_ret(ret, "cannot set CTF writer stream object's packet header field")
	227
	228	@property
	229	def packet_context_field(self):
50842bdc	230	field_ptr = native_bt.stream_get_packet_context(self._ptr)
81447b5b PP	231
	232	if field_ptr is None:
	233	return
	234
c4239792	235	return bt2.field._create_from_ptr(field_ptr)
81447b5b PP	236
	237	@packet_context_field.setter
	238	def packet_context_field(self, packet_context_field):
	239	packet_context_field_ptr = None
	240
	241	if packet_context_field is not None:
c4239792	242	utils._check_type(packet_context_field, bt2.field._Field)
81447b5b PP	243	packet_context_field_ptr = packet_context_field._ptr
81447b5b PP	244
50842bdc PP	245	ret = native_bt.stream_set_packet_context(self._ptr,
50842bdc PP	246	packet_context_field_ptr)
81447b5b PP	247	utils._handle_ret(ret, "cannot set CTF writer stream object's packet context field")
	248
	249	def __eq__(self, other):
	250	if type(other) is not type(self):
	251	return False
	252
	253	if self.addr == other.addr:
	254	return True
	255
	256	if not _StreamBase.__eq__(self, other):
	257	return False
	258
	259	self_props = (
	260	self.discarded_events_count,
	261	self.packet_header_field,
	262	self.packet_context_field,
	263	)
	264	other_props = (
	265	other.discarded_events_count,
	266	other.packet_header_field,
	267	other.packet_context_field,
	268	)
	269	return self_props == other_props
	270
	271	def _copy(self, copy_func):
	272	cpy = self.stream_class(self.name)
	273	cpy.append_discarded_events(self.discarded_events_count)
	274	cpy.packet_header_field = copy_func(self.packet_header_field)
	275	cpy.packet_context_field = copy_func(self.packet_context_field)
	276	return cpy
	277
	278	def __copy__(self):
	279	return self._copy(copy.copy)
	280
	281	def __deepcopy__(self, memo):
	282	cpy = self._copy(copy.deepcopy)
	283	memo[id(self)] = cpy
	284	return cpy
	285
	286
78288f58	287	class CtfWriter(bt2.object._SharedObject):
81447b5b PP	288	def __init__(self, path):
	289	utils._check_str(path)
	290	ptr = native_bt.ctf_writer_create(path)
	291
	292	if ptr is None:
	293	raise bt2.CreationError('cannot create CTF writer object')
	294
	295	super().__init__(ptr)
	296
	297	@property
	298	def trace(self):
	299	trace_ptr = native_bt.ctf_writer_get_trace(self._ptr)
811644b8	300	assert(trace_ptr)
81447b5b PP	301	return bt2.Trace._create_from_ptr(trace_ptr)
	302
	303	@property
	304	def metadata_string(self):
	305	metadata_string = native_bt.ctf_writer_get_metadata_string(self._ptr)
811644b8	306	assert(metadata_string is not None)
81447b5b PP	307	return metadata_string
	308
	309	def flush_metadata(self):
	310	native_bt.ctf_writer_flush_metadata(self._ptr)
	311
	312	def add_clock(self, clock):
	313	utils._check_type(clock, CtfWriterClock)
	314	ret = native_bt.ctf_writer_add_clock(self._ptr, clock._ptr)
	315	utils._handle_ret(ret, 'cannot add CTF writer clock object to CTF writer object')