Wake Heltec V4 buttons from power-saving sleep

test/test_esp32_usb_sleep.py

@@ -2,7 +2,9 @@
 """Run ESP32Board's real sleep/USB methods against host peripheral stubs."""
 
 from pathlib import Path
+import configparser
 import os
+import re
 import shutil
 import subprocess
 import tempfile
@@ -248,7 +250,305 @@ def board_method(signature: str) -> str:
     return source[start:end]
 
 
+def v4_profile_config():
+    config = configparser.ConfigParser(
+        interpolation=None, inline_comment_prefixes=("#", ";"))
+    config.read([ROOT / "platformio.ini",
+                 ROOT / "variants/heltec_v4/platformio.ini",
+                 ROOT / "variants/heltec_v4_r8/platformio.ini"])
+    return config
+
+
+def profile_option(config, section, option, seen=()):
+    key = (section, option)
+    if key in seen:
+        raise ValueError("cyclic profile interpolation")
+    if config.has_option(section, option):
+        value = config.get(section, option)
+    else:
+        parents = config.get(section, "extends", fallback="").replace(",", " ").split()
+        value = next(profile_option(config, parent, option, seen + (key,))
+                     for parent in parents if config.has_section(parent))
+    return re.sub(r"\x24[{]([^}]+)[}]",
+                  lambda match: profile_option(
+                      config, *match[1].rsplit(".", 1), seen + (key,)),
+                  value)
+
+
+def button_defines(config, section):
+    flags = profile_option(config, section, "build_flags")
+    values = {}
+    for match in re.finditer(
+            r"(?:^|\s)-(D|U)\s*([A-Za-z_]\w*)(?:=([^\s]+))?", flags):
+        operation, name, value = match.groups()
+        if operation == "U":
+            values.pop(name, None)
+        else:
+            values[name] = value if value is not None else "1"
+    return values
+
+
+def conditional_block(source, start):
+    """Keep the production preprocessor branches, including nested guards."""
+    depth = 0
+    end = start
+    for line in source[start:].splitlines(keepends=True):
+        if re.match(r"\s*#if(?:def|ndef)?\b", line):
+            depth += 1
+        elif re.match(r"\s*#endif\b", line):
+            depth -= 1
+        end += len(line)
+        if depth == 0:
+            return source[start:end]
+    raise ValueError("unterminated conditional block")
+
+
+V4_BUTTON_HARNESS = r'''
+#include <Arduino.h>
+#include <helpers/UsbHostSleepPolicy.h>
+#include <helpers/ui/MomentaryButton.h>
+#include <helpers/ui/DisplayPowerPolicy.h>
+#include <array>
+#include <stdexcept>
+#include <iostream>
+static void require(bool ok, const char* why) {
+  if (!ok) throw std::runtime_error(why);
+}
+struct SerialPort {
+  explicit operator bool() const { return false; }
+  bool isPlugged() const { return false; }
+} Serial;
+namespace mesh {
+bool isUsbLoggingEnabled() { return false; }
+struct Console { void printf(const char*) {} };
+static Console console;
+Console& usbConsolePort() { return console; }
+}
+using gpio_num_t = int;
+constexpr int USER_BTN_PRESSED = LOW;
+constexpr int GPIO_INTR_HIGH_LEVEL = 5;
+constexpr int GPIO_INTR_LOW_LEVEL = 4;
+constexpr int GPIO_INTR_POSEDGE = 1;
+constexpr int GPIO_INTR_DISABLE = 0;
+static std::array<int, 64> wake_levels{}, restored_interrupts{};
+static unsigned sleep_calls = 0;
+static int critical_depth = 0;
+static bool press_during_sleep = false, radio_irq_high = false;
+static void esp_sleep_enable_timer_wakeup(uint64_t) {}
+static void esp_sleep_enable_gpio_wakeup() {}
+static int gpio_get_level(gpio_num_t pin) {
+  return pin == PIN_USER_BTN ? digitalRead(pin) : (radio_irq_high ? HIGH : LOW);
+}
+static void gpio_wakeup_enable(gpio_num_t pin, int level) { wake_levels.at(pin) = level; }
+static void gpio_wakeup_disable(gpio_num_t pin) { wake_levels.at(pin) = 0; }
+static void gpio_set_intr_type(gpio_num_t pin, int type) { restored_interrupts.at(pin) = type; }
+static void esp_light_sleep_start() {
+  ++sleep_calls;
+  if (press_during_sleep) {
+    require(wake_levels[PIN_USER_BTN] == GPIO_INTR_LOW_LEVEL,
+            "resolved V4 profile did not arm active-low button wake");
+    require(wake_levels[P_LORA_DIO_1] == GPIO_INTR_HIGH_LEVEL,
+            "button wake replaced the radio wake source");
+    g_mock_pin_levels[PIN_USER_BTN] = LOW;
+    press_during_sleep = false;
+  }
+}
+#define portENTER_CRITICAL(mux) (++critical_depth)
+#define portEXIT_CRITICAL(mux) (--critical_depth)
+struct MainBoard {
+  virtual void sleep(uint32_t) = 0;
+  virtual bool isUsbDataConnected() = 0;
+  virtual bool isUsbHostConnected() = 0;
+  bool isRadioTestActive() const { return false; }
+};
+struct ESP32Board : MainBoard {
+  bool inhibit_sleep = false;
+  uint32_t getIRQGpio() { return P_LORA_DIO_1; }
+  bool isExternalPowered() const { return false; }
+@METHODS@
+} board;
+@BUTTON_CONSTRUCTOR@
+struct TestDisplay {
+  mesh::ui::DisplayPowerPrefs prefs;
+  mesh::ui::DisplayPowerPolicy policy;
+  bool servicePower(bool usb) {
+    const bool before = policy.on();
+    policy.update(prefs, usb, false, false, millis());
+    return before != policy.on();
+  }
+  bool isOn() const { return policy.on(); }
+  void wake(mesh::ui::DisplayWake reason) { policy.wake(reason, millis()); }
+} display;
+struct UITask {
+  TestDisplay* _display = &display;
+  uint32_t _next_refresh = 0, _powering_off_at = 0, _next_read = 0;
+  int _prevBtnState = HIGH;
+  void resetRadioProfileDisplayPage() {}
+  void advanceRadioProfileDisplayPage() {}
+  void loop() {
+@UI_BUTTON@
+  }
+} ui;
+static bool roleCanPowerSave() {
+  bool can_power_save = true;
+@ROLE_SLEEP_GUARD@
+  return can_power_save;
+}
+int main() {
+  try {
+    static_assert(PIN_USER_BTN == 0 && P_LORA_DIO_1 == 14,
+                  "fixture must use the actual V4 GPIOs");
+    resetArduinoMock();
+    g_mock_millis = 1000;
+    g_mock_pin_levels[PIN_USER_BTN] = HIGH;
+    user_btn.begin();
+    ui.loop();
+    require(!display.isOn(), "button-mode screen did not start blank");
+    require(roleCanPowerSave(), "idle button blocked initial power saving");
+    // A press already pending must be serviced instead of entering sleep.
+    g_mock_pin_levels[PIN_USER_BTN] = LOW;
+    board.sleep(30);
+    require(sleep_calls == 0 && critical_depth == 0,
+            "V4 slept over an already pressed button");
+    g_mock_pin_levels[PIN_USER_BTN] = HIGH;
+    // Simulate the physical edge arriving while the CPU is sleeping.
+    press_during_sleep = true;
+    board.sleep(30);
+    require(sleep_calls == 1 && digitalRead(PIN_USER_BTN) == LOW,
+            "fresh button press failed to wake the V4");
+    require(wake_levels[PIN_USER_BTN] == 0 && wake_levels[P_LORA_DIO_1] == 0
+            && restored_interrupts[PIN_USER_BTN] == GPIO_INTR_DISABLE
+            && restored_interrupts[P_LORA_DIO_1] == GPIO_INTR_POSEDGE
+            && critical_depth == 0, "button/radio wake cleanup regressed");
+    ui.loop();
+    require(user_btn.needsPolling() && !roleCanPowerSave(),
+            "role slept before button debounce completed");
+    g_mock_millis += 25;
+    ui.loop();
+    require(user_btn.needsPolling() && !roleCanPowerSave(),
+            "held button stopped gesture polling");
+    g_mock_millis += 50;
+    g_mock_pin_levels[PIN_USER_BTN] = HIGH;
+    ui.loop();
+    require(user_btn.needsPolling() && !roleCanPowerSave(),
+            "role slept before release debounce completed");
+    g_mock_millis += 25;
+    ui.loop();
+    require(user_btn.needsPolling() && !roleCanPowerSave(),
+            "role slept over the pending multi-click deadline");
+    g_mock_millis += MOMENTARY_BUTTON_MULTI_CLICK_MS;
+    ui.loop();
+    require(display.isOn(), "real UITask click did not wake the screen");
+    require(!user_btn.needsPolling() && roleCanPowerSave(),
+            "completed gesture permanently disabled power saving");
+    board.sleep(30);
+    require(sleep_calls == 2, "released button failed to allow sleep again");
+    g_mock_millis += 15000;
+    ui.loop();
+    require(!display.isOn(), "button screen failed to expire normally");
+  } catch (const std::exception& error) {
+    std::cerr << error.what() << '\n'; return 1;
+  }
+}
+'''
+
+
 class Esp32UsbSleepTest(unittest.TestCase):
