bindings/python/bt2/port.py

   1 # The MIT License (MIT)
   2 #
   3 # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
   4 #
   5 # Permission is hereby granted, free of charge, to any person obtaining a copy
   6 # of this software and associated documentation files (the "Software"), to deal
   7 # in the Software without restriction, including without limitation the rights
   8 # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   9 # copies of the Software, and to permit persons to whom the Software is
  10 # furnished to do so, subject to the following conditions:
  11 #
  12 # The above copyright notice and this permission notice shall be included in
  13 # all copies or substantial portions of the Software.
  14 #
  15 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  16 # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  17 # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  18 # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  19 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  20 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  21 # THE SOFTWARE.
  22
  23 from bt2 import native_bt, object, utils
  24 import collections.abc
  25 import bt2.component
  26 import bt2.connection
  27 import copy
  28 import bt2
  29
  30
  31 def _create_from_ptr(ptr):
  32     port_type = native_bt.port_get_type(ptr)
  33
  34     if port_type == native_bt.PORT_TYPE_INPUT:
  35         cls = _InputPort
  36     elif port_type == native_bt.PORT_TYPE_OUTPUT:
  37         cls = _OutputPort
  38     else:
  39         raise bt2.Error('unknown port type: {}'.format(port_type))
  40
  41     return cls._create_from_ptr(ptr)
  42
  43
  44 def _create_private_from_ptr(ptr):
  45     pub_ptr = native_bt.port_from_private_port(ptr)
  46     utils._handle_ptr(pub_ptr, 'cannot get port object from private port object')
  47     port_type = native_bt.port_get_type(pub_ptr)
  48     assert(port_type == native_bt.PORT_TYPE_INPUT or port_type == native_bt.PORT_TYPE_OUTPUT)
  49
  50     if port_type == native_bt.PORT_TYPE_INPUT:
  51         cls = _PrivateInputPort
  52     elif port_type == native_bt.PORT_TYPE_OUTPUT:
  53         cls = _PrivateOutputPort
  54
  55     obj = cls._create_from_ptr(ptr)
  56     obj._pub_ptr = pub_ptr
  57     return obj
  58
  59
  60 class _Port(object._Object):
  61     @staticmethod
  62     def _name(ptr):
  63         name = native_bt.port_get_name(ptr)
  64         assert(name is not None)
  65         return name
  66
  67     @staticmethod
  68     def _disconnect(ptr):
  69         status = native_bt.port_disconnect(ptr)
  70
  71         if status < 0:
  72             raise bt2.Error('cannot disconnect port')
  73
  74     @property
  75     def name(self):
  76         return self._name(self._ptr)
  77
  78     @property
  79     def component(self):
  80         comp_ptr = native_bt.port_get_component(self._ptr)
  81
  82         if comp_ptr is None:
  83             return
  84
  85         return bt2.component._create_generic_component_from_ptr(comp_ptr)
  86
  87     @property
  88     def connection(self):
  89         conn_ptr = native_bt.port_get_connection(self._ptr)
  90
  91         if conn_ptr is None:
  92             return
  93
  94         return bt2.connection._Connection._create_from_ptr(conn_ptr)
  95
  96     @property
  97     def is_connected(self):
  98         return self.connection is not None
  99
 100     def disconnect(self):
 101         self._disconnect(self._ptr)
 102
 103     def __eq__(self, other):
 104         if type(other) is not type(self):
 105             return False
 106
 107         return self.addr == other.addr
 108
 109
 110 class _InputPort(_Port):
 111     pass
 112
 113
 114 class _OutputPort(_Port):
 115     pass
 116
 117
 118 class _PrivatePort(object._PrivateObject, _Port):
 119     @property
 120     def name(self):
 121         return self._name(self._pub_ptr)
 122
 123     @property
 124     def component(self):
 125         comp_ptr = native_bt.private_port_get_private_component(self._ptr)
 126
 127         if comp_ptr is None:
 128             return
 129
 130         pub_comp_ptr = native_bt.component_from_private_component(comp_ptr)
 131         assert(pub_comp_ptr)
 132         comp = bt2.component._create_generic_component_from_ptr(pub_comp_ptr)
 133         native_bt.put(comp_ptr)
 134         return comp
 135
 136     @property
 137     def connection(self):
 138         conn_ptr = native_bt.private_port_get_private_connection(self._ptr)
 139
 140         if conn_ptr is None:
 141             return
 142
 143         return bt2.connection._create_private_from_ptr(conn_ptr)
 144
 145     def remove_from_component(self):
 146         status = native_bt.private_port_remove_from_component(self._ptr)
 147
 148         if status < 0:
 149             raise bt2.Error("cannot remove port from component")
 150
 151     def disconnect(self):
 152         self._disconnect(self._pub_ptr)
 153
 154
 155 class _PrivateInputPort(_PrivatePort, _InputPort):
 156     pass
 157
 158
 159 class _PrivateOutputPort(_PrivatePort, _OutputPort):
 160     pass