"""PolterHID application entry point.""" import network import time import urandom from display import StatusDisplay from hid import get_hid from settings import load, save from web import WebServer KEY_HOLD_MS = 80 def connect_wifi(settings): wlan = network.WLAN(network.STA_IF) wlan.active(True) set_mdns_hostname(wlan) if not wlan.isconnected(): wlan.connect(settings["wifi_ssid"], settings["wifi_password"]) return wlan def set_mdns_hostname(wlan): """Set the hostname used by ESP32's built-in mDNS responder.""" try: network.hostname("polterhid") except AttributeError: # Older ESP32 builds expose the same setting through the WLAN object. wlan.config(dhcp_hostname="polterhid") def seconds_to_ms(seconds): return seconds * 1000 def randint(minimum, maximum): """Return an inclusive random integer using MicroPython's core RNG API.""" return minimum + (urandom.getrandbits(30) % (maximum - minimum + 1)) def choose_due(now, minimum, maximum): return time.ticks_add(now, seconds_to_ms(randint(minimum, maximum))) def ip_address(wlan): """Return the station IPv4 address across supported MicroPython versions.""" try: return wlan.ipconfig("addr4")[0] except AttributeError: return wlan.ifconfig()[0] def main(): settings = load() display = None if settings["display_enabled"]: display = StatusDisplay() display.update(settings["wifi_ssid"]) hid = get_hid() wlan = connect_wifi(settings) volume_requests = [] def request_volume(action): # Keep the UI responsive but bound queued manual actions. if len(volume_requests) < 8: volume_requests.append(action) server = WebServer(settings, lambda: save(settings), request_volume) now = time.ticks_ms() next_return = choose_due(now, settings["return_min_seconds"], settings["return_max_seconds"]) 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) next_display_refresh = now while True: now = time.ticks_ms() server.poll() # 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) if display is not None and time.ticks_diff(now, next_display_refresh) >= 0: address = ip_address(wlan) if wlan.isconnected() else None display.update(settings["wifi_ssid"], address) next_display_refresh = time.ticks_add(now, 1000) if release_at is not None and time.ticks_diff(now, release_at) >= 0: hid.release_keys() release_at = None if consumer_release_at is not None and time.ticks_diff(now, consumer_release_at) >= 0: hid.release_consumer() consumer_release_at = None elif consumer_release_at is None and volume_requests: if volume_requests.pop(0) == "volume_up": hid.press_volume_up() else: hid.press_volume_down() consumer_release_at = time.ticks_add(now, KEY_HOLD_MS) if (settings["enabled"] and settings["return_enabled"] and release_at is None and time.ticks_diff(now, next_return) >= 0): hid.press_return() release_at = time.ticks_add(now, KEY_HOLD_MS) next_return = choose_due( now, settings["return_min_seconds"], settings["return_max_seconds"] ) if (settings["enabled"] and settings["mouse_enabled"] and time.ticks_diff(now, next_mouse) >= 0): # Never send (0, 0), so each scheduled jitter is observable. distance = settings["mouse_distance"] x = randint(-distance, distance) y = randint(-distance, distance) if x == 0 and y == 0: x = distance hid.move(x, y) next_mouse = choose_due( now, settings["mouse_min_seconds"], settings["mouse_max_seconds"] ) # Do not busy-loop; 25 ms keeps the 80 ms key hold responsive. time.sleep_ms(25) if __name__ == "__main__": main()