bind.py

# -*- coding: utf-8 -*-
########################################################################################################################
# NyxNode
# Author: Jérôme ODIER <jerome.odier@lpsc.in2p3.fr>
# SPDX-License-Identifier: GPL-3.0+
########################################################################################################################
 
import os
import sys
import enum
import ctypes
import typing
import pathlib
 
########################################################################################################################
# ERRORS                                                                                                               #
########################################################################################################################
 
class NyxError(RuntimeError):
    """!
    @brief Base error raised by the Python Nyx binding.
    """
 
class NyxLibraryError(NyxError):
    """!
    @brief Raised when the C shared library cannot be loaded or used.
    """
 
########################################################################################################################
# ALIASES                                                                                                              #
########################################################################################################################
 
c_void_p = ctypes.c_void_p
c_char_p = ctypes.c_char_p
c_bool = ctypes.c_bool
c_uint = ctypes.c_uint
c_int = ctypes.c_int
c_ulong = ctypes.c_ulong
c_long = ctypes.c_long
 
c_uint8 = ctypes.c_uint8
c_int8 = ctypes.c_int8
c_uint32 = ctypes.c_uint32
c_int32 = ctypes.c_int32
c_uint64 = ctypes.c_uint64
c_int64 = ctypes.c_int64
c_size_t = ctypes.c_size_t
 
c_float = ctypes.c_float
c_double = ctypes.c_double
 
########################################################################################################################
# TYPES                                                                                                                #
########################################################################################################################
 
class NyxObjectType(enum.IntEnum):
    """!
    @brief JSON object types.
    """
 
    NULL = 0x65656500
    BOOLEAN = 0x65656501
    NUMBER = 0x65656502
    STRING = 0x65656503
    DICT = 0x65656504
    LIST = 0x65656505
 
########################################################################################################################
 
# noinspection PyPep8Naming
class nyx_object_t(ctypes.Structure):
    """!
    @private
    """
 
    _fields_ = [
        ('type', c_int32),
        ('flags', c_uint64),
        ('ref', c_int32),
        ('node', c_void_p),
        ('parent', c_void_p),
        ('callback', c_void_p),
        ('ctx', c_void_p),
    ]
 
########################################################################################################################
 
nyx_object_p = ctypes.POINTER(nyx_object_t)
 
########################################################################################################################
 
# noinspection PyPep8Naming
class nyx_dict_t(ctypes.Structure):
    """!
    @private
    """
 
    _fields_ = [
        ('base', nyx_object_t),
        ('head', c_void_p),
        ('tail', c_void_p),
    ]
 
########################################################################################################################
 
nyx_dict_p = ctypes.POINTER(nyx_dict_t)
 
########################################################################################################################
 
# noinspection PyPep8Naming
class nyx_dict_iter_t(ctypes.Structure):
    """!
    @private
    """
 
    _fields_ = [
        ('idx', c_size_t),
        ('head', c_void_p),
    ]
 
########################################################################################################################
 
nyx_dict_iter_p = ctypes.POINTER(nyx_dict_iter_t)
 
########################################################################################################################
 
# noinspection PyPep8Naming
class nyx_list_t(ctypes.Structure):
    """!
    @private
    """
 
    _fields_ = [
        ('base', nyx_object_t),
        ('head', c_void_p),
        ('tail', c_void_p),
    ]
 
########################################################################################################################
 
nyx_list_p = ctypes.POINTER(nyx_list_t)
 
########################################################################################################################
 
# noinspection PyPep8Naming
class nyx_list_iter_t(ctypes.Structure):
    """!
    @private
    """
 
    _fields_ = [
        ('idx', c_size_t),
        ('head', c_void_p),
    ]
 
########################################################################################################################
 
nyx_list_iter_p = ctypes.POINTER(nyx_list_iter_t)
 
########################################################################################################################
 
nyx_callback_int_t = ctypes.CFUNCTYPE(
    c_bool,
    nyx_dict_p,
    nyx_dict_p,
    c_int,
    c_int,
)
 
nyx_callback_uint_t = ctypes.CFUNCTYPE(
    c_bool,
    nyx_dict_p,
    nyx_dict_p,
    c_uint,
    c_uint,
)
 
