Source code for caspia.meadow.services.button

import caspia.meadow.rules.handler
from caspia.meadow import errors
from caspia.meadow.rules import OnDoRule, OnDoRuleTrigger
from caspia.meadow.serialization import Deserializable, Serializable
from caspia.reactive import Observable, Observer

from .base import Characteristic, ServiceBase


class ButtonBase(ServiceBase):  # pylint: disable=abstract-method
    """Represents a button."""

    type = 'button'
    is_pushed = Characteristic('bool', 'RN')
    push = Characteristic('void', 'WN')
    release = Characteristic('void', 'WN')
    event = Characteristic('string', 'N')

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.events = list()
        self.requested_special_events = set()

    @property
    def all_observables(self):
        return self.events + super().all_observables

    def on(self, name, *args, **kwargs):
        if name == 'click':
            event = ButtonClickEvent(self)
        elif name == 'hold':
            trigger = args[0] if len(args) else 0.5
            event = ButtonHoldEvent(self, trigger, kwargs.get('cancelling', True))
        else:
            return super().on(name, *args, **kwargs)
        self.events.append(event)
        self.subscribe_events()
        return event

    def subscribe_events(self):
        if not getattr(self, '_events_subscribed', False):
            self._events_subscribed = True
            self.event.subscribe(self._event_fired)

    async def _event_fired(self, value, **kwargs):
        if value == 'click':
            for event in (e for e in self.events if isinstance(e, ButtonClickEvent)):
                await event.trigger('click')
        elif value == 'hold':
            interval = kwargs.get('extra', {}).get('interval')
            for event in (e for e in self.events if isinstance(e, ButtonHoldEvent)):
                if event.interval == interval:
                    await event.trigger('hold', **kwargs)


class ButtonEvent(Observable, Serializable, Deserializable):
    def __init__(self, button):
        super().__init__()
        self.button = button

    async def observe(self):
        raise errors.NotSupportedError()

    def do(self, *args, **kwargs):
        if self not in self.button.requested_special_events:
            rule = OnDoRule(on=self, do=ButtonTriggerEvent(), trigger=OnDoRuleTrigger.ON_UPDATE)
            caspia.meadow.rules.handler.rule_handler(rule)
        # pylint: disable=no-member
        return Observable.do(self, *args, **kwargs)


class ButtonClickEvent(ButtonEvent):
    serialized_type = 'button-click-event'

    def serialize(self):
        return dict(type=type(self).serialized_type, name=self.button.name)

    @classmethod
    def deserialize(cls, data, context):
        get_service = context['get_service']
        service = get_service(data['name'])
        return service.on('click')

    def __hash__(self):
        return hash((type(self).serialized_type, self.button.name))

    def __eq__(self, other):
        return hash(self) == hash(other)

    def __str__(self):
        return f'{self.button}:click'

    def __repr__(self):
        return f'<Event \'click\' of {self.button}>'


class ButtonHoldEvent(ButtonEvent):
    def __init__(self, button, interval, cancelling):
        super().__init__(button)
        self.interval = interval
        self.cancelling = cancelling

    serialized_type = 'button-hold-event'

    def serialize(self):
        return {
            'type': type(self).serialized_type,
            'name': self.button.name,
            'interval': self.interval,
            'cancelling': self.cancelling,
        }

    @classmethod
    def deserialize(cls, data, context):
        get_service = context['get_service']
        service = get_service(data['name'])
        return service.on('hold', data['interval'], cancelling=data['cancelling'])

    def __hash__(self):
        return hash((type(self).serialized_type, self.button.name, self.interval, self.cancelling))

    def __eq__(self, other):
        return hash(self) == hash(other)

    def __str__(self):
        return f'{self.button}:hold({self.interval}, cancelling={self.cancelling})'

    def __repr__(self):
        return (f'<Event \'hold\' of {self.button}, interval:{self.interval}, '
                f'cancelling:{self.cancelling}>')


class ButtonTriggerEvent(Observer, Serializable, Deserializable):
    """
    This class is to be used as "do" within an "on-do" rule.

    For example let's say, we want to have some Button trigger Button.event notification
    on 'hold' with interval 0.5. We must create "on-do" rule requesting this notification.
    Then the on-do rule will loke something like this:

    .. code::

        OnDoRule(
            on=ButtonHoldEvent(my_button, 0.5, cancelling=True),
            do=ButtonTriggerEvent()
            )


    If this `OnDoRule` gets optimized by some gateway, it means, that the button is going to
        send `Button.event` notification on the specified event. In this example, the `Button.event`
        notification is going to have value 'hold' and metadata of the notification will contain the
        `interval=0.5`.
    """

    serialized_type = 'button-trigger-event'

    def serialize(self):
        return dict(type=type(self).serialized_type)

    @classmethod
    def deserialize(cls, data, context):
        return ButtonTriggerEvent()

    async def on_next(self, value, **kwargs):
        pass

    async def on_error(self, error, **kwargs):
        pass

    def __repr__(self):
        return '<ButtonTriggerEvent>'