Add Babeltrace 2 Python bindings

[babeltrace.git] / bindings / python / bt2 / component.py

# The MIT License (MIT)
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.

from bt2 import native_bt, object, utils
import bt2.notification_iterator
import collections.abc
import bt2


# This class wraps a component class pointer. This component class could
# have been created by Python code, but since we only have the pointer,
# we can only wrap it in a generic way and lose the original Python
# class.
class _GenericComponentClass(object._Object):
    @property
    def name(self):
        return native_bt.component_class_get_name(self._ptr)

    @property
    def description(self):
        return native_bt.component_class_get_description(self._ptr)

    def __call__(self, params=None, name=None):
        params = bt2.create_value(params)
        comp_ptr = native_bt.component_create_with_init_method_data(self._ptr,
                                                                    name,
                                                                    params._ptr,
                                                                    None)

        if comp_ptr is None:
            raise bt2.CreationError('cannot create component object')

        return _create_generic_component_from_ptr(comp_ptr)


class _GenericSourceComponentClass(_GenericComponentClass):
    pass


class _GenericFilterComponentClass(_GenericComponentClass):
    pass


class _GenericSinkComponentClass(_GenericComponentClass):
    pass


# This class holds the methods which are common to both generic
# component objects and Python user component objects. They use the
# internal native _ptr, however it was set, to call native API
# functions.
class _CommonComponentMethods:
    @property
    def name(self):
        return native_bt.component_get_name(self._ptr)

    @property
    def component_class(self):
        cc_ptr = native_bt.component_get_class(self._ptr)
        utils._handle_ptr(cc_ptr, "cannot get component object's class object")
        return _create_generic_component_class_from_ptr(cc_ptr)

    def _handle_status(self, status, gen_error_msg):
        if status == native_bt.COMPONENT_STATUS_END:
            raise bt2.Stop
        elif status == native_bt.COMPONENT_STATUS_AGAIN:
            raise bt2.TryAgain
        elif status == native_bt.COMPONENT_STATUS_UNSUPPORTED:
            raise bt2.UnsupportedFeature
        elif status < 0:
            raise bt2.Error(gen_error_msg)


class _CommonSourceComponentMethods(_CommonComponentMethods):
    def create_notification_iterator(self):
        iter_ptr = native_bt.component_source_create_notification_iterator_with_init_method_data(self._ptr, None)

        if iter_ptr is None:
            raise bt2.CreationError('cannot create notification iterator object')

        return bt2.notification_iterator._GenericNotificationIterator._create_from_ptr(iter_ptr)


class _CommonFilterComponentMethods(_CommonComponentMethods):
    def create_notification_iterator(self):
        iter_ptr = native_bt.component_filter_create_notification_iterator_with_init_method_data(self._ptr, None)

        if iter_ptr is None:
            raise bt2.CreationError('cannot create notification iterator object')

        return bt2.notification_iterator._GenericNotificationIterator._create_from_ptr(iter_ptr)

    def add_notification_iterator(self, notif_iter):
        utils._check_type(notif_iter, bt2.notification_iterator._GenericNotificationIteratorMethods)
        status = native_bt.component_filter_add_iterator(self._ptr, notif_iter._ptr)
        self._handle_status(status, 'unexpected error: cannot add notification iterator to filter component object')


class _CommonSinkComponentMethods(_CommonComponentMethods):
    def add_notification_iterator(self, notif_iter):
        utils._check_type(notif_iter, bt2.notification_iterator._GenericNotificationIteratorMethods)
        status = native_bt.component_sink_add_iterator(self._ptr, notif_iter._ptr)
        self._handle_status(status, 'unexpected error: cannot add notification iterator to sink component object')

    def consume(self):
        status = native_bt.component_sink_consume(self._ptr)
        self._handle_status(status, 'unexpected error: cannot consume sink component object')


# This is analogous to _GenericSourceComponentClass, but for source
# component objects.
class _GenericSourceComponent(object._Object, _CommonSourceComponentMethods):
    pass


# This is analogous to _GenericFilterComponentClass, but for filter
# component objects.
class _GenericFilterComponent(object._Object, _CommonFilterComponentMethods):
    pass


