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