Initial

kmk/modules/__init__.py

@@ -0,0 +1,43 @@
+class InvalidExtensionEnvironment(Exception):
+    pass
+
+
+class Module:
+    '''
+    Modules differ from extensions in that they not only can read the state, but
+    are allowed to modify the state. The will be loaded on boot, and are not
+    allowed to be unloaded as they are required to continue functioning in a
+    consistant manner.
+    '''
+
+    # The below methods should be implemented by subclasses
+
+    def during_bootup(self, keyboard):
+        raise NotImplementedError
+
+    def before_matrix_scan(self, keyboard):
+        '''
+        Return value will be injected as an extra matrix update
+        '''
+        raise NotImplementedError
+
+    def after_matrix_scan(self, keyboard):
+        '''
+        Return value will be replace matrix update if supplied
+        '''
+        raise NotImplementedError
+
+    def process_key(self, keyboard, key, is_pressed, int_coord):
+        return key
+
+    def before_hid_send(self, keyboard):
+        raise NotImplementedError
+
+    def after_hid_send(self, keyboard):
+        raise NotImplementedError
+
+    def on_powersave_enable(self, keyboard):
+        raise NotImplementedError
+
+    def on_powersave_disable(self, keyboard):
+        raise NotImplementedError

kmk/modules/adns9800.py

@@ -0,0 +1,241 @@
+import busio
+import digitalio
+import microcontroller
+
+import time
+
+from kmk.modules import Module
+from kmk.modules.adns9800_firmware import firmware
+from kmk.modules.mouse_keys import PointingDevice
+
+
+class REG:
+    Product_ID = 0x0
+    Revision_ID = 0x1
+    MOTION = 0x2
+    DELTA_X_L = 0x3
+    DELTA_X_H = 0x4
+    DELTA_Y_L = 0x5
+    DELTA_Y_H = 0x6
+    SQUAL = 0x7
+    PIXEL_SUM = 0x8
+    Maximum_Pixel = 0x9
+    Minimum_Pixel = 0xA
+    Shutter_Lower = 0xB
+    Shutter_Upper = 0xC
+    Frame_Period_Lower = 0xD
+    Frame_Period_Upper = 0xE
+    Configuration_I = 0xF
+    Configuration_II = 0x10
+    Frame_Capture = 0x12
+    SROM_Enable = 0x13
+    Run_Downshift = 0x14
+    Rest1_Rate = 0x15
+    Rest1_Downshift = 0x16
+    Rest2_Rate = 0x17
+    Rest2_Downshift = 0x18
+    Rest3_Rate = 0x19
+    Frame_Period_Max_Bound_Lower = 0x1A
+    Frame_Period_Max_Bound_Upper = 0x1B
+    Frame_Period_Min_Bound_Lower = 0x1C
+    Frame_Period_Min_Bound_Upper = 0x1D
+    Shutter_Max_Bound_Lower = 0x1E
+    Shutter_Max_Bound_Upper = 0x1F
+    LASER_CTRL0 = 0x20
+    Observation = 0x24
+    Data_Out_Lower = 0x25
+    Data_Out_Upper = 0x26
+    SROM_ID = 0x2A
+    Lift_Detection_Thr = 0x2E
+    Configuration_V = 0x2F
+    Configuration_IV = 0x39
+    Power_Up_Reset = 0x3A
+    Shutdown = 0x3B
+    Inverse_Product_ID = 0x3F
+    Snap_Angle = 0x42
+    Motion_Burst = 0x50
+    SROM_Load_Burst = 0x62
+    Pixel_Burst = 0x64
+
+
+class ADNS9800(Module):
+    tswr = tsww = 120
+    tsrw = tsrr = 20
+    tsrad = 100
+    tbexit = 1
+    baud = 2000000
+    cpol = 1
+    cpha = 1
+    DIR_WRITE = 0x80
+    DIR_READ = 0x7F
+
+    def __init__(self, cs, sclk, miso, mosi, invert_x=False, invert_y=False):
+        self.pointing_device = PointingDevice()
+        self.cs = digitalio.DigitalInOut(cs)
+        self.cs.direction = digitalio.Direction.OUTPUT
+        self.spi = busio.SPI(clock=sclk, MOSI=mosi, MISO=miso)
+        self.invert_x = invert_x
+        self.invert_y = invert_y
+
+    def adns_start(self):
+        self.cs.value = False
+
+    def adns_stop(self):
+        self.cs.value = True
+
+    def adns_write(self, reg, data):
+        while not self.spi.try_lock():
+            pass
+        try:
+            self.spi.configure(baudrate=self.baud, polarity=self.cpol, phase=self.cpha)
+            self.adns_start()
+            self.spi.write(bytes([reg | self.DIR_WRITE, data]))
+        finally:
+            self.spi.unlock()
+            self.adns_stop()
+
+    def adns_read(self, reg):
+        result = bytearray(1)
+        while not self.spi.try_lock():
+            pass
+        try:
+            self.spi.configure(baudrate=self.baud, polarity=self.cpol, phase=self.cpha)
+            self.adns_start()
+            self.spi.write(bytes([reg & self.DIR_READ]))
+            microcontroller.delay_us(self.tsrad)
+            self.spi.readinto(result)
+        finally:
+            self.spi.unlock()
+            self.adns_stop()
+
+        return result[0]
+
+    def adns_upload_srom(self):
+        while not self.spi.try_lock():
+            pass
+        try:
+            self.spi.configure(baudrate=self.baud, polarity=self.cpol, phase=self.cpha)
+            self.adns_start()
+            self.spi.write(bytes([REG.SROM_Load_Burst | self.DIR_WRITE]))
+            for b in firmware:
+                self.spi.write(bytes([b]))
+        finally:
+            self.spi.unlock()
+            self.adns_stop()
+
+    def delta_to_int(self, high, low):
+        comp = (high << 8) | low
+        if comp & 0x8000:
+            return (-1) * (0xFFFF + 1 - comp)
+        return comp
+
+    def adns_read_motion(self):
+        result = bytearray(14)
+        while not self.spi.try_lock():
+            pass
+        try:
+            self.spi.configure(baudrate=self.baud, polarity=self.cpol, phase=self.cpha)
+            self.adns_start()
+            self.spi.write(bytes([REG.Motion_Burst & self.DIR_READ]))
+            microcontroller.delay_us(self.tsrad)
+            self.spi.readinto(result)
+        finally:
+            self.spi.unlock()
+            self.adns_stop()
+        microcontroller.delay_us(self.tbexit)
+        self.adns_write(REG.MOTION, 0x0)
+        return result
+
+    def during_bootup(self, keyboard):
+
+        self.adns_write(REG.Power_Up_Reset, 0x5A)
+        time.sleep(0.1)
+        self.adns_read(REG.MOTION)
+        microcontroller.delay_us(self.tsrr)
+        self.adns_read(REG.DELTA_X_L)
+        microcontroller.delay_us(self.tsrr)
+        self.adns_read(REG.DELTA_X_H)
+        microcontroller.delay_us(self.tsrr)
+        self.adns_read(REG.DELTA_Y_L)
+        microcontroller.delay_us(self.tsrr)
+        self.adns_read(REG.DELTA_Y_H)
+        microcontroller.delay_us(self.tsrw)
+
+        self.adns_write(REG.Configuration_IV, 0x2)
+        microcontroller.delay_us(self.tsww)
+        self.adns_write(REG.SROM_Enable, 0x1D)
+        microcontroller.delay_us(1000)
+        self.adns_write(REG.SROM_Enable, 0x18)
+        microcontroller.delay_us(self.tsww)
+
+        self.adns_upload_srom()
+        microcontroller.delay_us(2000)
+
+        laser_ctrl0 = self.adns_read(REG.LASER_CTRL0)
+        microcontroller.delay_us(self.tsrw)
+        self.adns_write(REG.LASER_CTRL0, laser_ctrl0 & 0xF0)
+        microcontroller.delay_us(self.tsww)
+        self.adns_write(REG.Configuration_I, 0x10)
+        microcontroller.delay_us(self.tsww)
+
+        if keyboard.debug_enabled:
+            print('ADNS: Product ID ', hex(self.adns_read(REG.Product_ID)))
+            microcontroller.delay_us(self.tsrr)
+            print('ADNS: Revision ID ', hex(self.adns_read(REG.Revision_ID)))
+            microcontroller.delay_us(self.tsrr)
+            print('ADNS: SROM ID ', hex(self.adns_read(REG.SROM_ID)))
+            microcontroller.delay_us(self.tsrr)
+            if self.adns_read(REG.Observation) & 0x20:
+                print('ADNS: Sensor is running SROM')
+            else:
+                print('ADNS: Error! Sensor is not runnin SROM!')
+
+        return
+
+    def before_matrix_scan(self, keyboard):
+        motion = self.adns_read_motion()
+        if motion[0] & 0x80:
+            delta_x = self.delta_to_int(motion[3], motion[2])
+            delta_y = self.delta_to_int(motion[5], motion[4])
+
+            if self.invert_x:
+                delta_x *= -1
+            if self.invert_y:
+                delta_y *= -1
+
+            if delta_x < 0:
+                self.pointing_device.report_x[0] = (delta_x & 0xFF) | 0x80
+            else:
+                self.pointing_device.report_x[0] = delta_x & 0xFF
+
+            if delta_y < 0:
+                self.pointing_device.report_y[0] = (delta_y & 0xFF) | 0x80
+            else:
+                self.pointing_device.report_y[0] = delta_y & 0xFF
+
+            if keyboard.debug_enabled:
+                print('Delta: ', delta_x, ' ', delta_y)
+            self.pointing_device.hid_pending = True
+
+        if self.pointing_device.hid_pending:
+            keyboard._hid_helper.hid_send(self.pointing_device._evt)
+            self.pointing_device.hid_pending = False
+            self.pointing_device.report_x[0] = 0
+            self.pointing_device.report_y[0] = 0
+
+        return
+
+    def after_matrix_scan(self, keyboard):
+        return
+
+    def before_hid_send(self, keyboard):
+        return
+
+    def after_hid_send(self, keyboard):
+        return
+
+    def on_powersave_enable(self, keyboard):
+        return
+
+    def on_powersave_disable(self, keyboard):
+        return

kmk/modules/capsword.py

@@ -0,0 +1,99 @@
+from kmk.keys import FIRST_KMK_INTERNAL_KEY, KC, ModifierKey, make_key
+from kmk.modules import Module
+
+
+class CapsWord(Module):
+    # default timeout is 8000
+    # alphabets, numbers and few more keys will not disable capsword
+    def __init__(self, timeout=8000):
+        self._alphabets = range(KC.A.code, KC.Z.code)
+        self._numbers = range(KC.N1.code, KC.N0.code)
+        self.keys_ignored = [
+            KC.MINS,
+            KC.BSPC,
+            KC.UNDS,
+        ]
+        self._timeout_key = False
+        self._cw_active = False
+        self.timeout = timeout
+        make_key(
+            names=(
+                'CAPSWORD',
+                'CW',
+            ),
+            on_press=self.cw_pressed,
+        )
+
+    def during_bootup(self, keyboard):
+        return
+
+    def before_matrix_scan(self, keyboard):
+        return
+
+    def process_key(self, keyboard, key, is_pressed, int_coord):
+        if self._cw_active and key != KC.CW:
+            continue_cw = False
+            # capitalize alphabets
+            if key.code in self._alphabets:
+                continue_cw = True
+                keyboard.process_key(KC.LSFT, is_pressed)
+            elif (
+                key.code in self._numbers
+                or isinstance(key, ModifierKey)
+                or key in self.keys_ignored
+                or key.code
+                >= FIRST_KMK_INTERNAL_KEY  # user defined keys are also ignored
+            ):
+                continue_cw = True
+            # requests and cancels existing timeouts
+            if is_pressed:
+                if continue_cw:
+                    self.discard_timeout(keyboard)
+                    self.request_timeout(keyboard)
+                else:
+                    self.process_timeout()
+
+        return key
+
+    def before_hid_send(self, keyboard):
+        return
+
+    def after_hid_send(self, keyboard):
+        return
+
+    def on_powersave_enable(self, keyboard):
+        return
+
+    def on_powersave_disable(self, keyboard):
+        return
+
+    def after_matrix_scan(self, keyboard):
+        return
+
+    def process_timeout(self):
+        self._cw_active = False
+        self._timeout_key = False
+
+    def request_timeout(self, keyboard):
+        if self._cw_active:
+            if self.timeout:
+                self._timeout_key = keyboard.set_timeout(
+                    self.timeout, lambda: self.process_timeout()
+                )
+
+    def discard_timeout(self, keyboard):
+        if self._timeout_key:
+            if self.timeout:
+                keyboard.cancel_timeout(self._timeout_key)
+            self._timeout_key = False
+
+    def cw_pressed(self, key, keyboard, *args, **kwargs):
+        # enables/disables capsword
+        if key == KC.CW:
+            if not self._cw_active:
+                self._cw_active = True
+                self.discard_timeout(keyboard)
+                self.request_timeout(keyboard)
+            else:
+                self.discard_timeout(keyboard)
+                self.process_timeout()

