Source code for caspia.pan.reactive.presence

from caspia import reactive

_presence = None


def register_presence(presence_service):
    global _presence
    _presence = presence_service


def get_presence():
    if _presence is None:
        raise RuntimeError('Presence service not set. Use register_presence()')
    return _presence


def user_state(user):
    return reactive.lambda_(lambda state: state.get(user, 'unknown'), get_presence().state)


def present(user):
    def _present(state):
        if state == 'present':
            return True
        if state != 'unknown':
            return False
        else:
            raise reactive.errors.ValueNotReady('Presence state of user %s is not known' % user)

    return reactive.lambda_(_present, user_state(user))


def away(user):
    def _away(state):
        if state == 'away':
            return True
        elif state != 'unknown':
            return False
        else:
            raise reactive.errors.ValueNotReady('Presence state of user %s is not known' % user)

    return reactive.lambda_(_away, user_state(user))


def everyone_away(users=None):
    def _everyone_away(state):
        if users is not None:
            state = {user: state.get(user, 'unknown') for user in users}
        if any(state[user] == 'present' for user in state):
            return False
        elif all(state[user] == 'away' for user in state):
            return True
        else:
            raise reactive.errors.ValueNotReady('Presence does not hold sufficient information')

    return reactive.lambda_(_everyone_away, get_presence().state)


def someone_present(users=None):
    # pylint: disable=singleton-comparison
    return everyone_away(users) == False


def user_transition(user):
    filtered = reactive.filter(lambda state: state in ['present', 'away'],
                               user_state(user),
                               mode='none')
    return reactive.impuls(reactive.transition(filtered))


def leaving_of(user):
    return reactive.filter(lambda transition: transition == ('present', 'away'),
                           user_transition(user))


def arrival_of(user):
    return reactive.filter(lambda transition: transition == ('away', 'present'),
                           user_transition(user))