[babeltrace.git] / bindings / python / bt2 / bt2 / stream_class.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_class
import bt2.event_class
import collections.abc
import bt2.stream
import bt2


class _StreamClass(object._SharedObject, collections.abc.Mapping):
    _get_ref = staticmethod(native_bt.stream_class_get_ref)
    _put_ref = staticmethod(native_bt.stream_class_put_ref)

    def __getitem__(self, key):
        utils._check_int64(key)
        ec_ptr = native_bt.stream_class_borrow_event_class_by_id(self._ptr, key)

        if ec_ptr is None:
            raise KeyError(key)

        return bt2.event_class._EventClass._create_from_ptr_and_get_ref(ec_ptr)

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

    def __iter__(self):
        for idx in range(len(self)):
            ec_ptr = native_bt.stream_class_borrow_event_class_by_index_const(self._ptr, idx)
            assert ec_ptr is not None

            id = native_bt.event_class_get_id(ec_ptr)
            assert id >= 0

            yield id

    def create_event_class(self, id=None, name=None, log_level=None, emf_uri=None,
                           specific_context_field_class=None,
                           payload_field_class=None):
        if self.assigns_automatic_event_class_id:
            if id is not None:
                raise ValueError('id provided, but stream class assigns automatic event class ids')

            ec_ptr = native_bt.event_class_create(self._ptr)
        else:
            if id is None:
                raise ValueError('id not provided, but stream class does not assign automatic event class ids')

            utils._check_uint64(id)
            ec_ptr = native_bt.event_class_create_with_id(self._ptr, id)

        event_class = bt2.event_class._EventClass._create_from_ptr(ec_ptr)

        if name is not None:
            event_class._name = name

        if log_level is not None:
            event_class._log_level = log_level

        if emf_uri is not None:
            event_class._emf_uri = emf_uri

        if specific_context_field_class is not None:
            event_class._specific_context_field_class = specific_context_field_class

        if payload_field_class is not None:
            event_class._payload_field_class = payload_field_class

        return event_class

    @property
    def trace_class(self):
        tc_ptr = native_bt.stream_class_borrow_trace_class_const(self._ptr)

        if tc_ptr is not None:
            return bt2._TraceClass._create_from_ptr_and_get_ref(tc_ptr)

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

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

    _name = property(fset=_name)

    @property
    def assigns_automatic_event_class_id(self):
        return native_bt.stream_class_assigns_automatic_event_class_id(self._ptr)

    def _assigns_automatic_event_class_id(self, auto_id):
        utils._check_bool(auto_id)
        return native_bt.stream_class_set_assigns_automatic_event_class_id(self._ptr, auto_id)

    _assigns_automatic_event_class_id = property(fset=_assigns_automatic_event_class_id)

    @property
    def assigns_automatic_stream_id(self):
        return native_bt.stream_class_assigns_automatic_stream_id(self._ptr)

    def _assigns_automatic_stream_id(self, auto_id):
        utils._check_bool(auto_id)
        return native_bt.stream_class_set_assigns_automatic_stream_id(self._ptr, auto_id)

    _assigns_automatic_stream_id = property(fset=_assigns_automatic_stream_id)

    @property
    def packets_have_beginning_default_clock_snapshot(self):
        return native_bt.stream_class_packets_have_beginning_default_clock_snapshot(self._ptr)

    def _packets_have_beginning_default_clock_snapshot(self, value):
        utils._check_bool(value)
        native_bt.stream_class_set_packets_have_beginning_default_clock_snapshot(self._ptr, value)

    _packets_have_beginning_default_clock_snapshot = property(fset=_packets_have_beginning_default_clock_snapshot)

    @property
    def packets_have_end_default_clock_snapshot(self):
        return native_bt.stream_class_packets_have_end_default_clock_snapshot(self._ptr)

    def _packets_have_end_default_clock_snapshot(self, value):
        utils._check_bool(value)
        native_bt.stream_class_set_packets_have_end_default_clock_snapshot(self._ptr, value)

    _packets_have_end_default_clock_snapshot = property(fset=_packets_have_end_default_clock_snapshot)

    @property
    def supports_discarded_events(self):
        return native_bt.stream_class_supports_discarded_events(self._ptr)

    def _set_supports_discarded_events(self, supports, with_cs=False):
        utils._check_bool(supports)
        utils._check_bool(with_cs)

        if not supports and with_cs:
            raise ValueError('cannot not support discarded events, but have default clock snapshots')

        native_bt.stream_class_set_supports_discarded_events(self._ptr, supports, with_cs)

    @property
    def discarded_events_have_default_clock_snapshots(self):
        return native_bt.stream_class_discarded_events_have_default_clock_snapshots(self._ptr)

    @property
    def supports_discarded_packets(self):
        return native_bt.stream_class_supports_discarded_packets(self._ptr)

    def _set_supports_discarded_packets(self, supports, with_cs):
        utils._check_bool(supports)
        utils._check_bool(with_cs)

        if not supports and with_cs:
            raise ValueError('cannot not support discarded packets, but have default clock snapshots')

        native_bt.stream_class_set_supports_discarded_packets(self._ptr, supports, with_cs)

    @property
    def discarded_packets_have_default_clock_snapshots(self):
        return native_bt.stream_class_discarded_packets_have_default_clock_snapshots(self._ptr)

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

        if id < 0:
            return

        return id

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

    @property
    def packet_context_field_class(self):
        fc_ptr = native_bt.stream_class_borrow_packet_context_field_class_const(self._ptr)

        if fc_ptr is None:
            return

        return bt2.field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)

    def _packet_context_field_class(self, packet_context_field_class):
        if packet_context_field_class is not None:
            utils._check_type(packet_context_field_class,
                              bt2.field_class._StructureFieldClass)
            ret = native_bt.stream_class_set_packet_context_field_class(self._ptr,
                                                                        packet_context_field_class._ptr)
            utils._handle_ret(ret, "cannot set stream class' packet context field class")

    _packet_context_field_class = property(fset=_packet_context_field_class)

    @property
    def event_common_context_field_class(self):
        fc_ptr = native_bt.stream_class_borrow_event_common_context_field_class_const(self._ptr)

        if fc_ptr is None:
            return

        return bt2.field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)

    def _event_common_context_field_class(self, event_common_context_field_class):
        if event_common_context_field_class is not None:
            utils._check_type(event_common_context_field_class,
                              bt2.field_class._StructureFieldClass)

            set_context_fn = native_bt.stream_class_set_event_common_context_field_class
            ret = set_context_fn(self._ptr, event_common_context_field_class._ptr)

            utils._handle_ret(ret, "cannot set stream class' event context field type")

    _event_common_context_field_class = property(fset=_event_common_context_field_class)

    @property
    def default_clock_class(self):
        cc_ptr = native_bt.stream_class_borrow_default_clock_class(self._ptr)
        if cc_ptr is None:
            return

        return bt2.clock_class._ClockClass._create_from_ptr_and_get_ref(cc_ptr)

    def _default_clock_class(self, clock_class):
        utils._check_type(clock_class, bt2.clock_class._ClockClass)
        native_bt.stream_class_set_default_clock_class(
            self._ptr, clock_class._ptr)

    _default_clock_class = property(fset=_default_clock_class)
