Source code for caspia.gateway.rules.activate_button

# pylint: disable=line-too-long,too-many-return-statements,too-many-branches
from caspia.gateway import services as gateway_services
from caspia.meadow import observers, rules, services
from caspia.node import components, pollen_action


def activate_on_do_rule(rule: rules.OnDoRule, context) -> bool:
    # on button push, release
    is_btn_char = isinstance(rule.on, services.Characteristic) and isinstance(
        rule.on.service, gateway_services.Button)
    is_btn_char = is_btn_char and (rule.on.name in ['push', 'release'])
    if is_btn_char:
        return activate_rule_on_button_characteristic_notification(rule, context)
    # on button hold, click
    is_btn_evt = isinstance(rule.on, (services.ButtonHoldEvent, services.ButtonClickEvent))
    if is_btn_evt:
        return activate_rule_on_button_event(rule, context)
    return None


def activate_rule_on_button_characteristic_notification(rule, context):
    assert isinstance(rule, rules.OnDoRule)
    is_btn_char = isinstance(rule.on, services.Characteristic)
    is_btn_char = is_btn_char and isinstance(rule.on.service, gateway_services.Button)
    is_btn_char = is_btn_char and (rule.on.name in ['push', 'release'])
    if not is_btn_char:
        return False

    pollen_data = create_pollen_data_for_button_observer(rule.do, rule.on, context)
    if pollen_data is None:
        return False

    button_s = rule.on.service
    if rule.on.name == 'push':
        trigger = components.Button.EventConfig.TRIGGER_PUSH
    elif rule.on.name == 'release':
        trigger = components.Button.EventConfig.TRIGGER_RELEASE
    button_s.button_c.config.add_event(create_button_event(trigger, pollen_data))
    return True


def activate_rule_on_button_event(rule, context):
    """Activate rule, where rule.on is ButtonClickEvent or ButtonHoldEvent."""
    pollen_data = create_pollen_data_for_button_observer(observer=rule.do,
                                                         observable=rule.on,
                                                         context=context)
    if pollen_data is None:
        return None
    button_s = rule.on.button
    if isinstance(rule.on, services.ButtonClickEvent):
        event = create_button_event(0, pollen_data, cancelling=True)
    else:  # ButtonHoldEvent
        event = create_button_event(rule.on.interval, pollen_data, cancelling=rule.on.cancelling)
    button_s.button_c.config.add_event(event)
    return True


def create_button_event(trigger, pollen_data, cancelling=True):
    ptype, address, data = pollen_data
    if ptype == 'request':
        evt_type = components.Button.EventConfig.TYPE_REQUEST
    elif ptype == 'broadcast':
        evt_type = components.Button.EventConfig.TYPE_BROADCAST
    else:
        raise ValueError('Could not create button event with type %r' % ptype)
    return components.Button.EventConfig(trigger=trigger,
                                         evt_type=evt_type,
                                         addr=address,
                                         data=data,
                                         canceling=cancelling)


def create_pollen_data_for_button_observer(observer, observable, context):
    """Return (type, address, data), for example ('request', 0x1C, b'test') or None if not supported."""
    if isinstance(observer, observers.CharacteristicObserver):
        characteristic = observer.characteristic
        lightgroup_behavior = context.get_lightgroup_behavior(characteristic.service, create=False)
        if lightgroup_behavior:
            if characteristic.name == 'toggle':
                return pollen_action.build(components.relay.Relay.group_toggle,
                                           lightgroup_behavior.identifier)
            elif characteristic.name == 'is_on' and observer.value:
                return pollen_action.build(components.relay.Relay.group_on,
                                           lightgroup_behavior.identifier)
            elif characteristic.name == 'is_on' and not observer.value:
                return pollen_action.build(components.relay.Relay.group_off,
                                           lightgroup_behavior.identifier)
        elif isinstance(characteristic.service, gateway_services.RelayBasedServiceMixin):
            if characteristic.name == 'toggle':
                return pollen_action.build(components.relay.Relay.toggle,
                                           characteristic.service.relay_c)
            elif characteristic.name == 'is_on' and observer.value:
                return pollen_action.build(components.relay.Relay.on,
                                           characteristic.service.relay_c)
            elif characteristic.name == 'is_on' and not observer.value:
                return pollen_action.build(components.relay.Relay.off,
                                           characteristic.service.relay_c)
        elif isinstance(characteristic.service, gateway_services.Blinds):
            if characteristic.name == 'move_up':
                return pollen_action.build(components.blinds.Blinds.move_up,
                                           characteristic.service.blinds_c)
            elif characteristic.name == 'move_down':
                return pollen_action.build(components.blinds.Blinds.move_down,
                                           characteristic.service.blinds_c)
            elif characteristic.name == 'stop_movement':
                return pollen_action.build(components.blinds.Blinds.stop_movement,
                                           characteristic.service.blinds_c)
            elif characteristic.name == 'calibrate':
                return pollen_action.build(components.blinds.Blinds.calibrate,
                                           characteristic.service.blinds_c)
        return None
    elif isinstance(observer, services.button.ButtonTriggerEvent):
        button_c = observable.button.button_c
        if isinstance(observable, services.button.ButtonClickEvent):
            evt_type = components.Button.ButtonEvent.EVENT_CLICK
            interval = None
        elif isinstance(observable, services.button.ButtonHoldEvent):
            evt_type = components.Button.ButtonEvent.EVENT_HOLD
            interval = observable.interval
        else:
            raise ValueError('Invalid observable class %r' % observable)
        data = components.Button.ButtonEvent.create_raw_broadcast_data(button_c.identifier,
                                                                       evt_type,
                                                                       interval=interval)
        return ('broadcast', components.button.Button.type | 0x100, data)
    return None