Cruft cleaned, timers more accurate
This commit is contained in:
@@ -1,3 +1,4 @@
|
||||
'''Enables splitting keyboards wirelessly'''
|
||||
from adafruit_ble import BLERadio
|
||||
from adafruit_ble.advertising.standard import ProvideServicesAdvertisement
|
||||
from adafruit_ble.services.nordic import UARTService
|
||||
@@ -8,6 +9,8 @@ from kmk.matrix import intify_coordinate
|
||||
|
||||
|
||||
class BLE_Split(Extension):
|
||||
'''Enables splitting keyboards wirelessly'''
|
||||
|
||||
def __init__(self, split_flip=True, split_side=None, hid_type=HIDModes.BLE):
|
||||
self._is_target = True
|
||||
self._uart_buffer = []
|
||||
@@ -28,7 +31,20 @@ class BLE_Split(Extension):
|
||||
return f'BLE_SPLIT({self._to_dict()})'
|
||||
|
||||
def _to_dict(self):
|
||||
return f'BLE_Split( _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} )'
|
||||
return f'''
|
||||
BLE_Split( _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 on_runtime_enable(self, keyboard):
|
||||
return
|
||||
|
||||
def on_runtime_disable(self, keyboard):
|
||||
return
|
||||
|
||||
def during_bootup(self, keyboard):
|
||||
self._is_target = bool(self.split_side == 'Left')
|
||||
@@ -49,32 +65,32 @@ class BLE_Split(Extension):
|
||||
for cidx in range(cols_to_calc):
|
||||
keyboard.coord_mapping.append(intify_coordinate(ridx, cidx))
|
||||
|
||||
def before_matrix_scan(self, keyboard_state):
|
||||
self.check_all_connections()
|
||||
def before_matrix_scan(self, keyboard):
|
||||
self._check_all_connections()
|
||||
return self._receive()
|
||||
|
||||
def after_matrix_scan(self, keyboard_state, matrix_update):
|
||||
def after_matrix_scan(self, keyboard, matrix_update):
|
||||
if matrix_update:
|
||||
matrix_update = self._send(matrix_update)
|
||||
return matrix_update
|
||||
return None
|
||||
|
||||
def check_all_connections(self):
|
||||
def before_hid_send(self, keyboard):
|
||||
return
|
||||
|
||||
def after_hid_send(self, keyboard):
|
||||
return
|
||||
|
||||
def _check_all_connections(self):
|
||||
'''Validates the correct number of BLE connections'''
|
||||
self._connection_count = len(self._ble.connections)
|
||||
if self._is_target and self._connection_count < 2:
|
||||
self.target_advertise()
|
||||
self._target_advertise()
|
||||
elif not self._is_target and self._connection_count < 1:
|
||||
self.initiator_scan()
|
||||
self._initiator_scan()
|
||||
|
||||
def connect(self):
|
||||
if not self.check_all_connections() and self.ble_rescan_timer:
|
||||
if self.split_side == 'Left':
|
||||
self._is_target = True
|
||||
self.target_advertise()
|
||||
elif self.split_side == 'Right':
|
||||
self._is_target = False
|
||||
self.initiator_scan()
|
||||
|
||||
def initiator_scan(self):
|
||||
def _initiator_scan(self):
|
||||
'''Scans for target device'''
|
||||
self._uart = None
|
||||
self._uart_connection = None
|
||||
# See if any existing connections are providing UARTService.
|
||||
@@ -83,6 +99,7 @@ class BLE_Split(Extension):
|
||||
for connection in self._ble.connections:
|
||||
if UARTService in connection:
|
||||
self._uart_connection = connection
|
||||
self._uart_connection.connection_interval = 11.25
|
||||
self._uart = self._uart_connection[UARTService]
|
||||
break
|
||||
|
||||
@@ -93,14 +110,15 @@ class BLE_Split(Extension):
|
||||
print('Scanning')
|
||||
if UARTService in adv.services:
|
||||
self._uart_connection = self._ble.connect(adv)
|
||||
self._uart_connection.connection_interval = 11.25
|
||||
self._uart = self._uart_connection[UARTService]
|
||||
self._ble.stop_scan()
|
||||
print('Scan complete')
|
||||
break
|
||||
self._ble.stop_scan()
|
||||
return
|
||||
|
||||
def target_advertise(self):
|
||||
def _target_advertise(self):
|
||||
'''Advertises the target for the initiator to find'''
|
||||
self._ble.stop_advertising()
|
||||
print('Advertising')
|
||||
# Uart must not change on this connection if reconnecting
|
||||
@@ -108,10 +126,8 @@ class BLE_Split(Extension):
|
||||
self._uart = UARTService()
|
||||
advertisement = ProvideServicesAdvertisement(self._uart)
|
||||
|
||||
try:
|
||||
self._ble.start_advertising(advertisement)
|
||||
except Exception as e:
|
||||
print(e)
|
||||
self._ble.stop_advertising()
|
||||
self._ble.start_advertising(advertisement)
|
||||
|
||||
self.ble_time_reset()
|
||||
while not self.ble_rescan_timer():
|
||||
@@ -123,11 +139,12 @@ class BLE_Split(Extension):
|
||||
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()
|
||||
return self
|
||||
|
||||
def _send(self, update):
|
||||
if self._uart:
|
||||
@@ -154,4 +171,3 @@ class BLE_Split(Extension):
|
||||
return update
|
||||
|
||||
return None
|
||||
|
||||
|
@@ -1,8 +1,20 @@
|
||||
from micropython import const
|
||||
|
||||
import kmk.handlers.modtap as modtap
|
||||
from kmk.extensions import Extension
|
||||
from kmk.key_validators import layer_key_validator, mod_tap_validator
|
||||
from kmk.keys import make_argumented_key
|
||||
from kmk.kmktime import ticks_diff, ticks_ms
|
||||
from kmk.kmktime import accurate_ticks, accurate_ticks_diff
|
||||
|
||||
|
||||
class LayerType:
|
||||
# These number must be reserved for layer timers.
|
||||
MO = const(0)
|
||||
DF = const(1)
|
||||
LM = const(2)
|
||||
LT = const(3)
|
||||
TG = const(4)
|
||||
TT = const(5)
|
||||
|
||||
|
||||
class Layers(Extension):
|
||||
@@ -49,6 +61,13 @@ class Layers(Extension):
|
||||
on_release=modtap.mt_released,
|
||||
)
|
||||
|
||||
start_time = {
|
||||
LayerType.LT: None,
|
||||
LayerType.TG: None,
|
||||
LayerType.TT: None,
|
||||
LayerType.LM: None,
|
||||
}
|
||||
|
||||
def df_pressed(self, key, state, *args, **kwargs):
|
||||
'''
|
||||
Switches the default layer
|
||||
@@ -86,7 +105,6 @@ class Layers(Extension):
|
||||
'''
|
||||
state._hid_pending = True
|
||||
# Sets the timer start and acts like MO otherwise
|
||||
state._start_time['lm'] = ticks_ms()
|
||||
state._keys_pressed.add(key.meta.kc)
|
||||
return self.mo_pressed(key, state, *args, **kwargs)
|
||||
|
||||
@@ -96,24 +114,25 @@ class Layers(Extension):
|
||||
'''
|
||||
state._hid_pending = True
|
||||
state._keys_pressed.discard(key.meta.kc)
|
||||
state._start_time['lm'] = None
|
||||
return self.mo_released(key, state, *args, **kwargs)
|
||||
|
||||
def lt_pressed(self, key, state, *args, **kwargs):
|
||||
# Sets the timer start and acts like MO otherwise
|
||||
state._start_time['lt'] = ticks_ms()
|
||||
self.start_time[LayerType.LT] = accurate_ticks()
|
||||
return self.mo_pressed(key, state, *args, **kwargs)
|
||||
|
||||
def lt_released(self, key, state, *args, **kwargs):
|
||||
# On keyup, check timer, and press key if needed.
|
||||
if state._start_time['lt'] and (
|
||||
ticks_diff(ticks_ms(), state._start_time['lt']) < state.tap_time
|
||||
if self.start_time[LayerType.LT] and (
|
||||
accurate_ticks_diff(
|
||||
accurate_ticks(), self.start_time[LayerType.LT], state.tap_time
|
||||
)
|
||||
):
|
||||
state._hid_pending = True
|
||||
state._tap_key(key.meta.kc)
|
||||
|
||||
self.mo_released(key, state, *args, **kwargs)
|
||||
state._start_time['lt'] = None
|
||||
self.start_time[LayerType.LT] = None
|
||||
return state
|
||||
|
||||
def tg_pressed(self, key, state, *args, **kwargs):
|
||||
@@ -143,22 +162,23 @@ class Layers(Extension):
|
||||
Momentarily activates layer if held, toggles it if tapped repeatedly
|
||||
'''
|
||||
# TODO Make this work with tap dance to function more correctly, but technically works.
|
||||
if state._start_time['tt'] is None:
|
||||
if self.start_time[LayerType.TT] is None:
|
||||
# Sets the timer start and acts like MO otherwise
|
||||
state._start_time['tt'] = ticks_ms()
|
||||
self.start_time[LayerType.TT] = accurate_ticks()
|
||||
return self.mo_pressed(key, state, *args, **kwargs)
|
||||
elif ticks_diff(ticks_ms(), state._start_time['tt']) < state.tap_time:
|
||||
state._start_time['tt'] = None
|
||||
elif accurate_ticks_diff(
|
||||
accurate_ticks(), self.start_time[LayerType.TT], state.tap_time
|
||||
):
|
||||
self.start_time[LayerType.TT] = None
|
||||
return self.tg_pressed(key, state, *args, **kwargs)
|
||||
|
||||
def tt_released(self, key, state, *args, **kwargs):
|
||||
tap_timed_out = (
|
||||
ticks_diff(ticks_ms(), state._start_time['tt']) >= state.tap_time
|
||||
)
|
||||
if state._start_time['tt'] is None or tap_timed_out:
|
||||
if self.start_time[LayerType.TT] is None or not accurate_ticks_diff(
|
||||
accurate_ticks(), self.start_time[LayerType.TT], state.tap_time
|
||||
):
|
||||
# On first press, works like MO. On second press, does nothing unless let up within
|
||||
# time window, then acts like TG.
|
||||
state._start_time['tt'] = None
|
||||
self.start_time[LayerType.TT] = None
|
||||
return self.mo_released(key, state, *args, **kwargs)
|
||||
|
||||
return state
|
||||
|
Reference in New Issue
Block a user