Luke/Esp32-Bulk-Keyboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Upload files to "Reciver"

Browse Source
This commit is contained in:
Luke
2025-11-13 21:50:25 +00:00
parent ded5b958fc
commit 47cf864860
3 changed files with 604 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
Reciver/boot.py Normal file
View File

@@ -0,0 +1,37 @@
import storage
import board
import digitalio
# --- Configuration ---
# Set up the two pins we want to check
pin16 = digitalio.DigitalInOut(board.IO16)
pin16.direction = digitalio.Direction.INPUT
pin16.pull = digitalio.Pull.UP
pin17 = digitalio.DigitalInOut(board.IO17)
pin17.direction = digitalio.Direction.INPUT
pin17.pull = digitalio.Pull.UP
# --- Logic ---
# By default, the pull-up resistors hold the pins HIGH (True).
# If we short a pin to GND, its value will become LOW (False).
# We check if BOTH pins are HIGH.
# If they are, it's a "normal boot," so we hide the drive.
if pin16.value and pin17.value:
print("Normal boot: Disabling CIRCUITPY drive.")
storage.disable_usb_drive()
else:
# One of the pins was LOW (grounded), so we
# enable the drive for maintenance.
print("Storage boot: CIRCUITPY drive is ENABLED.")
# We don't need to call storage.enable_usb_drive()
# because it's enabled by default.
# Clean up the pins so they can be used in code.py if needed
pin16.deinit()
pin17.deinit()
# The boot.py file finishes, and CircuitPython will
# either mount the drive or not, based on what we did.
# After this, code.py will run.

560
Reciver/code.py Normal file
View File