nyx_callback_long_t = ctypes.CFUNCTYPE(
    c_bool,
    nyx_dict_p,
    nyx_dict_p,
    c_long,
    c_long,
)
 
nyx_callback_ulong_t = ctypes.CFUNCTYPE(
    c_bool,
    nyx_dict_p,
    nyx_dict_p,
    c_ulong,
    c_ulong,
)
 
nyx_callback_double_t = ctypes.CFUNCTYPE(
    c_bool,
    nyx_dict_p,
    nyx_dict_p,
    c_double,
    c_double,
)
 
nyx_callback_str_t = ctypes.CFUNCTYPE(
    c_bool,
    nyx_dict_p,
    nyx_dict_p,
    c_char_p,
    c_char_p,
)
 
nyx_callback_buffer_t = ctypes.CFUNCTYPE(
    c_bool,
    nyx_dict_p,
    nyx_dict_p,
    c_size_t,
    c_void_p,
)
 
nyx_callback_vector_t = ctypes.CFUNCTYPE(
    None,
    nyx_dict_p,
    c_bool,
)
 
########################################################################################################################
 
# noinspection PyPep8Naming
class nyx_opts_t(ctypes.Structure):
    """!
    @private
    """
 
    _fields_ = [
        ('group', c_char_p),
        ('label', c_char_p),
        ('hints', c_char_p),
        ('message', c_char_p),
        ('timeout', c_double),
    ]
 
########################################################################################################################
 
class nyx_node_t(ctypes.Structure):
    """!
    @private
    """
 
    pass
 
########################################################################################################################
 
nyx_node_p = ctypes.POINTER(nyx_node_t)
 
########################################################################################################################
 
nyx_timer_callback_t = ctypes.CFUNCTYPE(
    None,
    c_void_p
)
 
########################################################################################################################
 
nyx_mqtt_handler_t = ctypes.CFUNCTYPE(
    None,
    nyx_node_p,
    c_int,
    c_size_t,
    c_void_p,
    c_size_t,
    c_void_p,
)
 
########################################################################################################################
# HELPERS                                                                                                              #
########################################################################################################################
 
# noinspection PyOverloads
@typing.overload
def as_bytes(value: str | bytes, *, allow_none: bool = True) -> bytes:
    ...
 
# noinspection PyOverloads
@typing.overload
def as_bytes(value: None, *, allow_none: typing.Literal[True] = True) -> None:
    ...
 
# noinspection PyOverloads
@typing.overload
def as_bytes(value: None, *, allow_none: typing.Literal[False]) -> typing.NoReturn:
    ...
 
# noinspection PyOverloads
@typing.overload
def as_bytes(value: str | bytes | None, *, allow_none: typing.Literal[False]) -> bytes:
    ...
 
# noinspection PyOverloads
@typing.overload
def as_bytes(value: str | bytes | None, *, allow_none: typing.Literal[True] = True) -> bytes | None:
    ...
 
########################################################################################################################
 
def as_bytes(value: str | bytes | None, *, allow_none: bool) -> bytes | None:
 
    ####################################################################################################################
 
    if value is None:
 
        if not allow_none:
 
            raise TypeError('None is not allowed here')
 
        return None
 
    ####################################################################################################################
 
    if isinstance(value, str):
 
        return value.encode('utf-8')
 
    if isinstance(value, bytes):
 
        return value#.encode('utf-8')
 
    ####################################################################################################################
 
    raise TypeError(f'expected str or bytes, got {type(value).__name__}')
 
########################################################################################################################
 
def as_opts(opts: dict[str, typing.Any] | None) -> nyx_opts_t | None:
 
    if opts is None:
 
        return None
 
    return nyx_opts_t(
        as_bytes(opts.get('group'), allow_none = True),
        as_bytes(opts.get('label'), allow_none = True),
        as_bytes(opts.get('hints'), allow_none = True),
        as_bytes(opts.get('message'), allow_none = True),
        float(opts.get('timeout', 0.0)),
    )
 
########################################################################################################################
 