kmk/modules/cg_swap.py

@@ -0,0 +1,70 @@
+from kmk.keys import KC, ModifierKey, make_key
+from kmk.modules import Module
+
+
+class CgSwap(Module):
+    # default cg swap is disabled, can be eanbled too if needed
+    def __init__(self, cg_swap_enable=False):
+        self.cg_swap_enable = cg_swap_enable
+        self._cg_mapping = {
+            KC.LCTL: KC.LGUI,
+            KC.RCTL: KC.RGUI,
+            KC.LGUI: KC.LCTL,
+            KC.RGUI: KC.RCTL,
+        }
+        make_key(
+            names=('CG_SWAP',),
+        )
+        make_key(
+            names=('CG_NORM',),
+        )
+        make_key(
+            names=('CG_TOGG',),
+        )
+
+    def during_bootup(self, keyboard):
+        return
+
+    def matrix_detected_press(self, keyboard):
+        return keyboard.matrix_update is None
+
+    def before_matrix_scan(self, keyboard):
+        return
+
+    def process_key(self, keyboard, key, is_pressed, int_coord):
+        if is_pressed:
+            # enables or disables or toggles cg swap
+            if key == KC.CG_SWAP:
+                self.cg_swap_enable = True
+            elif key == KC.CG_NORM:
+                self.cg_swap_enable = False
+            elif key == KC.CG_TOGG:
+                if not self.cg_swap_enable:
+                    self.cg_swap_enable = True
+                else:
+                    self.cg_swap_enable = False
+            # performs cg swap
+            if (
+                self.cg_swap_enable
+                and key not in (KC.CG_SWAP, KC.CG_NORM, KC.CG_TOGG)
+                and isinstance(key, ModifierKey)
+                and key in self._cg_mapping
+            ):
+                key = self._cg_mapping.get(key)
+
+        return key
+
+    def before_hid_send(self, keyboard):
+        return
+
+    def after_hid_send(self, keyboard):
+        return
+
+    def on_powersave_enable(self, keyboard):
+        return
+
+    def on_powersave_disable(self, keyboard):
+        return
+
+    def after_matrix_scan(self, keyboard):
+        return

kmk/modules/combos.py

@@ -0,0 +1,206 @@
+import kmk.handlers.stock as handlers
+from kmk.keys import make_key
+from kmk.modules import Module
+
+
+class Combo:
+    timeout = 50
+    per_key_timeout = False
+    _timeout = None
+    _remaining = []
+
+    def __init__(self, match, result, timeout=None, per_key_timeout=None):
+        '''
+        match: tuple of keys (KC.A, KC.B)
+        result: key KC.C
+        '''
+        self.match = match
+        self.result = result
+        if timeout:
+            self.timeout = timeout
+        if per_key_timeout:
+            self.per_key_timeout = per_key_timeout
+
+    def matches(self, key):
+        raise NotImplementedError
+
+    def reset(self):
+        self._remaining = list(self.match)
+
+
+class Chord(Combo):
+    def matches(self, key):
+        try:
+            self._remaining.remove(key)
+            return True
+        except ValueError:
+            return False
+
+
+class Sequence(Combo):
+    timeout = 1000
+    per_key_timeout = True
+
+    def matches(self, key):
+        try:
+            return key == self._remaining.pop(0)
+        except IndexError:
+            return False
+
+
+class Combos(Module):
+    def __init__(self, combos=[]):
+        self.combos = combos
+        self._active = []
+        self._matching = []
+        self._reset = set()
+        self._key_buffer = []
+
+        make_key(
+            names=('LEADER',),
+            on_press=handlers.passthrough,
+            on_release=handlers.passthrough,
+        )
+
+    def during_bootup(self, keyboard):
+        self.reset(keyboard)
+
+    def before_matrix_scan(self, keyboard):
+        return
+
+    def after_matrix_scan(self, keyboard):
+        return
+
+    def before_hid_send(self, keyboard):
+        return
+
+    def after_hid_send(self, keyboard):
+        return
+
+    def on_powersave_enable(self, keyboard):
+        return
+
+    def on_powersave_disable(self, keyboard):
+        return
+
+    def process_key(self, keyboard, key, is_pressed, int_coord):
+        if is_pressed:
+            return self.on_press(keyboard, key, int_coord)
+        else:
+            return self.on_release(keyboard, key, int_coord)
+
+    def on_press(self, keyboard, key, int_coord):
+        # refill potential matches from timed-out matches
+        if not self._matching:
+            self._matching = list(self._reset)
+            self._reset = set()
+
+        # filter potential matches
+        for combo in self._matching.copy():
+            if combo.matches(key):
+                continue
+            self._matching.remove(combo)
+            if combo._timeout:
+                keyboard.cancel_timeout(combo._timeout)
+            combo._timeout = keyboard.set_timeout(
+                combo.timeout, lambda c=combo: self.reset_combo(keyboard, c)
+            )
+
+        if self._matching:
+            # At least one combo matches current key: append key to buffer.
+            self._key_buffer.append((int_coord, key, True))
+            key = None
+
+            # Start or reset individual combo timeouts.
+            for combo in self._matching:
+                if combo._timeout:
+                    if combo.per_key_timeout:
+                        keyboard.cancel_timeout(combo._timeout)
+                    else:
+                        continue
+                combo._timeout = keyboard.set_timeout(
+                    combo.timeout, lambda c=combo: self.on_timeout(keyboard, c)
+                )
+        else:
+            # There's no matching combo: send and reset key buffer
+            self.send_key_buffer(keyboard)
+            self._key_buffer = []
+            key = keyboard._find_key_in_map(int_coord)
+
+        return key
+
+    def on_release(self, keyboard, key, int_coord):
+        for combo in self._active:
+            if key in combo.match:
+                # Deactivate combo if it matches current key.
+                self.deactivate(keyboard, combo)
+                self.reset_combo(keyboard, combo)
+                key = combo.result
+                break
+
+        # Don't propagate key-release events for keys that have been buffered.
+        # Append release events only if corresponding press is in buffer.
+        else:
+            pressed = self._key_buffer.count((int_coord, key, True))
+            released = self._key_buffer.count((int_coord, key, False))
+            if (pressed - released) > 0:
+                self._key_buffer.append((int_coord, key, False))
+                key = None
+
+        return key
+
+    def on_timeout(self, keyboard, combo):
+        # If combo reaches timeout and has no remaining keys, activate it;
+        # else, drop it from the match list.
+        combo._timeout = None
+        self._matching.remove(combo)
+
+        if not combo._remaining:
+            self.activate(keyboard, combo)
+            if any([not pressed for (int_coord, key, pressed) in self._key_buffer]):
+                # At least one of the combo keys has already been released:
+                # "tap" the combo result.
+                keyboard._send_hid()
+                self.deactivate(keyboard, combo)
+            self.reset(keyboard)
+            self._key_buffer = []
+        else:
+            if not self._matching:
+                # This was the last pending combo: flush key buffer.
+                self.send_key_buffer(keyboard)
+                self._key_buffer = []
+            self.reset_combo(keyboard, combo)
+
+    def send_key_buffer(self, keyboard):
+        for (int_coord, key, is_pressed) in self._key_buffer:
+            try:
+                new_key = keyboard._coordkeys_pressed[int_coord]
+            except KeyError:
+                new_key = None
+            if new_key is None:
+                new_key = keyboard._find_key_in_map(int_coord)
+
+            keyboard._coordkeys_pressed[int_coord] = new_key
+
+            keyboard.process_key(new_key, is_pressed)
+            keyboard._send_hid()
+
+    def activate(self, keyboard, combo):
+        combo.result.on_press(keyboard)
+        self._active.append(combo)
+
+    def deactivate(self, keyboard, combo):
+        combo.result.on_release(keyboard)
+        self._active.remove(combo)
+
+    def reset_combo(self, keyboard, combo):
+        combo.reset()
+        if combo._timeout is not None:
+            keyboard.cancel_timeout(combo._timeout)
+            combo._timeout = None
+        self._reset.add(combo)
+
+    def reset(self, keyboard):
+        self._matching = []
+        for combo in self.combos:
+            self.reset_combo(keyboard, combo)

kmk/modules/easypoint.py

@@ -0,0 +1,145 @@
+'''
+Extension handles usage of AS5013 by AMS
+'''
+
+from supervisor import ticks_ms
+
+from kmk.modules import Module
+from kmk.modules.mouse_keys import PointingDevice
+
+I2C_ADDRESS = 0x40
+I2X_ALT_ADDRESS = 0x41
+
+X = 0x10
+Y_RES_INT = 0x11
+
+XP = 0x12
+XN = 0x13
+YP = 0x14
+YN = 0x15
+
+M_CTRL = 0x2B
+T_CTRL = 0x2D
+
+Y_OFFSET = 17
+X_OFFSET = 7
+
+DEAD_X = 5
+DEAD_Y = 5
+
+
+class Easypoint(Module):
+    '''Module handles usage of AS5013 by AMS'''
+
+    def __init__(
+        self,
+        i2c,
+        address=I2C_ADDRESS,
+        y_offset=Y_OFFSET,
+        x_offset=X_OFFSET,
+        dead_x=DEAD_X,
+        dead_y=DEAD_Y,
+    ):
+        self._i2c_address = address
+        self._i2c_bus = i2c
+
+        # HID parameters
+        self.pointing_device = PointingDevice()
+        self.polling_interval = 20
+        self.last_tick = ticks_ms()
+
+        # Offsets for poor soldering
+        self.y_offset = y_offset
+        self.x_offset = x_offset
+
+        # Deadzone
+        self.dead_x = DEAD_X
+        self.dead_y = DEAD_Y
+
+    def during_bootup(self, keyboard):
+        return
+
+    def before_matrix_scan(self, keyboard):
+        '''
+        Return value will be injected as an extra matrix update
+        '''
+        now = ticks_ms()
+        if now - self.last_tick < self.polling_interval:
+            return
+        self.last_tick = now
+
+        x, y = self._read_raw_state()
+
+        # I'm a shit coder, so offset is handled in software side
+        s_x = self.getSignedNumber(x, 8) - self.x_offset
+        s_y = self.getSignedNumber(y, 8) - self.y_offset
+
+        # Evaluate Deadzone
+        if s_x in range(-self.dead_x, self.dead_x) and s_y in range(
+            -self.dead_y, self.dead_y
+        ):
+            # Within bounds, just die
+            return
+        else:
+            # Set the X/Y from easypoint
+            self.pointing_device.report_x[0] = x
+            self.pointing_device.report_y[0] = y
+
+            self.pointing_device.hid_pending = x != 0 or y != 0
+
+        return
+
+    def after_matrix_scan(self, keyboard):
+        return
+
+    def before_hid_send(self, keyboard):
+        return
+
+    def after_hid_send(self, keyboard):
+        if self.pointing_device.hid_pending:
+            keyboard._hid_helper.hid_send(self.pointing_device._evt)
+            self._clear_pending_hid()
+        return
+
+    def on_powersave_enable(self, keyboard):
+        return
+
+    def on_powersave_disable(self, keyboard):
+        return
+
+    def _clear_pending_hid(self):
+        self.pointing_device.hid_pending = False
+        self.pointing_device.report_x[0] = 0
+        self.pointing_device.report_y[0] = 0
+        self.pointing_device.report_w[0] = 0
+        self.pointing_device.button_status[0] = 0
+
+    def _read_raw_state(self):
+        '''Read data from AS5013'''
+        x, y = self._i2c_rdwr([X], length=2)
+        return x, y
+
+    def getSignedNumber(self, number, bitLength=8):
+        mask = (2 ** bitLength) - 1
+        if number & (1 << (bitLength - 1)):
+            return number | ~mask
+        else:
+            return number & mask
+
+    def _i2c_rdwr(self, data, length=1):
+        '''Write and optionally read I2C data.'''
+        while not self._i2c_bus.try_lock():
+            pass
+
+        try:
+            if length > 0:
+                result = bytearray(length)
+                self._i2c_bus.writeto_then_readfrom(
+                    self._i2c_address, bytes(data), result
+                )
+                return result
+            else:
+                self._i2c_bus.writeto(self._i2c_address, bytes(data))
+            return []
+        finally:
+            self._i2c_bus.unlock()

