path: root/klippy/extras/buttons.py

# Support for button detection and callbacks
#
# Copyright (C) 2018-2023  Kevin O'Connor <kevin@koconnor.net>
#
# This file may be distributed under the terms of the GNU GPLv3 license.
import logging


######################################################################
# Button state tracking
######################################################################

QUERY_TIME = 0.002
RETRANSMIT_COUNT = 50


class MCU_buttons:
    def __init__(self, printer, mcu):
        self.reactor = printer.get_reactor()
        self.mcu = mcu
        self.mcu.register_config_callback(self.build_config)
        self.pin_list = []
        self.callbacks = []
        self.invert = self.last_button = 0
        self.ack_cmd = None
        self.ack_count = 0

    def setup_buttons(self, pins, callback):
        mask = 0
        shift = len(self.pin_list)
        for pin_params in pins:
            if pin_params["invert"]:
                self.invert |= 1 << len(self.pin_list)
            mask |= 1 << len(self.pin_list)
            self.pin_list.append((pin_params["pin"], pin_params["pullup"]))
        self.callbacks.append((mask, shift, callback))

    def build_config(self):
        if not self.pin_list:
            return
        self.oid = self.mcu.create_oid()
        self.mcu.add_config_cmd(
            "config_buttons oid=%d button_count=%d" % (self.oid, len(self.pin_list))
        )
        for i, (pin, pull_up) in enumerate(self.pin_list):
            self.mcu.add_config_cmd(
                "buttons_add oid=%d pos=%d pin=%s pull_up=%d"
                % (self.oid, i, pin, pull_up),
                is_init=True,
            )
        cmd_queue = self.mcu.alloc_command_queue()
        self.ack_cmd = self.mcu.lookup_command(
            "buttons_ack oid=%c count=%c", cq=cmd_queue
        )
        clock = self.mcu.get_query_slot(self.oid)
        rest_ticks = self.mcu.seconds_to_clock(QUERY_TIME)
        self.mcu.add_config_cmd(
            "buttons_query oid=%d clock=%d"
            " rest_ticks=%d retransmit_count=%d invert=%d"
            % (self.oid, clock, rest_ticks, RETRANSMIT_COUNT, self.invert),
            is_init=True,
        )
        self.mcu.register_response(self.handle_buttons_state, "buttons_state", self.oid)

    def handle_buttons_state(self, params):
        # Expand the message ack_count from 8-bit
        ack_count = self.ack_count
        ack_diff = (params["ack_count"] - ack_count) & 0xFF
        ack_diff -= (ack_diff & 0x80) << 1
        msg_ack_count = ack_count + ack_diff
        # Determine new buttons
        buttons = bytearray(params["state"])
        new_count = msg_ack_count + len(buttons) - self.ack_count
        if new_count <= 0:
            return
        new_buttons = buttons[-new_count:]
        # Send ack to MCU
        self.ack_cmd.send([self.oid, new_count])
        self.ack_count += new_count
        # Invoke callbacks with this event in main thread
        btime = params["#receive_time"]
        for button in new_buttons:
            button ^= self.invert
            changed = button ^ self.last_button
            self.last_button = button
            for mask, shift, callback in self.callbacks:
                if changed & mask:
                    state = (button & mask) >> shift
                    self.reactor.register_async_callback(
                        (lambda et, c=callback, bt=btime, s=state: c(bt, s))
                    )


######################################################################
# ADC button tracking
######################################################################

ADC_REPORT_TIME = 0.015
ADC_DEBOUNCE_TIME = 0.025
ADC_SAMPLE_TIME = 0.001
ADC_SAMPLE_COUNT = 6


class MCU_ADC_buttons:
    def __init__(self, printer, pin, pullup):
        self.reactor = printer.get_reactor()
        self.buttons = []
        self.last_button = None
        self.last_pressed = None
        self.last_debouncetime = 0
        self.pullup = pullup
        self.pin = pin
        self.min_value = 999999999999.9
        self.max_value = 0.0
        ppins = printer.lookup_object("pins")
        self.mcu_adc = ppins.setup_pin("adc", self.pin)
        self.mcu_adc.setup_adc_sample(ADC_SAMPLE_TIME, ADC_SAMPLE_COUNT)
        self.mcu_adc.setup_adc_callback(ADC_REPORT_TIME, self.adc_callback)
        query_adc = printer.lookup_object("query_adc")
        query_adc.register_adc("adc_button:" + pin.strip(), self.mcu_adc)

    def setup_button(self, min_value, max_value, callback):
        self.min_value = min(self.min_value, min_value)
        self.max_value = max(self.max_value, max_value)
        self.buttons.append((min_value, max_value, callback))

    def adc_callback(self, read_time, read_value):
        adc = max(0.00001, min(0.99999, read_value))
        value = self.pullup * adc / (1.0 - adc)

        # Determine button pressed
        btn = None
        if self.min_value <= value <= self.max_value:
            for i, (min_value, max_value, cb) in enumerate(self.buttons):
                if min_value < value < max_value:
                    btn = i
                    break

        # If the button changed, due to noise or pressing:
        if btn != self.last_button:
            # reset the debouncing timer
            self.last_debouncetime = read_time

        # button debounce check & new button pressed
        if (
            (read_time - self.last_debouncetime) >= ADC_DEBOUNCE_TIME
            and self.last_button == btn
            and self.last_pressed != btn
        ):
            # release last_pressed
            if self.last_pressed is not None:
                self.call_button(self.last_pressed, False)
                self.last_pressed = None
            if btn is not None:
                self.call_button(btn, True)
                self.last_pressed = btn

        self.last_button = btn

    def call_button(self, button, state):
        minval, maxval, callback = self.buttons[button]
        self.reactor.register_async_callback((lambda e, cb=callback, s=state: cb(e, s)))