def check_ptr(ptr: int | c_void_p | None, what: str = 'C object') -> c_void_p:
 
    ####################################################################################################################
 
    if isinstance(ptr, c_void_p):
 
        if ptr.value:
 
            return ptr
 
    elif ptr:
 
        # noinspection PyTypeChecker
        return c_void_p(ptr)
 
    ####################################################################################################################
 
    raise TypeError(f'{what} is NULL')
 
########################################################################################################################
 
def take_bytes(ptr: int | c_void_p | None, size: int) -> bytes:
 
    ptr = check_ptr(ptr, 'C buffer')
 
    try:
        return ctypes.string_at(ptr, size)
    finally:
        lib.nyx_memory_free(ptr)
 
########################################################################################################################
 
def take_string(ptr: int | c_void_p | None, size: int | None = None) -> str:
 
    ptr = check_ptr(ptr, 'C string')
 
    try:
        if size is None:
            return ctypes.string_at(ptr).decode('utf-8')
        else:
            return ctypes.string_at(ptr, size).decode('utf-8')
    finally:
        lib.nyx_memory_free(ptr)
 
########################################################################################################################
# LOAD LIBRARY                                                                                                         #
########################################################################################################################
 
def _load_library() -> ctypes.CDLL:
 
    ####################################################################################################################
 
    if os.name == 'nt':
        names = ('libnyx-node.dll', 'nyx-node.dll')
        directories = ()
    elif sys.platform == 'darwin':
        names = ('libnyx-node.dylib', 'nyx-node.dylib')
        directories = (
            pathlib.Path('/usr/local/lib'),
            pathlib.Path('/usr/lib'),
            pathlib.Path('/lib'),
        )
    else:
        names = ('libnyx-node.so', 'nyx-node.so')
        directories = (
            pathlib.Path('/usr/local/lib'),
            pathlib.Path('/usr/lib'),
            pathlib.Path('/lib'),
        )
 
    ####################################################################################################################
 
    candidates = (
        *(str(directory / name) for directory in directories for name in names),
        *names,
    )
 
    ####################################################################################################################
 
    errors: list[str] = []
 
    for candidate in dict.fromkeys(filter(None, candidates)):
 
        if not os.path.dirname(candidate) or pathlib.Path(candidate).exists():
 
            try:
 
                return ctypes.CDLL(candidate)
 
            except OSError as exc:
 
                errors.append(f'{candidate}: {exc}')
 
    ####################################################################################################################
 
    message = 'Unable to load libnyx-node. Install it in a system library directory.'
 
    details = '\n'.join(errors)
 
    if details:
 
        message += f'\nTried:\n{details}'
 
    ####################################################################################################################
 
    raise NyxLibraryError(message)
 
########################################################################################################################
 
lib = _load_library()
 
########################################################################################################################
# BINDINGS                                                                                                             #
########################################################################################################################
 
def _bind(name: str, restype, argtypes: typing.Sequence[object]) -> None:
 
    ####################################################################################################################
 
    argtypes = list(argtypes)
 
    ####################################################################################################################
 
    try:
 
        func = getattr(lib, name)
 
        func.argtypes = argtypes
        func.restype = restype
 
    except AttributeError as e:
 
        raise NyxLibraryError(f'Missing C symbol: {name}') from e
 
########################################################################################################################
 
## LOG ##
 
_bind('nyx_set_log_level', None, [c_int])
 
########################################################################################################################
 
## MEMORY ##
 
_bind('nyx_memory_initialize', None, [])
_bind('nyx_memory_finalize', c_bool, [])
 
_bind('nyx_memory_free', c_size_t, [c_void_p])
 
_bind('nyx_string_ndup', c_void_p, [c_void_p, c_size_t])
_bind('nyx_buffer_ndup', c_void_p, [c_void_p, c_size_t])
 
########################################################################################################################
 
## UTILS ##
 
_bind('nyx_hash', ctypes.c_uint32, [c_size_t, c_void_p, c_uint32])
 
_bind('nyx_generate_mac_addr', None, [c_uint8 * 6, c_uint8, c_uint8, c_char_p])
 
_bind('nyx_base64_encode', c_void_p, [ctypes.POINTER(c_size_t), c_size_t, c_void_p])
_bind('nyx_base64_decode', c_void_p, [ctypes.POINTER(c_size_t), c_size_t, c_void_p])
 
########################################################################################################################
 
## OBJECT ##
 