kmk/modules/encoder.py

@@ -0,0 +1,289 @@
+# See docs/encoder.md for how to use
+
+import busio
+import digitalio
+from supervisor import ticks_ms
+
+from kmk.modules import Module
+
+# NB : not using rotaryio as it requires the pins to be consecutive
+
+
+class BaseEncoder:
+
+    VELOCITY_MODE = True
+
+    def __init__(self, is_inverted=False):
+
+        self.is_inverted = is_inverted
+
+        self._state = None
+        self._direction = None
+        self._pos = 0
+        self._button_state = True
+        self._button_held = None
+        self._velocity = 0
+
+        self._movement = 0
+        self._timestamp = ticks_ms()
+
+        # callback functions on events. Need to be defined externally
+        self.on_move_do = None
+        self.on_button_do = None
+
+    def get_state(self):
+        return {
+            'direction': self.is_inverted and -self._direction or self._direction,
+            'position': self.is_inverted and -self._pos or self._pos,
+            'is_pressed': not self._button_state,
+            'velocity': self._velocity,
+        }
+
+        # Called in a loop to refresh encoder state
+
+    def update_state(self):
+        # Rotation events
+        new_state = (self.pin_a.get_value(), self.pin_b.get_value())
+
+        if new_state != self._state:
+            # it moves !
+            self._movement += 1
+            # false / false and true / true are common half steps
+            # looking on the step just before helps determining
+            # the direction
+            if new_state[0] == new_state[1] and self._state[0] != self._state[1]:
+                if new_state[1] == self._state[0]:
+                    self._direction = 1
+                else:
+                    self._direction = -1
+
+            # when the encoder settles on a position (every 2 steps)
+            if new_state == (True, True):
+                if self._movement > 2:
+                    # 1 full step is 4 movements, however, when rotated quickly,
+                    # some steps may be missed. This makes it behaves more
+                    # naturally
+                    real_movement = round(self._movement / 4)
+                    self._pos += self._direction * real_movement
+                    if self.on_move_do is not None:
+                        for i in range(real_movement):
+                            self.on_move_do(self.get_state())
+                # Reinit to properly identify new movement
+                self._movement = 0
+                self._direction = 0
+
+            self._state = new_state
+
+        # Velocity
+        self.velocity_event()
+
+        # Button event
+        self.button_event()
+
+    def velocity_event(self):
+        if self.VELOCITY_MODE:
+            new_timestamp = ticks_ms()
+            self._velocity = new_timestamp - self._timestamp
+            self._timestamp = new_timestamp
+
+    def button_event(self):
+        raise NotImplementedError('subclasses must override button_event()!')
+
+    # return knob velocity as milliseconds between position changes (detents)
+    # for backwards compatibility
+    def vel_report(self):
+        # print(self._velocity)
+        return self._velocity
+
+
+class GPIOEncoder(BaseEncoder):
+    def __init__(self, pin_a, pin_b, pin_button=None, is_inverted=False):
+        super().__init__(is_inverted)
+
+        self.pin_a = EncoderPin(pin_a)
+        self.pin_b = EncoderPin(pin_b)
+        self.pin_button = (
+            EncoderPin(pin_button, button_type=True) if pin_button is not None else None
+        )
+
+        self._state = (self.pin_a.get_value(), self.pin_b.get_value())
+
+    def button_event(self):
+        if self.pin_button:
+            new_button_state = self.pin_button.get_value()
+            if new_button_state != self._button_state:
+                self._button_state = new_button_state
+                if self.on_button_do is not None:
+                    self.on_button_do(self.get_state())
+
+
+class EncoderPin:
+    def __init__(self, pin, button_type=False):
+        self.pin = pin
+        self.button_type = button_type
+        self.prepare_pin()
+
+    def prepare_pin(self):
+        if self.pin is not None:
+            self.io = digitalio.DigitalInOut(self.pin)
+            self.io.direction = digitalio.Direction.INPUT
+            self.io.pull = digitalio.Pull.UP
+        else:
+            self.io = None
+
+    def get_value(self):
+        return self.io.value
+
+
+class I2CEncoder(BaseEncoder):
+    def __init__(self, i2c, address, is_inverted=False):
+
+        try:
+            from adafruit_seesaw import digitalio, neopixel, rotaryio, seesaw
+        except ImportError:
+            print('seesaw missing')
+            return
+
+        super().__init__(is_inverted)
+
+        self.seesaw = seesaw.Seesaw(i2c, address)
+
+        # Check for correct product
+
+        seesaw_product = (self.seesaw.get_version() >> 16) & 0xFFFF
+        if seesaw_product != 4991:
+            print('Wrong firmware loaded?  Expected 4991')
+
+        self.encoder = rotaryio.IncrementalEncoder(self.seesaw)
+        self.seesaw.pin_mode(24, self.seesaw.INPUT_PULLUP)
+        self.switch = digitalio.DigitalIO(self.seesaw, 24)
+        self.pixel = neopixel.NeoPixel(self.seesaw, 6, 1)
+
+        self._state = self.encoder.position
+
+    def update_state(self):
+
+        # Rotation events
+        new_state = self.encoder.position
+        if new_state != self._state:
+            # it moves !
+            self._movement += 1
+            # false / false and true / true are common half steps
+            # looking on the step just before helps determining
+            # the direction
+            if self.encoder.position > self._state:
+                self._direction = 1
+            else:
+                self._direction = -1
+            self._state = new_state
+            self.on_move_do(self.get_state())
+
+        # Velocity
+        self.velocity_event()
+
+        # Button events
+        self.button_event()
+
+    def button_event(self):
+        if not self.switch.value and not self._button_held:
+            # Pressed
+            self._button_held = True
+            if self.on_button_do is not None:
+                self.on_button_do(self.get_state())
+
+        if self.switch.value and self._button_held:
+            # Released
+            self._button_held = False
+
+    def get_state(self):
+        return {
+            'direction': self.is_inverted and -self._direction or self._direction,
+            'position': self._state,
+            'is_pressed': not self.switch.value,
+            'is_held': self._button_held,
+            'velocity': self._velocity,
+        }
+
+
+class EncoderHandler(Module):
+    def __init__(self):
+        self.encoders = []
+        self.pins = None
+        self.map = None
+
+    def on_runtime_enable(self, keyboard):
+        return
+
+    def on_runtime_disable(self, keyboard):
+        return
+
+    def during_bootup(self, keyboard):
+        if self.pins and self.map:
+            for idx, pins in enumerate(self.pins):
+                try:
+                    # Check for busio.I2C
+                    if isinstance(pins[0], busio.I2C):
+                        new_encoder = I2CEncoder(*pins)
+
+                    # Else fall back to GPIO
+                    else:
+                        gpio_pins = pins[:3]
+                        new_encoder = GPIOEncoder(*gpio_pins)
+
+                    # In our case, we need to define keybord and encoder_id for callbacks
+                    new_encoder.on_move_do = lambda x, bound_idx=idx: self.on_move_do(
+                        keyboard, bound_idx, x
+                    )
+                    new_encoder.on_button_do = (
+                        lambda x, bound_idx=idx: self.on_button_do(
+                            keyboard, bound_idx, x
+                        )
+                    )
+                    self.encoders.append(new_encoder)
+                except Exception as e:
+                    print(e)
+        return
+
+    def on_move_do(self, keyboard, encoder_id, state):
+        if self.map:
+            layer_id = keyboard.active_layers[0]
+            # if Left, key index 0 else key index 1
+            if state['direction'] == -1:
+                key_index = 0
+            else:
+                key_index = 1
+            key = self.map[layer_id][encoder_id][key_index]
+            keyboard.tap_key(key)
+
+    def on_button_do(self, keyboard, encoder_id, state):
+        if state['is_pressed'] is True:
+            layer_id = keyboard.active_layers[0]
+            key = self.map[layer_id][encoder_id][2]
+            keyboard.tap_key(key)
+
+    def before_matrix_scan(self, keyboard):
+        '''
+        Return value will be injected as an extra matrix update
+        '''
+        for encoder in self.encoders:
+            encoder.update_state()
+
+        return keyboard
+
+    def after_matrix_scan(self, keyboard):
+        '''
+        Return value will be replace matrix update if supplied
+        '''
+        return
+
+    def before_hid_send(self, keyboard):
+        return
+
+    def after_hid_send(self, keyboard):
+        return
+
+    def on_powersave_enable(self, keyboard):
+        return
+
+    def on_powersave_disable(self, keyboard):
+        return

kmk/modules/holdtap.py

