[espidf] Switch the Windows console to UTF-8 while idf.py runs (#19352)

esphome/espidf/runner.py

@@ -74,6 +74,64 @@ FILTER_IDF_LINES: list[str] = [
     r"Stopping at filesystem boundary",
 ]
 
+# Windows code page identifier for UTF-8, as used by ``chcp 65001``.
+UTF8_CODEPAGE = 65001
+
+
+def _get_kernel32():
+    """Return the Windows kernel32 module, or None on any other platform."""
+    if sys.platform != "win32":
+        return None
+    import ctypes
+
+    return ctypes.windll.kernel32
+
+
+class _Utf8Console:
+    """Keep an attached Windows console on UTF-8 for the length of the build.
+
+    The build tree runs in UTF-8 mode, so esp_idf_size draws its table with
+    Unicode box characters. ``idf.py size`` reaches it through ``cmake -P``,
+    and CMake re-decodes the child's output with the console code page before
+    printing it, which turns the table into mojibake on any code page but
+    UTF-8. Every process in the build shares this console, so switching it
+    here covers CMake too. The old code pages go back on exit so the user's
+    terminal is left as it was.
+
+    A console that is already on UTF-8 is left alone. The code page belongs to
+    the console, not to this process, so a build that overlaps another one
+    must not save UTF-8 as the page to go back to.
+    """
+
+    def __init__(self, kernel32) -> None:
+        self._kernel32 = kernel32
+        self._codepages: tuple[int, int] | None = None
+
+    def __enter__(self) -> None:
+        kernel32 = self._kernel32
+        if kernel32 is None:
+            return
+        old_in = kernel32.GetConsoleCP()
+        old_out = kernel32.GetConsoleOutputCP()
+        # Both calls return 0 when no console is attached.
+        if not old_in or not old_out:
+            return
+        if old_in == UTF8_CODEPAGE and old_out == UTF8_CODEPAGE:
+            return
+        # Record the old pages first so a switch that fails part way through
+        # still gets put back on exit.
+        self._codepages = (old_in, old_out)
+        kernel32.SetConsoleCP(UTF8_CODEPAGE)
+        kernel32.SetConsoleOutputCP(UTF8_CODEPAGE)
+
+    def __exit__(self, *exc_info: object) -> None:
+        if self._codepages is None:
+            return
+        old_in, old_out = self._codepages
+        self._codepages = None
+        self._kernel32.SetConsoleCP(old_in)
+        self._kernel32.SetConsoleOutputCP(old_out)
+
 
 def main() -> int:
     # ---- sys.path fix-up ---------------------------------------------------
@@ -269,8 +327,29 @@ def main() -> int:
     is_verbose = any(arg in ("-v", "--verbose") for arg in sys.argv[2:])
     filter_lines = None if is_verbose else FILTER_IDF_LINES or None
 
-    stdout_shim = sys.stdout = _FilteringTTYStream(sys.stdout, filter_lines)  # type: ignore[assignment]
-    stderr_shim = sys.stderr = _FilteringTTYStream(sys.stderr, filter_lines)  # type: ignore[assignment]
+    class _FilteredStreams:
+        """Route ``sys.stdout`` and ``sys.stderr`` through the filtering shims.
+
+        On exit each shim releases a last line that never got its
+        terminator. The shims made here are drained rather than whatever
+        ``sys.stdout`` holds by then, which the script is free to replace.
+        """
+
+        def __init__(self, filter_lines: list[str] | None) -> None:
+            self._stdout = _FilteringTTYStream(sys.stdout, filter_lines)
+            self._stderr = _FilteringTTYStream(sys.stderr, filter_lines)
+
+        def __enter__(self) -> None:
+            sys.stdout = self._stdout  # type: ignore[assignment]
+            sys.stderr = self._stderr  # type: ignore[assignment]
+
+        def __exit__(self, *exc_info: object) -> None:
+            # Drain stderr from a finally so a surprise from the first one
+            # cannot strand the second.
+            try:
+                self._stdout.drain()
+            finally:
+                self._stderr.drain()
 
     # Shift argv so the target script sees its own path as argv[0] and
     # its own arguments starting at argv[1]. runpy.run_path does not
@@ -288,19 +367,11 @@ def main() -> int:
 
     # If idf.py calls sys.exit(), SystemExit propagates out of run_path
     # and carries the exit code back to our caller. For normal returns,
-    # fall through and exit with 0. Either way the streams get a chance to
-    # release a last line that never got its terminator. Drain the shims we
-    # made rather than sys.stdout, which the script is free to replace, and
-    # report instead of raising so cleanup cannot bury the real exit code.
-    try:
+    # fall through and exit with 0. Either way the context managers drain
+    # the streams and put the console back on the way out, and they report
+    # instead of raising so cleanup cannot bury the real exit code.
+    with _FilteredStreams(filter_lines), _Utf8Console(_get_kernel32()):
         runpy.run_path(script_path, run_name="__main__")
-    finally:
-        # Drain stderr from a finally so a surprise from the first one cannot
-        # strand the second.
-        try:
-            stdout_shim.drain()
-        finally:
-            stderr_shim.drain()
     return 0
 
 

tests/unit_tests/test_espidf_runner.py

@@ -209,3 +209,107 @@ def test_runner_streams_output_before_the_build_finishes(
             # Join before leaving the block, so the reader is done rather than
             # racing ``Popen`` closing the pipe under it.
             reader.join(1.0)
+
+
+class _FakeKernel32:
+    """Stand-in for the Windows kernel32 console code page calls."""
+
+    def __init__(self, input_cp: int, output_cp: int) -> None:
+        self.input_cp = input_cp
+        self.output_cp = output_cp
+        self.calls: list[tuple[str, int]] = []
+
+    def GetConsoleCP(self) -> int:  # noqa: N802
+        return self.input_cp
+
+    def GetConsoleOutputCP(self) -> int:  # noqa: N802
+        return self.output_cp
+
+    def SetConsoleCP(self, codepage: int) -> int:  # noqa: N802
+        self.calls.append(("SetConsoleCP", codepage))
+        self.input_cp = codepage
+        return 1
+
+    def SetConsoleOutputCP(self, codepage: int) -> int:  # noqa: N802
+        self.calls.append(("SetConsoleOutputCP", codepage))
+        self.output_cp = codepage
+        return 1
+
+
+def test_main_runs_the_build_with_a_utf8_console(
+    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
+) -> None:
+    """An attached console is switched to UTF-8 and then put back."""
+    kernel32 = _FakeKernel32(850, 850)
+    monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32)
+
+    _run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py")
+
+    assert kernel32.calls == [
+        ("SetConsoleCP", runner.UTF8_CODEPAGE),
+        ("SetConsoleOutputCP", runner.UTF8_CODEPAGE),
+        ("SetConsoleCP", 850),
+        ("SetConsoleOutputCP", 850),
+    ]
+
+
+def test_main_restores_the_console_when_the_build_dies(
+    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
+) -> None:
+    """A failing build must not leave the user's console on UTF-8."""
+    kernel32 = _FakeKernel32(437, 437)
+    monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32)
+    _prepare_main(monkeypatch, fixture_path / "espidf" / "crashing_probe.py")
+
+    with pytest.raises(SystemExit):
+        runner.main()
+
+    assert (kernel32.input_cp, kernel32.output_cp) == (437, 437)
+
+
+def test_main_restores_the_console_when_the_switch_fails_part_way(
+    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
+) -> None:
+    """A failed output page switch must not strand the changed input page."""
+    kernel32 = _FakeKernel32(850, 850)
+
+    def _refuse(codepage: int) -> int:
+        kernel32.calls.append(("SetConsoleOutputCP", codepage))
+        return 0
+
+    kernel32.SetConsoleOutputCP = _refuse  # type: ignore[method-assign]
+    monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32)
+
+    _run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py")
+
+    assert kernel32.input_cp == 850
+    assert kernel32.calls[-2:] == [("SetConsoleCP", 850), ("SetConsoleOutputCP", 850)]
+
+
+def test_main_leaves_the_console_alone_when_there_is_none(
+    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
+) -> None:
+    """Without a console the code page calls return 0 and nothing is set."""
+    kernel32 = _FakeKernel32(0, 0)
+    monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32)
+
+    _run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py")
+
+    assert kernel32.calls == []
+
+
+def test_main_leaves_a_console_already_on_utf8_alone(
+    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
+) -> None:
+    """An overlapping build must not save UTF-8 as the page to restore."""
+    kernel32 = _FakeKernel32(runner.UTF8_CODEPAGE, runner.UTF8_CODEPAGE)
+    monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32)
+
+    _run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py")
+
+    assert kernel32.calls == []
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="kernel32 exists on Windows")
+def test_get_kernel32_is_none_off_windows() -> None:
+    assert runner._get_kernel32() is None