_bind('nyx_object_parse', c_void_p, [c_char_p])
_bind('nyx_object_ref', c_void_p, [c_void_p])
_bind('nyx_object_unref', c_void_p, [c_void_p])
_bind('nyx_object_get_type', c_int32, [c_void_p])
_bind('nyx_object_notify', c_bool, [c_void_p])
_bind('nyx_object_to_string', c_void_p, [c_void_p])
_bind('nyx_object_to_cstring', c_void_p, [c_void_p])
 
########################################################################################################################
 
## XMLDOC ##
 
_bind('nyx_xmldoc_parse', c_void_p, [c_char_p])
_bind('nyx_xmldoc_free', None, [c_void_p])
_bind('nyx_xmldoc_to_string', c_void_p, [c_void_p])
 
########################################################################################################################
 
## JSON ##
 
_bind('nyx_null_new', c_void_p, [])
 
_bind('nyx_boolean_new', c_void_p, [])
_bind('nyx_boolean_get', c_bool, [c_void_p])
_bind('nyx_boolean_set', c_bool, [c_void_p, c_bool])
 
_bind('nyx_string_new', c_void_p, [])
_bind('nyx_string_get', c_char_p, [c_void_p])
_bind('nyx_string_set', c_bool, [c_void_p, c_char_p, c_bool])
 
_bind('nyx_number_new', c_void_p, [])
_bind('nyx_number_get', c_double, [c_void_p])
_bind('nyx_number_set', c_bool, [c_void_p, c_double])
 
_bind('nyx_dict_new', c_void_p, [])
_bind('nyx_dict_clear', None, [c_void_p])
_bind('nyx_dict_del', None, [c_void_p, c_char_p])
_bind('nyx_dict_iterate', c_bool, [nyx_dict_iter_p, ctypes.POINTER(c_char_p), ctypes.POINTER(c_void_p)])
_bind('nyx_dict_get', c_void_p, [c_void_p, c_char_p])
_bind('nyx_dict_set', c_bool, [c_void_p, c_char_p, c_void_p])
_bind('nyx_dict_size', c_size_t, [c_void_p])
 
_bind('nyx_list_new', c_void_p, [])
_bind('nyx_list_clear', None, [c_void_p])
_bind('nyx_list_del', None, [c_void_p, c_size_t])
_bind('nyx_list_iterate', c_bool, [nyx_list_iter_p, ctypes.POINTER(c_size_t), ctypes.POINTER(c_void_p)])
_bind('nyx_list_get', c_void_p, [c_void_p, c_size_t])
_bind('nyx_list_set', c_bool, [c_void_p, c_size_t, c_void_p])
_bind('nyx_list_size', c_size_t, [c_void_p])
 
########################################################################################################################
 
## NYX MESSAGE ##
 
_bind('nyx_message_new', c_void_p, [c_char_p, c_char_p])
 
## NYX DEL PROPERTY ##
 
_bind('nyx_del_property_new', c_void_p, [c_char_p, c_char_p, c_char_p])
 
## NYX NUMBER ##
 
_bind('nyx_number_prop_new_int', c_void_p, [c_char_p, c_char_p, c_char_p, c_int32, c_int32, c_int32, c_int32])
_bind('nyx_number_prop_new_uint', c_void_p, [c_char_p, c_char_p, c_char_p, c_uint32, c_uint32, c_uint32, c_uint32])
_bind('nyx_number_prop_new_long', c_void_p, [c_char_p, c_char_p, c_char_p, c_int64, c_int64, c_int64, c_int64])
_bind('nyx_number_prop_new_ulong', c_void_p, [c_char_p, c_char_p, c_char_p, c_uint64, c_uint64, c_uint64, c_uint64])
_bind('nyx_number_prop_new_double', c_void_p, [c_char_p, c_char_p, c_char_p, c_double, c_double, c_double, c_double])
 
_bind('nyx_number_prop_set_int', c_bool, [c_void_p, c_int32])
_bind('nyx_number_prop_set_uint', c_bool, [c_void_p, c_uint32])
_bind('nyx_number_prop_set_long', c_bool, [c_void_p, c_int64])
_bind('nyx_number_prop_set_ulong', c_bool, [c_void_p, c_uint64])
_bind('nyx_number_prop_set_double', c_bool, [c_void_p, c_double])
 