@@ -0,0 +1,164 @@
+from micropython import const
+
+from kmk.modules import Module
+
+
+class ActivationType:
+    PRESSED = const(0)
+    RELEASED = const(1)
+    HOLD_TIMEOUT = const(2)
+    INTERRUPTED = const(3)
+
+
+class HoldTapKeyState:
+    def __init__(self, timeout_key, *args, **kwargs):
+        self.timeout_key = timeout_key
+        self.args = args
+        self.kwargs = kwargs
+        self.activated = ActivationType.PRESSED
+
+
+class HoldTap(Module):
+    tap_time = 300
+
+    def __init__(self):
+        self.key_buffer = []
+        self.key_states = {}
+
+    def during_bootup(self, keyboard):
+        return
+
+    def before_matrix_scan(self, keyboard):
+        return
+
+    def after_matrix_scan(self, keyboard):
+        return
+
+    def process_key(self, keyboard, key, is_pressed, int_coord):
+        '''Handle holdtap being interrupted by another key press/release.'''
+        current_key = key
+        for key, state in self.key_states.items():
+            if key == current_key:
+                continue
+            if state.activated != ActivationType.PRESSED:
+                continue
+
+            # holdtap is interrupted by another key event.
+            if (is_pressed and not key.meta.tap_interrupted) or (
+                not is_pressed and key.meta.tap_interrupted and self.key_buffer
+            ):
+
+                keyboard.cancel_timeout(state.timeout_key)
+                self.key_states[key].activated = ActivationType.INTERRUPTED
+                self.ht_activate_on_interrupt(
+                    key, keyboard, *state.args, **state.kwargs
+                )
+
+                keyboard._send_hid()
+
+                self.send_key_buffer(keyboard)
+
+            # if interrupt on release: store interrupting keys until one of them
+            # is released.
+            if key.meta.tap_interrupted:
+                if is_pressed:
+                    self.key_buffer.append((int_coord, current_key))
+                    current_key = None
+
+        return current_key
+
+    def before_hid_send(self, keyboard):
+        return
+
+    def after_hid_send(self, keyboard):
+        return
+
+    def on_powersave_enable(self, keyboard):
+        return
+
+    def on_powersave_disable(self, keyboard):
+        return
+
+    def ht_pressed(self, key, keyboard, *args, **kwargs):
+        '''Do nothing yet, action resolves when key is released, timer expires or other key is pressed.'''
+        if key.meta.tap_time is None:
+            tap_time = self.tap_time
+        else:
+            tap_time = key.meta.tap_time
+        timeout_key = keyboard.set_timeout(
+            tap_time,
+            lambda: self.on_tap_time_expired(key, keyboard, *args, **kwargs),
+        )
+        self.key_states[key] = HoldTapKeyState(timeout_key, *args, **kwargs)
+        return keyboard
+
+    def ht_released(self, key, keyboard, *args, **kwargs):
+        '''On keyup, release mod or tap key.'''
+        if key in self.key_states:
+            state = self.key_states[key]
+            keyboard.cancel_timeout(state.timeout_key)
+            if state.activated == ActivationType.HOLD_TIMEOUT:
+                # release hold
+                self.ht_deactivate_hold(key, keyboard, *args, **kwargs)
+            elif state.activated == ActivationType.INTERRUPTED:
+                # release tap
+                self.ht_deactivate_on_interrupt(key, keyboard, *args, **kwargs)
+            elif state.activated == ActivationType.PRESSED:
+                # press and release tap because key released within tap time
+                self.ht_activate_tap(key, keyboard, *args, **kwargs)
+                keyboard.set_timeout(
+                    False,
+                    lambda: self.ht_deactivate_tap(key, keyboard, *args, **kwargs),
+                )
+                self.send_key_buffer(keyboard)
+            del self.key_states[key]
+        return keyboard
+
+    def on_tap_time_expired(self, key, keyboard, *args, **kwargs):
+        '''When tap time expires activate hold if key is still being pressed.
+        Remove key if ActivationType is RELEASED.'''
+        try:
+            state = self.key_states[key]
+        except KeyError:
+            if keyboard.debug_enabled:
+                print(f'HoldTap.on_tap_time_expired: no such key {key}')
+            return
+
+        if self.key_states[key].activated == ActivationType.PRESSED:
+            # press hold because timer expired after tap time
+            self.key_states[key].activated = ActivationType.HOLD_TIMEOUT
+            self.ht_activate_hold(key, keyboard, *args, **kwargs)
+            self.send_key_buffer(keyboard)
+        elif state.activated == ActivationType.RELEASED:
+            self.ht_deactivate_tap(key, keyboard, *args, **kwargs)
+            del self.key_states[key]
+
+    def send_key_buffer(self, keyboard):
+        for (int_coord, key) in self.key_buffer:
+            key.on_press(keyboard)
+            keyboard._send_hid()
+        self.key_buffer.clear()
+
+    def ht_activate_hold(self, key, keyboard, *args, **kwargs):
+        pass
+
+    def ht_deactivate_hold(self, key, keyboard, *args, **kwargs):
+        pass
+
+    def ht_activate_tap(self, key, keyboard, *args, **kwargs):
+        pass
+
+    def ht_deactivate_tap(self, key, keyboard, *args, **kwargs):
+        pass
+
+    def ht_activate_on_interrupt(self, key, keyboard, *args, **kwargs):
+        if key.meta.prefer_hold:
+            self.ht_activate_hold(key, keyboard, *args, **kwargs)
+        else:
+            self.ht_activate_tap(key, keyboard, *args, **kwargs)
+
+    def ht_deactivate_on_interrupt(self, key, keyboard, *args, **kwargs):
+        if key.meta.prefer_hold:
+            self.ht_deactivate_hold(key, keyboard, *args, **kwargs)
+        else:
+            self.ht_deactivate_tap(key, keyboard, *args, **kwargs)

kmk/modules/layers.py

@@ -0,0 +1,183 @@
+'''One layer isn't enough. Adds keys to get to more of them'''
+from micropython import const
+
+from kmk.key_validators import layer_key_validator
+from kmk.keys import make_argumented_key
+from kmk.modules.holdtap import ActivationType, HoldTap
+
+
+class LayerType:
+    '''Defines layer types to be passed on as on_press and on_release kwargs where needed'''
+
+    LT = const(0)
+    TT = const(1)
+
+
+def curry(fn, *args, **kwargs):
+    def curried(*fn_args, **fn_kwargs):
+        merged_args = args + fn_args
+        merged_kwargs = kwargs.copy()
+        merged_kwargs.update(fn_kwargs)
+        return fn(*merged_args, **merged_kwargs)
+
+    return curried
+
+
+class Layers(HoldTap):
+    '''Gives access to the keys used to enable the layer system'''
+
+    def __init__(self):
+        # Layers
+        super().__init__()
+        make_argumented_key(
+            validator=layer_key_validator,
+            names=('MO',),
+            on_press=self._mo_pressed,
+            on_release=self._mo_released,
+        )
+        make_argumented_key(
+            validator=layer_key_validator, names=('DF',), on_press=self._df_pressed
+        )
+        make_argumented_key(
+            validator=layer_key_validator,
+            names=('LM',),
+            on_press=self._lm_pressed,
+            on_release=self._lm_released,
+        )
+        make_argumented_key(
+            validator=layer_key_validator,
+            names=('LT',),
+            on_press=curry(self.ht_pressed, key_type=LayerType.LT),
+            on_release=curry(self.ht_released, key_type=LayerType.LT),
+        )
+        make_argumented_key(
+            validator=layer_key_validator, names=('TG',), on_press=self._tg_pressed
+        )
+        make_argumented_key(
+            validator=layer_key_validator, names=('TO',), on_press=self._to_pressed
+        )
+        make_argumented_key(
+            validator=curry(layer_key_validator, prefer_hold=True),
+            names=('TT',),
+            on_press=curry(self.ht_pressed, key_type=LayerType.TT),
+            on_release=curry(self.ht_released, key_type=LayerType.TT),
+        )
+
+    def process_key(self, keyboard, key, is_pressed, int_coord):
+        current_key = super().process_key(keyboard, key, is_pressed, int_coord)
+
+        for key, state in self.key_states.items():
+            if key == current_key:
+                continue
+
+            # on interrupt: key must be translated here, because it was asigned
+            # before the layer shift happend.
+            if state.activated == ActivationType.INTERRUPTED:
+                current_key = keyboard._find_key_in_map(int_coord)
+
+        return current_key
+
+    def send_key_buffer(self, keyboard):
+        for (int_coord, old_key) in self.key_buffer:
+            new_key = keyboard._find_key_in_map(int_coord)
+
+            # adding keys late to _coordkeys_pressed isn't pretty,
+            # but necessary to mitigate race conditions when multiple
+            # keys are pressed during a tap-interrupted hold-tap.
+            keyboard._coordkeys_pressed[int_coord] = new_key
+            new_key.on_press(keyboard)
+
+            keyboard._send_hid()
+
+        self.key_buffer.clear()
+
+    def _df_pressed(self, key, keyboard, *args, **kwargs):
+        '''
+        Switches the default layer
+        '''
+        keyboard.active_layers[-1] = key.meta.layer
+
+    def _mo_pressed(self, key, keyboard, *args, **kwargs):
+        '''
+        Momentarily activates layer, switches off when you let go
+        '''
+        keyboard.active_layers.insert(0, key.meta.layer)
+
+    @staticmethod
+    def _mo_released(key, keyboard, *args, **kwargs):
+        # remove the first instance of the target layer
+        # from the active list
+        # under almost all normal use cases, this will
+        # disable the layer (but preserve it if it was triggered
+        # as a default layer, etc.)
+        # this also resolves an issue where using DF() on a layer
+        # triggered by MO() and then defaulting to the MO()'s layer
+        # would result in no layers active
+        try:
+            del_idx = keyboard.active_layers.index(key.meta.layer)
+            del keyboard.active_layers[del_idx]
+        except ValueError:
+            pass
+
+    def _lm_pressed(self, key, keyboard, *args, **kwargs):
+        '''
+        As MO(layer) but with mod active
+        '''
+        keyboard.hid_pending = True
+        # Sets the timer start and acts like MO otherwise
+        keyboard.keys_pressed.add(key.meta.kc)
+        self._mo_pressed(key, keyboard, *args, **kwargs)
+
+    def _lm_released(self, key, keyboard, *args, **kwargs):
+        '''
+        As MO(layer) but with mod active
+        '''
+        keyboard.hid_pending = True
+        keyboard.keys_pressed.discard(key.meta.kc)
+        self._mo_released(key, keyboard, *args, **kwargs)
+
+    def _tg_pressed(self, key, keyboard, *args, **kwargs):
+        '''
+        Toggles the layer (enables it if not active, and vise versa)
+        '''
+        # See mo_released for implementation details around this
+        try:
+            del_idx = keyboard.active_layers.index(key.meta.layer)
+            del keyboard.active_layers[del_idx]
+        except ValueError:
+            keyboard.active_layers.insert(0, key.meta.layer)
+
+    def _to_pressed(self, key, keyboard, *args, **kwargs):
+        '''
+        Activates layer and deactivates all other layers
+        '''
+        keyboard.active_layers.clear()
+        keyboard.active_layers.insert(0, key.meta.layer)
+
+    def ht_activate_hold(self, key, keyboard, *args, **kwargs):
+        key_type = kwargs['key_type']
+        if key_type == LayerType.LT:
+            self._mo_pressed(key, keyboard, *args, **kwargs)
+        elif key_type == LayerType.TT:
+            self._tg_pressed(key, keyboard, *args, **kwargs)
+
+    def ht_deactivate_hold(self, key, keyboard, *args, **kwargs):
+        key_type = kwargs['key_type']
+        if key_type == LayerType.LT:
+            self._mo_released(key, keyboard, *args, **kwargs)
+        elif key_type == LayerType.TT:
+            self._tg_pressed(key, keyboard, *args, **kwargs)
+
+    def ht_activate_tap(self, key, keyboard, *args, **kwargs):
+        key_type = kwargs['key_type']
+        if key_type == LayerType.LT:
+            keyboard.hid_pending = True
+            keyboard.keys_pressed.add(key.meta.kc)
+        elif key_type == LayerType.TT:
+            self._tg_pressed(key, keyboard, *args, **kwargs)
+
+    def ht_deactivate_tap(self, key, keyboard, *args, **kwargs):
+        key_type = kwargs['key_type']
+        if key_type == LayerType.LT:
+            keyboard.hid_pending = True
+            keyboard.keys_pressed.discard(key.meta.kc)

kmk/modules/midi.py

@@ -0,0 +1,103 @@
+import adafruit_midi
+import usb_midi
+from adafruit_midi.control_change import ControlChange
+from adafruit_midi.note_off import NoteOff
+from adafruit_midi.note_on import NoteOn
+from adafruit_midi.pitch_bend import PitchBend
+from adafruit_midi.program_change import ProgramChange
+from adafruit_midi.start import Start
+from adafruit_midi.stop import Stop
+
+from kmk.keys import make_argumented_key
+from kmk.modules import Module
+
+
+class midiNoteValidator:
+    def __init__(self, note=69, velocity=64, channel=None):
+        self.note = note
+        self.velocity = velocity
+        self.channel = channel
+
+
+class MidiKeys(Module):
+    def __init__(self):
+        make_argumented_key(
+            names=('MIDI_CC',),
+            validator=ControlChange,
+            on_press=self.on_press,
+        )
+
+        make_argumented_key(
+            names=('MIDI_NOTE',),
+            validator=midiNoteValidator,
+            on_press=self.note_on,
+            on_release=self.note_off,
+        )
+
+        make_argumented_key(
+            names=('MIDI_PB',),
+            validator=PitchBend,
+            on_press=self.on_press,
+        )
+
+        make_argumented_key(
+            names=('MIDI_PC',),
+            validator=ProgramChange,
+            on_press=self.on_press,
+        )
+
+        make_argumented_key(
+            names=('MIDI_START',),
+            validator=Start,
+            on_press=self.on_press,
+        )
+
+        make_argumented_key(
+            names=('MIDI_STOP',),
+            validator=Stop,
+            on_press=self.on_press,
+        )
+
+        try:
+            self.midi = adafruit_midi.MIDI(midi_out=usb_midi.ports[1], out_channel=0)
+        except IndexError:
+            self.midi = None
+            # if debug_enabled:
+            print('No midi device found.')
+
+    def during_bootup(self, keyboard):
+        return None
+
+    def before_matrix_scan(self, keyboard):
+        return None
+
+    def after_matrix_scan(self, keyboard):
+        return None
+
+    def process_key(self, keyboard, key, is_pressed, int_coord):
+        return key
+
+    def before_hid_send(self, keyboard):
+        return None
+
+    def after_hid_send(self, keyboard):
+        return None
+
+    def on_powersave_enable(self, keyboard):
+        return None
+
+    def on_powersave_disable(self, keyboard):
+        return None
+
+    def send(self, message):
+        if self.midi:
+            self.midi.send(message)
+
+    def on_press(self, key, keyboard, *args, **kwargs):
+        self.send(key.meta)
+
+    def note_on(self, key, keyboard, *args, **kwargs):
+        self.send(NoteOn(key.meta.note, key.meta.velocity, channel=key.meta.channel))
+
+    def note_off(self, key, keyboard, *args, **kwargs):
+        self.send(NoteOff(key.meta.note, key.meta.velocity, channel=key.meta.channel))

