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