_bind('nyx_number_prop_get_int', c_int32, [c_void_p])
_bind('nyx_number_prop_get_uint', c_uint32, [c_void_p])
_bind('nyx_number_prop_get_long', c_int64, [c_void_p])
_bind('nyx_number_prop_get_ulong', c_uint64, [c_void_p])
_bind('nyx_number_prop_get_double', c_double, [c_void_p])
 
_bind('nyx_number_vector_new', c_void_p, [c_char_p, c_char_p, c_int, c_int, ctypes.POINTER(nyx_dict_p), ctypes.POINTER(nyx_opts_t)])
 
## NYX TEXT ##
 
_bind("nyx_text_prop_new", c_void_p, [c_char_p, c_char_p, c_char_p, c_bool])
_bind("nyx_text_vector_new", c_void_p, [c_char_p, c_char_p, c_int, c_int, ctypes.POINTER(nyx_dict_p), ctypes.POINTER(nyx_opts_t)])
 
## NYX LIGHT ##
 
_bind("nyx_light_prop_new", c_void_p, [c_char_p, c_char_p, c_int])
_bind("nyx_light_vector_new", c_void_p, [c_char_p, c_char_p, c_int, ctypes.POINTER(nyx_dict_p), ctypes.POINTER(nyx_opts_t)])
 
## NYX TEXT ##
 
_bind("nyx_switch_prop_new", c_void_p, [c_char_p, c_char_p, c_int])
_bind("nyx_switch_vector_new", c_void_p, [c_char_p, c_char_p, c_int, c_int, c_int, ctypes.POINTER(nyx_dict_p), ctypes.POINTER(nyx_opts_t)])
 
## NYX BLOB ##
 
_bind("nyx_blob_prop_new", c_void_p, [c_char_p, c_char_p, c_char_p, c_size_t, c_void_p, c_bool])
_bind("nyx_blob_vector_new", c_void_p, [c_char_p, c_char_p, c_int, c_int, ctypes.POINTER(nyx_dict_p), ctypes.POINTER(nyx_opts_t)])
 
## NYX STREAM ##
 
_bind("nyx_stream_prop_new", c_void_p, [c_char_p, c_char_p])
_bind("nyx_stream_vector_new", c_void_p, [c_char_p, c_char_p, c_int, ctypes.POINTER(nyx_dict_p), ctypes.POINTER(nyx_opts_t)])
 
_bind('nyx_stream_pub', c_bool, [c_void_p, c_size_t, ctypes.POINTER(c_size_t), ctypes.POINTER(c_void_p)])
 
########################################################################################################################
 
## NYX NODE ##
 
_bind('nyx_node_initialize', nyx_node_p, [c_char_p, ctypes.POINTER(nyx_dict_p), c_char_p, c_char_p, c_char_p, c_char_p, c_char_p, nyx_mqtt_handler_t, c_uint32, c_bool])
_bind('nyx_node_finalize', None, [nyx_node_p, c_bool])
 
_bind('nyx_node_add_timer', None, [nyx_node_p, c_uint32, nyx_timer_callback_t, c_void_p])
 
_bind('nyx_node_poll', None, [nyx_node_p, c_uint32])
 
_bind('nyx_node_enable', None, [nyx_node_p, c_char_p, c_char_p, c_char_p])
_bind('nyx_node_disable', None, [nyx_node_p, c_char_p, c_char_p, c_char_p])
 
_bind('nyx_node_send_message', None, [nyx_node_p, c_char_p, c_char_p])
_bind('nyx_node_send_del_property', None, [nyx_node_p, c_char_p, c_char_p, c_char_p])
 
_bind('nyx_mqtt_sub', None, [c_void_p, c_char_p, c_int])
_bind('nyx_mqtt_pub', None, [c_void_p, c_char_p, c_size_t, c_void_p, c_int])
 
_bind('nyx_nss_pub', None, [c_void_p, c_char_p, c_char_p, c_size_t, ctypes.POINTER(c_uint32), ctypes.POINTER(c_size_t), ctypes.POINTER(c_void_p)])
 
########################################################################################################################