kmk/modules/modtap.py

@@ -0,0 +1,27 @@
+import kmk.handlers.stock as handlers
+from kmk.key_validators import mod_tap_validator
+from kmk.keys import make_argumented_key
+from kmk.modules.holdtap import HoldTap
+
+
+class ModTap(HoldTap):
+    def __init__(self):
+        super().__init__()
+        make_argumented_key(
+            validator=mod_tap_validator,
+            names=('MT',),
+            on_press=self.ht_pressed,
+            on_release=self.ht_released,
+        )
+
+    def ht_activate_hold(self, key, keyboard, *args, **kwargs):
+        handlers.default_pressed(key.meta.mods, keyboard, None)
+
+    def ht_deactivate_hold(self, key, keyboard, *args, **kwargs):
+        handlers.default_released(key.meta.mods, keyboard, None)
+
+    def ht_activate_tap(self, key, keyboard, *args, **kwargs):
+        handlers.default_pressed(key.meta.kc, keyboard, None)
+
+    def ht_deactivate_tap(self, key, keyboard, *args, **kwargs):
+        handlers.default_released(key.meta.kc, keyboard, None)

kmk/modules/mouse_keys.py

@@ -0,0 +1,245 @@
+from supervisor import ticks_ms
+
+from kmk.hid import HID_REPORT_SIZES, HIDReportTypes
+from kmk.keys import make_key
+from kmk.modules import Module
+
+
+class PointingDevice:
+    MB_LMB = 1
+    MB_RMB = 2
+    MB_MMB = 4
+    _evt = bytearray(HID_REPORT_SIZES[HIDReportTypes.MOUSE] + 1)
+
+    def __init__(self):
+        self.key_states = {}
+        self.hid_pending = False
+        self.report_device = memoryview(self._evt)[0:1]
+        self.report_device[0] = HIDReportTypes.MOUSE
+        self.button_status = memoryview(self._evt)[1:2]
+        self.report_x = memoryview(self._evt)[2:3]
+        self.report_y = memoryview(self._evt)[3:4]
+        self.report_w = memoryview(self._evt)[4:]
+
+
+class MouseKeys(Module):
+    def __init__(self):
+        self.pointing_device = PointingDevice()
+        self._nav_key_activated = 0
+        self._up_activated = False
+        self._down_activated = False
+        self._left_activated = False
+        self._right_activated = False
+        self.max_speed = 10
+        self.ac_interval = 100  # Delta ms to apply acceleration
+        self._next_interval = 0  # Time for next tick interval
+        self.move_step = 1
+
+        make_key(
+            names=('MB_LMB',),
+            on_press=self._mb_lmb_press,
+            on_release=self._mb_lmb_release,
+        )
+        make_key(
+            names=('MB_MMB',),
+            on_press=self._mb_mmb_press,
+            on_release=self._mb_mmb_release,
+        )
+        make_key(
+            names=('MB_RMB',),
+            on_press=self._mb_rmb_press,
+            on_release=self._mb_rmb_release,
+        )
+        make_key(
+            names=('MW_UP',),
+            on_press=self._mw_up_press,
+            on_release=self._mw_up_release,
+        )
+        make_key(
+            names=(
+                'MW_DOWN',
+                'MW_DN',
+            ),
+            on_press=self._mw_down_press,
+            on_release=self._mw_down_release,
+        )
+        make_key(
+            names=('MS_UP',),
+            on_press=self._ms_up_press,
+            on_release=self._ms_up_release,
+        )
+        make_key(
+            names=(
+                'MS_DOWN',
+                'MS_DN',
+            ),
+            on_press=self._ms_down_press,
+            on_release=self._ms_down_release,
+        )
+        make_key(
+            names=(
+                'MS_LEFT',
+                'MS_LT',
+            ),
+            on_press=self._ms_left_press,
+            on_release=self._ms_left_release,
+        )
+        make_key(
+            names=(
+                'MS_RIGHT',
+                'MS_RT',
+            ),
+            on_press=self._ms_right_press,
+            on_release=self._ms_right_release,
+        )
+
+    def during_bootup(self, keyboard):
+        return
+
+    def matrix_detected_press(self, keyboard):
+        return keyboard.matrix_update is None
+
+    def before_matrix_scan(self, keyboard):
+        return
+
+    def after_matrix_scan(self, keyboard):
+        if self._nav_key_activated:
+            if self._next_interval <= ticks_ms():
+                # print("hello: ")
+                # print(ticks_ms())
+                self._next_interval = ticks_ms() + self.ac_interval
+                # print(self._next_interval)
+                if self.move_step < self.max_speed:
+                    self.move_step = self.move_step + 1
+                    if self._right_activated:
+                        self.pointing_device.report_x[0] = self.move_step
+                    if self._left_activated:
+                        self.pointing_device.report_x[0] = 0xFF & (0 - self.move_step)
+                    if self._up_activated:
+                        self.pointing_device.report_y[0] = 0xFF & (0 - self.move_step)
+                    if self._down_activated:
+                        self.pointing_device.report_y[0] = self.move_step
+                    # self.pointing_device.hid_pending = True
+        return
+
+    def before_hid_send(self, keyboard):
+        if self.pointing_device.hid_pending:
+            keyboard._hid_helper.hid_send(self.pointing_device._evt)
+        return
+
+    def after_hid_send(self, keyboard):
+        return
+
+    def on_powersave_enable(self, keyboard):
+        return
+
+    def on_powersave_disable(self, keyboard):
+        return
+
+    def _mb_lmb_press(self, key, keyboard, *args, **kwargs):
+        self.pointing_device.button_status[0] |= self.pointing_device.MB_LMB
+        self.pointing_device.hid_pending = True
+
+    def _mb_lmb_release(self, key, keyboard, *args, **kwargs):
+        self.pointing_device.button_status[0] &= ~self.pointing_device.MB_LMB
+        self.pointing_device.hid_pending = True
+
+    def _mb_mmb_press(self, key, keyboard, *args, **kwargs):
+        self.pointing_device.button_status[0] |= self.pointing_device.MB_MMB
+        self.pointing_device.hid_pending = True
+
+    def _mb_mmb_release(self, key, keyboard, *args, **kwargs):
+        self.pointing_device.button_status[0] &= ~self.pointing_device.MB_MMB
+        self.pointing_device.hid_pending = True
+
+    def _mb_rmb_press(self, key, keyboard, *args, **kwargs):
+        self.pointing_device.button_status[0] |= self.pointing_device.MB_RMB
+        self.pointing_device.hid_pending = True
+
+    def _mb_rmb_release(self, key, keyboard, *args, **kwargs):
+        self.pointing_device.button_status[0] &= ~self.pointing_device.MB_RMB
+        self.pointing_device.hid_pending = True
+
+    def _mw_up_press(self, key, keyboard, *args, **kwargs):
+        self.pointing_device.report_w[0] = self.move_step
+        self.pointing_device.hid_pending = True
+
+    def _mw_up_release(self, key, keyboard, *args, **kwargs):
+        self.pointing_device.report_w[0] = 0
+        self.pointing_device.hid_pending = True
+
+    def _mw_down_press(self, key, keyboard, *args, **kwargs):
+        self.pointing_device.report_w[0] = 0xFF
+        self.pointing_device.hid_pending = True
+
+    def _mw_down_release(self, key, keyboard, *args, **kwargs):
+        self.pointing_device.report_w[0] = 0
+        self.pointing_device.hid_pending = True
+
+    # Mouse movement
+    def _reset_next_interval(self):
+        if self._nav_key_activated == 1:
+            self._next_interval = ticks_ms() + self.ac_interval
+            self.move_step = 1
+
+    def _check_last(self):
+        if self._nav_key_activated == 0:
+            self.move_step = 1
+
+    def _ms_up_press(self, key, keyboard, *args, **kwargs):
+        self._nav_key_activated += 1
+        self._reset_next_interval()
+        self._up_activated = True
+        self.pointing_device.report_y[0] = 0xFF & (0 - self.move_step)
+        self.pointing_device.hid_pending = True
+
+    def _ms_up_release(self, key, keyboard, *args, **kwargs):
+        self._up_activated = False
+        self._nav_key_activated -= 1
+        self._check_last()
+        self.pointing_device.report_y[0] = 0
+        self.pointing_device.hid_pending = False
+
+    def _ms_down_press(self, key, keyboard, *args, **kwargs):
+        self._nav_key_activated += 1
+        self._reset_next_interval()
+        self._down_activated = True
+        # if not self.x_activated and not self.y_activated:
+        #     self.next_interval = ticks_ms() + self.ac_intervalle
+        self.pointing_device.report_y[0] = self.move_step
+        self.pointing_device.hid_pending = True
+
+    def _ms_down_release(self, key, keyboard, *args, **kwargs):
+        self._down_activated = False
+        self._nav_key_activated -= 1
+        self._check_last()
+        self.pointing_device.report_y[0] = 0
+        self.pointing_device.hid_pending = False
+
+    def _ms_left_press(self, key, keyboard, *args, **kwargs):
+        self._nav_key_activated += 1
+        self._reset_next_interval()
+        self._left_activated = True
+        self.pointing_device.report_x[0] = 0xFF & (0 - self.move_step)
+        self.pointing_device.hid_pending = True
+
+    def _ms_left_release(self, key, keyboard, *args, **kwargs):
+        self._nav_key_activated -= 1
+        self._left_activated = False
+        self._check_last()
+        self.pointing_device.report_x[0] = 0
+        self.pointing_device.hid_pending = False
+
+    def _ms_right_press(self, key, keyboard, *args, **kwargs):
+        self._nav_key_activated += 1
+        self._reset_next_interval()
+        self._right_activated = True
+        self.pointing_device.report_x[0] = self.move_step
+        self.pointing_device.hid_pending = True
+
+    def _ms_right_release(self, key, keyboard, *args, **kwargs):
+        self._nav_key_activated -= 1
+        self._right_activated = False
+        self._check_last()
+        self.pointing_device.report_x[0] = 0
+        self.pointing_device.hid_pending = False

kmk/modules/oled.py

@@ -0,0 +1,85 @@
+import adafruit_ssd1306
+import busio
+import time
+
+from kmk.modules import Module
+
+class Oled(Module):
+    def __init__(self, pin1=None, pin2=None, i2c=None, startup_image = None, anim1 = None, anim2 = None):
+        print("Init OLED")
+        if i2c is None:
+            self.i2c = busio.I2C(scl=pin1, sda=pin2)
+        else:
+            self.i2c = i2c
+        self.startup_image = startup_image
+        self.anims = [anim1, anim2]
+        self.anim_state = 0
+        self.last_state = None
+        self.oled = adafruit_ssd1306.SSD1306_I2C(128, 32, self.i2c)
+        self.oled.text("Init", 0, 0, 1)
+        self.oled.show()
+        self.inactive_timer = time.time()
+        self.timeout = 3
+        self.inactive_timeout = 30
+        self.last_anim = 0
+        
+
+    def _oled_blank(self):
+        self.oled.fill(0)
+        self.oled.show()
+
+    def _draw_image(self, image, size=(32, 128), offset=0, clear=False):
+        print("calling draw_image")
+        if clear:
+            self.oled.fill(0)
+        for i in range(size[0]):
+            for j in range(size[1]):
+                if image[i][j] is False:
+                    continue
+                # self.oled.buffer[i + j] = image[i][j]
+                self.oled.pixel(j+offset, i, 0 if image[i][j] is False else True)
+        # self.oled_changed = True
+        self.oled.show()
+
+    def during_bootup(self, keyboard):
+        if self.startup_image is not None:
+            self._draw_image(self.startup_image, clear=True)
+        
+        print("booting") 
+        keyboard.oled = self
+        return
+
+    def before_matrix_scan(self, keyboard):
+        return
+
+    def before_hid_send(self, keyboard):
+        return
+
+    def after_hid_send(self, keyboard):
+        return
+
+    def on_powersave_enable(self, keyboard):
+        return
+
+    def on_powersave_disable(self, keyboard):
+        return
+
+    def after_matrix_scan(self, keyboard):
+        curr_time = time.time()
+        # print(time.time(), self.timeout + self.last_anim )
+        if curr_time > self.timeout + self.last_anim and self.last_anim != 0:
+            self._draw_image(self.startup_image, clear = True)
+            self.last_anim = 0
+        if curr_time > self.inactive_timer + self.inactive_timeout:
+            self._oled_blank()
+        return
+
+    def process_key(self, keyboard, key, is_pressed):
+        curr_time = time.time()
+        if is_pressed == 1:
+            # Move the drawing out of process_key
+            # self._draw_image(self.anims[self.anim_state], (32,32), 64-16, True)
+            self.anim_state = 1 if self.anim_state == 0 else 0
+            self.last_anim = curr_time
+        self.inactive_timer = curr_time
+        return super().process_key(keyboard, key, is_pressed)

