time = property(fset=_time)
-class _CtfWriterStream(stream._StreamBase):
+class _CtfWriterStream:
@property
def discarded_events_count(self):
ret, count = native_bt.stream_get_discarded_events_count(self._ptr)
utils._handle_ret(ret, "cannot set field class object's byte order")
-class IntegerFieldClass(_FieldClass, _AlignmentProp, _ByteOrderProp):
- _NAME = 'Integer'
+class _IntegerFieldClass(_FieldClass):
def __init__(self, size, alignment=None, byte_order=None, is_signed=None,
base=None, encoding=None, mapped_clock_class=None):
utils._handle_ret(ret, "cannot set integer field class object's mapped clock class")
+class _SignedIntegerFieldClass(_IntegerFieldClass):
+ pass
+
+class SignedIntegerFieldClass(_SignedIntegerFieldClass):
+ _NAME = 'SignedInteger'
+
class FloatingPointNumberFieldClass(_FieldClass, _AlignmentProp, _ByteOrderProp):
_NAME = 'Floating point number'
return mapping
-class EnumerationFieldClass(IntegerFieldClass, collections.abc.Sequence):
+class EnumerationFieldClass(_IntegerFieldClass, collections.abc.Sequence):
_NAME = 'Enumeration'
def __init__(self, int_field_class=None, size=None, alignment=None,
def append_field(self, name, field_class):
utils._check_str(name)
utils._check_type(field_class, _FieldClass)
- ret = self._add_field(field_class._ptr, name)
+ ret = self._add_field(name, field_class._ptr)
utils._handle_ret(ret, "cannot add field to {} field class object".format(self._NAME.lower()))
def __iadd__(self, fields):
def _get_field_by_name(self, key):
return native_bt.field_class_structure_get_field_class_by_name(self._ptr, key)
- def _add_field(self, ptr, name):
- return native_bt.field_class_structure_add_field(self._ptr, ptr,
- name)
+ def _add_field(self, name, ptr):
+ return native_bt.field_class_structure_append_member(self._ptr, name, ptr)
def _at(self, index):
if index < 0 or index >= len(self):
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
-from bt2 import native_bt, object, utils
+from bt2 import native_bt, utils
import bt2.packet
import bt2.event
-import abc
import bt2
-def _create_from_ptr(stream_ptr):
- import bt2.ctf_writer
-
- if native_bt.stream_is_writer(stream_ptr):
- cls = bt2.ctf_writer._CtfWriterStream
- else:
- cls = _Stream
-
- return cls._create_from_ptr(stream_ptr)
-
+class _Stream(bt2.object._SharedObject):
+ _get_ref = staticmethod(native_bt.stream_get_ref)
+ _put_ref = staticmethod(native_bt.stream_put_ref)
-class _StreamBase(object._SharedObject):
@property
def stream_class(self):
- stream_class_ptr = native_bt.stream_get_class(self._ptr)
- assert(stream_class_ptr)
- return bt2.StreamClass._create_from_ptr(stream_class_ptr)
+ stream_class_ptr = native_bt.stream_borrow_class(self._ptr)
+ assert stream_class_ptr is not None
+ return bt2.stream_class.StreamClass._create_from_ptr_and_get_ref(stream_class_ptr)
@property
def name(self):
return native_bt.stream_get_name(self._ptr)
+ def _name(self, name):
+ utils._check_str(name)
+ native_bt.stream_set_name(self._ptr, name)
+
+ _name = property(fset=_name)
+
@property
def id(self):
id = native_bt.stream_get_id(self._ptr)
return id if id >= 0 else None
- def __eq__(self, other):
- if self.addr == other.addr:
- return True
-
- return (self.name, self.id) == (other.name, other.id)
-
-
-class _Stream(_StreamBase):
- _get_ref = staticmethod(native_bt.stream_get_ref)
- _put_ref = staticmethod(native_bt.stream_put_ref)
-
def create_packet(self):
packet_ptr = native_bt.packet_create(self._ptr)
raise bt2.CreationError('cannot create packet object')
return bt2.packet._Packet._create_from_ptr(packet_ptr)
-
- def __eq__(self, other):
- if type(other) is not type(self):
- return False
-
- return _StreamBase.__eq__(self, other)
-
- def _copy(self):
- return self.stream_class(self.name, self.id)
-
- def __copy__(self):
- return self._copy()
-
- def __deepcopy__(self, memo):
- cpy = self._copy()
- memo[id(self)] = cpy
- return cpy
yield id
- def create_event_class(self, id=None):
+ 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 bt2.CreationError('id provided, but stream class assigns automatic event class ids')
utils._check_uint64(id)
ec_ptr = native_bt.event_class_create_with_id(self._ptr, id)
- return bt2.event_class.EventClass._create_from_ptr(ec_ptr)
+ 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):
raise bt2.CreationError(
'cannot create {} field class'.format(type_name))
+ def _create_integer_field_class(self, create_func, py_cls, type_name, range, display_base):
+ field_class_ptr = create_func(self._ptr)
+ self._check_create_status(field_class_ptr, type_name)
+
+ field_class = py_cls._create_from_ptr(field_class_ptr)
+
+ if range is not None:
+ field_class._range = range
+
+ if display_base is not None:
+ field_class._display_base = display_base
+
+ return field_class
+
+ def create_signed_integer_field_class(self, range=None, display_base=None):
+ return self._create_integer_field_class(native_bt.field_class_signed_integer_create,
+ bt2.field_class.SignedIntegerFieldClass,
+ 'signed integer', range, display_base)
+
def create_structure_field_class(self):
field_class_ptr = native_bt.field_class_structure_create(self._ptr)
self._check_create_status(field_class_ptr, 'structure')
-from collections import OrderedDict
-from bt2 import value
import unittest
-import copy
-import bt2
+from utils import run_in_component_init
-@unittest.skip("this is broken")
class StreamTestCase(unittest.TestCase):
def setUp(self):
- self._stream = self._create_stream(stream_id=23)
+ def f(comp_self):
+ return comp_self._create_trace_class()
- def tearDown(self):
- del self._stream
+ self._tc = run_in_component_init(f)
+ self._sc = self._tc.create_stream_class(assigns_automatic_stream_id=True)
+ self._tr = self._tc()
- def _create_stream(self, name='my_stream', stream_id=None):
- # event header
- eh = bt2.StructureFieldClass()
- eh += OrderedDict((
- ('id', bt2.IntegerFieldClass(8)),
- ('ts', bt2.IntegerFieldClass(32)),
- ))
+ def test_create_default(self):
+ stream = self._tr.create_stream(self._sc)
+ self.assertIsNone(stream.name)
- # stream event context
- sec = bt2.StructureFieldClass()
- sec += OrderedDict((
- ('cpu_id', bt2.IntegerFieldClass(8)),
- ('stuff', bt2.FloatingPointNumberFieldClass()),
- ))
+ def test_name(self):
+ stream = self._tr.create_stream(self._sc, name='équidistant')
+ self.assertEqual(stream.name, 'équidistant')
- # packet context
- pc = bt2.StructureFieldClass()
- pc += OrderedDict((
- ('something', bt2.IntegerFieldClass(8)),
- ('something_else', bt2.FloatingPointNumberFieldClass()),
- ))
+ def test_invalid_name(self):
+ with self.assertRaises(TypeError):
+ self._tr.create_stream(self._sc, name=22)
- # stream class
- sc = bt2.StreamClass()
- sc.packet_context_field_class = pc
- sc.event_header_field_class = eh
- sc.event_context_field_class = sec
+ def test_stream_class(self):
+ stream = self._tr.create_stream(self._sc)
+ self.assertEqual(stream.stream_class, self._sc)
- # event context
- ec = bt2.StructureFieldClass()
- ec += OrderedDict((
- ('ant', bt2.IntegerFieldClass(16, is_signed=True)),
- ('msg', bt2.StringFieldClass()),
- ))
+ def test_invalid_id(self):
+ sc = self._tc.create_stream_class(assigns_automatic_stream_id=False)
- # event payload
- ep = bt2.StructureFieldClass()
- ep += OrderedDict((
- ('giraffe', bt2.IntegerFieldClass(32)),
- ('gnu', bt2.IntegerFieldClass(8)),
- ('mosquito', bt2.IntegerFieldClass(8)),
- ))
-
- # event class
- event_class = bt2.EventClass('ec')
- event_class.context_field_class = ec
- event_class.payload_field_class = ep
- sc.add_event_class(event_class)
-
- # packet header
- ph = bt2.StructureFieldClass()
- ph += OrderedDict((
- ('magic', bt2.IntegerFieldClass(32)),
- ('stream_id', bt2.IntegerFieldClass(16)),
- ))
-
- # trace c;ass
- tc = bt2.Trace()
- tc.packet_header_field_class = ph
- tc.add_stream_class(sc)
-
- # stream
- return sc(name=name, id=stream_id)
-
- def test_attr_stream_class(self):
- self.assertIsNotNone(self._stream.stream_class)
-
- def test_attr_name(self):
- self.assertEqual(self._stream.name, 'my_stream')
-
- def test_eq(self):
- stream1 = self._create_stream(stream_id=17)
- stream2 = self._create_stream(stream_id=17)
- self.assertEqual(stream1, stream2)
-
- def test_ne_name(self):
- stream1 = self._create_stream(stream_id=17)
- stream2 = self._create_stream('lel', 17)
- self.assertNotEqual(stream1, stream2)
-
- def test_ne_id(self):
- stream1 = self._create_stream(stream_id=17)
- stream2 = self._create_stream(stream_id=23)
- self.assertNotEqual(stream1, stream2)
-
- def test_eq_invalid(self):
- self.assertFalse(self._stream == 23)
-
- def _test_copy(self, func):
- stream = self._create_stream()
- cpy = func(stream)
- self.assertIsNot(stream, cpy)
- self.assertNotEqual(stream.addr, cpy.addr)
- self.assertEqual(stream, cpy)
-
- def test_copy(self):
- self._test_copy(copy.copy)
-
- def test_deepcopy(self):
- self._test_copy(copy.deepcopy)
+ with self.assertRaises(TypeError):
+ self._tr.create_stream(sc, id='string')
projects / babeltrace.git / commitdiff
