[babeltrace.git] / bindings / python / bt2 / stream_class.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.ctf_writer
import bt2.stream
import copy
import bt2


class _EventClassIterator(collections.abc.Iterator):
    def __init__(self, stream_class):
        self._stream_class = stream_class
        self._at = 0

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

        ec_ptr = native_bt.ctf_stream_class_get_event_class(self._stream_class._ptr,
                                                            self._at)
        utils._handle_ptr(ec_ptr, "cannot get stream class object's event class object")
        name = native_bt.ctf_event_class_get_name(ec_ptr)
        native_bt.put(ec_ptr)
        utils._handle_ptr(name, "cannot get event class object's name")
        self._at += 1
        return name


class StreamClass(object._Object, collections.abc.Mapping):
    def __init__(self, name=None, id=None, packet_context_field_type=None,
                 event_header_field_type=None, event_context_field_type=None,
                 event_classes=None):
        ptr = native_bt.ctf_stream_class_create(None)

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

        super().__init__(ptr)

        if name is not None:
            self.name = name

        if id is not None:
            self.id = id

        if packet_context_field_type is not None:
            self.packet_context_field_type = packet_context_field_type

        if event_header_field_type is not None:
            self.event_header_field_type = event_header_field_type

        if event_context_field_type is not None:
            self.event_context_field_type = event_context_field_type

        if event_classes is not None:
            for event_class in event_classes:
                self.add_event_class(event_class)

    def __getitem__(self, key):
        utils._check_str(key)
        ec_ptr = native_bt.ctf_stream_class_get_event_class_by_name(self._ptr,
                                                                    key)

        if ec_ptr is None:
            raise KeyError(key)

        return bt2.EventClass._create_from_ptr(ec_ptr)

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

    def __iter__(self):
        return _EventClassIterator(self)

    def event_class_with_id(self, id):
        utils._check_int64(id)
        ec_ptr = native_bt.ctf_stream_class_get_event_class_by_id(self._ptr, id)
        utils._handle_ptr(ec_ptr, "cannot get stream class object's event class object")
        return bt2.EventClass._create_from_ptr(ec_ptr)

    def add_event_class(self, event_class):
        utils._check_type(event_class, bt2.EventClass)
        ret = native_bt.ctf_stream_class_add_event_class(self._ptr, event_class._ptr)
        utils._handle_ret(ret, "cannot add event class object to stream class object's")

    @property
    def trace(self):
        tc_ptr = native_bt.ctf_stream_class_get_trace(self._ptr)

        if tc_ptr is not None:
            return bt2.Trace._create_from_ptr(tc_ptr)

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

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

    @property
    def id(self):
        id = native_bt.ctf_stream_class_get_id(self._ptr)

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

        return id

    @id.setter
    def id(self, id):
        utils._check_int64(id)
        ret = native_bt.ctf_stream_class_set_id(self._ptr, id)
        utils._handle_ret(ret, "cannot set stream class object's ID")

    @property
    def clock(self):
        clock_ptr = native_bt.ctf_stream_class_get_clock(self._ptr)

        if clock_ptr is None:
            return

        return bt2.ctf_writer.CtfWriterClock._create_from_ptr(clock_ptr)

    @clock.setter
    def clock(self, clock):
        utils._check_type(clock, bt2.ctf_writer.CtfWriterClock)
        ret = native_bt.ctf_stream_class_set_clock(self._ptr, clock._ptr)
        utils._handle_ret(ret, "cannot set stream class object's CTF writer clock object")

    @property
    def packet_context_field_type(self):
        ft_ptr = native_bt.ctf_stream_class_get_packet_context_type(self._ptr)

        if ft_ptr is None:
            return

        return bt2.field_types._create_from_ptr(ft_ptr)

    @packet_context_field_type.setter
    def packet_context_field_type(self, packet_context_field_type):
        packet_context_field_type_ptr = None

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

        ret = native_bt.ctf_stream_class_set_packet_context_type(self._ptr,
                                                                 packet_context_field_type_ptr)
        utils._handle_ret(ret, "cannot set stream class object's packet context field type")

    @property
    def event_header_field_type(self):
        ft_ptr = native_bt.ctf_stream_class_get_event_header_type(self._ptr)

        if ft_ptr is None:
            return

        return bt2.field_types._create_from_ptr(ft_ptr)

    @event_header_field_type.setter
    def event_header_field_type(self, event_header_field_type):
        event_header_field_type_ptr = None

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

        ret = native_bt.ctf_stream_class_set_event_header_type(self._ptr,
                                                               event_header_field_type_ptr)
        utils._handle_ret(ret, "cannot set stream class object's event header field type")

    @property
    def event_context_field_type(self):
        ft_ptr = native_bt.ctf_stream_class_get_event_context_type(self._ptr)

        if ft_ptr is None:
            return

        return bt2.field_types._create_from_ptr(ft_ptr)

    @event_context_field_type.setter
    def event_context_field_type(self, event_context_field_type):
        event_context_field_type_ptr = None

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

        ret = native_bt.ctf_stream_class_set_event_context_type(self._ptr,
                                                               event_context_field_type_ptr)
        utils._handle_ret(ret, "cannot set stream class object's event context field type")

    def __call__(self, name=None):
        if name is not None:
            utils._check_str(name)

        stream_ptr = native_bt.ctf_stream_create(self._ptr, name)

        if stream_ptr is None:
            raise bt2.CreationError('cannot create stream object')

        return bt2.stream._create_from_ptr(stream_ptr)

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

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

        self_event_classes = list(self.values())
        other_event_classes = list(other.values())
        self_props = (
            self_event_classes,
            self.name,
            self.id,
            self.clock,
            self.packet_context_field_type,
            self.event_header_field_type,
            self.event_context_field_type,
        )
        other_props = (
            other_event_classes,
            other.name,
            other.id,
            other.clock,
            other.packet_context_field_type,
            other.event_header_field_type,
            other.event_context_field_type,
        )
        return self_props == other_props

    def _copy(self, ft_copy_func, ev_copy_func):
        cpy = StreamClass()
        cpy.id = self.id
        cpy.name = self.name
        cpy.packet_context_field_type = ft_copy_func(self.packet_context_field_type)
        cpy.event_header_field_type = ft_copy_func(self.event_header_field_type)
        cpy.event_context_field_type = ft_copy_func(self.event_context_field_type)

        for event_class in self.values():
            cpy.add_event_class(ev_copy_func(event_class))

        return cpy

    def __copy__(self):
        return self._copy(lambda ft: ft, 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.ctf_writer
	27	import bt2.stream
	28	import copy
	29	import bt2
	30
	31
	32	class _EventClassIterator(collections.abc.Iterator):
	33	def __init__(self, stream_class):
	34	self._stream_class = stream_class
	35	self._at = 0
	36
	37	def __next__(self):
	38	if self._at == len(self._stream_class):
	39	raise StopIteration
	40
	41	ec_ptr = native_bt.ctf_stream_class_get_event_class(self._stream_class._ptr,
	42	self._at)
	43	utils._handle_ptr(ec_ptr, "cannot get stream class object's event class object")
	44	name = native_bt.ctf_event_class_get_name(ec_ptr)
	45	native_bt.put(ec_ptr)
	46	utils._handle_ptr(name, "cannot get event class object's name")
	47	self._at += 1
	48	return name
	49
	50
	51	class StreamClass(object._Object, collections.abc.Mapping):
	52	def __init__(self, name=None, id=None, packet_context_field_type=None,
	53	event_header_field_type=None, event_context_field_type=None,
	54	event_classes=None):
	55	ptr = native_bt.ctf_stream_class_create(None)
	56
	57	if ptr is None:
	58	raise bt2.CreationError('cannot create stream class object')
	59
	60	super().__init__(ptr)
	61
	62	if name is not None:
	63	self.name = name
	64
65	if id is not None:
66	self.id = id
67
68	if packet_context_field_type is not None:
69	self.packet_context_field_type = packet_context_field_type
70
71	if event_header_field_type is not None:
72	self.event_header_field_type = event_header_field_type
73
74	if event_context_field_type is not None:
75	self.event_context_field_type = event_context_field_type
76
77	if event_classes is not None:
78	for event_class in event_classes:
79	self.add_event_class(event_class)
80
81	def __getitem__(self, key):
82	utils._check_str(key)
83	ec_ptr = native_bt.ctf_stream_class_get_event_class_by_name(self._ptr,
84	key)
85
86	if ec_ptr is None:
87	raise KeyError(key)
88
89	return bt2.EventClass._create_from_ptr(ec_ptr)
90
91	def __len__(self):
92	count = native_bt.ctf_stream_class_get_event_class_count(self._ptr)
93	utils._handle_ret(count, "cannot get stream class object's event class count")
94	return count
95
96	def __iter__(self):
97	return _EventClassIterator(self)
98
99	def event_class_with_id(self, id):
100	utils._check_int64(id)
101	ec_ptr = native_bt.ctf_stream_class_get_event_class_by_id(self._ptr, id)
102	utils._handle_ptr(ec_ptr, "cannot get stream class object's event class object")
103	return bt2.EventClass._create_from_ptr(ec_ptr)
104
105	def add_event_class(self, event_class):
106	utils._check_type(event_class, bt2.EventClass)
107	ret = native_bt.ctf_stream_class_add_event_class(self._ptr, event_class._ptr)
108	utils._handle_ret(ret, "cannot add event class object to stream class object's")
109
110	@property
111	def trace(self):
112	tc_ptr = native_bt.ctf_stream_class_get_trace(self._ptr)
113
114	if tc_ptr is not None:
115	return bt2.Trace._create_from_ptr(tc_ptr)
116
117	@property
118	def name(self):
119	return native_bt.ctf_stream_class_get_name(self._ptr)
120
121	@name.setter
122	def name(self, name):
123	utils._check_str(name)
124	ret = native_bt.ctf_stream_class_set_name(self._ptr, name)
125	utils._handle_ret(ret, "cannot set stream class object's name")
126
127	@property
128	def id(self):
129	id = native_bt.ctf_stream_class_get_id(self._ptr)
130
131	if utils._is_m1ull(id):
132	raise bt2.Error("cannot get stream class object's ID")
133
134	return id
135
136	@id.setter
137	def id(self, id):
138	utils._check_int64(id)
139	ret = native_bt.ctf_stream_class_set_id(self._ptr, id)
140	utils._handle_ret(ret, "cannot set stream class object's ID")
141
142	@property
143	def clock(self):
144	clock_ptr = native_bt.ctf_stream_class_get_clock(self._ptr)
145
146	if clock_ptr is None:
147	return
148
149	return bt2.ctf_writer.CtfWriterClock._create_from_ptr(clock_ptr)
150
151	@clock.setter
152	def clock(self, clock):
153	utils._check_type(clock, bt2.ctf_writer.CtfWriterClock)
154	ret = native_bt.ctf_stream_class_set_clock(self._ptr, clock._ptr)
155	utils._handle_ret(ret, "cannot set stream class object's CTF writer clock object")
156
157	@property
158	def packet_context_field_type(self):
159	ft_ptr = native_bt.ctf_stream_class_get_packet_context_type(self._ptr)
160
161	if ft_ptr is None:
162	return
163
164	return bt2.field_types._create_from_ptr(ft_ptr)
165
166	@packet_context_field_type.setter
167	def packet_context_field_type(self, packet_context_field_type):
168	packet_context_field_type_ptr = None
169
170	if packet_context_field_type is not None:
171	utils._check_type(packet_context_field_type, bt2.field_types._FieldType)
172	packet_context_field_type_ptr = packet_context_field_type._ptr
173
174	ret = native_bt.ctf_stream_class_set_packet_context_type(self._ptr,
175	packet_context_field_type_ptr)
176	utils._handle_ret(ret, "cannot set stream class object's packet context field type")
177
178	@property
179	def event_header_field_type(self):
180	ft_ptr = native_bt.ctf_stream_class_get_event_header_type(self._ptr)
181
182	if ft_ptr is None:
183	return
184
185	return bt2.field_types._create_from_ptr(ft_ptr)
186
187	@event_header_field_type.setter
188	def event_header_field_type(self, event_header_field_type):
189	event_header_field_type_ptr = None
190
191	if event_header_field_type is not None:
192	utils._check_type(event_header_field_type, bt2.field_types._FieldType)
193	event_header_field_type_ptr = event_header_field_type._ptr
194
195	ret = native_bt.ctf_stream_class_set_event_header_type(self._ptr,
196	event_header_field_type_ptr)
197	utils._handle_ret(ret, "cannot set stream class object's event header field type")
198
199	@property
200	def event_context_field_type(self):
201	ft_ptr = native_bt.ctf_stream_class_get_event_context_type(self._ptr)
202
203	if ft_ptr is None:
204	return
205
206	return bt2.field_types._create_from_ptr(ft_ptr)
207
208	@event_context_field_type.setter
209	def event_context_field_type(self, event_context_field_type):
210	event_context_field_type_ptr = None
211
212	if event_context_field_type is not None:
213	utils._check_type(event_context_field_type, bt2.field_types._FieldType)
214	event_context_field_type_ptr = event_context_field_type._ptr
215
216	ret = native_bt.ctf_stream_class_set_event_context_type(self._ptr,
217	event_context_field_type_ptr)
218	utils._handle_ret(ret, "cannot set stream class object's event context field type")
219
220	def __call__(self, name=None):
221	if name is not None:
222	utils._check_str(name)
223
224	stream_ptr = native_bt.ctf_stream_create(self._ptr, name)
225
226	if stream_ptr is None:
227	raise bt2.CreationError('cannot create stream object')
228
229	return bt2.stream._create_from_ptr(stream_ptr)
230
231	def __eq__(self, other):
232	if type(other) is not type(self):
233	return False
234
235	if self.addr == other.addr:
236	return True
237
238	self_event_classes = list(self.values())
239	other_event_classes = list(other.values())
240	self_props = (
241	self_event_classes,
242	self.name,
243	self.id,
244	self.clock,
245	self.packet_context_field_type,
246	self.event_header_field_type,
247	self.event_context_field_type,
248	)
249	other_props = (
250	other_event_classes,
251	other.name,
252	other.id,
253	other.clock,
254	other.packet_context_field_type,
255	other.event_header_field_type,
256	other.event_context_field_type,
257	)
258	return self_props == other_props
259
260	def _copy(self, ft_copy_func, ev_copy_func):
261	cpy = StreamClass()
262	cpy.id = self.id
263	cpy.name = self.name
264	cpy.packet_context_field_type = ft_copy_func(self.packet_context_field_type)
265	cpy.event_header_field_type = ft_copy_func(self.event_header_field_type)
266	cpy.event_context_field_type = ft_copy_func(self.event_context_field_type)
267
268	for event_class in self.values():
269	cpy.add_event_class(ev_copy_func(event_class))
270
271	return cpy
272
273	def __copy__(self):
274	return self._copy(lambda ft: ft, copy.copy)
275
276	def __deepcopy__(self, memo):
277	cpy = self._copy(copy.deepcopy, copy.deepcopy)
278	memo[id(self)] = cpy
279	return cpy