(self, sock_addr, ifname, mcast_rx=True)
Instantiate UDP socket

        Args:
            sock_addr : Socket address string
            ifname    : Network interface name
            mcast_rx  : Use socket to receive multicast packets.

        Returns:
            Socket object
Instantiate UDP socket
Instantiate UDP socket
def __init__(self, sock_addr, ifname, mcast_rx=True):
        """Instantiate UDP socket

        Args:
            sock_addr : Socket address string
            ifname    : Network interface name
            mcast_rx  : Use socket to receive multicast packets.

        Returns:
            Socket object

        """
        assert (":" in sock_addr), "Socket address must be in ip:port format"
        self.ip = sock_addr.split(":")[0]
        assert (ipaddress.ip_address(self.ip))  # parse address
        self.port = int(sock_addr.split(":")[1])
        self.ifindex = None

        assert (self.ip is not None), "UDP source IP is not defined"
        assert (self.port is not None), "UDP port is not defined"
        logger.debug("Connect with UDP socket %s:%s" % (self.ip, self.port))

        try:
            # Open and bind socket to Blocksat API port
            self.sock = sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

            # Allow reuse and bind
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            sock.bind(('', self.port))

            # Get the network interface index
            self._get_ifindex(ifname)

            # Join multicast group if listening
            if (mcast_rx):
                self._join_mcast_group()

        except socket.error as e:
            if (e.errno == errno.EADDRNOTAVAIL):
                logger.error("Error on connection with UDP socket %s:%s" %
                             (self.ip, self.port))
                logger.info(
                    "Use argument `--sock-addr` to define the socket address.")
            raise
def __init__ ( self , sock_addr , ifname , mcast_rx = True ) : assert ( ":" in sock_addr ) , "Socket address must be in ip:port format" self . ip = sock_addr . split ( ":" ) [ 0 ] assert ( ipaddress . ip_address ( self . ip ) ) # parse address self . port = int ( sock_addr . split ( ":" ) [ 1 ] ) self . ifindex = None assert ( self . ip is not None ) , "UDP source IP is not defined" assert ( self . port is not None ) , "UDP port is not defined" logger . debug ( "Connect with UDP socket %s:%s" % ( self . ip , self . port ) ) try : # Open and bind socket to Blocksat API port self . sock = sock = socket . socket ( socket . AF_INET , socket . SOCK_DGRAM ) # Allow reuse and bind sock . setsockopt ( socket . SOL_SOCKET , socket . SO_REUSEADDR , 1 ) sock . bind ( ( '' , self . port ) ) # Get the network interface index self . _get_ifindex ( ifname ) # Join multicast group if listening if ( mcast_rx ) : self . _join_mcast_group ( ) except socket . error as e : if ( e . errno == errno . EADDRNOTAVAIL ) : logger . error ( "Error on connection with UDP socket %s:%s" % ( self . ip , self . port ) ) logger . info ( "Use argument `--sock-addr` to define the socket address." ) raise