corelink_client.py

Go to the documentation of this file.

"""
    @file corelink_client.py
     Contains client to server commands.
"""
import ctypes
import corelink_const
import corelink_exception
import corelink_send_stream
import corelink_stream_data
import corelink_recv_stream

# pylint: disable=too-many-arguments
# Reasonable in this case.

class Client:
    """
        Client class
    """
    @staticmethod
    def is_active():
        """
        check whether it is active
        """
        corelink_const.LIB.getClientState.restype = ctypes.c_bool
        ans = corelink_const.LIB.getClientState()
        return ans

    @staticmethod
    def connect(server_ip,port):
        """
            Connect
        """
        error_id_c = ctypes.c_int()
        server_ip_c = ctypes.create_string_buffer(server_ip.encode('utf-8'))
        port_c = ctypes.c_int(port)
        corelink_const.LIB.corelinkConnect.argtypes = [ctypes.c_char_p,
                                                        ctypes.c_int,
                                                        ctypes.POINTER(ctypes.c_int)]
        corelink_const.LIB.corelinkConnect(server_ip_c,port_c,ctypes.byref(error_id_c))
        corelink_exception.CorelinkException.get_dll_exception(error_id_c)

    @staticmethod
    def get_comm_response_data_helper(data):
        """
            get_comm_response_data_helper
        """
        data_parsed = []
        total_len = int.from_bytes(data[0],"little")
        int_size = ctypes.sizeof(ctypes.c_int)
        cur_p = int_size
        cur_len = int.from_bytes(data[cur_p],"little")
        cur_msg_p = (total_len+1)*int_size
        for _ in range(total_len):
            data_parsed.append(data[cur_msg_p:cur_msg_p+cur_len])
            cur_msg_p = cur_msg_p + cur_len
            cur_p = cur_p + int_size
            cur_len = int.from_bytes(data[cur_p],"little")
        return data_parsed

    @staticmethod
    def cleanup():
        """
            Clean up
        """
        if not corelink_const.LIB.getClientState():
            return
        streams = corelink_stream_data.StreamData.list_client_streams()
        for i in streams:
            if corelink_stream_data.StreamData.is_receiver(i):
                corelink_recv_stream.RecvStream.get_receiver_handler(i).set_on_receive()
        corelink_const.LIB.corelinkCleanup()

    @staticmethod
    def get_comm_response_data(comm_id_c):
        """
            get comm response data
        """
        response_len_c = ctypes.c_int()
        corelink_const.LIB.getCommData.argtypes = [ctypes.POINTER(ctypes.c_int),
                                                    ctypes.POINTER(ctypes.c_int)]
        corelink_const.LIB.getCommData.restype = ctypes.POINTER(ctypes.c_char)
        response_msg = corelink_const.LIB.getCommData(ctypes.byref(comm_id_c),
                                                    ctypes.byref(response_len_c))
        return Client.get_comm_response_data_helper(response_msg)

    @staticmethod
    def get_data_stream_info(comm_id_c):
        """
            get data stream info
        """
        data_out = Client.get_comm_response_data(comm_id_c)
        stream_id =  int.from_bytes(data_out[0],"little")
        state =  int.from_bytes(data_out[1],"little")
        mtu = int.from_bytes(data_out[2],"little")
        stream_id_c = ctypes.c_int(stream_id)
        corelink_const.LIB.getStreamRef.argtypes = [ctypes.POINTER(ctypes.c_int)]
        corelink_const.LIB.getStreamRef.restype = ctypes.c_int
        stream_ref = corelink_const.LIB.getStreamRef(ctypes.byref(stream_id_c))
        return corelink_stream_data.StreamData(stream_id,
                                                state,mtu,
                                                data_out[3],
                                                data_out[4],
                                                data_out[5],
                                                stream_ref,
                                                data_out[6])

    @staticmethod
    def create_sender(workspace,types,meta,echo,alert,protocol):
        """
            create sender
        """
        comm_id_c = ctypes.c_int()
        error_id_c = ctypes.c_int()
        workspace_c = ctypes.create_string_buffer(workspace.encode('utf-8'))
        type_c = ctypes.create_string_buffer(types.encode('utf-8'))
        meta_c = ctypes.create_string_buffer(meta.encode('utf-8'))
        echo_c = ctypes.c_bool(False)
        alert_c = ctypes.c_bool(False)
        if echo:
            echo_c = ctypes.c_bool(True)
        if alert:
            alert_c = ctypes.c_bool(True)
        protocol_c = ctypes.c_int(protocol)

        corelink_const.LIB.commAddSender.argtypes = [ctypes.c_char_p,
                                                    ctypes.c_char_p,
                                                    ctypes.c_char_p,
                                                    ctypes.c_bool,
                                                    ctypes.c_bool,
                                                    ctypes.c_int,
                                                    ctypes.POINTER(ctypes.c_int),
                                                    ctypes.POINTER(ctypes.c_int)]
        corelink_const.LIB.commAddSender(workspace_c,
                                        type_c,
                                        meta_c,
                                        echo_c,
                                        alert_c,
                                        protocol_c,
                                        ctypes.byref(comm_id_c),
                                        ctypes.byref(error_id_c))
        corelink_exception.CorelinkException.get_dll_exception(error_id_c)
        return corelink_send_stream.SendStream(Client.get_data_stream_info(comm_id_c))

    @staticmethod
    def subscribe(receiver_id,sender_id):
        """
            Subscribe
        """
        receiver_id = ctypes.c_int(receiver_id)
        sender_id = ctypes.c_int(sender_id)
        corelink_const.LIB.commSubscribe.argtypes = [ctypes.POINTER(ctypes.c_int),
                                                    ctypes.POINTER(ctypes.c_int)]
        corelink_const.LIB.commSubscribe.restype = ctypes.c_bool
        output = corelink_const.LIB.commSubscribe(receiver_id,sender_id)
        return output

    @staticmethod
    def un_subscribe(receiver_id,sender_id):
        """
            Un subscribe
        """
        receiver_id = ctypes.c_int(receiver_id)
        sender_id = ctypes.c_int(sender_id)
        corelink_const.LIB.commUnSubscribe.argtypes = [ctypes.POINTER(ctypes.c_int),
                                                        ctypes.POINTER(ctypes.c_int)]
        corelink_const.LIB.commUnSubscribe.restype = ctypes.c_bool
        output = corelink_const.LIB.commUnSubscribe(receiver_id,sender_id)
        return output

    @staticmethod
    def rm_stream(sender_id):
        """
            remove stream
        """
        sender_id = ctypes.c_int(sender_id)
        corelink_const.LIB.commDisconnect.argtypes = [ctypes.POINTER(ctypes.c_int)]
        corelink_const.LIB.commDisconnect.restype = ctypes.c_bool
        output = corelink_const.LIB.commDisconnect(sender_id)
        return output

    @staticmethod
    def vec_string_to_char_array(types):
        """
            vec_string_to_char_array
        """
        lp_c_char = ctypes.POINTER(ctypes.c_char)
        lp_lp_c_char = ctypes.POINTER(lp_c_char)
        types_c = (ctypes.c_char_p * len(types))()
        for (index,val) in enumerate(types):
            types_c[index] = val.encode("UTF-8")
        ans = ctypes.cast(types_c, lp_lp_c_char)
        return ans

    @staticmethod
    def create_receiver(workspace,types,meta,echo,alert,protocol):
        """
            create receiver
        """
        comm_id_c = ctypes.c_int()
        error_id_c = ctypes.c_int()
        workspace_c = ctypes.create_string_buffer(workspace.encode('utf-8'))
        types_c = Client.vec_string_to_char_array(types)
        meta_c = ctypes.create_string_buffer(meta.encode('utf-8'))
        echo_c = ctypes.c_bool(False)
        alert_c = ctypes.c_bool(False)
        if echo:
            echo_c = ctypes.c_bool(True)
        if alert:
            alert_c = ctypes.c_bool(True)
        protocol_c = ctypes.c_int(protocol)
        types_size_c = ctypes.c_int(len(types))
        corelink_const.LIB.commAddReceiver.argtypes = [ctypes.c_char_p,
                                                    ctypes.POINTER(ctypes.POINTER(ctypes.c_char)),
                                                    ctypes.c_int,
                                                    ctypes.c_char_p,
                                                    ctypes.c_bool,
                                                    ctypes.c_bool,
                                                    ctypes.c_int,
                                                    ctypes.POINTER(ctypes.c_int),
                                                    ctypes.POINTER(ctypes.c_int)]
        corelink_const.LIB.commAddReceiver(workspace_c,
                                            types_c,
                                            types_size_c,
                                            meta_c,
                                            echo_c,
                                            alert_c,
                                            protocol_c,
                                            ctypes.byref(comm_id_c),
                                            ctypes.byref(error_id_c))
        corelink_exception.CorelinkException.get_dll_exception(error_id_c)
        return corelink_recv_stream.RecvStream(Client.get_data_stream_info(comm_id_c))

    @staticmethod
    def connect_plugin():
        """
            connect plugin
        """
        error_id_c = ctypes.c_int()
        corelink_const.LIB.corelinkConnectPlugin.argtypes = [ctypes.POINTER(ctypes.c_int)]
        corelink_const.LIB.corelinkConnectPlugin(ctypes.byref(error_id_c))
        corelink_exception.CorelinkException.get_dll_exception(error_id_c)

    @staticmethod
    def get_token():
        """
            get token
        """
        corelink_const.LIB.getTokenStr.argtypes = [ctypes.POINTER(ctypes.c_char)]
        token = ctypes.create_string_buffer(corelink_const.LIB.getTokenLen())
        corelink_const.LIB.getTokenStr(token)
        return token.value.decode("utf-8")

    @staticmethod
    def list_server_functions():
        """
            list server functions
        """
        comm_id_c = ctypes.c_int()
        error_id_c = ctypes.c_int()
        corelink_const.LIB.commListFunctions.argtypes = [ctypes.POINTER(ctypes.c_int),
                                                        ctypes.POINTER(ctypes.c_int)]
        corelink_const.LIB.commListFunctions(ctypes.byref(comm_id_c), ctypes.byref(error_id_c))
        corelink_exception.CorelinkException.get_dll_exception(error_id_c)
        temp = Client.get_comm_response_data(comm_id_c)
        ans = [i.decode("utf-8") for i in temp]
        return ans

    @staticmethod
    def describe_server_function(function):
        """
            describe server function
        """
        comm_id_c = ctypes.c_int()
        error_id_c = ctypes.c_int()
        function = str(function)
        corelink_const.LIB.commGetFunctionInfo.argtypes = [ctypes.POINTER(ctypes.c_char),
                                                        ctypes.POINTER(ctypes.c_int),
                                                        ctypes.POINTER(ctypes.c_int)]
        corelink_const.LIB.commGetFunctionInfo(function.encode("utf-8"),
                                                ctypes.byref(comm_id_c),
                                                ctypes.byref(error_id_c))
        corelink_exception.CorelinkException.get_dll_exception(error_id_c)
        temp = Client.get_comm_response_data(comm_id_c)
        ans = [i.decode("utf-8") for i in temp]
        return ans

    @staticmethod
    def add_workspace(workspace):
        """
            add workspace
        """
        workspace = str(workspace)
        corelink_const.LIB.commAddWorkspace.argtypes = [ctypes.POINTER(ctypes.c_char)]
        corelink_const.LIB.commAddWorkspace(workspace.encode("utf-8"))

    @staticmethod
    def rm_workspace(workspace):
        """
            remove workspace
        """
        workspace = str(workspace)
        corelink_const.LIB.commRmWorkspace.argtypes = [ctypes.POINTER(ctypes.c_char)]
        corelink_const.LIB.commRmWorkspace(workspace.encode("utf-8"))

    @staticmethod
    def list_workspaces():
        """
            list workspace
        """
        comm_id_c = ctypes.c_int()
        error_id_c = ctypes.c_int()
        corelink_const.LIB.commListWorkspaces.argtypes = [ctypes.POINTER(ctypes.c_int),
                                                            ctypes.POINTER(ctypes.c_int)]
        corelink_const.LIB.commListWorkspaces(ctypes.byref(comm_id_c), ctypes.byref(error_id_c))
        corelink_exception.CorelinkException.get_dll_exception(error_id_c)
        temp = Client.get_comm_response_data(comm_id_c)
        ans = [i.decode("utf-8") for i in temp]
        return ans

    @staticmethod
    def stream_info(stream_id):
        """
            stream info
        """
        stream_id_c = ctypes.c_int(stream_id)
        comm_id_c = ctypes.c_int()
        error_id_c = ctypes.c_int()
        corelink_const.LIB.commGetStreamInfo.argtypes = [ctypes.POINTER(ctypes.c_int),
                                                        ctypes.POINTER(ctypes.c_int),
                                                        ctypes.POINTER(ctypes.c_int)]
        corelink_const.LIB.commGetStreamInfo(ctypes.byref(stream_id_c),
                                            ctypes.byref(comm_id_c),
                                            ctypes.byref(error_id_c))
        corelink_exception.CorelinkException.get_dll_exception(error_id_c)
        temp = Client.get_data_stream_info(comm_id_c)
        return temp