IPC.py

Go to the documentation of this file.

#
# Copyright 2007, MeVis Medical Solutions AG
#
# The user may use this file in accordance with the license agreement provided with
# the Software or, alternatively, in accordance with the terms contained in a
# written agreement between the user and MeVis Medical Solutions AG.
#
# For further information use the contact form at https://www.mevislab.de/contact
#
 
 
 
# -- system imports ----------------------------------------------------------------------------{{{-
import os
import socket
import select
import errno
import time
import sys
from enum import IntEnum
 
# ----------------------------------------------------------------------------------------------}}}-
 
# -- error constants ---------------------------------------------------------------------------{{{-
 
 
class Error(IntEnum):
    OK = -1
    ON_INIT = 1
    ON_CONNECTING = 2
    ON_DISCONNECTING = 3
    NOT_CONNECTED = 10
    ON_SENDING_TIMEOUT = 20
    ON_RECEIVING_TIMEOUT = 21
    ON_SENDING_1 = 30
    SENDING_BLOCKED = 31
    ON_SENDING_2 = 32
    ON_RECEIVING_1 = 40
    EMPTY_MESSAGE = 41
    ON_RECEIVING_2 = 42
 
 
# ----------------------------------------------------------------------------------------------}}}-
 
 
# -- class Communicator ------------------------------------------------------------------------{{{-
 