# This is analogous to _GenericSinkComponentClass, but for sink
# component objects.
class _GenericSinkComponent(object._Object, _CommonSinkComponentMethods):
    pass


_COMP_CLS_TYPE_TO_GENERIC_COMP_CLS = {
    native_bt.COMPONENT_CLASS_TYPE_SOURCE: _GenericSourceComponent,
    native_bt.COMPONENT_CLASS_TYPE_FILTER: _GenericFilterComponent,
    native_bt.COMPONENT_CLASS_TYPE_SINK: _GenericSinkComponent,
}


_COMP_CLS_TYPE_TO_GENERIC_COMP_CLS_CLS = {
    native_bt.COMPONENT_CLASS_TYPE_SOURCE: _GenericSourceComponentClass,
    native_bt.COMPONENT_CLASS_TYPE_FILTER: _GenericFilterComponentClass,
    native_bt.COMPONENT_CLASS_TYPE_SINK: _GenericSinkComponentClass,
}


def _create_generic_component_from_ptr(ptr):
    comp_cls_type = native_bt.component_get_class_type(ptr)
    return _COMP_CLS_TYPE_TO_GENERIC_COMP_CLS[comp_cls_type]._create_from_ptr(ptr)


def _create_generic_component_class_from_ptr(ptr):
    comp_cls_type = native_bt.component_class_get_type(ptr)
    return _COMP_CLS_TYPE_TO_GENERIC_COMP_CLS_CLS[comp_cls_type]._create_from_ptr(ptr)


