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     @staticmethod
  21     def _get_ref(ptr):
  22         native_bt.stream_get_ref(ptr)
  23
  24     @staticmethod
  25     def _put_ref(ptr):
  26         native_bt.stream_put_ref(ptr)
  27
  28     _borrow_class_ptr = staticmethod(native_bt.stream_borrow_class_const)
  29     _borrow_user_attributes_ptr = staticmethod(
  30         native_bt.stream_borrow_user_attributes_const
  31     )
  32     _create_value_from_ptr_and_get_ref = staticmethod(
  33         bt2_value._create_from_const_ptr_and_get_ref
  34     )
  35     _borrow_trace_ptr = staticmethod(native_bt.stream_borrow_trace_const)
  36     _stream_class_pycls = bt2_stream_class._StreamClassConst
  37     _trace_pycls = property(lambda _: _bt2_trace()._TraceConst)
  38
  39     @property
  40     def cls(self):
  41         stream_class_ptr = self._borrow_class_ptr(self._ptr)
  42         assert stream_class_ptr is not None
  43         return self._stream_class_pycls._create_from_ptr_and_get_ref(stream_class_ptr)
  44
  45     @property
  46     def name(self):
  47         return native_bt.stream_get_name(self._ptr)
  48
  49     @property
  50     def user_attributes(self):
  51         ptr = self._borrow_user_attributes_ptr(self._ptr)
  52         assert ptr is not None
  53         return self._create_value_from_ptr_and_get_ref(ptr)
  54
  55     @property
  56     def id(self):
  57         id = native_bt.stream_get_id(self._ptr)
  58         return id if id >= 0 else None
  59
  60     @property
  61     def trace(self):
  62         trace_ptr = self._borrow_trace_ptr(self._ptr)
  63         assert trace_ptr is not None
  64         return self._trace_pycls._create_from_ptr_and_get_ref(trace_ptr)
  65
  66
  67 class _Stream(_StreamConst):
  68     _borrow_class_ptr = staticmethod(native_bt.stream_borrow_class)
  69     _borrow_user_attributes_ptr = staticmethod(native_bt.stream_borrow_user_attributes)
  70     _create_value_from_ptr_and_get_ref = staticmethod(
  71         bt2_value._create_from_ptr_and_get_ref
  72     )
  73     _borrow_trace_ptr = staticmethod(native_bt.stream_borrow_trace)
  74     _stream_class_pycls = bt2_stream_class._StreamClass
  75     _trace_pycls = property(lambda _: _bt2_trace()._Trace)
  76
  77     def create_packet(self):
  78         if not self.cls.supports_packets:
  79             raise ValueError(
  80                 "cannot create packet: stream class does not support packets"
  81             )
  82
  83         packet_ptr = native_bt.packet_create(self._ptr)
  84
  85         if packet_ptr is None:
  86             raise bt2._MemoryError("cannot create packet object")
  87
  88         return bt2_packet._Packet._create_from_ptr(packet_ptr)
  89
  90     def _user_attributes(self, user_attributes):
  91         value = bt2_value.create_value(user_attributes)
  92         utils._check_type(value, bt2_value.MapValue)
  93         native_bt.stream_set_user_attributes(self._ptr, value._ptr)
  94
  95     _user_attributes = property(
  96         fget=_StreamConst.user_attributes.fget, fset=_user_attributes
  97     )
  98
  99     def _name(self, name):
 100         utils._check_str(name)
 101         native_bt.stream_set_name(self._ptr, name)
 102
 103     _name = property(fget=_StreamConst.name.fget, fset=_name)