src/bindings/python/bt2/bt2/port.py

   1 # SPDX-License-Identifier: MIT
   2 #
   3 # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
   4
   5 from bt2 import native_bt, object
   6
   7
   8 def _bt2_connection():
   9     from bt2 import connection as bt2_connection
  10
  11     return bt2_connection
  12
  13
  14 def _create_from_const_ptr_and_get_ref(ptr, port_type):
  15     cls = _PORT_TYPE_TO_PYCLS.get(port_type, None)
  16
  17     if cls is None:
  18         raise TypeError('unknown port type: {}'.format(port_type))
  19
  20     return cls._create_from_ptr_and_get_ref(ptr)
  21
  22
  23 def _create_self_from_ptr_and_get_ref(ptr, port_type):
  24     cls = _PORT_TYPE_TO_USER_PYCLS.get(port_type, None)
  25
  26     if cls is None:
  27         raise TypeError('unknown port type: {}'.format(port_type))
  28
  29     return cls._create_from_ptr_and_get_ref(ptr)
  30
  31
  32 class _PortConst(object._SharedObject):
  33     @classmethod
  34     def _get_ref(cls, ptr):
  35         ptr = cls._as_port_ptr(ptr)
  36         return native_bt.port_get_ref(ptr)
  37
  38     @classmethod
  39     def _put_ref(cls, ptr):
  40         ptr = cls._as_port_ptr(ptr)
  41         return native_bt.port_put_ref(ptr)
  42
  43     @property
  44     def name(self):
  45         ptr = self._as_port_ptr(self._ptr)
  46         name = native_bt.port_get_name(ptr)
  47         assert name is not None
  48         return name
  49
  50     @property
  51     def connection(self):
  52         ptr = self._as_port_ptr(self._ptr)
  53         conn_ptr = native_bt.port_borrow_connection_const(ptr)
  54
  55         if conn_ptr is None:
  56             return
  57
  58         return _bt2_connection()._ConnectionConst._create_from_ptr_and_get_ref(conn_ptr)
  59
  60     @property
  61     def is_connected(self):
  62         return self.connection is not None
  63
  64
  65 class _InputPortConst(_PortConst):
  66     _as_port_ptr = staticmethod(native_bt.port_input_as_port_const)
  67
  68
  69 class _OutputPortConst(_PortConst):
  70     _as_port_ptr = staticmethod(native_bt.port_output_as_port_const)
  71
  72
  73 class _UserComponentPort(_PortConst):
  74     @classmethod
  75     def _as_port_ptr(cls, ptr):
  76         ptr = cls._as_self_port_ptr(ptr)
  77         return native_bt.self_component_port_as_port(ptr)
  78
  79     @property
  80     def connection(self):
  81         ptr = self._as_port_ptr(self._ptr)
  82         conn_ptr = native_bt.port_borrow_connection_const(ptr)
  83
  84         if conn_ptr is None:
  85             return
  86
  87         return _bt2_connection()._ConnectionConst._create_from_ptr_and_get_ref(conn_ptr)
  88
  89     @property
  90     def user_data(self):
  91         ptr = self._as_self_port_ptr(self._ptr)
  92         return native_bt.self_component_port_get_data(ptr)
  93
  94
  95 class _UserComponentInputPort(_UserComponentPort, _InputPortConst):
  96     _as_self_port_ptr = staticmethod(
  97         native_bt.self_component_port_input_as_self_component_port
  98     )
  99
 100
 101 class _UserComponentOutputPort(_UserComponentPort, _OutputPortConst):
 102     _as_self_port_ptr = staticmethod(
 103         native_bt.self_component_port_output_as_self_component_port
 104     )
 105
 106
 107 _PORT_TYPE_TO_PYCLS = {
 108     native_bt.PORT_TYPE_INPUT: _InputPortConst,
 109     native_bt.PORT_TYPE_OUTPUT: _OutputPortConst,
 110 }
 111
 112
 113 _PORT_TYPE_TO_USER_PYCLS = {
 114     native_bt.PORT_TYPE_INPUT: _UserComponentInputPort,
 115     native_bt.PORT_TYPE_OUTPUT: _UserComponentOutputPort,
 116 }