# Metaclass for component classes defined by Python code.
#
# The Python user can create a standard Python class which inherits one
# of the three base classes (UserSourceComponent, UserFilterComponent,
# or UserSinkComponent). Those base classes set this class
# (_UserComponentType) as their metaclass.
#
# Once the body of a user-defined component class is executed, this
# metaclass is used to create and initialize the class. The metaclass
# creates a native BT component class of the corresponding type and
# associates it with this user-defined class. The metaclass also defines
# class methods like the `name` and `description` properties to match
# the _GenericComponentClass interface.
#
# The component class name which is used is either:
#
# * The `name` parameter of the class:
#
#       class MySink(bt2.SinkComponent, name='my-custom-sink'):
#           ...
#
# * If the `name` class parameter is not used: the name of the class
#   itself (`MySink` in the example above).
#
# The component class description which is used is the user-defined
# class's docstring:
#
#     class MySink(bt2.SinkComponent):
#         'Description goes here'
#         ...
#
# A user-defined Python component class can have an __init__() method
# which must at least accept the `params` and `name` arguments:
#
#     def __init__(self, params, name, something_else):
#         ...
#
# The user-defined component class can also have a _destroy() method
# (do NOT use __del__()) to be notified when the component object is
# (really) destroyed.
#
# User-defined source and filter component classes must use the
# `notification_iterator_class` class parameter to specify the
# notification iterator class to use for this component class:
#
#     class MyNotificationIterator(bt2.UserNotificationIterator):
#         ...
#
#     class MySource(bt2.UserSourceComponent,
#                    notification_iterator_class=MyNotificationIterator):
#         ...
#
# This notification iterator class must inherit
# bt2.UserNotificationIterator, and it must define the _get() and
# _next() methods. The notification iterator class can also define an
# __init__() method: this method has access to the original Python
# component object which was used to create it as the `component`
# property. The notification iterator class can also define a _destroy()
# method (again, do NOT use __del__()): this is called when the
# notification iterator is (really) destroyed.
#
# When the user-defined class is destroyed, this metaclass's __del__()
# method is called: the native BT component class pointer is put (not
# needed anymore, at least not by any Python code since all references
# are dropped for __del__() to be called).
class _UserComponentType(type):
    # __new__() is used to catch custom class parameters
    def __new__(meta_cls, class_name, bases, attrs, **kwargs):
        return super().__new__(meta_cls, class_name, bases, attrs)

    def __init__(cls, class_name, bases, namespace, **kwargs):
        super().__init__(class_name, bases, namespace)

        # skip our own bases; they are never directly instantiated by the user
        if class_name in ('_UserComponent', 'UserSourceComponent', 'UserFilterComponent', 'UserSinkComponent'):
            return

        comp_cls_name = kwargs.get('name', class_name)
        comp_cls_descr = getattr(cls, '__doc__', None)
        iter_cls = kwargs.get('notification_iterator_class')

        if UserSourceComponent in bases:
            _UserComponentType._set_iterator_class(cls, iter_cls)
            has_seek_time = _UserComponentType._has_seek_to_time_method(cls._iter_cls)
            cc_ptr = native_bt.py3_component_class_source_create(cls,
                                                                 comp_cls_name,
                                                                 comp_cls_descr,
                                                                 has_seek_time)
        elif UserFilterComponent in bases:
            _UserComponentType._set_iterator_class(cls, iter_cls)
            has_seek_time = _UserComponentType._has_seek_to_time_method(cls._iter_cls)
            cc_ptr = native_bt.py3_component_class_filter_create(cls,
                                                                 comp_cls_name,
                                                                 comp_cls_descr,
                                                                 has_seek_time)
        elif UserSinkComponent in bases:
            if not hasattr(cls, '_consume'):
                raise bt2.IncompleteUserClassError("cannot create component class '{}': missing a _consume() method".format(class_name))

            cc_ptr = native_bt.py3_component_class_sink_create(cls,
                                                               comp_cls_name,
                                                               comp_cls_descr)
        else:
            raise bt2.IncompleteUserClassError("cannot find a known component class base in the bases of '{}'".format(class_name))

        if cc_ptr is None:
            raise bt2.CreationError("cannot create component class '{}'".format(class_name))

        cls._cc_ptr = cc_ptr

    def __call__(cls, *args, **kwargs):
        # create instance
        self = cls.__new__(cls)

        # assign native component pointer received from caller
        self._ptr = kwargs.get('__comp_ptr')
        name = kwargs.get('name')

        if self._ptr is None:
            # called from Python code
            self._belongs_to_native_component = False

            # py3_component_create() will call self.__init__() with the
            # desired arguments and keyword arguments. This is needed
            # because functions such as
            # bt_component_sink_set_minimum_input_count() can only be
            # called _during_ the bt_component_create() call (in
            # Python words: during self.__init__()).
            #
            # The arguments and keyword arguments to use for
            # self.__init__() are put in the object itself to find them
            # from the bt_component_create() function.
            self._init_args = args
            self._init_kwargs = kwargs
            native_bt.py3_component_create(cls._cc_ptr, self, name)

            # At this point, self._ptr should be set to non-None. If
            # it's not, an error occured during the
            # native_bt.py3_component_create() call. We consider this a
            # creation error.
            if self._ptr is None:
                raise bt2.CreationError("cannot create component object from component class '{}'".format(cls.__name__))
        else:
            # Called from non-Python code (within
            # bt_component_create()): call __init__() here, after
            # removing the __comp_ptr keyword argument which is just for
            # this __call__() method.
            self._belongs_to_native_component = True
            del kwargs['__comp_ptr']

            # inject `name` into the keyword arguments
            kwargs['name'] = self.name
            self.__init__(*args, **kwargs)

        return self

    @staticmethod
    def _has_seek_to_time_method(iter_cls):
        return hasattr(iter_cls, '_seek_to_time')

    @staticmethod
    def _set_iterator_class(cls, iter_cls):
        if iter_cls is None:
            raise bt2.IncompleteUserClassError("cannot create component class '{}': missing notification iterator class".format(cls.__name__))

        if not issubclass(iter_cls, bt2.notification_iterator.UserNotificationIterator):
            raise bt2.IncompleteUserClassError("cannot create component class '{}': notification iterator class does not inherit bt2.UserNotificationIterator".format(cls.__name__))

        if not hasattr(iter_cls, '_get'):
            raise bt2.IncompleteUserClassError("cannot create component class '{}': notification iterator class is missing a _get() method".format(cls.__name__))

        if not hasattr(iter_cls, '_next'):
            raise bt2.IncompleteUserClassError("cannot create component class '{}': notification iterator class is missing a _next() method".format(cls.__name__))

        cls._iter_cls = iter_cls

    @property
    def name(cls):
        return native_bt.component_class_get_name(cls._cc_ptr)

    @property
    def description(cls):
        return native_bt.component_class_get_description(cls._cc_ptr)

    @property
    def addr(cls):
        return int(cls._cc_ptr)

    def __del__(cls):
        if hasattr(cls, '_cc_ptr'):
            native_bt.put(cls._cc_ptr)


