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        = 20000 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, u"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
  # --------------------------------------------------------------------------------------------}}}-
# ----------------------------------------------------------------------------------------------}}}-