+    def test_v4_profiles_inherit_button_wake_with_the_actual_gpio_pins(self):
+        config = v4_profile_config()
+        # These standalone partition-migration utilities deliberately bypass
+        # the V4 board/display/radio bases and have no user-button support.
+        utilities = {"env:heltec_v4_partition_migrator" + suffix for suffix in
+                     ("", "_lora_repeater", "_test_hold", "_test_usb_diagnostic")}
+        variants = configparser.ConfigParser(interpolation=None)
+        variants.read([ROOT / "variants/heltec_v4/platformio.ini",
+                       ROOT / "variants/heltec_v4_r8/platformio.ini"])
+        profiles = [section for section in variants.sections()
+                    if section.startswith("env:") and section not in utilities]
+        self.assertGreater(len(profiles), 30)
+        for section in profiles:
+            with self.subTest(profile=section):
+                defines = button_defines(config, section)
+                self.assertEqual(defines["MOMENTARY_BUTTON_WAKE_FROM_SLEEP"], "1")
+                self.assertEqual(defines["PIN_USER_BTN"], "0")
+                self.assertEqual(defines["P_LORA_DIO_1"], "14")
+
+    def test_v4_sleep_button_poller_and_real_ui_wake_the_screen(self):
+        config = v4_profile_config()
+        methods = "\n".join(board_method(signature) for signature in (
+            "void sleep(uint32_t secs) override",
+            "bool isUsbDataConnected() override",
+            "bool isUsbHostConnected() override"))
+        compiler = os.environ.get("CXX", "c++")
+        self.assertIsNotNone(shutil.which(compiler), "a C++17 compiler is required")
+        with tempfile.TemporaryDirectory(prefix="meshcore-v4-button-wake-") as directory:
+            for family in ("heltec_v4", "heltec_v4_r8"):
+                target = (ROOT / f"variants/{family}/target.cpp").read_text()
+                constructor = re.search(
+                    r"MomentaryButton user_btn\([\s\S]*?\);", target)[0]
+                for role, suffix in (("simple_repeater", "repeater"),
+                                     ("simple_room_server", "room_server"),
+                                     ("simple_sensor", "sensor")):
+                    section = f"env:{family}_{suffix}"
+                    defines = button_defines(config, section)
+                    ui_source = (ROOT / f"examples/{role}/UITask.cpp").read_text()
+                    loop = ui_source.index("void UITask::loop()")
+                    opening = ui_source.index("{", loop) + 1
+                    button_start = ui_source.index("#if defined(PIN_USER_BTN)", opening)
+                    button_block = conditional_block(ui_source, button_start)
+                    ui = ui_source[opening:button_start] + button_block
+                    main = (ROOT / f"examples/{role}/main.cpp").read_text()
+                    guard_start = main.index("#if defined(MOMENTARY_BUTTON_WAKE_FROM_SLEEP)")
+                    guard = conditional_block(main, guard_start)
+                    harness = V4_BUTTON_HARNESS.replace("@METHODS@", methods)
+                    harness = harness.replace("@BUTTON_CONSTRUCTOR@", constructor)
+                    harness = harness.replace("@UI_BUTTON@", ui)
+                    harness = harness.replace("@ROLE_SLEEP_GUARD@", guard)
+                    cases = [("profile", harness, defines, None)]
+                    if family == "heltec_v4" and role == "simple_repeater":
+                        # Removing the actual inherited flag reproduces the
+                        # missed GPIO wake using the same production methods.
+                        missing = dict(defines, MOMENTARY_BUTTON_WAKE_FROM_SLEEP="0")
+                        cases.append(("missing-flag", harness, missing,
+                                      "V4 slept over an already pressed button"))
+                        cases.append(("missing-poller", harness.replace(
+                            "can_power_save = can_power_save && !user_btn.needsPolling();",
+                            ""), defines, "role slept before button debounce completed"))
+                        cases.append(("missing-display-wake", harness.replace(
+                            "_display->wake(mesh::ui::DisplayWake::Button);",
+                            "(void)0;"), defines,
+                            "real UITask click did not wake the screen"))
+                    for name, text, selected, expected_error in cases:
+                        with self.subTest(profile=section, control=name):
+                            stem = Path(directory) / f"{family}-{suffix}-{name}"
+                            cpp = stem.with_suffix(".cpp")
+                            cpp.write_text(text, encoding="ascii")
+                            flags = [f"-D{key}={selected[key]}" for key in (
+                                "PIN_USER_BTN", "P_LORA_DIO_1",
+                                "MOMENTARY_BUTTON_WAKE_FROM_SLEEP", "DISPLAY_CLASS")]
+                            result = subprocess.run([
+                                compiler, "-std=c++17", "-Wall", "-Wextra",
+                                "-DARDUINO_USB_CDC_ON_BOOT=1", "-DARDUINO_USB_MODE=1",
+                                "-DMESH_ESP32_USB_CONSOLE_COOPERATIVE=0",
+                                "-DMESH_USB_LOGGING_AVAILABLE=0",
+                                "-DUSER_BTN_LONG_PRESS_MILLIS=" +
+                                selected.get("USER_BTN_LONG_PRESS_MILLIS", "1000"),
+                                "-DUSER_BTN_MULTICLICK=" +
+                                selected.get("USER_BTN_MULTICLICK", "1"),
+                                "-DPOWEROFF_DELAY=2000",
+                                *flags, "-I", str(ROOT / "test/mocks"),
+                                "-I", str(ROOT / "src"), str(cpp),
+                                str(ROOT / "src/helpers/ui/MomentaryButton.cpp"),
+                                "-o", str(stem)], capture_output=True, text=True, timeout=60)
+                            self.assertEqual(result.returncode, 0, result.stderr)
+                            result = subprocess.run(
+                                [str(stem)], capture_output=True, text=True, timeout=10)
+                            if expected_error is None:
+                                self.assertEqual(result.returncode, 0, result.stderr)
+                            else:
+                                self.assertNotEqual(result.returncode, 0)
+                                self.assertIn(expected_error, result.stderr)
+
     def test_raw_uart_driver_preserves_fresh_console_indefinitely_until_end(self):
         methods = "\n".join(board_method(signature) for signature in (
             "void sleep(uint32_t secs) override", "bool isUsbDataConnected() override",