Commit	Line	Data
	1	# The MIT License (MIT)
	2	#
	3	# Copyright (c) 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, utils
	24	import bt2.field_class
	25	import bt2.event_class
	26	import collections.abc
	27	import bt2.stream
	28	import bt2
	29
	30
	31	class _StreamClass(object._SharedObject, collections.abc.Mapping):
	32	_get_ref = staticmethod(native_bt.stream_class_get_ref)
	33	_put_ref = staticmethod(native_bt.stream_class_put_ref)
	34
	35	def __getitem__(self, key):
	36	utils._check_int64(key)
	37	ec_ptr = native_bt.stream_class_borrow_event_class_by_id(self._ptr, key)
	38
	39	if ec_ptr is None:
	40	raise KeyError(key)
	41
	42	return bt2.event_class._EventClass._create_from_ptr_and_get_ref(ec_ptr)
	43
	44	def __len__(self):
	45	count = native_bt.stream_class_get_event_class_count(self._ptr)
	46	assert count >= 0
	47	return count
	48
	49	def __iter__(self):
	50	for idx in range(len(self)):
	51	ec_ptr = native_bt.stream_class_borrow_event_class_by_index_const(self._ptr, idx)
	52	assert ec_ptr is not None
	53
	54	id = native_bt.event_class_get_id(ec_ptr)
	55	assert id >= 0
	56
	57	yield id
	58
	59	def create_event_class(self, id=None, name=None, log_level=None, emf_uri=None,
	60	specific_context_field_class=None,
	61	payload_field_class=None):
	62	if self.assigns_automatic_event_class_id:
	63	if id is not None:
	64	raise ValueError('id provided, but stream class assigns automatic event class ids')
	65
	66	ec_ptr = native_bt.event_class_create(self._ptr)
	67	else:
	68	if id is None:
	69	raise ValueError('id not provided, but stream class does not assign automatic event class ids')
	70
	71	utils._check_uint64(id)
	72	ec_ptr = native_bt.event_class_create_with_id(self._ptr, id)
	73
	74	event_class = bt2.event_class._EventClass._create_from_ptr(ec_ptr)
	75
	76	if name is not None:
	77	event_class._name = name
	78
	79	if log_level is not None:
	80	event_class._log_level = log_level
	81
	82	if emf_uri is not None:
	83	event_class._emf_uri = emf_uri
	84
	85	if specific_context_field_class is not None:
	86	event_class._specific_context_field_class = specific_context_field_class
	87
	88	if payload_field_class is not None:
	89	event_class._payload_field_class = payload_field_class
	90
	91	return event_class
	92
	93	@property
	94	def trace_class(self):
	95	tc_ptr = native_bt.stream_class_borrow_trace_class_const(self._ptr)
	96
	97	if tc_ptr is not None:
	98	return bt2._TraceClass._create_from_ptr_and_get_ref(tc_ptr)
	99
	100	@property
	101	def name(self):
	102	return native_bt.stream_class_get_name(self._ptr)
	103
	104	def _name(self, name):
	105	utils._check_str(name)
	106	ret = native_bt.stream_class_set_name(self._ptr, name)
	107	utils._handle_ret(ret, "cannot set stream class object's name")
	108
	109	_name = property(fset=_name)
	110
	111	@property
	112	def assigns_automatic_event_class_id(self):
	113	return native_bt.stream_class_assigns_automatic_event_class_id(self._ptr)
	114
	115	def _assigns_automatic_event_class_id(self, auto_id):
	116	utils._check_bool(auto_id)
	117	return native_bt.stream_class_set_assigns_automatic_event_class_id(self._ptr, auto_id)
	118
	119	_assigns_automatic_event_class_id = property(fset=_assigns_automatic_event_class_id)
	120
	121	@property
	122	def assigns_automatic_stream_id(self):
	123	return native_bt.stream_class_assigns_automatic_stream_id(self._ptr)
	124
	125	def _assigns_automatic_stream_id(self, auto_id):
	126	utils._check_bool(auto_id)
	127	return native_bt.stream_class_set_assigns_automatic_stream_id(self._ptr, auto_id)
	128
	129	_assigns_automatic_stream_id = property(fset=_assigns_automatic_stream_id)
	130
	131	@property
	132	def packets_have_beginning_default_clock_snapshot(self):
	133	return native_bt.stream_class_packets_have_beginning_default_clock_snapshot(self._ptr)
	134
	135	def _packets_have_beginning_default_clock_snapshot(self, value):
	136	utils._check_bool(value)
	137	native_bt.stream_class_set_packets_have_beginning_default_clock_snapshot(self._ptr, value)
	138
	139	_packets_have_beginning_default_clock_snapshot = property(fset=_packets_have_beginning_default_clock_snapshot)
	140
	141	@property
	142	def packets_have_end_default_clock_snapshot(self):
	143	return native_bt.stream_class_packets_have_end_default_clock_snapshot(self._ptr)
	144
	145	def _packets_have_end_default_clock_snapshot(self, value):
	146	utils._check_bool(value)
	147	native_bt.stream_class_set_packets_have_end_default_clock_snapshot(self._ptr, value)
	148
	149	_packets_have_end_default_clock_snapshot = property(fset=_packets_have_end_default_clock_snapshot)
	150
	151	@property
	152	def supports_discarded_events(self):
	153	return native_bt.stream_class_supports_discarded_events(self._ptr)
	154
	155	def _set_supports_discarded_events(self, supports, with_cs=False):
	156	utils._check_bool(supports)
	157	utils._check_bool(with_cs)
	158
	159	if not supports and with_cs:
	160	raise ValueError('cannot not support discarded events, but have default clock snapshots')
	161
	162	native_bt.stream_class_set_supports_discarded_events(self._ptr, supports, with_cs)
	163
	164	@property
	165	def discarded_events_have_default_clock_snapshots(self):
	166	return native_bt.stream_class_discarded_events_have_default_clock_snapshots(self._ptr)
	167
	168	@property
	169	def supports_discarded_packets(self):
	170	return native_bt.stream_class_supports_discarded_packets(self._ptr)
	171
	172	def _set_supports_discarded_packets(self, supports, with_cs):
	173	utils._check_bool(supports)
	174	utils._check_bool(with_cs)
	175
	176	if not supports and with_cs:
	177	raise ValueError('cannot not support discarded packets, but have default clock snapshots')
	178
	179	native_bt.stream_class_set_supports_discarded_packets(self._ptr, supports, with_cs)
	180
	181	@property
	182	def discarded_packets_have_default_clock_snapshots(self):
	183	return native_bt.stream_class_discarded_packets_have_default_clock_snapshots(self._ptr)
	184
	185	@property
	186	def id(self):
	187	id = native_bt.stream_class_get_id(self._ptr)
	188
	189	if id < 0:
	190	return
	191
	192	return id
	193
	194	@id.setter
	195	def id(self, id):
	196	utils._check_int64(id)
	197	ret = native_bt.stream_class_set_id(self._ptr, id)
	198	utils._handle_ret(ret, "cannot set stream class object's ID")
	199
	200	@property
	201	def packet_context_field_class(self):
	202	fc_ptr = native_bt.stream_class_borrow_packet_context_field_class_const(self._ptr)
	203
	204	if fc_ptr is None:
	205	return
	206
	207	return bt2.field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
	208
	209	def _packet_context_field_class(self, packet_context_field_class):
	210	if packet_context_field_class is not None:
	211	utils._check_type(packet_context_field_class,
	212	bt2.field_class._StructureFieldClass)
	213	ret = native_bt.stream_class_set_packet_context_field_class(self._ptr,
	214	packet_context_field_class._ptr)
	215	utils._handle_ret(ret, "cannot set stream class' packet context field class")
	216
	217	_packet_context_field_class = property(fset=_packet_context_field_class)
	218
	219	@property
	220	def event_common_context_field_class(self):
	221	fc_ptr = native_bt.stream_class_borrow_event_common_context_field_class_const(self._ptr)
	222
	223	if fc_ptr is None:
	224	return
	225
	226	return bt2.field_class._create_field_class_from_ptr_and_get_ref(fc_ptr)
	227
	228	def _event_common_context_field_class(self, event_common_context_field_class):
	229	if event_common_context_field_class is not None:
	230	utils._check_type(event_common_context_field_class,
	231	bt2.field_class._StructureFieldClass)
	232
	233	set_context_fn = native_bt.stream_class_set_event_common_context_field_class
	234	ret = set_context_fn(self._ptr, event_common_context_field_class._ptr)
	235
	236	utils._handle_ret(ret, "cannot set stream class' event context field type")
	237
	238	_event_common_context_field_class = property(fset=_event_common_context_field_class)
	239
	240	@property
	241	def default_clock_class(self):
	242	cc_ptr = native_bt.stream_class_borrow_default_clock_class(self._ptr)
	243	if cc_ptr is None:
	244	return
	245
	246	return bt2.clock_class._ClockClass._create_from_ptr_and_get_ref(cc_ptr)
	247
	248	def _default_clock_class(self, clock_class):
	249	utils._check_type(clock_class, bt2.clock_class._ClockClass)
	250	native_bt.stream_class_set_default_clock_class(
	251	self._ptr, clock_class._ptr)
	252
	253	_default_clock_class = property(fset=_default_clock_class)