class _ComponentInputNotificationIterators(collections.abc.Sequence):
    def __init__(self, comp, count_func, get_func):
        self._comp = comp
        self._count_func = count_func
        self._get_func = get_func

    def __len__(self):
        status, count = self._count_func(self._comp._ptr)
        utils._handle_ret(status, "cannot get component object's input notification iterator count")
        return count

    def __getitem__(self, index):
        utils._check_uint64(index)

        if index >= len(self):
            raise IndexError

        notif_iter_ptr = self._get_func(self._comp._ptr, index)
        utils._handle_ptr(notif_iter_ptr, "cannot get component object's input notification iterator")
        return bt2.notification_iterator._GenericNotificationIterator._create_from_ptr(notif_iter_ptr)


class _UserComponent(metaclass=_UserComponentType):
    @property
    def addr(self):
        return int(self._ptr)

    def __init__(self, *args, **kwargs):
        pass

    def _destroy(self):
        pass

    def __del__(self):
        if not self._belongs_to_native_component:
            self._belongs_to_native_component = True
            native_bt.py3_component_on_del(self)
            native_bt.put(self._ptr)


class UserSourceComponent(_UserComponent, _CommonSourceComponentMethods):
    pass


class UserFilterComponent(_UserComponent, _CommonFilterComponentMethods):
    def _set_minimum_input_notification_iterator_count(self, count):
        utils._check_uint64(count)
        status = native_bt.component_filter_set_minimum_input_count(self._ptr, count)
        self._handle_status(status, "unexpected error: cannot set filter component object's minimum input notification iterator count")

    _minimum_input_notification_iterator_count = property(fset=_set_minimum_input_notification_iterator_count)

    def _set_maximum_input_notification_iterator_count(self, count):
        utils._check_uint64(count)
        status = native_bt.component_filter_set_maximum_input_count(self._ptr, count)
        self._handle_status(status, "unexpected error: cannot set filter component object's maximum input notification iterator count")

    _maximum_input_notification_iterator_count = property(fset=_set_maximum_input_notification_iterator_count)

    @property
    def _input_notification_iterators(self):
        return _ComponentInputNotificationIterators(self,
                                                    native_bt.component_filter_get_input_count,
                                                    native_bt.component_filter_get_input_iterator_private)

    def _add_iterator_from_bt(self, notif_iter_ptr):
        notif_iter = bt2.notification_iterator._GenericNotificationIterator._create_from_ptr(notif_iter_ptr)
        self._add_notification_iterator(notif_iter)

    def _add_notification_iterator(self, notif_iter):
        pass


class UserSinkComponent(_UserComponent, _CommonSinkComponentMethods):
    def _set_minimum_input_notification_iterator_count(self, count):
        utils._check_uint64(count)
        status = native_bt.component_sink_set_minimum_input_count(self._ptr, count)
        self._handle_status(status, "unexpected error: cannot set sink component object's minimum input notification iterator count")

    _minimum_input_notification_iterator_count = property(fset=_set_minimum_input_notification_iterator_count)

    def _set_maximum_input_notification_iterator_count(self, count):
        utils._check_uint64(count)
        status = native_bt.component_sink_set_maximum_input_count(self._ptr, count)
        self._handle_status(status, "unexpected error: cannot set sink component object's maximum input notification iterator count")

    _maximum_input_notification_iterator_count = property(fset=_set_maximum_input_notification_iterator_count)

    @property
    def _input_notification_iterators(self):
        return _ComponentInputNotificationIterators(self,
                                                    native_bt.component_sink_get_input_count,
                                                    native_bt.component_sink_get_input_iterator_private)

    def _add_iterator_from_bt(self, notif_iter_ptr):
        notif_iter = bt2.notification_iterator._GenericNotificationIterator._create_from_ptr(notif_iter_ptr)
        self._add_notification_iterator(notif_iter)

    def _add_notification_iterator(self, notif_iter):
        pass