PolterHID/main.py

"""PolterHID application entry point."""

import network
import time
import urandom

import maintenance
from display import StatusDisplay
from hid import get_hid
from led import ActivityLed
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 mac_address(wlan):
    """Return the station interface MAC address in conventional display form."""
    try:
        mac = wlan.config("mac")
    except (AttributeError, OSError, ValueError):
        return None
    if not isinstance(mac, (bytes, bytearray)) or len(mac) != 6:
        return None
    return "%02X:%02X:%02X:%02X:%02X:%02X" % tuple(mac)


def wifi_rssi(wlan):
    """Return RSSI in dBm when the current ESP32 port exposes it."""
    if not wlan.isconnected():
        return None
    try:
        return wlan.status("rssi")
    except (AttributeError, OSError, ValueError):
        return None


def main():
    # boot.py leaves USB CDC enabled in maintenance mode; do not activate Wi-Fi,
    # the display, web server, or injection scheduler in that local service mode.
    if maintenance.active:
        return

    settings = load()
    display = None
    if settings["display_enabled"]:
        display = StatusDisplay()
    hid = get_hid()
    activity_led = ActivityLed()
    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_network_refresh = now
    display_address = None
    display_mac = None
    display_rssi = None

    while True:
        now = time.ticks_ms()
        activity_led.tick(now)
        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:
            # 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_mac = mac_address(wlan)
                display_rssi = wifi_rssi(wlan)
                next_display_network_refresh = time.ticks_add(now, 1000)
            display.tick(
                now, settings["wifi_ssid"], display_address, display_mac,
                display_rssi, maintenance.button_pressed()
            )


        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()
            activity_led.pulse(0, 0, 255, now)
            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()
            activity_led.pulse(255, 0, 0, now)
            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)
            activity_led.pulse(0, 255, 0, now)
            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()