bindings/python/bt2/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 bt2.message_iterator
  28 import bt2.message
  29 import copy
  30 import bt2
  31
  32
  33 def _create_from_ptr(ptr):
  34     port_type = native_bt.port_get_type(ptr)
  35
  36     if port_type == native_bt.PORT_TYPE_INPUT:
  37         cls = _InputPort
  38     elif port_type == native_bt.PORT_TYPE_OUTPUT:
  39         cls = _OutputPort
  40     else:
  41         raise bt2.Error('unknown port type: {}'.format(port_type))
  42
  43     return cls._create_from_ptr(ptr)
  44
  45
  46 def _create_private_from_ptr(ptr):
  47     pub_ptr = native_bt.port_from_private(ptr)
  48     utils._handle_ptr(pub_ptr, 'cannot get port object from private port object')
  49     port_type = native_bt.port_get_type(pub_ptr)
  50     assert(port_type == native_bt.PORT_TYPE_INPUT or port_type == native_bt.PORT_TYPE_OUTPUT)
  51
  52     if port_type == native_bt.PORT_TYPE_INPUT:
  53         cls = _PrivateInputPort
  54     elif port_type == native_bt.PORT_TYPE_OUTPUT:
  55         cls = _PrivateOutputPort
  56
  57     obj = cls._create_from_ptr(ptr)
  58     obj._pub_ptr = pub_ptr
  59     return obj
  60
  61
  62 class _Port(object._Object):
  63     @staticmethod
  64     def _name(ptr):
  65         name = native_bt.port_get_name(ptr)
  66         assert(name is not None)
  67         return name
  68
  69     @staticmethod
  70     def _disconnect(ptr):
  71         status = native_bt.port_disconnect(ptr)
  72
  73         if status < 0:
  74             raise bt2.Error('cannot disconnect port')
  75
  76     @property
  77     def name(self):
  78         return self._name(self._ptr)
  79
  80     @property
  81     def component(self):
  82         comp_ptr = native_bt.port_get_component(self._ptr)
  83
  84         if comp_ptr is None:
  85             return
  86
  87         return bt2.component._create_generic_component_from_ptr(comp_ptr)
  88
  89     @property
  90     def connection(self):
  91         conn_ptr = native_bt.port_get_connection(self._ptr)
  92
  93         if conn_ptr is None:
  94             return
  95
  96         return bt2.connection._Connection._create_from_ptr(conn_ptr)
  97
  98     @property
  99     def is_connected(self):
 100         return self.connection is not None
 101
 102     def disconnect(self):
 103         self._disconnect(self._ptr)
 104
 105     def __eq__(self, other):
 106         if type(other) is not type(self):
 107             return False
 108
 109         return self.addr == other.addr
 110
 111
 112 class _InputPort(_Port):
 113     pass
 114
 115
 116 class _OutputPort(_Port):
 117     def create_message_iterator(self, message_types=None,
 118                                      colander_component_name=None):
 119         msg_types = bt2.message._msg_types_from_msg_classes(message_types)
 120
 121         if colander_component_name is not None:
 122             utils._check_str(colander_component_name)
 123
 124         msg_iter_ptr = native_bt.py3_create_output_port_msg_iter(int(self._ptr),
 125                                                                      colander_component_name,
 126                                                                      msg_types)
 127
 128         if msg_iter_ptr is None:
 129             raise bt2.CreationError('cannot create output port message iterator')
 130
 131         return bt2.message_iterator._OutputPortMessageIterator._create_from_ptr(msg_iter_ptr)
 132
 133
 134 class _PrivatePort(object._PrivateObject, _Port):
 135     @property
 136     def name(self):
 137         return self._name(self._pub_ptr)
 138
 139     @property
 140     def component(self):
 141         comp_ptr = native_bt.private_port_get_private_component(self._ptr)
 142
 143         if comp_ptr is None:
 144             return
 145
 146         pub_comp_ptr = native_bt.component_from_private(comp_ptr)
 147         assert(pub_comp_ptr)
 148         comp = bt2.component._create_generic_component_from_ptr(pub_comp_ptr)
 149         native_bt.put(comp_ptr)
 150         return comp
 151
 152     @property
 153     def connection(self):
 154         conn_ptr = native_bt.private_port_get_private_connection(self._ptr)
 155
 156         if conn_ptr is None:
 157             return
 158
 159         return bt2.connection._create_private_from_ptr(conn_ptr)
 160
 161     def remove_from_component(self):
 162         status = native_bt.private_port_remove_from_component(self._ptr)
 163
 164         if status < 0:
 165             raise bt2.Error("cannot remove port from component")
 166
 167     def disconnect(self):
 168         self._disconnect(self._pub_ptr)
 169
 170
 171 class _PrivateInputPort(_PrivatePort, _InputPort):
 172     pass
 173
 174
 175 class _PrivateOutputPort(_PrivatePort, _OutputPort):
 176     pass