kmk/modules/oled2.py

@@ -0,0 +1,78 @@
+import busio
+import terminalio
+import displayio
+import adafruit_displayio_ssd1306
+import time
+from adafruit_display_text import label
+from kmk.modules import Module
+
+
+
+class Oled(Module):
+    def __init__(self):
+        self.pins = None
+        self.label = label.Label(font=terminalio.FONT, text="", color = 0xFFFFFF)
+        displayio.release_displays()
+        self.lock_values = ['x', 'x', 'x']
+        self.old_locks_values = None
+        self.locks = None
+        self.layer_labels = {
+            0: "Def",
+            1: "RGB"
+        }
+        self.refresh_rate = 0.2
+        self.last_update = time.monotonic()
+        
+        
+    
+    def _init_oled(self, pins):
+        self.bus = None
+        self._I2C = busio.I2C(scl=pins[0], sda=pins[1])
+        self.display_bus = displayio.I2CDisplay(i2c_bus=self._I2C, device_address=60)
+        self.display = adafruit_displayio_ssd1306.SSD1306(self.display_bus, width=128, height=32)
+        self.label.text = "INIT"
+        self.label.x = 1
+        self.label.y = 5
+        self.display.rotation = 90
+        self.display.show(self.label)
+
+    def _update_locks(self):
+        self.lock_values[0] = 'N' if self.locks.get_num_lock() else 'n'
+        self.lock_values[1] = 'C' if self.locks.get_caps_lock() else 'c'
+        self.lock_values[2] = 'S' if self.locks.get_scroll_lock() else 's'
+
+    def write_text(self, text, label):
+        pass
+
+    def during_bootup(self, keyboard):
+        self._init_oled(self.pins)
+        return
+
+    def after_hid_send(self, keyboard):
+        t = time.monotonic()
+        if self.last_update + self.refresh_rate < t:
+            self._update_locks()
+            self.label.text = f"{self.lock_values[0]}|{self.lock_values[1]}|{self.lock_values[2]}\n{self.layer_labels[keyboard.sandbox.active_layers[0]]}"
+            self.last_update = t
+        # if self.lock_values != self.old_locks_values:
+        # self.label.text = f"{self.lock_values[0]} {self.lock_values[1]} {self.lock_values[2]}\n{self.layer_labels[keyboard.sandbox.active_layers[0]]}"
+            # self.old_locks_values = self.lock_values
+        return
+
+    def before_hid_send(self, keyboard):
+        return
+
+    def before_matrix_scan(self, keyboard):
+        
+        return
+
+    def after_matrix_scan(self, keyboard):
+        return
+
+    def on_powersave_enable(self, keyboard):
+        self.display.sleep()
+        return super()
+    
+    def on_powersave_disable(self, keyboard):
+        self.display.wake()
+        return

kmk/modules/oneshot.py

@@ -0,0 +1,66 @@
+from kmk.keys import make_argumented_key
+from kmk.modules.holdtap import ActivationType, HoldTap
+from kmk.types import HoldTapKeyMeta
+
+
+def oneshot_validator(kc, tap_time=None):
+    return HoldTapKeyMeta(kc=kc, prefer_hold=False, tap_time=tap_time)
+
+
+class OneShot(HoldTap):
+    tap_time = 1000
+
+    def __init__(self):
+        super().__init__()
+        make_argumented_key(
+            validator=oneshot_validator,
+            names=('OS', 'ONESHOT'),
+            on_press=self.osk_pressed,
+            on_release=self.osk_released,
+        )
+
+    def process_key(self, keyboard, current_key, is_pressed, int_coord):
+        '''Release os key after interrupting keyup.'''
+        for key, state in self.key_states.items():
+            if key == current_key:
+                continue
+
+            if state.activated == ActivationType.PRESSED and is_pressed:
+                state.activated = ActivationType.HOLD_TIMEOUT
+            elif state.activated == ActivationType.RELEASED and is_pressed:
+                state.activated = ActivationType.INTERRUPTED
+            elif state.activated == ActivationType.INTERRUPTED:
+                self.ht_released(key, keyboard)
+
+        return current_key
+
+    def osk_pressed(self, key, keyboard, *args, **kwargs):
+        '''Register HoldTap mechanism and activate os key.'''
+        self.ht_pressed(key, keyboard, *args, **kwargs)
+        self.ht_activate_tap(key, keyboard, *args, **kwargs)
+        return keyboard
+
+    def osk_released(self, key, keyboard, *args, **kwargs):
+        '''On keyup, mark os key as released or handle HoldTap.'''
+        try:
+            state = self.key_states[key]
+        except KeyError:
+            if keyboard.debug_enabled:
+                print(f'OneShot.osk_released: no such key {key}')
+            return keyboard
+
+        if state.activated == ActivationType.PRESSED:
+            state.activated = ActivationType.RELEASED
+        else:
+            self.ht_released(key, keyboard, *args, **kwargs)
+
+        return keyboard
+
+    def ht_activate_tap(self, key, keyboard, *args, **kwargs):
+        keyboard.process_key(key.meta.kc, True)
+
+    def ht_deactivate_tap(self, key, keyboard, *args, **kwargs):
+        keyboard.process_key(key.meta.kc, False)
+
+    def ht_deactivate_hold(self, key, keyboard, *args, **kwargs):
+        keyboard.process_key(key.meta.kc, False)

kmk/modules/pimoroni_trackball.py

@@ -0,0 +1,218 @@
+'''
+Extension handles usage of Trackball Breakout by Pimoroni
+Product page: https://shop.pimoroni.com/products/trackball-breakout
+'''
+from micropython import const
+
+import math
+import struct
+
+from kmk.keys import make_key
+from kmk.kmktime import PeriodicTimer
+from kmk.modules import Module
+from kmk.modules.mouse_keys import PointingDevice
+
+I2C_ADDRESS = 0x0A
+I2C_ADDRESS_ALTERNATIVE = 0x0B
+
+CHIP_ID = 0xBA11
+VERSION = 1
+
+REG_LED_RED = 0x00
+REG_LED_GRN = 0x01
+REG_LED_BLU = 0x02
+REG_LED_WHT = 0x03
+
+REG_LEFT = 0x04
+REG_RIGHT = 0x05
+REG_UP = 0x06
+REG_DOWN = 0x07
+REG_SWITCH = 0x08
+MSK_SWITCH_STATE = 0b10000000
+
+REG_USER_FLASH = 0xD0
+REG_FLASH_PAGE = 0xF0
+REG_INT = 0xF9
+MSK_INT_TRIGGERED = 0b00000001
+MSK_INT_OUT_EN = 0b00000010
+REG_CHIP_ID_L = 0xFA
+RED_CHIP_ID_H = 0xFB
+REG_VERSION = 0xFC
+REG_I2C_ADDR = 0xFD
+REG_CTRL = 0xFE
+MSK_CTRL_SLEEP = 0b00000001
+MSK_CTRL_RESET = 0b00000010
+MSK_CTRL_FREAD = 0b00000100
+MSK_CTRL_FWRITE = 0b00001000
+
+ANGLE_OFFSET = 0
+
+
+class TrackballMode:
+    '''Behaviour mode of trackball: mouse movement or vertical scroll'''
+
+    MOUSE_MODE = const(0)
+    SCROLL_MODE = const(1)
+
+
+class Trackball(Module):
+    '''Module handles usage of Trackball Breakout by Pimoroni'''
+
+    def __init__(self, i2c, mode=TrackballMode.MOUSE_MODE, address=I2C_ADDRESS):
+        self._i2c_address = address
+        self._i2c_bus = i2c
+
+        self.pointing_device = PointingDevice()
+        self.mode = mode
+        self.previous_state = False  # click state
+        self.polling_interval = 20
+
+        chip_id = struct.unpack('<H', bytearray(self._i2c_rdwr([REG_CHIP_ID_L], 2)))[0]
+        if chip_id != CHIP_ID:
+            raise RuntimeError(
+                'Invalid chip ID: 0x{:04X}, expected 0x{:04X}'.format(chip_id, CHIP_ID)
+            )
+
+        make_key(
+            names=('TB_MODE',),
+            on_press=self._tb_mode_press,
+            on_release=self._tb_mode_press,
+        )
+
+    def during_bootup(self, keyboard):
+        self._timer = PeriodicTimer(self.polling_interval)
+
+    def before_matrix_scan(self, keyboard):
+        '''
+        Return value will be injected as an extra matrix update
+        '''
+        if not self._timer.tick():
+            return
+
+        up, down, left, right, switch, state = self._read_raw_state()
+
+        if self.mode == TrackballMode.MOUSE_MODE:
+            x_axis, y_axis = self._calculate_movement(right - left, down - up)
+            if x_axis >= 0:
+                self.pointing_device.report_x[0] = x_axis
+            else:
+                self.pointing_device.report_x[0] = 0xFF & x_axis
+            if y_axis >= 0:
+                self.pointing_device.report_y[0] = y_axis
+            else:
+                self.pointing_device.report_y[0] = 0xFF & y_axis
+            self.pointing_device.hid_pending = x_axis != 0 or y_axis != 0
+        else:  # SCROLL_MODE
+            if up >= 0:
+                self.pointing_device.report_w[0] = up
+            if down > 0:
+                self.pointing_device.report_w[0] = 0xFF & (0 - down)
+            self.pointing_device.hid_pending = up != 0 or down != 0
+
+        if switch == 1:  # Button pressed
+            self.pointing_device.button_status[0] |= self.pointing_device.MB_LMB
+            self.pointing_device.hid_pending = True
+
+        if not state and self.previous_state is True:  # Button released
+            self.pointing_device.button_status[0] &= ~self.pointing_device.MB_LMB
+            self.pointing_device.hid_pending = True
+
+        self.previous_state = state
+        return
+
+    def after_matrix_scan(self, keyboard):
+        return
+
+    def before_hid_send(self, keyboard):
+        return
+
+    def after_hid_send(self, keyboard):
+        if self.pointing_device.hid_pending:
+            keyboard._hid_helper.hid_send(self.pointing_device._evt)
+            self._clear_pending_hid()
+        return
+
+    def on_powersave_enable(self, keyboard):
+        return
+
+    def on_powersave_disable(self, keyboard):
+        return
+
+    def set_rgbw(self, r, g, b, w):
+        '''Set all LED brightness as RGBW.'''
+        self._i2c_rdwr([REG_LED_RED, r, g, b, w])
+
+    def set_red(self, value):
+        '''Set brightness of trackball red LED.'''
+        self._i2c_rdwr([REG_LED_RED, value & 0xFF])
+
+    def set_green(self, value):
+        '''Set brightness of trackball green LED.'''
+        self._i2c_rdwr([REG_LED_GRN, value & 0xFF])
+
+    def set_blue(self, value):
+        '''Set brightness of trackball blue LED.'''
+        self._i2c_rdwr([REG_LED_BLU, value & 0xFF])
+
+    def set_white(self, value):
+        '''Set brightness of trackball white LED.'''
+        self._i2c_rdwr([REG_LED_WHT, value & 0xFF])
+
+    def _clear_pending_hid(self):
+        self.pointing_device.hid_pending = False
+        self.pointing_device.report_x[0] = 0
+        self.pointing_device.report_y[0] = 0
+        self.pointing_device.report_w[0] = 0
+        self.pointing_device.button_status[0] = 0
+
+    def _read_raw_state(self):
+        '''Read up, down, left, right and switch data from trackball.'''
+        left, right, up, down, switch = self._i2c_rdwr([REG_LEFT], 5)
+        switch, switch_state = (
+            switch & ~MSK_SWITCH_STATE,
+            (switch & MSK_SWITCH_STATE) > 0,
+        )
+        return up, down, left, right, switch, switch_state
+
+    def _calculate_movement(self, raw_x, raw_y):
+        '''Calculate accelerated movement vector from raw data'''
+        if raw_x == 0 and raw_y == 0:
+            return 0, 0
+
+        var_accel = 1
+        power = 2.5
+
+        angle_rad = math.atan2(raw_y, raw_x) + ANGLE_OFFSET
+        vector_length = math.sqrt(pow(raw_x, 2) + pow(raw_y, 2))
+        vector_length = pow(vector_length * var_accel, power)
+        x = math.floor(vector_length * math.cos(angle_rad))
+        y = math.floor(vector_length * math.sin(angle_rad))
+
+        limit = 127  # hid size limit
+        x_clamped = max(min(limit, x), -limit)
+        y_clamped = max(min(limit, y), -limit)
+
+        return x_clamped, y_clamped
+
+    def _i2c_rdwr(self, data, length=0):
+        '''Write and optionally read I2C data.'''
+        while not self._i2c_bus.try_lock():
+            pass
+
+        try:
+            if length > 0:
+                result = bytearray(length)
+                self._i2c_bus.writeto_then_readfrom(
+                    self._i2c_address, bytes(data), result
+                )
+                return list(result)
+            else:
+                self._i2c_bus.writeto(self._i2c_address, bytes(data))
+
+            return []
+
+        finally:
+            self._i2c_bus.unlock()
+
+    def _tb_mode_press(self, key, keyboard, *args, **kwargs):
+        self.mode = not self.mode

