Source code for caspia.meadow.observers

from caspia.meadow.serialization import Deserializable, Serializable
from caspia.meadow.services import Characteristic
from caspia.reactive import FunctionObserver, Observer

__all__ = ('CharacteristicObserver', 'FunctionObserver')


class CharacteristicObserver(Observer, Serializable, Deserializable):
    @staticmethod
    @Observer.register_creator
    def creator(*args, **kwargs):
        if len(args) >= 1 and isinstance(args[0], Characteristic):
            value = args[1] if len(args) > 1 else None
            return CharacteristicObserver(args[0], value)
        return None

    def __init__(self, characteristic, value=None):
        self.characteristic = characteristic
        self.value = value

    async def on_next(self, value, **kwargs):
        await self.characteristic.write(self.value)

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

    serialized_type = 'characteristic-write'

    @classmethod
    def deserialize(cls, data, context):
        get_service = context['get_service']
        service = get_service(data['name'])
        characteristic = service[data['characteristic']]
        return CharacteristicObserver(characteristic, data.get('value'))

    def serialize(self):
        return {
            'type': type(self).serialized_type,
            'name': self.characteristic.service.name,
            'characteristic': self.characteristic.name,
            'value': self.value,
        }

    def __hash__(self):
        return hash((self.characteristic.service.name, self.characteristic.name, self.value))

    def __str__(self):
        if self.value is not None:
            return f'{str(self.characteristic)}={self.value}'
        else:
            return f'{str(self.characteristic)}()'

    def __repr__(self):
        return (f'<CharacteristicObserver characteristic={self.characteristic}'
                f' value={self.value}')

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