src/bindings/python/bt2/bt2/stream.py

   1 # SPDX-License-Identifier: MIT
   2 #
   3 # Copyright (c) 2016-2017 Philippe Proulx <pproulx@efficios.com>
   4
   5 from bt2 import native_bt, utils
   6 from bt2 import object as bt2_object
   7 from bt2 import packet as bt2_packet
   8 from bt2 import stream_class as bt2_stream_class
   9 from bt2 import value as bt2_value
  10 import bt2
  11
  12
  13 def _bt2_trace():
  14     from bt2 import trace as bt2_trace
  15
  16     return bt2_trace
  17
  18
  19 class _StreamConst(bt2_object._SharedObject):
  20     _get_ref = staticmethod(native_bt.stream_get_ref)
  21     _put_ref = staticmethod(native_bt.stream_put_ref)
  22     _borrow_class_ptr = staticmethod(native_bt.stream_borrow_class_const)
  23     _borrow_user_attributes_ptr = staticmethod(
  24         native_bt.stream_borrow_user_attributes_const
  25     )
  26     _create_value_from_ptr_and_get_ref = staticmethod(
  27         bt2_value._create_from_const_ptr_and_get_ref
  28     )
  29     _borrow_trace_ptr = staticmethod(native_bt.stream_borrow_trace_const)
  30     _stream_class_pycls = bt2_stream_class._StreamClassConst
  31     _trace_pycls = property(lambda _: _bt2_trace()._TraceConst)
  32
  33     @property
  34     def cls(self):
  35         stream_class_ptr = self._borrow_class_ptr(self._ptr)
  36         assert stream_class_ptr is not None
  37         return self._stream_class_pycls._create_from_ptr_and_get_ref(stream_class_ptr)
  38
  39     @property
  40     def name(self):
  41         return native_bt.stream_get_name(self._ptr)
  42
  43     @property
  44     def user_attributes(self):
  45         ptr = self._borrow_user_attributes_ptr(self._ptr)
  46         assert ptr is not None
  47         return self._create_value_from_ptr_and_get_ref(ptr)
  48
  49     @property
  50     def id(self):
  51         id = native_bt.stream_get_id(self._ptr)
  52         return id if id >= 0 else None
  53
  54     @property
  55     def trace(self):
  56         trace_ptr = self._borrow_trace_ptr(self._ptr)
  57         assert trace_ptr is not None
  58         return self._trace_pycls._create_from_ptr_and_get_ref(trace_ptr)
  59
  60
  61 class _Stream(_StreamConst):
  62     _borrow_class_ptr = staticmethod(native_bt.stream_borrow_class)
  63     _borrow_user_attributes_ptr = staticmethod(native_bt.stream_borrow_user_attributes)
  64     _create_value_from_ptr_and_get_ref = staticmethod(
  65         bt2_value._create_from_ptr_and_get_ref
  66     )
  67     _borrow_trace_ptr = staticmethod(native_bt.stream_borrow_trace)
  68     _stream_class_pycls = bt2_stream_class._StreamClass
  69     _trace_pycls = property(lambda _: _bt2_trace()._Trace)
  70
  71     def create_packet(self):
  72         if not self.cls.supports_packets:
  73             raise ValueError(
  74                 "cannot create packet: stream class does not support packets"
  75             )
  76
  77         packet_ptr = native_bt.packet_create(self._ptr)
  78
  79         if packet_ptr is None:
  80             raise bt2._MemoryError("cannot create packet object")
  81
  82         return bt2_packet._Packet._create_from_ptr(packet_ptr)
  83
  84     def _user_attributes(self, user_attributes):
  85         value = bt2_value.create_value(user_attributes)
  86         utils._check_type(value, bt2_value.MapValue)
  87         native_bt.stream_set_user_attributes(self._ptr, value._ptr)
  88
  89     _user_attributes = property(
  90         fget=_StreamConst.user_attributes.fget, fset=_user_attributes
  91     )
  92
  93     def _name(self, name):
  94         utils._check_str(name)
  95         native_bt.stream_set_name(self._ptr, name)
  96
  97     _name = property(fget=_StreamConst.name.fget, fset=_name)