kmk/modules/power.py

@@ -0,0 +1,149 @@
+import board
+import digitalio
+from supervisor import ticks_ms
+
+from time import sleep
+
+from kmk.handlers.stock import passthrough as handler_passthrough
+from kmk.keys import make_key
+from kmk.kmktime import check_deadline
+from kmk.modules import Module
+
+
+class Power(Module):
+    def __init__(self, powersave_pin=None):
+        self.enable = False
+        self.powersave_pin = powersave_pin  # Powersave pin board object
+        self._powersave_start = ticks_ms()
+        self._usb_last_scan = ticks_ms() - 5000
+        self._psp = None  # Powersave pin object
+        self._i2c = 0
+        self._loopcounter = 0
+
+        make_key(
+            names=('PS_TOG',), on_press=self._ps_tog, on_release=handler_passthrough
+        )
+        make_key(
+            names=('PS_ON',), on_press=self._ps_enable, on_release=handler_passthrough
+        )
+        make_key(
+            names=('PS_OFF',), on_press=self._ps_disable, on_release=handler_passthrough
+        )
+
+    def __repr__(self):
+        return f'Power({self._to_dict()})'
+
+    def _to_dict(self):
+        return {
+            'enable': self.enable,
+            'powersave_pin': self.powersave_pin,
+            '_powersave_start': self._powersave_start,
+            '_usb_last_scan': self._usb_last_scan,
+            '_psp': self._psp,
+        }
+
+    def during_bootup(self, keyboard):
+        self._i2c_scan()
+
+    def before_matrix_scan(self, keyboard):
+        return
+
+    def after_matrix_scan(self, keyboard):
+        if keyboard.matrix_update or keyboard.secondary_matrix_update:
+            self.psave_time_reset()
+
+    def before_hid_send(self, keyboard):
+        return
+
+    def after_hid_send(self, keyboard):
+        if self.enable:
+            self.psleep()
+
+    def on_powersave_enable(self, keyboard):
+        '''Gives 10 cycles to allow other extensions to clean up before powersave'''
+        if self._loopcounter > 10:
+            self.enable_powersave(keyboard)
+            self._loopcounter = 0
+        else:
+            self._loopcounter += 1
+        return
+
+    def on_powersave_disable(self, keyboard):
+        self.disable_powersave(keyboard)
+        return
+
+    def enable_powersave(self, keyboard):
+        '''Enables power saving features'''
+        if keyboard.i2c_deinit_count >= self._i2c and self.powersave_pin:
+            # Allows power save to prevent RGB drain.
+            # Example here https://docs.nicekeyboards.com/#/nice!nano/pinout_schematic
+
+            if not self._psp:
+                self._psp = digitalio.DigitalInOut(self.powersave_pin)
+                self._psp.direction = digitalio.Direction.OUTPUT
+            if self._psp:
+                self._psp.value = True
+
+        self.enable = True
+        keyboard._trigger_powersave_enable = False
+        return
+
+    def disable_powersave(self, keyboard):
+        '''Disables power saving features'''
+        if self._psp:
+            self._psp.value = False
+            # Allows power save to prevent RGB drain.
+            # Example here https://docs.nicekeyboards.com/#/nice!nano/pinout_schematic
+
+        keyboard._trigger_powersave_disable = False
+        self.enable = False
+        return
+
+    def psleep(self):
+        '''
+        Sleeps longer and longer to save power the more time in between updates.
+        '''
+        if check_deadline(ticks_ms(), self._powersave_start) <= 60000:
+            sleep(8 / 1000)
+        elif check_deadline(ticks_ms(), self._powersave_start) >= 240000:
+            sleep(180 / 1000)
+        return
+
+    def psave_time_reset(self):
+        self._powersave_start = ticks_ms()
+
+    def _i2c_scan(self):
+        i2c = board.I2C()
+        while not i2c.try_lock():
+            pass
+        try:
+            self._i2c = len(i2c.scan())
+        finally:
+            i2c.unlock()
+        return
+
+    def usb_rescan_timer(self):
+        return bool(check_deadline(ticks_ms(), self._usb_last_scan) > 5000)
+
+    def usb_time_reset(self):
+        self._usb_last_scan = ticks_ms()
+        return
+
+    def usb_scan(self):
+        # TODO Add USB detection here. Currently lies that it's connected
+        # https://github.com/adafruit/circuitpython/pull/3513
+        return True
+
+    def _ps_tog(self, key, keyboard, *args, **kwargs):
+        if self.enable:
+            keyboard._trigger_powersave_disable = True
+        else:
+            keyboard._trigger_powersave_enable = True
+
+    def _ps_enable(self, key, keyboard, *args, **kwargs):
+        if not self.enable:
+            keyboard._trigger_powersave_enable = True
+
+    def _ps_disable(self, key, keyboard, *args, **kwargs):
+        if self.enable:
+            keyboard._trigger_powersave_disable = True

kmk/modules/split.py