@@ -0,0 +1,560 @@
import wifi
import socketpool
import os
import time
import errno
import supervisor
# --- HID Imports ---
import usb_hid
from adafruit_hid.keyboard import Keyboard
from adafruit_hid.keyboard_layout_us import KeyboardLayoutUS
from adafruit_hid.keycode import Keycode
from adafruit_hid.mouse import Mouse
from adafruit_hid.consumer_control import ConsumerControl
from adafruit_hid.consumer_control_code import ConsumerControlCode
# --- Configuration ---
CONTROLLER_IP = "192.168.4.1" # Corrected IP
CONTROLLER_PORT = 5000
CLIENT_NAME = "ESP32-S2-Keyboard-Mouse"
RECONNECT_DELAY = 5
# --- Global Settings ---
global_poll_delay_sec = 0.14
auto_enter_enabled = False
# --- Command Queue & Flow Control ---
command_queue = [] # (Fast) From network, holds raw strings
hid_action_queue = [] # (Slow) To USB, holds atomic HID actions
receiver_is_busy = False
BUSY_THRESHOLD = 50
RESUME_THRESHOLD = 10
# --- HID State Machine ---
g_last_hid_action_time = 0
g_hid_wait_delay = 0
g_last_keys_pressed = set() # NEW: Tracks the "physical" keyboard state
# --- Keycode Mapping (Unchanged) ---
KEY_MAP = {
# Special Keys
"ENTER": Keycode.ENTER, "TAB": Keycode.TAB, "ESC": Keycode.ESCAPE,
"DEL": Keycode.DELETE, "BACKSPACE": Keycode.BACKSPACE, "INSERT": Keycode.INSERT,
"PRTSCR": Keycode.PRINT_SCREEN, "UP": Keycode.UP_ARROW, "DOWN": Keycode.DOWN_ARROW,
"LEFT": Keycode.LEFT_ARROW, "RIGHT": Keycode.RIGHT_ARROW,
"CAPS_LOCK": Keycode.CAPS_LOCK, "SCROLL_LOCK": Keycode.SCROLL_LOCK,
"NUM_LOCK": Keycode.KEYPAD_NUMLOCK, "F1": Keycode.F1, "F2": Keycode.F2,
"F3": Keycode.F3, "F4": Keycode.F4, "F5": Keycode.F5, "F6": Keycode.F6,
"F7": Keycode.F7, "F8": Keycode.F8, "F9": Keycode.F9, "F10": Keycode.F10,
"F11": Keycode.F11, "F12": Keycode.F12, "WIN": Keycode.GUI,
# Modifiers
"L_CTRL": Keycode.LEFT_CONTROL, "L_SHIFT": Keycode.LEFT_SHIFT,
"L_ALT": Keycode.LEFT_ALT, "L_WIN": Keycode.LEFT_GUI,
"R_CTRL": Keycode.RIGHT_CONTROL, "R_SHIFT": Keycode.RIGHT_SHIFT,
"R_ALT": Keycode.RIGHT_ALT, "R_WIN": Keycode.RIGHT_GUI,
# Aliases
"CTRL": Keycode.LEFT_CONTROL, "SHIFT": Keycode.LEFT_SHIFT, "ALT": Keycode.LEFT_ALT,
# Alphanumeric Keys
"A": Keycode.A, "B": Keycode.B, "C": Keycode.C, "D": Keycode.D, "E": Keycode.E,
"F": Keycode.F, "G": Keycode.G, "H": Keycode.H, "I": Keycode.I, "J": Keycode.J,
"K": Keycode.K, "L": Keycode.L, "M": Keycode.M, "N": Keycode.N, "O": Keycode.O,
"P": Keycode.P, "Q": Keycode.Q, "R": Keycode.R, "S": Keycode.S, "T": Keycode.T,
"U": Keycode.U, "V": Keycode.V, "W": Keycode.W, "X": Keycode.X, "Y": Keycode.Y,
"Z": Keycode.Z,
"1": Keycode.ONE, "2": Keycode.TWO, "3": Keycode.THREE, "4": Keycode.FOUR,
"5": Keycode.FIVE, "6": Keycode.SIX, "7": Keycode.SEVEN, "8": Keycode.EIGHT,
"9": Keycode.NINE, "0": Keycode.ZERO,
}
MOUSE_MAP = {
"L_CLICK": Mouse.LEFT_BUTTON, "R_CLICK": Mouse.RIGHT_BUTTON,
"M_CLICK": Mouse.MIDDLE_BUTTON, "BACK_CLICK": Mouse.BACK_BUTTON,
"FWD_CLICK": Mouse.FORWARD_BUTTON,
}
CC_MAP = {
"VOL_UP": ConsumerControlCode.VOLUME_INCREMENT,
"VOL_DOWN": ConsumerControlCode.VOLUME_DECREMENT,
"MUTE": ConsumerControlCode.MUTE,
"PLAY_PAUSE": ConsumerControlCode.PLAY_PAUSE,
"NEXT_TRACK": ConsumerControlCode.SCAN_NEXT_TRACK,
"PREV_TRACK": ConsumerControlCode.SCAN_PREVIOUS_TRACK,
"STOP": ConsumerControlCode.STOP,
}
# --- Setup (Unchanged) ---
try:
kbd = Keyboard(usb_hid.devices)
kbd_layout = KeyboardLayoutUS(kbd)
mouse = Mouse(usb_hid.devices)
cc = ConsumerControl(usb_hid.devices)
print("HID Keyboard, Mouse & ConsumerControl Initialized")
except Exception as e:
print(f"Error initializing HID: {e}")
supervisor.reload()
pool = socketpool.SocketPool(wifi.radio)
sock = None
read_buffer = bytearray(256)
line_buffer = ""
last_led_status = kbd.led_status
# --- Helper Functions (Unchanged) ---
def send_to_controller(message):
global sock
if sock:
try:
sock.send(f"{message}\n".encode('utf-8'))
except Exception as e:
pass # Don't print, this can get spammy
def connect_to_controller():
global g_last_keys_pressed
ssid = os.getenv("CIRCUITPY_WIFI_SSID")
password = os.getenv("CIRCUITPY_WIFI_PASSWORD")
if not ssid:
print("ERROR: CIRCUITPY_WIFI_SSID not set in settings.toml")
return None
print(f"Connecting to Wi-Fi AP: {ssid}...")
while not wifi.radio.connected:
try:
wifi.radio.connect(ssid, password)
except ConnectionError as e:
print(f"Failed to connect to Wi-Fi, retrying... ({e})")
time.sleep(RECONNECT_DELAY)
print(f"Wi-Fi Connected! IP: {wifi.radio.ipv4_address}")
# --- FIXED IP TYPO ---
print(f"Connecting to server at {CONTROLLER_IP}:{CONTROLLER_PORT}...")
try:
new_sock = pool.socket(pool.AF_INET, pool.SOCK_STREAM)
new_sock.connect((CONTROLLER_IP, CONTROLLER_PORT))
mac_addr = ":".join([f"{b:02X}" for b in wifi.radio.mac_address])
handshake = f"HANDSHAKE:NAME={CLIENT_NAME}:MAC={mac_addr}\n"
new_sock.send(handshake.encode('utf-8'))
new_sock.setblocking(False)
g_last_keys_pressed.clear() # Reset key state on connect
print("TCP Connected and Handshake Sent!")
return new_sock
except Exception as e:
print(f"Failed to connect to TCP server: {e}")
if new_sock: new_sock.close()
return None
def parse_keys(key_string_list):
keycodes = []
for key_str in key_string_list:
key_upper = key_str.upper()
if key_upper.startswith("//"):
key_name = key_upper[2:]
else:
key_name = key_upper
if key_name in KEY_MAP:
keycodes.append(KEY_MAP[key_name])
else:
print(f"Warning: Unknown key '{key_str}'")
return keycodes
# --- NEW: `_parse_and_queue_hid_actions` ---
# This function is now the heart of the receiver.
# It's a "fast parser" that translates one command string
# into a series of *atomic* HID actions, optimized for speed.
def _parse_and_queue_hid_actions(command):
global auto_enter_enabled, global_poll_delay_sec, hid_action_queue, g_last_keys_pressed, kbd_layout
try:
command_stripped = command.lstrip()
if not command_stripped:
# It's just whitespace. We must release any pressed keys.
if g_last_keys_pressed:
hid_action_queue.append( ('update_keys', ((), tuple(g_last_keys_pressed))) )
hid_action_queue.append( ('delay', global_poll_delay_sec) )
g_last_keys_pressed.clear()
_queue_string_reliable(command) # Type the whitespace
if auto_enter_enabled:
print("Auto-ENTER: Queuing ENTER")
_queue_simple_command("ENTER")
return
cmd_parts = command_stripped.split()
cmd_first_word = cmd_parts[0].lower()
# --- 1. Handle Multi-part HID Commands ---
if cmd_first_word == "//hold":
keys_to_hold = set(parse_keys(cmd_parts[1:]))
keys_to_press = tuple(keys_to_hold - g_last_keys_pressed)
if keys_to_press:
print(f"Queuing Hold: {keys_to_press}")
hid_action_queue.append( ('update_keys', (keys_to_press, ())) )
g_last_keys_pressed.update(keys_to_press)
elif cmd_first_word == "//release":
if len(cmd_parts) > 1 and cmd_parts[1].lower() == "all":
print("Queuing Release All")
if g_last_keys_pressed:
hid_action_queue.append( ('update_keys', ((), tuple(g_last_keys_pressed))) )
g_last_keys_pressed.clear()
else:
keys_to_release = set(parse_keys(cmd_parts[1:]))
keys_that_can_be_released = tuple(keys_to_release & g_last_keys_pressed)
if keys_that_can_be_released:
print(f"Queuing Release: {keys_that_can_be_released}")
hid_action_queue.append( ('update_keys', ((), keys_that_can_be_released)) )
g_last_keys_pressed.difference_update(keys_that_can_be_released)
elif cmd_first_word == "//delay":
if len(cmd_parts) > 1:
hid_action_queue.append( ('delay', int(cmd_parts[1]) / 1000.0) )
elif cmd_first_word == "//relx":
if len(cmd_parts) > 1: hid_action_queue.append( ('mouse_move', (int(cmd_parts[1]), 0, 0)) )
elif cmd_first_word == "//rely":
if len(cmd_parts) > 1: hid_action_queue.append( ('mouse_move', (0, int(cmd_parts[1]), 0)) )
elif cmd_first_word == "//scroll":
if len(cmd_parts) > 1: hid_action_queue.append( ('mouse_move', (0, 0, int(cmd_parts[1]))) )
# --- 2. Setting Commands ---
elif cmd_first_word == "//auto_enter_on":
auto_enter_enabled = True
print("Auto-ENTER mode: ENABLED")
send_to_controller("TRIGGER:Auto-ENTER ENABLED")
elif cmd_first_word == "//auto_enter_off":
auto_enter_enabled = False
print("Auto-ENTER mode: DISABLED")
send_to_controller("TRIGGER:Auto-ENTER DISABLED")
elif cmd_first_word == "//set_poll_delay":
if len(cmd_parts) > 1:
try:
delay_ms = int(cmd_parts[1])
if delay_ms < 1: delay_ms = 1
global_poll_delay_sec = delay_ms / 1000.0
print(f"Setting POLL_DELAY to {delay_ms}ms")
send_to_controller(f"TRIGGER:POLL_DELAY set to {delay_ms}ms")
except ValueError:
print("Invalid delay value")
# --- 3. Handle In-line Commands and Text ---
else:
print(f"Parsing: {command}")
typed_text = False
current_pos = 0
while current_pos < len(command):
next_cmd_pos = command.find("//", current_pos)
# --- Part 1: Handle text (if any) ---
text_chunk = ""
if next_cmd_pos == -1: # No more commands, type rest of string
text_chunk = command[current_pos:]
current_pos = len(command)
elif next_cmd_pos > current_pos: # Text exists before the command
text_chunk = command[current_pos:next_cmd_pos]
current_pos = next_cmd_pos
if text_chunk:
typed_text = True
# This is the new fast typing loop
for char in text_chunk:
try:
keycodes = set(kbd_layout.keycodes(char))
except Exception:
continue # Can't type this char
# Handle "hello" (same key twice)
if keycodes == g_last_keys_pressed:
# Release, wait
hid_action_queue.append( ('update_keys', ((), tuple(g_last_keys_pressed))) )
hid_action_queue.append( ('delay', global_poll_delay_sec) )
g_last_keys_pressed.clear()
keys_to_press = tuple(keycodes - g_last_keys_pressed)
keys_to_release = tuple(g_last_keys_pressed - keycodes)
hid_action_queue.append( ('update_keys', (keys_to_press, keys_to_release)) )
hid_action_queue.append( ('delay', global_poll_delay_sec) )
g_last_keys_pressed = keycodes
# --- Part 2: Handle command (if any) ---
if next_cmd_pos != -1: # We found a "//"
end_cmd_pos = command.find(" ", next_cmd_pos)
if end_cmd_pos == -1: end_cmd_pos = len(command)
cmd_word = command[next_cmd_pos:end_cmd_pos]
cmd_upper = cmd_word[2:].upper().strip()
if cmd_upper in KEY_MAP or cmd_upper in MOUSE_MAP or cmd_upper in CC_MAP:
# Release any typed chars before running a simple command
if g_last_keys_pressed:
hid_action_queue.append( ('update_keys', ((), tuple(g_last_keys_pressed))) )
hid_action_queue.append( ('delay', global_poll_delay_sec) )
g_last_keys_pressed.clear()
_queue_simple_command(cmd_upper)
else:
# Unknown command, type it literally
typed_text = True
for char in cmd_word: # Type "//unknown"
try:
keycodes = set(kbd_layout.keycodes(char))
except Exception: continue
if keycodes == g_last_keys_pressed:
hid_action_queue.append( ('update_keys', ((), tuple(g_last_keys_pressed))) )
hid_action_queue.append( ('delay', global_poll_delay_sec) )
g_last_keys_pressed.clear()
keys_to_press = tuple(keycodes - g_last_keys_pressed)
keys_to_release = tuple(g_last_keys_pressed - keycodes)
hid_action_queue.append( ('update_keys', (keys_to_press, keys_to_release)) )
hid_action_queue.append( ('delay', global_poll_delay_sec) )
g_last_keys_pressed = keycodes
current_pos = end_cmd_pos
if current_pos < len(command) and command[current_pos] == " ":
current_pos += 1
# After line is done, check for auto-enter
if auto_enter_enabled and typed_text:
print("Auto-ENTER: Queuing ENTER")
# Release any lingering keys from typing
if g_last_keys_pressed:
hid_action_queue.append( ('update_keys', ((), tuple(g_last_keys_pressed))) )
hid_action_queue.append( ('delay', global_poll_delay_sec) )
g_last_keys_pressed.clear()
_queue_simple_command("ENTER")
except Exception as e:
print(f"Error parsing command '{command}': {e}")
hid_action_queue.append( ('release_all', None) )
# --- NEW: Helper function, formerly merged ---
def _queue_simple_command(cmd_upper_no_slash):
global hid_action_queue, global_poll_delay_sec, g_last_keys_pressed
# Simple commands must release any held keys first
if g_last_keys_pressed:
hid_action_queue.append( ('update_keys', ((), tuple(g_last_keys_pressed))) )
hid_action_queue.append( ('delay', global_poll_delay_sec) )
g_last_keys_pressed.clear()
if cmd_upper_no_slash in KEY_MAP:
key = KEY_MAP[cmd_upper_no_slash]
hid_action_queue.append( ('update_keys', ((key,), ())) )
hid_action_queue.append( ('delay', global_poll_delay_sec) )
hid_action_queue.append( ('update_keys', ((), (key,))) )
hid_action_queue.append( ('delay', global_poll_delay_sec) )
elif cmd_upper_no_slash in MOUSE_MAP:
hid_action_queue.append( ('mouse_click', MOUSE_MAP[cmd_upper_no_slash]) )
elif cmd_upper_no_slash in CC_MAP:
hid_action_queue.append( ('cc_send', CC_MAP[cmd_upper_no_slash]) )
# --- NEW: Fast Network Packet Handler (Modified) ---
def _queue_command(line):
global command_queue, kbd, mouse, g_last_keys_pressed
line = line.rstrip()
if not line: return
try:
parts = line.split(':', 2)
if len(parts) < 3 or parts[0] != "SEQ":
print(f"Received malformed line: {line}")
return
command = parts[2]
cmd_word = command.lstrip().split(" ", 1)[0].lower()
if cmd_word == "//stop" or cmd_word == "//s":
print("[!] E-STOP received. Clearing queues and releasing keys.")
command_queue.clear()
hid_action_queue.clear()
g_last_keys_pressed.clear()
kbd.release_all()
mouse.release_all()
elif cmd_word == "//ping":
send_to_controller("PONG:OK")
elif cmd_word == "//heartbeat":
pass
elif cmd_word == "//exit":
print("Controller commanded disconnect. Closing socket.")
sock.close()
else:
command_queue.append(command)
except Exception as e:
print(f"Error queuing command '{line}': {e}")
# --- NEW: The "Slow" HID Executor (Modified) ---
def _process_hid_action_queue():
global hid_action_queue, kbd, mouse, cc
global g_last_hid_action_time, g_hid_wait_delay
now = time.monotonic()
if (now - g_last_hid_action_time) < g_hid_wait_delay:
return
if not hid_action_queue:
# After all actions, release any dangling keys
global g_last_keys_pressed
if g_last_keys_pressed:
print("Queue empty, releasing straggler keys.")
kbd.release(*g_last_keys_pressed)
g_last_keys_pressed.clear()
return
action, data = hid_action_queue.pop(0)
g_last_hid_action_time = now
g_hid_wait_delay = 0
try:
# --- NEW: Handle the 'update_keys' action ---
if action == 'update_keys':
press_keys, release_keys = data
if release_keys:
kbd.release(*release_keys)
if press_keys:
kbd.press(*press_keys)
# --- End New ---
elif action == 'release_all':
kbd.release_all()
elif action == 'mouse_click':
mouse.click(data)
elif action == 'mouse_move':
mouse.move(x=data[0], y=data[1], wheel=data[2])
elif action == 'cc_send':
cc.send(data)
elif action == 'delay':
g_hid_wait_delay = data
except Exception as e:
print(f"Error executing HID action '{action}': {e}")
kbd.release_all()
mouse.release_all()
# --- Network Loop (Unchanged) ---
def _handle_network():
global sock, line_buffer
if sock is None:
return
try:
bytes_recvd = sock.recv_into(read_buffer)
if bytes_recvd > 0:
data = read_buffer[:bytes_recvd].decode('utf-8')
line_buffer += data
while '\n' in line_buffer:
line, line_buffer = line_buffer.split('\n', 1)
_queue_command(line)
elif bytes_recvd == 0:
print("Server closed connection.")
sock.close()
sock = None
except OSError as ex:
if ex.errno == errno.EAGAIN:
pass
else:
print(f"Socket Error: {ex}")
sock.close()
sock = None
# --- Flow Control Loop (Unchanged) ---
def _handle_flow_control():
global receiver_is_busy, command_queue, hid_action_queue
total_queue_len = len(command_queue) + len(hid_action_queue)
if not receiver_is_busy and total_queue_len > BUSY_THRESHOLD:
receiver_is_busy = True
send_to_controller("TRIGGER:BUSY:ON")
print(f"Queue > {BUSY_THRESHOLD}, sending BUSY ON")
elif receiver_is_busy and total_queue_len < RESUME_THRESHOLD:
receiver_is_busy = False
send_to_controller("TRIGGER:BUSY:OFF")
print(f"Queue < {RESUME_THRESHOLD}, sending BUSY OFF")
# --- check_led_status (Unchanged) ---
def check_led_status():
global last_led_status, kbd
current_led_status = kbd.led_status
if current_led_status != last_led_status:
caps_changed = (current_led_status & Keyboard.LED_CAPS_LOCK) != (last_led_status & Keyboard.LED_CAPS_LOCK)
num_changed = (current_led_status & Keyboard.LED_NUM_LOCK) != (last_led_status & Keyboard.LED_NUM_LOCK)
scroll_changed = (current_led_status & Keyboard.LED_SCROLL_LOCK) != (last_led_status & Keyboard.LED_SCROLL_LOCK)
caps_on = (current_led_status & Keyboard.LED_CAPS_LOCK) > 0
num_on = (current_led_status & Keyboard.LED_NUM_LOCK) > 0
scroll_on = (current_led_status & Keyboard.LED_SCROLL_LOCK) > 0
trigger_msg = None
if caps_changed:
trigger_msg = f"TRIGGER:LED_STATUS CapsLock:{'On' if caps_on else 'Off'}"
elif num_changed:
trigger_msg = f"TRIGGER:LED_STATUS NumLock:{'On' if num_on else 'Off'}"
elif scroll_changed:
trigger_msg = f"TRIGGER:LED_STATUS ScrollLock:{'On' if scroll_on else 'Off'}"
if trigger_msg:
print(trigger_msg)
send_to_controller(trigger_msg)
last_led_status = current_led_status
# --- Main Loop (RE-ARCHITECTED) ---
try:
while True:
# --- 1. Handle Connection ---
if sock is None:
sock = connect_to_controller()
if sock is None:
time.sleep(RECONNECT_DELAY)
continue
line_buffer = ""
last_led_status = kbd.led_status
command_queue.clear()
hid_action_queue.clear()
g_last_keys_pressed.clear() # Reset state
receiver_is_busy = False
g_last_hid_action_time = 0
g_hid_wait_delay = 0
# --- 2. Network Read (Fast) ---
_handle_network()
# --- 3. HID & LED Check (Fast) ---
check_led_status()
# --- 4. Flow Control Check (Fast) ---
_handle_flow_control()
# --- 5. Parse ONE Command (Fast, if HID queue is empty) ---
if command_queue and not hid_action_queue:
command_to_parse = command_queue.pop(0)
_parse_and_queue_hid_actions(command_to_parse)
# --- 6. Process ONE HID Action (Can be slow, but is non-blocking) ---
_process_hid_action_queue()
# --- 7. Sleep (NEW: 1ms tick) ---
time.sleep(0.001)
except Exception as e:
print("\n" * 5)
print("=" * 40)
print(f"CRITICAL ERROR in Receiver main loop: {e}")
print("=" * 40)
print("Rebooting in 10 seconds...")
time.sleep(10)
supervisor.reload()

7
Reciver/settings.toml Normal file
View File

@@ -0,0 +1,7 @@
# This tells CircuitPython to connect to your controller's hotspot
CIRCUITPY_WIFI_SSID = "keyboard-controller"
CIRCUITPY_WIFI_PASSWORD = "lWz4Bho2vb2JaIK9jXt0ctrSPglqjJe9YWlHlls0ifCQsL7tEXpFCRpAyIAf"
# These are not used for this project, but can be left in
CIRCUITPY_WEB_API_PASSWORD = "passw0rd"
CIRCUITPY_WEB_API_PORT = 80