Commander1024/PolterHID
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Harden HID, Wi-Fi, and web server handling

Browse Source
This commit is contained in:
Commander1024
2026-08-31 11:54:11 +02:00
parent 072cd793c4
commit 3db3c20a7b
3 changed files with 137 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hid.py
+11
View File
@@ -47,6 +47,7 @@ CONFIG_DESCRIPTOR = bytes((
STRINGS = [None, "PolterHID", "Awareness trainer", "PHID-001"]
_GET_DESCRIPTOR = 0x06
_DESCRIPTOR_TYPE_REPORT = 0x22
_MAX_PENDING_REPORTS = 8
_INSTANCE = None
@@ -117,6 +118,16 @@ class HID:
self._busy = False
def _send(self, report):
if not self._ready:
return
report_id = report[0]
for index, pending in enumerate(self._pending):
if pending[0] == report_id:
self._pending[index] = report
self._flush()
return
if len(self._pending) >= _MAX_PENDING_REPORTS:
self._pending.pop(0)
self._pending.append(report)
self._flush()
main.py
+87 -18
View File
@@ -1,5 +1,6 @@
"""PolterHID application entry point."""
import machine
import network
import time
import urandom
@@ -12,14 +13,24 @@ from settings import load, save
from web import WebServer
KEY_HOLD_MS = 80
WIFI_RETRY_MS = 10000
WIFI_RETRY_SETTLE_MS = 50
WDT_TIMEOUT_MS = 15000
def connect_wifi(settings):
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
try:
wlan.active(True)
except (AttributeError, OSError, ValueError):
pass
set_mdns_hostname(wlan)
if not wlan.isconnected():
wlan.connect(settings["wifi_ssid"], settings["wifi_password"])
disable_wifi_power_save(wlan)
if not wifi_is_connected(wlan):
try:
wlan.connect(settings["wifi_ssid"], settings["wifi_password"])
except (AttributeError, OSError, ValueError):
pass
return wlan
@@ -27,9 +38,57 @@ def set_mdns_hostname(wlan):
"""Set the hostname used by ESP32's built-in mDNS responder."""
try:
network.hostname("polterhid")
except AttributeError:
return
except (AttributeError, OSError, ValueError):
pass
try:
# Older ESP32 builds expose the same setting through the WLAN object.
wlan.config(dhcp_hostname="polterhid")
except (AttributeError, OSError, ValueError):
pass
def disable_wifi_power_save(wlan):
"""Keep the station responsive when the port exposes the PM setting."""
try:
wlan.config(pm=network.WLAN.PM_NONE)
except (AttributeError, OSError, ValueError):
pass
def wifi_is_connected(wlan):
"""Return a safe station connection state during transient WLAN errors."""
try:
return bool(wlan.isconnected())
except (AttributeError, OSError, ValueError):
return False
def reconnect_wifi(wlan, settings):
"""Force a fresh non-blocking connection attempt after a short settle."""
try:
wlan.disconnect()
except (AttributeError, OSError, ValueError):
pass
time.sleep_ms(WIFI_RETRY_SETTLE_MS)
try:
wlan.active(True)
except (AttributeError, OSError, ValueError):
pass
set_mdns_hostname(wlan)
disable_wifi_power_save(wlan)
try:
wlan.connect(settings["wifi_ssid"], settings["wifi_password"])
except (AttributeError, OSError, ValueError):
pass
def start_watchdog():
"""Start the application watchdog when supported by the board/port."""
try:
return machine.WDT(timeout=WDT_TIMEOUT_MS)
except (AttributeError, OSError, ValueError):
return None
def seconds_to_ms(seconds):
@@ -47,10 +106,15 @@ def choose_due(now, minimum, maximum):
def ip_address(wlan):
"""Return the station IPv4 address across supported MicroPython versions."""
if not wifi_is_connected(wlan):
return None
try:
return wlan.ipconfig("addr4")[0]
except AttributeError:
return wlan.ifconfig()[0]
except (AttributeError, OSError, ValueError):
try:
return wlan.ifconfig()[0]
except (AttributeError, OSError, ValueError):
return None
def mac_address(wlan):
@@ -66,7 +130,7 @@ def mac_address(wlan):
def wifi_rssi(wlan):
"""Return RSSI in dBm when the current ESP32 port exposes it."""
if not wlan.isconnected():
if not wifi_is_connected(wlan):
return None
try:
return wlan.status("rssi")
@@ -101,31 +165,36 @@ def main():
next_mouse = choose_due(now, settings["mouse_min_seconds"], settings["mouse_max_seconds"])
release_at = None
consumer_release_at = None
# connect_wifi() already started the first attempt; do not call connect()
# again while the ESP32 station is still in STAT_CONNECTING state.
next_wifi_retry = time.ticks_add(now, 10000)
# connect_wifi() already started the first non-blocking attempt.
next_wifi_retry = time.ticks_add(now, WIFI_RETRY_MS)
next_display_network_refresh = now
display_address = None
display_mac = None
display_rssi = None
watchdog = start_watchdog()
while True:
now = time.ticks_ms()
if watchdog is not None:
try:
watchdog.feed()
except (AttributeError, OSError, ValueError):
watchdog = None
activity_led.tick(now)
server.poll()
server.poll(now)
# Reconnect without blocking injections or the web server indefinitely.
if (not wlan.isconnected() and
time.ticks_diff(now, next_wifi_retry) >= 0 and
wlan.status() != network.STAT_CONNECTING):
wlan.connect(settings["wifi_ssid"], settings["wifi_password"])
next_wifi_retry = time.ticks_add(now, 10000)
# Reset even a stuck STAT_CONNECTING attempt every ten seconds. The only
# deliberate blocking is the 50 ms settle between disconnect/connect.
if (not wifi_is_connected(wlan) and
time.ticks_diff(now, next_wifi_retry) >= 0):
next_wifi_retry = time.ticks_add(now, WIFI_RETRY_MS)
reconnect_wifi(wlan, settings)
if display is not None:
# The display itself is frame-driven; query WLAN state only once per
# second rather than for every 25 ms application-loop iteration.
if time.ticks_diff(now, next_display_network_refresh) >= 0:
display_address = ip_address(wlan) if wlan.isconnected() else None
display_address = ip_address(wlan)
display_mac = mac_address(wlan)
display_rssi = wifi_rssi(wlan)
next_display_network_refresh = time.ticks_add(now, 1000)
web.py
+39 -15
View File
@@ -1,7 +1,13 @@
"""Small non-blocking password-protected HTTP configuration interface."""
import ubinascii
import gc
import socket
import time
import ubinascii
CLIENT_TIMEOUT_MS = 5000
MAX_REQUEST_BYTES = 4096
class WebServer:
@@ -11,32 +17,46 @@ class WebServer:
self.on_volume = on_volume
self._client = None
self._buffer = b""
self._last_activity = None
self._socket = socket.socket()
self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self._socket.bind(("0.0.0.0", 80))
self._socket.listen(1)
self._socket.settimeout(0)
def poll(self):
def poll(self, now=None):
if now is None:
now = time.ticks_ms()
if self._client is None:
try:
self._client, _ = self._socket.accept()
self._client.settimeout(0)
self._buffer = b""
except OSError:
if self._client is not None:
self._close()
return
self._buffer = b""
self._last_activity = now
try:
data = self._client.recv(512)
if data:
self._buffer += data
elif not self._buffer:
self._close()
return
except OSError:
pass
data = None
if len(self._buffer) > 4096:
self._reply(413, "Request too large")
if data == b"":
self._close()
return
if data:
self._last_activity = now
if len(self._buffer) + len(data) > MAX_REQUEST_BYTES:
self._reply(413, "Request too large")
return
self._buffer += data
elif time.ticks_diff(now, self._last_activity) >= CLIENT_TIMEOUT_MS:
self._close()
return
else:
return
marker = self._buffer.find(b"\r\n\r\n")
if marker < 0:
@@ -108,12 +128,16 @@ class WebServer:
self._close()
def _close(self):
try:
self._client.close()
except OSError:
pass
client = self._client
self._client = None
self._buffer = b""
self._last_activity = None
if client is not None:
try:
client.close()
except OSError:
pass
gc.collect()
def _parse_form(body):