Ported BLE Splits to this branch
ble-splits/corne working
This commit is contained in:
2
boot.py
2
boot.py
@@ -1,3 +1,3 @@
|
||||
import supervisor
|
||||
|
||||
supervisor.set_next_stack_limit(4096 + 1024)
|
||||
supervisor.set_next_stack_limit(4096 + 4096)
|
||||
|
@@ -1,7 +1,6 @@
|
||||
import board
|
||||
|
||||
from kmk.extensions.layers import Layers
|
||||
from kmk.extensions.rgb import RGB
|
||||
from kmk.extensions.split import Split
|
||||
from kmk.kmk_keyboard import KMKKeyboard as _KMKKeyboard
|
||||
from kmk.matrix import DiodeOrientation
|
||||
@@ -35,7 +34,6 @@ class KMKKeyboard(_KMKKeyboard):
|
||||
# And now, to handle R3, which at this point is down to just six keys
|
||||
coord_mapping.extend(ic(3, x) for x in range(3, 9))
|
||||
|
||||
rgb_ext = RGB(pixel_pin=rgb_pixel_pin, num_pixels=12)
|
||||
layers_ext = Layers()
|
||||
split = Split(uart_pin=uart_pin, split_offsets=split_offsets)
|
||||
extensions = [rgb_ext, split, layers_ext]
|
||||
extensions = [layers_ext]
|
||||
|
158
kmk/extensions/ble_split.py
Normal file
158
kmk/extensions/ble_split.py
Normal file
@@ -0,0 +1,158 @@
|
||||
from adafruit_ble import BLERadio
|
||||
from adafruit_ble.advertising.standard import ProvideServicesAdvertisement
|
||||
from adafruit_ble.services.nordic import UARTService
|
||||
from kmk.extensions import Extension
|
||||
from kmk.hid import HIDModes
|
||||
from kmk.kmktime import ticks_diff, ticks_ms
|
||||
from kmk.matrix import intify_coordinate
|
||||
|
||||
|
||||
class BLE_Split(Extension):
|
||||
def __init__(
|
||||
self, split_flip=True, split_side=None, split_offsets=[], hid_type=HIDModes.BLE
|
||||
):
|
||||
self._is_target = True
|
||||
self._uart_buffer = []
|
||||
self.hid_type = hid_type
|
||||
self.split_flip = split_flip
|
||||
self.split_side = split_side
|
||||
self.split_offsets = split_offsets
|
||||
self._ble = BLERadio()
|
||||
self._ble_last_scan = ticks_ms() - 5000
|
||||
self._is_target = True
|
||||
self._connection_count = 0
|
||||
self._uart = None
|
||||
self._advertisment = None
|
||||
self._advertising = False
|
||||
|
||||
def __repr__(self):
|
||||
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} _split_offsets={self.split_offsets} )'
|
||||
|
||||
def during_bootup(self, keyboard):
|
||||
if self.split_side == 'Left':
|
||||
self._is_target = True
|
||||
else:
|
||||
self._is_target = False
|
||||
|
||||
if self.split_flip and not self._is_target:
|
||||
keyboard.col_pins = list(reversed(keyboard.col_pins))
|
||||
if self.split_side == 'Left':
|
||||
self._is_target = True
|
||||
self.target_advertise()
|
||||
elif self.split_side == 'Right':
|
||||
self._is_target = False
|
||||
self.initiator_scan()
|
||||
|
||||
# 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)
|
||||
cols_to_calc = len(keyboard.col_pins)
|
||||
|
||||
if self.split_offsets:
|
||||
rows_to_calc *= 2
|
||||
cols_to_calc *= 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_state):
|
||||
self.check_all_connections()
|
||||
if self._is_target:
|
||||
return self._receive_from_initiator()
|
||||
|
||||
def after_matrix_scan(self, keyboard_state, matrix_update):
|
||||
if matrix_update is not None and not self._is_target:
|
||||
self._send_to_target(matrix_update)
|
||||
|
||||
def check_all_connections(self):
|
||||
self._connection_count = len(self._ble.connections)
|
||||
if self._is_target 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):
|
||||
self._uart = 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 UARTService in connection:
|
||||
self._uart = connection
|
||||
break
|
||||
|
||||
if not self._uart:
|
||||
print('Scanning...')
|
||||
for adv in self._ble.start_scan(ProvideServicesAdvertisement, timeout=20):
|
||||
print('Scanning...')
|
||||
if UARTService in adv.services:
|
||||
self._uart = self._ble.connect(adv)
|
||||
self._ble.stop_scan()
|
||||
print('Scan complete')
|
||||
break
|
||||
self._ble.stop_scan()
|
||||
return
|
||||
|
||||
def send(self, data):
|
||||
if self._uart and self._uart.connected:
|
||||
try:
|
||||
self._uart[UARTService].write(data)
|
||||
except OSError:
|
||||
try:
|
||||
self._uart.disconnect()
|
||||
except: # noqa: E722
|
||||
print('UART disconnect failed')
|
||||
print('Connection error')
|
||||
self._uart = None
|
||||
return
|
||||
|
||||
def target_advertise(self):
|
||||
self._ble.stop_advertising()
|
||||
print('Advertising')
|
||||
# Uart must not change on this connection if reconnecting
|
||||
if not self._uart:
|
||||
self._uart = UARTService()
|
||||
advertisement = ProvideServicesAdvertisement(self._uart)
|
||||
|
||||
try:
|
||||
self._ble.start_advertising(advertisement)
|
||||
except Exception as e:
|
||||
print(e)
|
||||
|
||||
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()
|
||||
print('Advertising complete')
|
||||
break
|
||||
self._ble.stop_advertising()
|
||||
|
||||
def ble_rescan_timer(self):
|
||||
return bool(ticks_diff(ticks_ms(), self._ble_last_scan) > 5000)
|
||||
|
||||
def ble_time_reset(self):
|
||||
self._ble_last_scan = ticks_ms()
|
||||
return self
|
||||
|
||||
def _send_to_target(self, update):
|
||||
update[1] += self.split_offsets[update[0]]
|
||||
if self._uart is not None and self._uart.connected:
|
||||
self.send(update)
|
||||
|
||||
def _receive_from_initiator(self):
|
||||
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:
|
||||
update = bytearray(self._uart_buffer.pop(0))
|
||||
|
||||
return update
|
||||
|
||||
return None
|
@@ -36,6 +36,7 @@ class LED(Extension):
|
||||
self._brightness = 0
|
||||
self._pos = 0
|
||||
self._effect_init = False
|
||||
self._enabled = True
|
||||
|
||||
self.brightness_step = brightness_step
|
||||
self.brightness_limit = brightness_limit
|
||||
@@ -63,7 +64,18 @@ class LED(Extension):
|
||||
return 'LED({})'.format(self._to_dict())
|
||||
|
||||
def _to_dict(self):
|
||||
# TODO FIXME remove
|
||||
return (
|
||||
'LED('
|
||||
f'_brightness={self._brightness} '
|
||||
f'_pos={self._pos} '
|
||||
f'brightness_step={self.brightness_step} '
|
||||
f'brightness_limit={self.brightness_limit} '
|
||||
f'animation_mode={self.animation_mode} '
|
||||
f'animation_speed={self.animation_speed} '
|
||||
f'breathe_center={self.breathe_center} '
|
||||
f'val={self.val} '
|
||||
')'
|
||||
)
|
||||
pass
|
||||
|
||||
def _init_effect(self):
|
||||
|
@@ -8,12 +8,12 @@ class SplitType:
|
||||
UART = 1
|
||||
I2C = 2 # unused
|
||||
ONEWIRE = 3 # unused
|
||||
BLE = 4 # unused
|
||||
|
||||
|
||||
class Split(Extension):
|
||||
def __init__(
|
||||
self,
|
||||
is_target=True,
|
||||
extra_data_pin=None,
|
||||
split_offsets=None,
|
||||
split_flip=True,
|
||||
@@ -24,6 +24,7 @@ class Split(Extension):
|
||||
uart_pin=None,
|
||||
uart_timeout=20,
|
||||
):
|
||||
self._is_target = is_target
|
||||
self.extra_data_pin = extra_data_pin
|
||||
self.split_offsets = split_offsets
|
||||
self.split_flip = split_flip
|
||||
@@ -77,7 +78,7 @@ class Split(Extension):
|
||||
keyboard.coord_mapping.append(intify_coordinate(ridx, cidx))
|
||||
|
||||
def before_matrix_scan(self, keyboard_state):
|
||||
if self.split_type is not None and self._is_target:
|
||||
if self._is_target:
|
||||
return self._receive_from_initiator()
|
||||
|
||||
def after_matrix_scan(self, keyboard_state, matrix_update):
|
||||
|
@@ -110,7 +110,7 @@ class KMKKeyboard:
|
||||
|
||||
#####
|
||||
# SPLICE: INTERNAL STATE
|
||||
# FIXME CLEAN THIS
|
||||
# TODO FIXME CLEAN THIS
|
||||
#####
|
||||
|
||||
def _find_key_in_map(self, row, col):
|
||||
@@ -305,6 +305,11 @@ class KMKKeyboard:
|
||||
x.__class__.__module__ == 'kmk.extensions.split' for x in self._extensions
|
||||
):
|
||||
return
|
||||
if any(
|
||||
x.__class__.__module__ == 'kmk.extensions.ble_split'
|
||||
for x in self.extensions
|
||||
):
|
||||
return
|
||||
|
||||
if not self.coord_mapping:
|
||||
self.coord_mapping = []
|
||||
@@ -359,11 +364,10 @@ class KMKKeyboard:
|
||||
try:
|
||||
ext.during_bootup(self)
|
||||
except Exception:
|
||||
# TODO FIXME log the exceptions or something
|
||||
print('Failed to load ', ext)
|
||||
print('Failed to load extention', ext)
|
||||
import time
|
||||
|
||||
time.sleep(30)
|
||||
time.sleep(5)
|
||||
|
||||
self._init_matrix()
|
||||
|
||||
@@ -390,9 +394,8 @@ class KMKKeyboard:
|
||||
for ext in self._extensions:
|
||||
try:
|
||||
ext.before_hid_send(self)
|
||||
except Exception:
|
||||
# TODO FIXME log the exceptions or something
|
||||
pass
|
||||
except Exception as e:
|
||||
print('Failed to run pre hid function: ', e)
|
||||
|
||||
if self._hid_pending:
|
||||
self._send_hid()
|
||||
@@ -410,9 +413,8 @@ class KMKKeyboard:
|
||||
for ext in self._extensions:
|
||||
try:
|
||||
ext.after_hid_send(self)
|
||||
except Exception:
|
||||
# TODO FIXME log the exceptions or something
|
||||
pass
|
||||
except Exception as e:
|
||||
print('Failed to run post hid function: ', e)
|
||||
|
||||
if self.state_changed:
|
||||
self._print_debug_cycle()
|
||||
|
@@ -1,36 +1,41 @@
|
||||
import gc
|
||||
import board
|
||||
|
||||
from kmk.boards.nice_nano.crkbd import KMKKeyboard
|
||||
from kmk.extensions.ble_split import BLE_Split
|
||||
from kmk.extensions.leader import Leader, LeaderMode
|
||||
from kmk.extensions.rgb import RGB
|
||||
from kmk.handlers.sequences import send_string, simple_key_sequence
|
||||
from kmk.hid import HIDModes
|
||||
from kmk.keys import KC
|
||||
from kmk_side import split_side
|
||||
|
||||
keyboard = KMKKeyboard()
|
||||
|
||||
_______ = KC.TRNS
|
||||
XXXXXXX = KC.NO
|
||||
|
||||
LT1_SP = KC.LT(2, KC.SPC)
|
||||
LT1_SP = KC.MO(2)
|
||||
LT2_SP = KC.LT(3, KC.SPC)
|
||||
TAB_SB = KC.LT(5, KC.TAB)
|
||||
SUPER_L = KC.LM(4, KC.LGUI)
|
||||
|
||||
keyboard.tap_time = 150
|
||||
keyboard.tap_time = 300
|
||||
keyboard.leader_timeout = 2000
|
||||
keyboard.debug_enabled = False
|
||||
|
||||
# RGB Config (underglow)
|
||||
keyboard.rgb_config['num_pixels'] = 27
|
||||
keyboard.rgb_config['val_limit'] = 150
|
||||
keyboard.rgb_config['hue_step'] = 10
|
||||
keyboard.rgb_config['sat_step'] = 5
|
||||
keyboard.rgb_config['val_step'] = 5
|
||||
keyboard.rgb_config['hue_default'] = 260
|
||||
keyboard.rgb_config['sat_default'] = 100
|
||||
keyboard.rgb_config['val_default'] = 40
|
||||
keyboard.rgb_config['knight_effect_length'] = 4
|
||||
keyboard.rgb_config['animation_mode'] = 'static'
|
||||
keyboard.rgb_config['animation_speed'] = 1
|
||||
leader_ext = Leader(mode=LeaderMode.ENTER, sequences={
|
||||
'hello': send_string('hello world from kmk macros'),
|
||||
'ls': KC.LGUI(KC.HOME),
|
||||
'dbg': KC.DBG,
|
||||
})
|
||||
|
||||
# TODO Get this out of here
|
||||
split_offsets = [6, 6, 6, 6]
|
||||
rgb_pixel_pin = board.P0_06
|
||||
rgb_ext = RGB(pixel_pin=rgb_pixel_pin, num_pixels=27, val_limit=100, hue_default=190, sat_default=100, val_default=5)
|
||||
|
||||
split = BLE_Split(split_offsets=split_offsets, split_side=split_side)
|
||||
keyboard.extensions = [leader_ext, split, rgb_ext]
|
||||
|
||||
keyboard.keymap = [
|
||||
# DVORAK
|
||||
@@ -154,4 +159,4 @@ keyboard.keymap = [
|
||||
]
|
||||
|
||||
if __name__ == '__main__':
|
||||
keyboard.go()
|
||||
keyboard.go(hid_type=HIDModes.BLE)
|
||||
|
Reference in New Issue
Block a user