Allow using SX1509 and MCP23017 IO expander for buttons

klippy/extras/mmu_i2c_inputs.py

@@ -0,0 +1,122 @@
+# Simulated I2C Inputs support
+#
+# Copyright (C) 2024  Ivan Dubrov <[email protected]>
+#
+# This file may be distributed under the terms of the GNU GPLv3 license.
+import pins
+from . import bus
+
+# Used to ignore acks from the buttons code
+class MmuI2cInputsSendAck():
+    def send(self, data=(), minclock=0, reqclock=0):
+        return
+
+class MmuI2cInputs(object):
+    def __init__(self, config):
+        self._printer = config.get_printer()
+        name = config.get_name().split()[1]
+        i2c = bus.MCU_I2C_from_config(config, default_speed=400000)
+        mcu = i2c.get_mcu()
+        mcu.register_config_callback(self._build_inputs_config)
+        # Input pins support
+        self._init_cmds = []
+        self._restart_cmds = []
+        self._config_cmds = []
+        self._config_callbacks = []
+        self._oid_to_pins = []
+        self._oid_to_callbacks = []
+        self._ack_count = 0
+
+        # Interrupt pin triggers button events
+        interrupt_pin = config.get('interrupt_pin', None)
+        if interrupt_pin is not None:
+            buttons = self._printer.load_object(config, 'buttons')
+            buttons.register_button_push(interrupt_pin, self._send_buttons_state)
+        self._printer.register_event_handler("klippy:ready", self._ready)
+    def _ready(self):
+        self._send_buttons_state(self._printer.get_reactor().monotonic())
+    def _build_inputs_config(self):
+        # Build config commands
+        for cb in self._config_callbacks:
+            cb()
+
+        # Interpret the commands to handle the buttons
+        for cmdlist in (self._config_cmds, self._init_cmds):
+            for cmd in cmdlist:
+                self._interpret_init_cmd(cmd)
+    def add_config_cmd(self, cmd, is_init=False, on_restart=False):
+        if is_init:
+            self._init_cmds.append(cmd)
+        elif on_restart:
+            self._restart_cmds.append(cmd)
+        else:
+            self._config_cmds.append(cmd)
+    def _interpret_init_cmd(self, cmd):
+        parts = cmd.split()
+        name = parts[0]
+        argparts = dict(arg.split('=', 1) for arg in parts[1:])
+        if name == "config_buttons":
+            # Setup button pos to pin mapping
+            oid = int(argparts['oid'])
+            count = int(argparts['button_count'])
+            self._oid_to_pins[oid] = [-1] * count
+        elif name == "buttons_add":
+            # Setup pin in the button pos to pin mapping
+            oid = int(argparts['oid'])
+            pos = int(argparts['pos'])
+            pull_up = int(argparts['pull_up'])
+            if argparts['pin'][0:4] != "PIN_":
+                raise pins.error("Wrong pin: %s!" % (argparts['pin'][0:4]))
+            pin = int(argparts['pin'][4:])
+            self._oid_to_pins[oid][pos] = pin
+            self.setup_input_pin(pin, pull_up)
+        elif name == "buttons_query":
+            # FIXME: What should we do here?
+            pass
+        else:
+            raise pins.error("Command is not supported by I2C inputs: %s" % name)
+    def register_config_callback(self, cb):
+        self._config_callbacks.append(cb)
+    def create_oid(self):
+        self._oid_to_pins.append([])
+        self._oid_to_callbacks.append(None)
+        return len(self._oid_to_pins) - 1
+    def alloc_command_queue(self):
+        return None
+    def get_query_slot(self, oid):
+        return 0
+    def seconds_to_clock(self, time):
+        return 0
+    def register_response(self, callback, name, oid):
+        if name != "buttons_state":
+            raise pins.error("I2C inputs only supports buttons_state response callback")
+        self._oid_to_callbacks[oid] = callback
+    def lookup_command(self, msgformat, cq=None):
+        parts = msgformat.split()
+        name = parts[0]
+        if name != "buttons_ack":
+            raise pins.error("I2C inputs does not support '%s' command" % msgformat)
+        return MmuI2cInputsSendAck()
+    def setup_input_pin(self, pin_idx, pullup):
+        # Must be overridden to handle setup of the I2C input pin
+        pass
+    def read_input_pins(self):
+        return 0
+    def _send_buttons_state(self, eventtime):
+        pins = self.read_input_pins()
+
+        # Send updated buttons state through the callbacks
+        self._ack_count = (self._ack_count + 1) & 0xff
+        for oid, cb in enumerate(self._oid_to_callbacks):
+            state = 0
+            for i, pin_idx in enumerate(self._oid_to_pins[oid]):
+                state |= ((pins >> pin_idx) & 1) << i
+
+            if cb is not None:
+                params = {
+                    'ack_count': self._ack_count,
+                    'oid': oid,
+                    'state': [state],
+                    '#receive_time': eventtime
+                }
+                cb(params)

klippy/extras/mmu_mcp23017.py