@@ -0,0 +1,386 @@
+'''Enables splitting keyboards wirelessly or wired'''
+import busio
+from micropython import const
+from supervisor import runtime, ticks_ms
+
+from keypad import Event as KeyEvent
+from storage import getmount
+
+from kmk.hid import HIDModes
+from kmk.kmktime import check_deadline
+from kmk.modules import Module
+
+
+class SplitSide:
+    LEFT = const(1)
+    RIGHT = const(2)
+
+
+class SplitType:
+    UART = const(1)
+    I2C = const(2)  # unused
+    ONEWIRE = const(3)  # unused
+    BLE = const(4)
+
+
+class Split(Module):
+    '''Enables splitting keyboards wirelessly, or wired'''
+
+    def __init__(
+        self,
+        split_flip=True,
+        split_side=None,
+        split_type=SplitType.UART,
+        split_target_left=True,
+        uart_interval=20,
+        data_pin=None,
+        data_pin2=None,
+        target_left=True,
+        uart_flip=True,
+        use_pio=False,
+        debug_enabled=False,
+    ):
+        self._is_target = True
+        self._uart_buffer = []
+        self.split_flip = split_flip
+        self.split_side = split_side
+        self.split_type = split_type
+        self.split_target_left = split_target_left
+        self.split_offset = None
+        self.data_pin = data_pin
+        self.data_pin2 = data_pin2
+        self.target_left = target_left
+        self.uart_flip = uart_flip
+        self._use_pio = use_pio
+        self._uart = None
+        self._uart_interval = uart_interval
+        self._debug_enabled = debug_enabled
+        self.uart_header = bytearray([0xB2])  # Any non-zero byte should work
+
+        if self.split_type == SplitType.BLE:
+            try:
+                from adafruit_ble import BLERadio
+                from adafruit_ble.advertising.standard import (
+                    ProvideServicesAdvertisement,
+                )
+                from adafruit_ble.services.nordic import UARTService
+
+                self.BLERadio = BLERadio
+                self.ProvideServicesAdvertisement = ProvideServicesAdvertisement
+                self.UARTService = UARTService
+            except ImportError:
+                print('BLE Import error')
+                return  # BLE isn't supported on this platform
+            self._ble_last_scan = ticks_ms() - 5000
+            self._connection_count = 0
+            self._split_connected = False
+            self._uart_connection = None
+            self._advertisment = None  # Seems to not be used anywhere
+            self._advertising = False
+            self._psave_enable = False
+
+        if self._use_pio:
+            from kmk.transports.pio_uart import PIO_UART
+
+            self.PIO_UART = PIO_UART
+
+    def during_bootup(self, keyboard):
+        # Set up name for target side detection and BLE advertisment
+        name = str(getmount('/').label)
+        if self.split_type == SplitType.BLE:
+            if keyboard.hid_type == HIDModes.BLE:
+                self._ble = keyboard._hid_helper.ble
+            else:
+                self._ble = self.BLERadio()
+                self._ble.name = name
+        else:
+            # Try to guess data pins if not supplied
+            if not self.data_pin:
+                self.data_pin = keyboard.data_pin
+
+        # if split side was given, find target from split_side.
+        if self.split_side == SplitSide.LEFT:
+            self._is_target = bool(self.split_target_left)
+        elif self.split_side == SplitSide.RIGHT:
+            self._is_target = not bool(self.split_target_left)
+        else:
+            # Detect split side from name
+            if (
+                self.split_type == SplitType.UART
+                or self.split_type == SplitType.ONEWIRE
+            ):
+                self._is_target = runtime.usb_connected
+            elif self.split_type == SplitType.BLE:
+                self._is_target = name.endswith('L') == self.split_target_left
+
+            if name.endswith('L'):
+                self.split_side = SplitSide.LEFT
+            elif name.endswith('R'):
+                self.split_side = SplitSide.RIGHT
+
+        if not self._is_target:
+            keyboard._hid_send_enabled = False
+
+        if self.split_offset is None:
+            self.split_offset = keyboard.matrix[-1].coord_mapping[-1] + 1
+
+        if self.split_type == SplitType.UART and self.data_pin is not None:
+            if self._is_target or not self.uart_flip:
+                if self._use_pio:
+                    self._uart = self.PIO_UART(tx=self.data_pin2, rx=self.data_pin)
+                else:
+                    self._uart = busio.UART(
+                        tx=self.data_pin2, rx=self.data_pin, timeout=self._uart_interval
+                    )
+            else:
+                if self._use_pio:
+                    self._uart = self.PIO_UART(tx=self.data_pin, rx=self.data_pin2)
+                else:
+                    self._uart = busio.UART(
+                        tx=self.data_pin, rx=self.data_pin2, timeout=self._uart_interval
+                    )
+
+        # Attempt to sanely guess a coord_mapping if one is not provided.
+        if not keyboard.coord_mapping:
+            cm = []
+
+            rows_to_calc = len(keyboard.row_pins)
+            cols_to_calc = len(keyboard.col_pins)
+
+            # Flips the col order if PCB is the same but flipped on right
+            cols_rhs = list(range(cols_to_calc))
+            if self.split_flip:
+                cols_rhs = list(reversed(cols_rhs))
+
+            for ridx in range(rows_to_calc):
+                for cidx in range(cols_to_calc):
+                    cm.append(cols_to_calc * ridx + cidx)
+                for cidx in cols_rhs:
+                    cm.append(cols_to_calc * (rows_to_calc + ridx) + cidx)
+
+            keyboard.coord_mapping = tuple(cm)
+
+        if self.split_side == SplitSide.RIGHT:
+            offset = self.split_offset
+            for matrix in keyboard.matrix:
+                matrix.offset = offset
+                offset += matrix.key_count
+
+    def before_matrix_scan(self, keyboard):
+        if self.split_type == SplitType.BLE:
+            self._check_all_connections(keyboard)
+            self._receive_ble(keyboard)
+        elif self.split_type == SplitType.UART:
+            if self._is_target or self.data_pin2:
+                self._receive_uart(keyboard)
+        elif self.split_type == SplitType.ONEWIRE:
+            pass  # Protocol needs written
+        return
+
+    def after_matrix_scan(self, keyboard):
+        if keyboard.matrix_update:
+            if self.split_type == SplitType.UART:
+                if not self._is_target or self.data_pin2:
+                    self._send_uart(keyboard.matrix_update)
+                else:
+                    pass  # explicit pass just for dev sanity...
+            elif self.split_type == SplitType.BLE:
+                self._send_ble(keyboard.matrix_update)
+            elif self.split_type == SplitType.ONEWIRE:
+                pass  # Protocol needs written
+            else:
+                print('Unexpected case in after_matrix_scan')
+
+        return
+
+    def before_hid_send(self, keyboard):
+        if not self._is_target:
+            keyboard.hid_pending = False
+
+        return
+
+    def after_hid_send(self, keyboard):
+        return
+
+    def on_powersave_enable(self, keyboard):
+        if self.split_type == SplitType.BLE:
+            if self._uart_connection and not self._psave_enable:
+                self._uart_connection.connection_interval = self._uart_interval
+                self._psave_enable = True
+
+    def on_powersave_disable(self, keyboard):
+        if self.split_type == SplitType.BLE:
+            if self._uart_connection and self._psave_enable:
+                self._uart_connection.connection_interval = 11.25
+                self._psave_enable = False
+
+    def _check_all_connections(self, keyboard):
+        '''Validates the correct number of BLE connections'''
+        self._previous_connection_count = self._connection_count
+        self._connection_count = len(self._ble.connections)
+        if self._is_target:
+            if self._advertising or not self._check_if_split_connected():
+                self._target_advertise()
+            elif self._connection_count < 2 and keyboard.hid_type == HIDModes.BLE:
+                keyboard._hid_helper.start_advertising()
+
+        elif not self._is_target and self._connection_count < 1:
+            self._initiator_scan()
+
+    def _check_if_split_connected(self):
+        # I'm looking for a way how to recognize which connection is on and which one off
+        # For now, I found that service name relation to having other CP device
+        if self._connection_count == 0:
+            return False
+        if self._connection_count == 2:
+            self._split_connected = True
+            return True
+
+        # Polling this takes some time so I check only if connection_count changed
+        if self._previous_connection_count == self._connection_count:
+            return self._split_connected
+
+        bleio_connection = self._ble.connections[0]._bleio_connection
+        connection_services = bleio_connection.discover_remote_services()
+        for service in connection_services:
+            if str(service.uuid).startswith("UUID('adaf0001"):
+                self._split_connected = True
+                return True
+        return False
+
+    def _initiator_scan(self):
+        '''Scans for target device'''
+        self._uart = None
+        self._uart_connection = None
+        # See if any existing connections are providing UARTService.
+        self._connection_count = len(self._ble.connections)
+        if self._connection_count > 0 and not self._uart:
+            for connection in self._ble.connections:
+                if self.UARTService in connection:
+                    self._uart_connection = connection
+                    self._uart_connection.connection_interval = 11.25
+                    self._uart = self._uart_connection[self.UARTService]
+                    break
+
+        if not self._uart:
+            if self._debug_enabled:
+                print('Scanning')
+            self._ble.stop_scan()
+            for adv in self._ble.start_scan(
+                self.ProvideServicesAdvertisement, timeout=20
+            ):
+                if self._debug_enabled:
+                    print('Scanning')
+                if self.UARTService in adv.services and adv.rssi > -70:
+                    self._uart_connection = self._ble.connect(adv)
+                    self._uart_connection.connection_interval = 11.25
+                    self._uart = self._uart_connection[self.UARTService]
+                    self._ble.stop_scan()
+                    if self._debug_enabled:
+                        print('Scan complete')
+                    break
+        self._ble.stop_scan()
+
+    def _target_advertise(self):
+        '''Advertises the target for the initiator to find'''
+        # Give previous advertising some time to complete
+        if self._advertising:
+            if self._check_if_split_connected():
+                if self._debug_enabled:
+                    print('Advertising complete')
+                self._ble.stop_advertising()
+                self._advertising = False
+                return
+
+            if not self.ble_rescan_timer():
+                return
+
+            if self._debug_enabled:
+                print('Advertising not answered')
+
+        self._ble.stop_advertising()
+        if self._debug_enabled:
+            print('Advertising')
+        # Uart must not change on this connection if reconnecting
+        if not self._uart:
+            self._uart = self.UARTService()
+        advertisement = self.ProvideServicesAdvertisement(self._uart)
+
+        self._ble.start_advertising(advertisement)
+        self._advertising = True
+        self.ble_time_reset()
+
+    def ble_rescan_timer(self):
+        '''If true, the rescan timer is up'''
+        return not bool(check_deadline(ticks_ms(), self._ble_last_scan, 5000))
+
+    def ble_time_reset(self):
+        '''Resets the rescan timer'''
+        self._ble_last_scan = ticks_ms()
+
+    def _serialize_update(self, update):
+        buffer = bytearray(2)
+        buffer[0] = update.key_number
+        buffer[1] = update.pressed
+        return buffer
+
+    def _deserialize_update(self, update):
+        kevent = KeyEvent(key_number=update[0], pressed=update[1])
+        return kevent
+
+    def _send_ble(self, update):
+        if self._uart:
+            try:
+                self._uart.write(self._serialize_update(update))
+            except OSError:
+                try:
+                    self._uart.disconnect()
+                except:  # noqa: E722
+                    if self._debug_enabled:
+                        print('UART disconnect failed')
+
+                if self._debug_enabled:
+                    print('Connection error')
+                self._uart_connection = None
+                self._uart = None
+
+    def _receive_ble(self, keyboard):
+        if self._uart is not None and self._uart.in_waiting > 0 or self._uart_buffer:
+            while self._uart.in_waiting >= 2:
+                update = self._deserialize_update(self._uart.read(2))
+                self._uart_buffer.append(update)
+            if self._uart_buffer:
+                keyboard.secondary_matrix_update = self._uart_buffer.pop(0)
+
+    def _checksum(self, update):
+        checksum = bytes([sum(update) & 0xFF])
+
+        return checksum
+
+    def _send_uart(self, update):
+        # Change offsets depending on where the data is going to match the correct
+        # matrix location of the receiever
+        if self._uart is not None:
+            update = self._serialize_update(update)
+            self._uart.write(self.uart_header)
+            self._uart.write(update)
+            self._uart.write(self._checksum(update))
+
+    def _receive_uart(self, keyboard):
+        if self._uart is not None and self._uart.in_waiting > 0 or self._uart_buffer:
+            if self._uart.in_waiting >= 60:
+                # This is a dirty hack to prevent crashes in unrealistic cases
+                import microcontroller
+
+                microcontroller.reset()
+
+            while self._uart.in_waiting >= 4:
+                # Check the header
+                if self._uart.read(1) == self.uart_header:
+                    update = self._uart.read(2)
+
+                    # check the checksum
+                    if self._checksum(update) == self._uart.read(1):
+                        self._uart_buffer.append(self._deserialize_update(update))
+            if self._uart_buffer:
+                keyboard.secondary_matrix_update = self._uart_buffer.pop(0)

kmk/modules/tapdance.py

@@ -0,0 +1,124 @@
+from kmk.key_validators import tap_dance_key_validator
+from kmk.keys import make_argumented_key
+from kmk.modules import Module
+from kmk.types import TapDanceKeyMeta
+
+
+class TapDance(Module):
+    # User-configurable
+    tap_time = 300
+
+    # Internal State
+    _tapping = False
+    _tap_dance_counts = {}
+    _tap_timeout = None
+    _tap_side_effects = {}
+
+    def __init__(self):
+        make_argumented_key(
+            validator=tap_dance_key_validator,
+            names=('TAP_DANCE', 'TD'),
+            on_press=self.td_pressed,
+            on_release=self.td_released,
+        )
+
+    def during_bootup(self, keyboard):
+        return
+
+    def before_matrix_scan(self, keyboard):
+        return
+
+    def after_matrix_scan(self, keyboard):
+        return
+
+    def before_hid_send(self, keyboard):
+        return
+
+    def after_hid_send(self, keyboard):
+        return
+
+    def on_powersave_enable(self, keyboard):
+        return
+
+    def on_powersave_disable(self, keyboard):
+        return
+
+    def process_key(self, keyboard, key, is_pressed, int_coord):
+        if self._tapping and is_pressed and not isinstance(key.meta, TapDanceKeyMeta):
+            for k, v in self._tap_dance_counts.items():
+                if v:
+                    self._end_tap_dance(k, keyboard, hold=True)
+                    keyboard.hid_pending = True
+                    keyboard._send_hid()
+                    keyboard.set_timeout(
+                        False, lambda: keyboard.process_key(key, is_pressed)
+                    )
+                    return None
+
+        return key
+
+    def td_pressed(self, key, keyboard, *args, **kwargs):
+        if key not in self._tap_dance_counts or not self._tap_dance_counts[key]:
+            self._tap_dance_counts[key] = 1
+            self._tapping = True
+        else:
+            keyboard.cancel_timeout(self._tap_timeout)
+            self._tap_dance_counts[key] += 1
+
+        if key not in self._tap_side_effects:
+            self._tap_side_effects[key] = None
+
+        self._tap_timeout = keyboard.set_timeout(
+            self.tap_time, lambda: self._end_tap_dance(key, keyboard, hold=True)
+        )
+
+        return self
+
+    def td_released(self, key, keyboard, *args, **kwargs):
+        has_side_effects = self._tap_side_effects[key] is not None
+        hit_max_defined_taps = self._tap_dance_counts[key] == len(key.meta.codes)
+
+        keyboard.cancel_timeout(self._tap_timeout)
+        if has_side_effects or hit_max_defined_taps:
+            self._end_tap_dance(key, keyboard)
+        else:
+            self._tap_timeout = keyboard.set_timeout(
+                self.tap_time, lambda: self._end_tap_dance(key, keyboard)
+            )
+
+        return self
+
+    def _end_tap_dance(self, key, keyboard, hold=False):
+        v = self._tap_dance_counts[key] - 1
+
+        if v < 0:
+            return self
+
+        if key in keyboard.keys_pressed:
+            key_to_press = key.meta.codes[v]
+            keyboard.add_key(key_to_press)
+            self._tap_side_effects[key] = key_to_press
+        elif self._tap_side_effects[key]:
+            keyboard.remove_key(self._tap_side_effects[key])
+            self._tap_side_effects[key] = None
+            self._cleanup_tap_dance(key)
+        elif hold is False:
+            if key.meta.codes[v] in keyboard.keys_pressed:
+                keyboard.remove_key(key.meta.codes[v])
+            else:
+                keyboard.tap_key(key.meta.codes[v])
+            self._cleanup_tap_dance(key)
+        else:
+            key_to_press = key.meta.codes[v]
+            keyboard.add_key(key_to_press)
+            self._tap_side_effects[key] = key_to_press
+            self._tapping = 0
+
+        keyboard.hid_pending = True
+
+        return self
+
+    def _cleanup_tap_dance(self, key):
+        self._tap_dance_counts[key] = 0
+        self._tapping = any(count > 0 for count in self._tap_dance_counts.values())
+        return self