######################################################################
# Rotary Encoders
######################################################################


# Rotary encoder handler https://github.com/brianlow/Rotary
# Copyright 2011 Ben Buxton (bb@cactii.net).
# Licenced under the GNU GPL Version 3.
class BaseRotaryEncoder:
    R_START = 0x0
    R_DIR_CW = 0x10
    R_DIR_CCW = 0x20
    R_DIR_MSK = 0x30

    def __init__(self, cw_callback, ccw_callback):
        self.cw_callback = cw_callback
        self.ccw_callback = ccw_callback
        self.encoder_state = self.R_START

    def encoder_callback(self, eventtime, state):
        es = self.ENCODER_STATES[self.encoder_state & 0xF][state & 0x3]
        self.encoder_state = es
        if es & self.R_DIR_MSK == self.R_DIR_CW:
            self.cw_callback(eventtime)
        elif es & self.R_DIR_MSK == self.R_DIR_CCW:
            self.ccw_callback(eventtime)


class FullStepRotaryEncoder(BaseRotaryEncoder):
    R_CW_FINAL = 0x1
    R_CW_BEGIN = 0x2
    R_CW_NEXT = 0x3
    R_CCW_BEGIN = 0x4
    R_CCW_FINAL = 0x5
    R_CCW_NEXT = 0x6

    # Use the full-step state table (emits a code at 00 only)
    ENCODER_STATES = (
        # R_START
        (BaseRotaryEncoder.R_START, R_CW_BEGIN, R_CCW_BEGIN, BaseRotaryEncoder.R_START),
        # R_CW_FINAL
        (
            R_CW_NEXT,
            BaseRotaryEncoder.R_START,
            R_CW_FINAL,
            BaseRotaryEncoder.R_START | BaseRotaryEncoder.R_DIR_CW,
        ),
        # R_CW_BEGIN
        (R_CW_NEXT, R_CW_BEGIN, BaseRotaryEncoder.R_START, BaseRotaryEncoder.R_START),
        # R_CW_NEXT
        (R_CW_NEXT, R_CW_BEGIN, R_CW_FINAL, BaseRotaryEncoder.R_START),
        # R_CCW_BEGIN
        (R_CCW_NEXT, BaseRotaryEncoder.R_START, R_CCW_BEGIN, BaseRotaryEncoder.R_START),
        # R_CCW_FINAL
        (
            R_CCW_NEXT,
            R_CCW_FINAL,
            BaseRotaryEncoder.R_START,
            BaseRotaryEncoder.R_START | BaseRotaryEncoder.R_DIR_CCW,
        ),
        # R_CCW_NEXT
        (R_CCW_NEXT, R_CCW_FINAL, R_CCW_BEGIN, BaseRotaryEncoder.R_START),
    )


class HalfStepRotaryEncoder(BaseRotaryEncoder):
    # Use the half-step state table (emits a code at 00 and 11)
    R_CCW_BEGIN = 0x1
    R_CW_BEGIN = 0x2
    R_START_M = 0x3
    R_CW_BEGIN_M = 0x4
    R_CCW_BEGIN_M = 0x5

    ENCODER_STATES = (
        # R_START (00)
        (R_START_M, R_CW_BEGIN, R_CCW_BEGIN, BaseRotaryEncoder.R_START),
        # R_CCW_BEGIN
        (
            R_START_M | BaseRotaryEncoder.R_DIR_CCW,
            BaseRotaryEncoder.R_START,
            R_CCW_BEGIN,
            BaseRotaryEncoder.R_START,
        ),
        # R_CW_BEGIN
        (
            R_START_M | BaseRotaryEncoder.R_DIR_CW,
            R_CW_BEGIN,
            BaseRotaryEncoder.R_START,
            BaseRotaryEncoder.R_START,
        ),
        # R_START_M (11)
        (R_START_M, R_CCW_BEGIN_M, R_CW_BEGIN_M, BaseRotaryEncoder.R_START),
        # R_CW_BEGIN_M
        (
            R_START_M,
            R_START_M,
            R_CW_BEGIN_M,
            BaseRotaryEncoder.R_START | BaseRotaryEncoder.R_DIR_CW,
        ),
        # R_CCW_BEGIN_M
        (
            R_START_M,
            R_CCW_BEGIN_M,
            R_START_M,
            BaseRotaryEncoder.R_START | BaseRotaryEncoder.R_DIR_CCW,
        ),
    )


