feat(extensions): most of the extensions implementation, by kdb424
This commit is contained in:
Kyle Brown
Josh Klar
40
kmk/modules/__init__.py
Normal file
40
kmk/modules/__init__.py
Normal file
@@ -0,0 +1,40 @@
|
||||
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 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
|
||||
191
kmk/modules/layers.py
Normal file
191
kmk/modules/layers.py
Normal file
@@ -0,0 +1,191 @@
|
||||
'''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.kmktime import accurate_ticks, accurate_ticks_diff
|
||||
from kmk.modules import Module
|
||||
|
||||
|
||||
class LayerType:
|
||||
'''Defines layer type values for readability'''
|
||||
|
||||
MO = const(0)
|
||||
DF = const(1)
|
||||
LM = const(2)
|
||||
LT = const(3)
|
||||
TG = const(4)
|
||||
TT = const(5)
|
||||
|
||||
|
||||
class Layers(Module):
|
||||
'''Gives access to the keys used to enable the layer system'''
|
||||
|
||||
def __init__(self):
|
||||
# Layers
|
||||
self.start_time = {
|
||||
LayerType.LT: None,
|
||||
LayerType.TG: None,
|
||||
LayerType.TT: None,
|
||||
LayerType.LM: None,
|
||||
}
|
||||
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=self._lt_pressed,
|
||||
on_release=self._lt_released,
|
||||
)
|
||||
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=layer_key_validator,
|
||||
names=('TT',),
|
||||
on_press=self._tt_pressed,
|
||||
on_release=self._tt_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 _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 _lt_pressed(self, key, keyboard, *args, **kwargs):
|
||||
# Sets the timer start and acts like MO otherwise
|
||||
self.start_time[LayerType.LT] = accurate_ticks()
|
||||
self._mo_pressed(key, keyboard, *args, **kwargs)
|
||||
|
||||
def _lt_released(self, key, keyboard, *args, **kwargs):
|
||||
# On keyup, check timer, and press key if needed.
|
||||
if self.start_time[LayerType.LT] and (
|
||||
accurate_ticks_diff(
|
||||
accurate_ticks(), self.start_time[LayerType.LT], keyboard.tap_time
|
||||
)
|
||||
):
|
||||
keyboard.hid_pending = True
|
||||
keyboard.tap_key(key.meta.kc)
|
||||
|
||||
self._mo_released(key, keyboard, *args, **kwargs)
|
||||
self.start_time[LayerType.LT] = None
|
||||
|
||||
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 _tt_pressed(self, key, keyboard, *args, **kwargs):
|
||||
'''
|
||||
Momentarily activates layer if held, toggles it if tapped repeatedly
|
||||
'''
|
||||
if self.start_time[LayerType.TT] is None:
|
||||
# Sets the timer start and acts like MO otherwise
|
||||
self.start_time[LayerType.TT] = accurate_ticks()
|
||||
self._mo_pressed(key, keyboard, *args, **kwargs)
|
||||
elif accurate_ticks_diff(
|
||||
accurate_ticks(), self.start_time[LayerType.TT], keyboard.tap_time
|
||||
):
|
||||
self.start_time[LayerType.TT] = None
|
||||
self._tg_pressed(key, keyboard, *args, **kwargs)
|
||||
return
|
||||
return
|
||||
|
||||
def _tt_released(self, key, keyboard, *args, **kwargs):
|
||||
if self.start_time[LayerType.TT] is None or not accurate_ticks_diff(
|
||||
accurate_ticks(), self.start_time[LayerType.TT], keyboard.tap_time
|
||||
):
|
||||
# On first press, works like MO. On second press, does nothing unless let up within
|
||||
# time window, then acts like TG.
|
||||
self.start_time[LayerType.TT] = None
|
||||
self._mo_released(key, keyboard, *args, **kwargs)
|
||||
57
kmk/modules/modtap.py
Normal file
57
kmk/modules/modtap.py
Normal file
@@ -0,0 +1,57 @@
|
||||
from kmk.key_validators import mod_tap_validator
|
||||
from kmk.keys import make_argumented_key
|
||||
from kmk.kmktime import accurate_ticks, accurate_ticks_diff
|
||||
from kmk.modules import Module
|
||||
|
||||
|
||||
class ModTap(Module):
|
||||
def __init__(self):
|
||||
self._mod_tap_timer = None
|
||||
make_argumented_key(
|
||||
validator=mod_tap_validator,
|
||||
names=('MT',),
|
||||
on_press=self.mt_pressed,
|
||||
on_release=self.mt_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 mt_pressed(self, key, keyboard, *args, **kwargs):
|
||||
'''Sets the timer start and acts like a modifier otherwise'''
|
||||
keyboard.keys_pressed.add(key.meta.mods)
|
||||
|
||||
self._mod_tap_timer = accurate_ticks()
|
||||
return keyboard
|
||||
|
||||
def mt_released(self, key, keyboard, *args, **kwargs):
|
||||
''' On keyup, check timer, and press key if needed.'''
|
||||
keyboard.keys_pressed.discard(key.meta.mods)
|
||||
if self._mod_tap_timer and (
|
||||
accurate_ticks_diff(
|
||||
accurate_ticks(), self._mod_tap_timer, keyboard.tap_time
|
||||
)
|
||||
):
|
||||
keyboard.hid_pending = True
|
||||
keyboard.tap_key(key.meta.kc)
|
||||
|
||||
self._mod_tap_timer = None
|
||||
return keyboard
|
||||
146
kmk/modules/power.py
Normal file
146
kmk/modules/power.py
Normal file
@@ -0,0 +1,146 @@
|
||||
import board
|
||||
import digitalio
|
||||
|
||||
from kmk.handlers.stock import passthrough as handler_passthrough
|
||||
from kmk.keys import make_key
|
||||
from kmk.kmktime import sleep_ms, ticks_diff, ticks_ms
|
||||
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 extentions 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 ticks_diff(ticks_ms(), self._powersave_start) <= 60000:
|
||||
sleep_ms(8)
|
||||
elif ticks_diff(ticks_ms(), self._powersave_start) >= 240000:
|
||||
sleep_ms(180)
|
||||
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(ticks_diff(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
|
||||
308
kmk/modules/split.py
Normal file
308
kmk/modules/split.py
Normal file
@@ -0,0 +1,308 @@
|
||||
'''Enables splitting keyboards wirelessly or wired'''
|
||||
import busio
|
||||
from micropython import const
|
||||
|
||||
from kmk.hid import HIDModes
|
||||
from kmk.kmktime import ticks_diff, ticks_ms
|
||||
from kmk.matrix import intify_coordinate
|
||||
from kmk.modules import Module
|
||||
from storage import getmount
|
||||
|
||||
|
||||
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,
|
||||
):
|
||||
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._is_target = True
|
||||
self._uart = None
|
||||
self._uart_interval = uart_interval
|
||||
self._debug_enabled = False
|
||||
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.ProvideServicesAdvertisement = ProvideServicesAdvertisement
|
||||
self.UARTService = UARTService
|
||||
except ImportError:
|
||||
pass # BLE isn't supported on this platform
|
||||
self._ble = BLERadio()
|
||||
self._ble_last_scan = ticks_ms() - 5000
|
||||
self._connection_count = 0
|
||||
self._uart_connection = None
|
||||
self._advertisment = None
|
||||
self._advertising = False
|
||||
self._psave_enable = False
|
||||
|
||||
def __repr__(self):
|
||||
return f'BLE_SPLIT({self._to_dict()})'
|
||||
|
||||
def _to_dict(self):
|
||||
return {
|
||||
'_ble': self._ble,
|
||||
'_ble_last_scan': self._ble_last_scan,
|
||||
'_is_target': self._is_target,
|
||||
'uart_buffer': self._uart_buffer,
|
||||
'_split_flip': self.split_flip,
|
||||
'_split_side': self.split_side,
|
||||
}
|
||||
|
||||
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:
|
||||
self._ble.name = name
|
||||
else:
|
||||
# Try to guess data pins if not supplied
|
||||
if not self.data_pin:
|
||||
self.data_pin = keyboard.data_pin
|
||||
|
||||
# Detect split side from name
|
||||
if self.split_side is None:
|
||||
if name.endswith('L'):
|
||||
# If name ends in 'L' assume left and strip from name
|
||||
self._is_target = bool(self.split_target_left)
|
||||
self.split_side = SplitSide.LEFT
|
||||
elif name.endswith('R'):
|
||||
# If name ends in 'R' assume right and strip from name
|
||||
self._is_target = not bool(self.split_target_left)
|
||||
self.split_side = SplitSide.RIGHT
|
||||
|
||||
# if split side was given, find master from split_side.
|
||||
elif 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)
|
||||
|
||||
# Flips the col pins if PCB is the same but flipped on right
|
||||
if self.split_flip and self.split_side == SplitSide.RIGHT:
|
||||
keyboard.col_pins = list(reversed(keyboard.col_pins))
|
||||
|
||||
self.split_offset = len(keyboard.col_pins)
|
||||
|
||||
if self.split_type == SplitType.UART and self.data_pin is not None:
|
||||
if self._is_target:
|
||||
self._uart = busio.UART(
|
||||
tx=self.data_pin2, rx=self.data_pin, timeout=self._uart_interval
|
||||
)
|
||||
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:
|
||||
keyboard.coord_mapping = []
|
||||
|
||||
rows_to_calc = len(keyboard.row_pins) * 2
|
||||
cols_to_calc = len(keyboard.col_pins) * 2
|
||||
|
||||
for ridx in range(rows_to_calc):
|
||||
for cidx in range(cols_to_calc):
|
||||
keyboard.coord_mapping.append(intify_coordinate(ridx, cidx))
|
||||
|
||||
def before_matrix_scan(self, keyboard):
|
||||
if self.split_type == SplitType.BLE:
|
||||
self._check_all_connections(keyboard._hid_helper)
|
||||
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.BLE:
|
||||
self._send_ble(keyboard.matrix_update)
|
||||
elif self.split_type == SplitType.UART and self.data_pin2:
|
||||
self._send_uart(keyboard.matrix_update)
|
||||
elif self.split_type == SplitType.ONEWIRE:
|
||||
pass # Protocol needs written
|
||||
|
||||
return
|
||||
|
||||
def before_hid_send(self, keyboard):
|
||||
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, hid_type):
|
||||
'''Validates the correct number of BLE connections'''
|
||||
self._connection_count = len(self._ble.connections)
|
||||
if self._is_target and hid_type == HIDModes.BLE and self._connection_count < 2:
|
||||
self._target_advertise()
|
||||
elif not self._is_target and self._connection_count < 1:
|
||||
self._initiator_scan()
|
||||
|
||||
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'''
|
||||
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.ble_time_reset()
|
||||
while not self.ble_rescan_timer():
|
||||
self._connection_count = len(self._ble.connections)
|
||||
if self._connection_count > 1:
|
||||
self.ble_time_reset()
|
||||
if self._debug_enabled:
|
||||
print('Advertising complete')
|
||||
break
|
||||
self._ble.stop_advertising()
|
||||
|
||||
def ble_rescan_timer(self):
|
||||
'''If true, the rescan timer is up'''
|
||||
return bool(ticks_diff(ticks_ms(), self._ble_last_scan) > 5000)
|
||||
|
||||
def ble_time_reset(self):
|
||||
'''Resets the rescan timer'''
|
||||
self._ble_last_scan = ticks_ms()
|
||||
|
||||
def _send_ble(self, update):
|
||||
if self._uart:
|
||||
try:
|
||||
if not self._is_target:
|
||||
update[1] += self.split_offset
|
||||
self._uart.write(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 >= 3:
|
||||
self._uart_buffer.append(self._uart.read(3))
|
||||
if self._uart_buffer:
|
||||
keyboard.secondary_matrix_update = bytearray(self._uart_buffer.pop(0))
|
||||
return
|
||||
|
||||
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._is_target:
|
||||
if self.split_target_left:
|
||||
update[1] += self.split_offset
|
||||
else:
|
||||
update[1] -= self.split_offset
|
||||
else:
|
||||
if self.split_target_left:
|
||||
update[1] -= self.split_offset
|
||||
else:
|
||||
update[1] += self.split_offset
|
||||
|
||||
if self._uart is not None:
|
||||
self._uart.write(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 >= 3:
|
||||
self._uart_buffer.append(self._uart.read(3))
|
||||
if self._uart_buffer:
|
||||
keyboard.secondary_matrix_update = bytearray(self._uart_buffer.pop(0))
|
||||
|
||||
return
|
||||
Reference in New Issue
Block a user