indi_stream.py

# -*- coding: utf-8 -*-
########################################################################################################################
# NyxNode
# Author: Jérôme ODIER <jerome.odier@lpsc.in2p3.fr>
# SPDX-License-Identifier: GPL-3.0+
########################################################################################################################
 
import ctypes
import typing
 
########################################################################################################################
 
from .. import bind
from .. import json
 
from . import enums
from . import utils
 
########################################################################################################################
 
## @defgroup STREAM_MESSAGE_PY Nyx Stream Message
#  @brief Nyx Stream Message.
 
########################################################################################################################
 
@utils.nyx_property(
    'name',
    '@name',
)
@utils.nyx_property(
    'label',
    '@label',
)
class NyxStreamProp(json.json_dict.NyxDict):
    """!
    @ingroup STREAM_MESSAGE_PY
    @brief Nyx Stream property.
    """
 
    ####################################################################################################################
 
    def __init__(self, name: str, label: str | None = None):
        """!
        @brief Allocates a new Nyx Stream property.
 
        @param name Property name.
        @param label Property label.
 
        @note If the property name ends with `.b`, the payload is automatically Base64-encoded.
        @note If the property name ends with `.z`, the payload is automatically ZLib-compressed.
        """
 
        super().__init__(bind.lib.nyx_stream_prop_new(
            bind.as_bytes(name, allow_none = False),
            bind.as_bytes(label, allow_none = True),
        ))
 
########################################################################################################################
 
@utils.nyx_property(
    'device',
    '@device',
)
@utils.nyx_property(
    'name',
    '@name',
)
@utils.nyx_property(
    'state',
    '@state',
    kind = enums.NyxState,
    getter = enums.NyxState.to_int,
    setter = enums.NyxState.to_str,
)
class NyxStreamVector(json.json_dict.NyxDict):
    """!
    @ingroup STREAM_MESSAGE_PY
    @brief Nyx Stream vector.
    """
 
    ####################################################################################################################
 
    def __init__(self, device: str, name: str, state: enums.NyxState | int | str, props: typing.Iterable[NyxStreamProp], **opts: typing.Any):
        """!
        @brief Allocates a new Nyx Stream vector.
 
        @param device Device name.
        @param name Vector name.
        @param state Vector state.
        @param props Properties.
        @param opts Options (group, label, hints, timeout, message).
        """
 
        ################################################################################################################
 
        super().__init__(bind.lib.nyx_stream_vector_new(
            bind.as_bytes(device, allow_none = False),
            bind.as_bytes(name, allow_none = False),
            enums.NyxState.to_int(state),
            bind.nyx_dict_p(),
            bind.as_opts(opts),
        ))
 
        ################################################################################################################
 
        # noinspection PyTypeChecker
        children: json.NyxList = self['children']
 
        for prop in props:
 
            if not isinstance(prop, NyxStreamProp):
 
                raise TypeError('Expected NyxStreamProp')
 
            children.append(prop)
 
    ####################################################################################################################
 
    def stream_pub(self, field_values: typing.Sequence[bytes]) -> bool:
        """!
        @brief If Nyx Stream is enabled, publishes an entry to a stream.
 
        @param field_values Field payloads, one per field.
        @return @c True if the provided fields match the vector content, @c False otherwise.
 
        @note Field payloads may contain arbitrary binary data.
        """
 
        ################################################################################################################
 
        field_values = [bind.as_bytes(value, allow_none = False) for value in field_values]
 
        ################################################################################################################
 
        cast = lambda value: ctypes.cast(ctypes.c_char_p(value), bind.c_void_p)
 
        ################################################################################################################
 
        n_fields = len(field_values)
 
        field_sizes = (bind.c_size_t * n_fields)(
            *(len(value) for value in field_values),
        )
 
        field_buffs = (bind.c_void_p * n_fields)(
            *(cast(value) for value in field_values),
        )
 
        ################################################################################################################
 
        return bool(bind.lib.nyx_stream_pub(
            self.ptr,
            n_fields,
            field_sizes,
            field_buffs,
        ))
 
########################################################################################################################
 
__all__ = [name for name in globals() if name.lower().startswith('nyx')]
 
########################################################################################################################