class Communicator:
 
    
    _mMsgLength = 8192
 
    def __init__(self):
        # -- member variables ------------------------------------------------------------------------{{{-
        
        self._mHost = "localhost"
        
        self._mPort = 21319 if not "MeVisLabTestCenterPort" in os.environ else int(os.environ["MeVisLabTestCenterPort"])
        
        self._mConnected = False
 
        
        self._msocket = None
        
        self._csocket = None
 
        
        self._mErrorCode = Error.OK
        
        self._mErrorMsg = ""
        # --------------------------------------------------------------------------------------------}}}-
 
    # -- def getPort -----------------------------------------------------------------------------{{{-
    
    def getPort(self):
        return self._mPort
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def isConnected -------------------------------------------------------------------------{{{-
    
    def isConnected(self):
        return self._mConnected
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def send --------------------------------------------------------------------------------{{{-
    
    def send(self, data, timeout):
        # Sending requires to be connected.
        if not self._mConnected:
            self._handleError(Error.NOT_CONNECTED, "Not connected.")
            return False
 
        if data == "":
            data == "##[empty]"
 
        # Send length of following data.
        length = len(data)
        if not self._send(str(length).ljust(16), 16, timeout):
            return False
 
        # Send actual data.
        return self._send(data, length, 1)
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def _send -------------------------------------------------------------------------------{{{-
    
    def _send(self, data, length, timeout):
        if not isinstance(data, bytes):
            data = data.encode("UTF-8")
        i = 0
        while i < length:
            # Wait for socket to get ready.
            try:
                (rlist, wlist, xlist) = select.select([], [self._csocket], [], timeout)
                if self._csocket not in wlist:
                    self._handleError(Error.ON_SENDING_TIMEOUT, "Connection error (sending).")
                    return False
            except select.error as msg:
                self._handleError(Error.ON_SENDING_1, "Error sending: %s" % (msg))
                return False
 
            # Send data.
            try:
                # Send next data junk (of max length given in _mMsgLength).
                j = self._csocket.send(data[i : min(length, i + self._mMsgLength)])
                if j == -1:
                    self._handleError(Error.SENDING_BLOCKED, "Error sending: couldn't send.")
                    return -1
                i += j
            except OSError as msg:
                self._handleError(Error.ON_SENDING_2, "Error sending: %s" % (msg))
                return False
        return i == length
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def recv --------------------------------------------------------------------------------{{{-
    
    def recv(self, timeout):
        # Link must be up to receive data.
        if not self._mConnected:
            self._handleError(Error.NOT_CONNECTED, "Not connected.")
            return None
 
        # Receive the length of the data.
        data = self._recv(16, timeout)
        if data in (b"", None):
            return None
        length = int(data)
 
        # Receive the actual data.
        data = self._recv(length, 1)
        if isinstance(data, bytes):
            data = data.decode("UTF-8")
        if data == "##[empty]":
            return ""
        return data
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def _recv -------------------------------------------------------------------------------{{{-
    
    def _recv(self, length, timeout):
        data = b""
        while len(data) < length:
            # Wait for socket to get ready.
            try:
                (rlist, wlist, xlist) = select.select([self._csocket], [], [], timeout)
                if self._csocket not in rlist:
                    self._handleError(Error.ON_RECEIVING_TIMEOUT, "Connection error (receiving).")
                    return None
            except OSError as msg:
                self._handleError(Error.ON_RECEIVING_1, "Error receiving: %s" % (msg))
                return None
 
            # Receive data of given length.
            try:
                tdata = self._csocket.recv(length - len(data))
                if tdata in (b"", None):
                    self._handleError(Error.EMPTY_MESSAGE, "Error receiving: empty message received")
                    return None
                data += tdata
            except OSError as msg:
                self._handleError(Error.ON_RECEIVING_2, "Error receiving: %s" % (msg))
                return None
        return data
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def getLastError ------------------------------------------------------------------------{{{-
    
    def getLastError(self):
        return self._mErrorCode
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def getLastErrorWithMessage -------------------------------------------------------------{{{-
    
    def getLastErrorWithMessage(self):
        return (self._mErrorCode, self._mErrorMsg)
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def _handleError ------------------------------------------------------------------------{{{-
    
    def _handleError(self, errorCode, errorMsg, silent=False):
        if not silent:
            print(errorMsg)
        self._mErrorCode = errorCode
        self._mErrorMsg = errorMsg
        self.disconnect()
 
    # --------------------------------------------------------------------------------------------}}}-
 
 
# ----------------------------------------------------------------------------------------------}}}-
 
 
# -- class ComMaster ---------------------------------------------------------------------------{{{-
 
class ComMaster(Communicator):
    # -- def __init__ ----------------------------------------------------------------------------{{{-
    
    def __init__(self, connectTimeout=10):
        super().__init__()
        try:
            self._connectTimeout = connectTimeout
            self._msocket_msocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self._msocket_msocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            self._msocket_msocket.settimeout(self._connectTimeout)
            port = self._mPort_mPort
            while port < 65535:
                try:
                    self._msocket_msocket.bind((self._mHost_mHost, port))
                    if port != self._mPort_mPort:
                        print(f"Port {self._mPort} was blocked, using {port} instead.")
                        self._mPort_mPort = port
                    break
                except OSError as e:
                    # error code 10013 happens on windows, even though it is about missing access permissions
                    # for the socket, it actually indicates that the port is already in use
                    if e.errno in (errno.EADDRINUSE, errno.EADDRNOTAVAIL, 10013):
                        port += 1
                    else:
                        raise
            self._msocket_msocket.listen(1)
        except OSError as msg:
            self._handleError(Error.ON_INIT, "Socket error (server): %s " % (msg))
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def __del__ -----------------------------------------------------------------------------{{{-
    
    def __del__(self):
        if self._mConnected_mConnected:
            self._csocket_csocket.close()
            del self._csocket_csocket
 
        self._msocket_msocket.close()
        del self._msocket_msocket
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def connect -----------------------------------------------------------------------------{{{-
    
    def connect(self, timeout=2):
        # Reset error handler.
        self._mErrorCode_mErrorCode = Error.OK
        self._mErrorMsg_mErrorMsg = ""
        try:
            (self._csocket_csocket, addr) = self._msocket_msocket.accept()
            self._csocket_csocket.settimeout(timeout)
            self._mConnected_mConnected = True
        except OSError as msg:
            # can happen at the beginning, there are several re-tries, so be silent about it
            self._handleError(Error.ON_CONNECTING, "Error connecting (server): %s" % (msg), silent=True)
            return False
        return True
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def disconnect --------------------------------------------------------------------------{{{-
    
    def disconnect(self):
        if self._mConnected_mConnected:
            try:
                self._mConnected_mConnected = False
                self._csocket_csocket.close()
                del self._csocket_csocket
            except OSError as msg:
                self._handleError(Error.ON_DISCONNECTING, "Error disconnecting: %s" % (msg), silent=True)
                return False
        return True
 
    # --------------------------------------------------------------------------------------------}}}-
 
    def getConnectTimeout(self):
        """Returns the timeout used by the Master while establishing the connection to the slave."""
        return self._connectTimeout
 
 
# ----------------------------------------------------------------------------------------------}}}-
 
 
# -- class ComSlave ----------------------------------------------------------------------------{{{-
 
class ComSlave(Communicator):
    # -- def __init__ ----------------------------------------------------------------------------{{{-
    
    def __init__(self):
        super().__init__()
        try:
            self._csocket_csocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        except OSError as msg:
            self._handleError(Error.ON_INIT, "Socket error (client): %s" % (msg))
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def __del__ -----------------------------------------------------------------------------{{{-
    
    def __del__(self):
        if self._csocket_csocket != None:
            self._csocket_csocket.close()
            del self._csocket_csocket
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def connect -----------------------------------------------------------------------------{{{-
    
    def connect(self):
        try:
            try:
                self._csocket_csocket.connect((self._mHost_mHost, self._mPort_mPort))
            except OSError as msg:
                if sys.platform.startswith("win") or msg.errno != errno.EINTR:
                    raise
                poll = select.poll()
                poll.register(self._csocket_csocket, select.POLLOUT)
                while poll.poll(-1) == -1:
                    if msg.errno != errno.EINTR:
                        print("poll() failed", file=sys.stderr)
                        raise
                    time.sleep(0.5)
                error = self._csocket_csocket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
                if error:
                    print("getsockopt() failed %s" % (error), file=sys.stderr)
                    raise
 
            self._csocket_csocket.settimeout(2)
            self._csocket_csocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            self._mConnected_mConnected = True
        except OSError as msg:
            self._handleError(Error.ON_CONNECTING, "Error connecting (client): %s" % (msg))
            return False
        return True
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def disconnect --------------------------------------------------------------------------{{{-
    
    def disconnect(self):
        if self._mConnected_mConnected:
            try:
                self._mConnected_mConnected = False
                self._csocket_csocket.close()
            except OSError as msg:
                self._handleError(Error.ON_DISCONNECTING, "Error disconnecting: %s" % (msg), silent=True)
                return False
        return True
 
    # --------------------------------------------------------------------------------------------}}}-
 
 
# ----------------------------------------------------------------------------------------------}}}-