corelink_const.py

Go to the documentation of this file.

"""
    @file corelink_const.py
     Copies the constants from the DLL to wrapper
"""
import ctypes
from sys import platform
import config

# pylint: disable=global-statement
# global declaration is essential
DLL_PATH = config.DLL_PATH
LIB = None
STATE_CODES = dict()
ERROR_CODES = dict()
DATA_CODES = dict()
CALL_BACK = dict()

def init_state_codes():
    """
        initialize State Codes
    """
    global LIB,STATE_CODES
    arr = ["STREAM_STATE_NONE",
            "STREAM_STATE_SEND_UDP",
            "STREAM_STATE_SEND_TCP",
            "STREAM_STATE_RECV_UDP",
            "STREAM_STATE_RECV_TCP",
            "STREAM_STATE_SEND",
            "STREAM_STATE_RECV",
            "STREAM_STATE_UDP",
            "STREAM_STATE_TCP",
            "STREAM_STATE_ALL"]
    for string in arr:
        STATE_CODES[string] = ctypes.c_int.in_dll(LIB, string).value

def init_error_codes():
    """
        Initialize Error Codes
    """
    global LIB
    global ERROR_CODES
    arr = ["ERROR_CODE_NONE",
           "ERROR_CODE_STATE",
           "ERROR_CODE_VALUE",
           "ERROR_CODE_SOCKET",
           "ERROR_CODE_COMM",
           "ERROR_CODE_NO_TOKEN",
           ]
    for string in arr:
        ERROR_CODES[string] = ctypes.c_int.in_dll(LIB,string).value

def init_call_back():
    """
        Initialize CallBack Codes
    """
    global LIB,CALL_BACK
    arr = ["CALLBACK_DROPPED",
           "CALLBACK_STALE",
           "CALLBACK_SUBSCRIBE",
           "CALLBACK_UPDATE",
            ]
    for string in arr:
        CALL_BACK[string] = ctypes.c_int.in_dll(LIB,string).value

def init_data_codes():
    """
        Initialize Data Codes
    """
    global LIB,DATA_CODES
    arr = ["DATA_FUNCTION_LIST",
           "DATA_FUNCTION_INFO",
           "DATA_WORK_SPACES",
           "DATA_STREAM_INFO",
           "DATA_STREAM_LIST",
           "DATA_MAX",
           ]
    for string in arr:
        DATA_CODES[string] = ctypes.c_int.in_dll(LIB,string).value

def stream_state_to_bit_index(state):
    """
        get stream_state_to_bit_index
    """
    return LIB.stream_state_to_bit_index(state)

def stream_state_name(state):
    """
        get stream_state_name
    """
    global LIB
    length = LIB.stream_state_nameLen(state)
    if length == 0:
        return ""
    buffer = ctypes.create_string_buffer(length)
    LIB.stream_state_nameStr(state, buffer)
    name = buffer.value
    name = name.decode('UTF-8')
    # LIB.stream_state_name.argtypes = [ctypes.c_int]
    # LIB.stream_state_name.restype = ctypes.c_char_p
    # ans = (LIB.stream_state_name(ctypes.c_int(state))).decode('UTF-8')
    return name

def error_code_name(error_code):
    """
        Get error_code_name
    """
    global LIB
    length = LIB.errorCodeNameLen(error_code)
    if length == 0:
        return ""
    buffer = ctypes.create_string_buffer(length)
    LIB.errorCodeNameStr(error_code, buffer)
    name = buffer.value
    name = name.decode('UTF-8')
    # LIB.error_code_name.argtypes = [ctypes.c_int]
    # LIB.error_code_name.restype = ctypes.c_char_p
    # ans = (LIB.error_code_name(ctypes.c_int(error_code))).decode('UTF-8')
    return name

def init_lib():
    """
        Initialize LIB depending on Operating Sysytem
    """
    global LIB
    global DLL_PATH
    if LIB is not None:
        return
    if platform.startswith('win'):
        LIB = ctypes.CDLL(DLL_PATH+"Corelink.dll")
    if platform.startswith('linux'):
        LIB = ctypes.CDLL(DLL_PATH+"libCorelink.so")
    if platform.startswith('darwin'):
        LIB = ctypes.CDLL(DLL_PATH+"libCorelink.dylib")

init_lib()
init_state_codes()
init_error_codes()

ERROR_CODE_STRING= [error_code_name(0),
                    error_code_name(1),
                    error_code_name(2),
                    error_code_name(3),
                    error_code_name(4),
                    error_code_name(5)]