@@ -0,0 +1,131 @@
+# MCP23017 Extra
+#
+# Copyright (C) 2024  Ivan Dubrov <[email protected]>
+#
+# This file may be distributed under the terms of the GNU GPLv3 license.
+from . import bus, mmu_i2c_inputs
+
+# Registers
+REG_IODIR = 0x00
+REG_IPOL = 0x02
+REG_GPINTEN = 0x04
+REG_DEFVAL = 0x06
+REG_INTCON = 0x08
+REG_IOCON = 0x0A
+REG_GPPU = 0x0C
+REG_INTF = 0x0E
+REG_INTCAP = 0x10
+REG_GPIO = 0x12
+REG_OLAT = 0x14
+
+class MmuMCP23017(mmu_i2c_inputs.MmuI2cInputs):
+    def __init__(self, config):
+        super().__init__(config)
+        self._printer = config.get_printer()
+        self._name = config.get_name().split()[1]
+        self._i2c = bus.MCU_I2C_from_config(config, default_speed=400000)
+        self._ppins = self._printer.lookup_object("pins")
+        self._ppins.register_chip("mmu_mcp23017_" + self._name, self)
+        self._mcu = self._i2c.get_mcu()
+        self._mcu.register_config_callback(self._build_config)
+        self._oid = self._i2c.get_oid()
+        self._last_clock = 0
+        # Set up registers default values
+        self.reg_dict = {
+            # FIXME: direction
+            REG_IODIR: 0xFFFF,
+            REG_IPOL: 0,
+            REG_GPINTEN: 0,
+            REG_DEFVAL: 0,
+            # FIXME: interrup
+            REG_INTCON: 0,
+            # FIXME: pullup
+            REG_GPPU: 0,
+            REG_GPIO: 0,
+        }
+    def _build_config(self):
+        # Reset the defalut configuration
+        # MIRROR = 1, The INT pins are internally connected
+        # ODR = 1, Configures the INT pin as an open-drain output
+        self._mcu.add_config_cmd("i2c_write oid=%d data=%02x%02x" % (self._oid, REG_IOCON, 0x44))
+        # Transfer all regs with their initial cached state
+        for _reg, _data in self.reg_dict.items():
+            self._mcu.add_config_cmd("i2c_write oid=%d data=%02x%02x%02x" % (
+                self._oid, _reg, _data & 0xFF, (_data >> 8) & 0xFF), is_init=True)
+
+    def setup_input_pin(self, pin_idx, pullup):
+        if pullup == 1:
+            self.set_bits_in_register(REG_GPPU, 1 << pin_idx)
+        elif pullup == -1:
+            raise self._ppins.error("Can not pulldown MCP23017 pins")
+        self.set_bits_in_register(REG_IODIR, 1 << pin_idx)
+        self.set_bits_in_register(REG_GPINTEN, 1 << pin_idx)
+    def read_input_pins(self):
+        params = self._i2c.i2c_read([REG_GPIO], 2)
+        response = bytearray(params['response'])
+        return (response[1] << 8) | response[0]
+#     def setup_pin(self, pin_type, pin_params):
+#         if pin_type == 'digital_out' and pin_params['pin'][0:4] == "PIN_":
+#             return MCP23017_digital_out(self, pin_params)
+#         raise pins.error("Wrong pin or incompatible type: %s with type %s! " % (
+#             pin_params['pin'][0:4], pin_type))
+#     def get_mcu(self):
+#         return self._mcu
+    def get_oid(self):
+        return self._oid
+    def clear_bits_in_register(self, reg, bitmask):
+        if reg in self.reg_dict:
+            self.reg_dict[reg] &= ~(bitmask)
+    def set_bits_in_register(self, reg, bitmask):
+        if reg in self.reg_dict:
+            self.reg_dict[reg] |= bitmask
+    def set_register(self, reg, value):
+        if reg in self.reg_dict:
+            self.reg_dict[reg] = value
+    def send_register(self, reg, print_time):
+        data = [reg & 0xFF, self.reg_dict[reg] & 0xFF, (self.reg_dict[reg] >> 8) & 0xFF]
+        clock = self._mcu.print_time_to_clock(print_time)
+        self._i2c.i2c_write(data, minclock=self._last_clock, reqclock=clock)
+        self._last_clock = clock
+
+# class MCP23017_digital_out(object):
+#     def __init__(self, mcp23017, pin_params):
+#         self._mcp23017 = mcp23017
+#         self._mcu = mcp23017.get_mcu()
+#         self._sxpin = int(pin_params['pin'].split('_')[1])
+#         self._bitmask = 1 << self._sxpin
+#         self._pin = pin_params['pin']
+#         self._invert = pin_params['invert']
+#         self._mcu.register_config_callback(self._build_config)
+#         self._start_value = self._shutdown_value = self._invert
+#         self._max_duration = 2.
+#         self._set_cmd = self._clear_cmd = None
+#         # Set direction to output
+#         self._mcp23017.clear_bits_in_register(REG_DIR, self._bitmask)
+#     def _build_config(self):
+#         if self._max_duration:
+#             raise pins.error("MCP23017 pins are not suitable for heaters")
+#     def get_mcu(self):
+#         return self._mcu
+#     def setup_max_duration(self, max_duration):
+#         self._max_duration = max_duration
+#     def setup_start_value(self, start_value, shutdown_value):
+#         self._start_value = (not not start_value) ^ self._invert
+#         self._shutdown_value = self._invert
+#         # We need to set the start value here so the register is
+#         # updated before the MCP23017 class writes it.
+#         if self._start_value:
+#             self._mcp23017.set_bits_in_register(REG_DATA, self._bitmask)
+#         else:
+#             self._mcp23017.clear_bits_in_register(REG_DATA, self._bitmask)
+#     def set_digital(self, print_time, value):
+#         if int(value) ^ self._invert:
+#             self._mcp23017.set_bits_in_register(REG_DATA, self._bitmask)
+#         else:
+#             self._mcp23017.clear_bits_in_register(REG_DATA, self._bitmask)
+#         self._mcp23017.send_register(REG_DATA, print_time)
+#     def set_pwm(self, print_time, value, cycle_time=None):
+#         self.set_digital(print_time, value >= 0.5)
+
+def load_config_prefix(config):
+    return MmuMCP23017(config)