class DebounceButton:
    def __init__(self, config, button_action):
        self.printer = config.get_printer()
        self.reactor = self.printer.get_reactor()
        self.button_action = button_action
        self.debounce_delay = config.getfloat("debounce_delay", 0.0, minval=0.0)
        self.logical_state = None
        self.physical_state = None
        self.latest_eventtime = None

    def button_handler(self, eventtime, state):
        self.physical_state = state
        self.latest_eventtime = eventtime
        # if there would be no state transition, ignore the event:
        if self.logical_state == self.physical_state:
            return
        trigger_time = eventtime + self.debounce_delay
        self.reactor.register_callback(self._debounce_event, trigger_time)

    def _debounce_event(self, eventtime):
        # if there would be no state transition, ignore the event:
        if self.logical_state == self.physical_state:
            return
        # if there were more recent events, they supersede this one:
        if (eventtime - self.debounce_delay) < self.latest_eventtime:
            return
        # enact state transition and trigger action
        self.logical_state = self.physical_state
        self.button_action(self.latest_eventtime, self.logical_state)


######################################################################
# Button registration code
######################################################################


class PrinterButtons:
    def __init__(self, config):
        self.printer = config.get_printer()
        self.printer.load_object(config, "query_adc")
        self.mcu_buttons = {}
        self.adc_buttons = {}

    def register_adc_button(self, pin, min_val, max_val, pullup, callback):
        adc_buttons = self.adc_buttons.get(pin)
        if adc_buttons is None:
            self.adc_buttons[pin] = adc_buttons = MCU_ADC_buttons(
                self.printer, pin, pullup
            )
        adc_buttons.setup_button(min_val, max_val, callback)

    def register_debounce_button(self, pin, callback, config):
        debounce = DebounceButton(config, callback)
        return self.register_buttons([pin], debounce.button_handler)

    def register_debounce_adc_button(
        self, pin, min_val, max_val, pullup, callback, config
    ):
        debounce = DebounceButton(config, callback)
        return self.register_adc_button(
            pin, min_val, max_val, pullup, debounce.button_handler
        )

    def register_adc_button_push(self, pin, min_val, max_val, pullup, callback):
        def helper(eventtime, state, callback=callback):
            if state:
                callback(eventtime)

        self.register_adc_button(pin, min_val, max_val, pullup, helper)

    def register_buttons(self, pins, callback):
        # Parse pins
        ppins = self.printer.lookup_object("pins")
        mcu = mcu_name = None
        pin_params_list = []
        for pin in pins:
            pin_params = ppins.lookup_pin(pin, can_invert=True, can_pullup=True)
            if mcu is not None and pin_params["chip"] != mcu:
                raise ppins.error("button pins must be on same mcu")
            mcu = pin_params["chip"]
            mcu_name = pin_params["chip_name"]
            pin_params_list.append(pin_params)
        # Register pins and callback with the appropriate MCU
        mcu_buttons = self.mcu_buttons.get(mcu_name)
        if mcu_buttons is None or len(mcu_buttons.pin_list) + len(pin_params_list) > 8:
            self.mcu_buttons[mcu_name] = mcu_buttons = MCU_buttons(self.printer, mcu)
        mcu_buttons.setup_buttons(pin_params_list, callback)

    def register_rotary_encoder(
        self, pin1, pin2, cw_callback, ccw_callback, steps_per_detent
    ):
        if steps_per_detent == 2:
            re = HalfStepRotaryEncoder(cw_callback, ccw_callback)
        elif steps_per_detent == 4:
            re = FullStepRotaryEncoder(cw_callback, ccw_callback)
        else:
            raise self.printer.config_error(
                "%d steps per detent not supported" % steps_per_detent
            )
        self.register_buttons([pin1, pin2], re.encoder_callback)

    def register_button_push(self, pin, callback):
        def helper(eventtime, state, callback=callback):
            if state:
                callback(eventtime)

        self.register_buttons([pin], helper)


def load_config(config):
    return PrinterButtons(config)