X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=bindings%2Fpython%2Fbt2%2Fbt2%2Fport.py;h=f7c236692e23d35951c00b85375ef58a887d9831;hb=2c6f8520ad68e1ca0c9554d705c39790e7d3ef5f;hp=ebd2765045a198423d939c461d5406a3fbe161b7;hpb=78288f581343ec033cc38898777fe43a02380719;p=babeltrace.git diff --git a/bindings/python/bt2/bt2/port.py b/bindings/python/bt2/bt2/port.py index ebd27650..f7c23669 100644 --- a/bindings/python/bt2/bt2/port.py +++ b/bindings/python/bt2/bt2/port.py @@ -20,157 +20,117 @@ # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN # THE SOFTWARE. -from bt2 import native_bt, object, utils -import collections.abc +from bt2 import native_bt, object import bt2.component import bt2.connection import bt2.message_iterator import bt2.message -import copy import bt2 -def _create_from_ptr(ptr): - port_type = native_bt.port_get_type(ptr) +def _create_from_ptr_and_get_ref(ptr, port_type): + cls = _PORT_TYPE_TO_PYCLS.get(port_type, None) - if port_type == native_bt.PORT_TYPE_INPUT: - cls = _InputPort - elif port_type == native_bt.PORT_TYPE_OUTPUT: - cls = _OutputPort - else: + if cls is None: raise bt2.Error('unknown port type: {}'.format(port_type)) - return cls._create_from_ptr(ptr) + return cls._create_from_ptr_and_get_ref(ptr) -def _create_private_from_ptr(ptr): - pub_ptr = native_bt.port_from_private(ptr) - utils._handle_ptr(pub_ptr, 'cannot get port object from private port object') - port_type = native_bt.port_get_type(pub_ptr) - assert(port_type == native_bt.PORT_TYPE_INPUT or port_type == native_bt.PORT_TYPE_OUTPUT) +def _create_self_from_ptr_and_get_ref(ptr, port_type): + cls = _PORT_TYPE_TO_USER_PYCLS.get(port_type, None) - if port_type == native_bt.PORT_TYPE_INPUT: - cls = _PrivateInputPort - elif port_type == native_bt.PORT_TYPE_OUTPUT: - cls = _PrivateOutputPort + if cls is None: + raise bt2.Error('unknown port type: {}'.format(port_type)) - obj = cls._create_from_ptr(ptr) - obj._pub_ptr = pub_ptr - return obj + return cls._create_from_ptr_and_get_ref(ptr) class _Port(object._SharedObject): - @staticmethod - def _name(ptr): - name = native_bt.port_get_name(ptr) - assert(name is not None) - return name - - @staticmethod - def _disconnect(ptr): - status = native_bt.port_disconnect(ptr) + @classmethod + def _get_ref(cls, ptr): + ptr = cls._as_port_ptr(ptr) + return native_bt.port_get_ref(ptr) - if status < 0: - raise bt2.Error('cannot disconnect port') + @classmethod + def _put_ref(cls, ptr): + ptr = cls._as_port_ptr(ptr) + return native_bt.port_put_ref(ptr) @property def name(self): - return self._name(self._ptr) - - @property - def component(self): - comp_ptr = native_bt.port_get_component(self._ptr) - - if comp_ptr is None: - return - - return bt2.component._create_generic_component_from_ptr(comp_ptr) + ptr = self._as_port_ptr(self._ptr) + name = native_bt.port_get_name(ptr) + assert name is not None + return name @property def connection(self): - conn_ptr = native_bt.port_get_connection(self._ptr) + ptr = self._as_port_ptr(self._ptr) + conn_ptr = native_bt.port_borrow_connection_const(ptr) if conn_ptr is None: return - return bt2.connection._Connection._create_from_ptr(conn_ptr) + return bt2.connection._Connection._create_from_ptr_and_get_ref(conn_ptr) @property def is_connected(self): return self.connection is not None - def disconnect(self): - self._disconnect(self._ptr) - - def __eq__(self, other): - if type(other) is not type(self): - return False - - return self.addr == other.addr - class _InputPort(_Port): - pass + _as_port_ptr = staticmethod(native_bt.port_input_as_port_const) class _OutputPort(_Port): - def create_message_iterator(self, message_types=None, - colander_component_name=None): - msg_types = bt2.message._msg_types_from_msg_classes(message_types) + _as_port_ptr = staticmethod(native_bt.port_output_as_port_const) - if colander_component_name is not None: - utils._check_str(colander_component_name) - msg_iter_ptr = native_bt.py3_create_output_port_msg_iter(int(self._ptr), - colander_component_name, - msg_types) - - if msg_iter_ptr is None: - raise bt2.CreationError('cannot create output port message iterator') - - return bt2.message_iterator._OutputPortMessageIterator._create_from_ptr(msg_iter_ptr) - - -class _PrivatePort(_Port): - @property - def name(self): - return self._name(self._pub_ptr) +class _UserComponentPort(_Port): + @classmethod + def _as_port_ptr(cls, ptr): + ptr = cls._as_self_port_ptr(ptr) + return native_bt.self_component_port_as_port(ptr) @property - def component(self): - comp_ptr = native_bt.private_port_get_private_component(self._ptr) + def connection(self): + ptr = self._as_port_ptr(self._ptr) + conn_ptr = native_bt.port_borrow_connection_const(ptr) - if comp_ptr is None: + if conn_ptr is None: return - pub_comp_ptr = native_bt.component_from_private(comp_ptr) - assert(pub_comp_ptr) - comp = bt2.component._create_generic_component_from_ptr(pub_comp_ptr) - native_bt.put(comp_ptr) - return comp + return bt2.connection._Connection._create_from_ptr_and_get_ref(conn_ptr) @property - def connection(self): - conn_ptr = native_bt.private_port_get_private_connection(self._ptr) + def user_data(self): + ptr = self._as_self_port_ptr(self._ptr) + return native_bt.self_component_port_get_data(ptr) - if conn_ptr is None: - return - return bt2.connection._create_private_from_ptr(conn_ptr) +class _UserComponentInputPort(_UserComponentPort, _InputPort): + _as_self_port_ptr = staticmethod(native_bt.self_component_port_input_as_self_component_port) + + def create_message_iterator(self): + msg_iter_ptr = native_bt.self_component_port_input_message_iterator_create(self._ptr) + if msg_iter_ptr is None: + raise bt2.CreationError('cannot create message iterator object') - def remove_from_component(self): - status = native_bt.private_port_remove_from_component(self._ptr) + return bt2.message_iterator._UserComponentInputPortMessageIterator(msg_iter_ptr) - if status < 0: - raise bt2.Error("cannot remove port from component") - def disconnect(self): - self._disconnect(self._pub_ptr) +class _UserComponentOutputPort(_UserComponentPort, _OutputPort): + _as_self_port_ptr = staticmethod(native_bt.self_component_port_output_as_self_component_port) -class _PrivateInputPort(_PrivatePort, _InputPort): - pass +_PORT_TYPE_TO_PYCLS = { + native_bt.PORT_TYPE_INPUT: _InputPort, + native_bt.PORT_TYPE_OUTPUT: _OutputPort, +} -class _PrivateOutputPort(_PrivatePort, _OutputPort): - pass +_PORT_TYPE_TO_USER_PYCLS = { + native_bt.PORT_TYPE_INPUT: _UserComponentInputPort, + native_bt.PORT_TYPE_OUTPUT: _UserComponentOutputPort, +}