Tell the post-upload probe results apart and blame no connection for a device refusal

Locates the current key on its yaml or secrets.yaml line, writes the new
one with the editor from #20602, keeps the replaced key as old_key, checks
the device with the probe from #20601, builds, uploads with the current key
and confirms with the new one; anything short of a reached device puts the
files back.

esphome/__main__.py

@@ -1305,7 +1305,10 @@ def upload_program(
                 f"{option_string} is only supported with the esphome OTA platform; "
                 "the web_server OTA path can only update the firmware image."
             )
-        return _upload_via_web_server(config, network_devices, _app_ota_image(args))
+        binary = CORE.firmware_bin
+        if getattr(args, "file", None) is not None:
+            binary = Path(args.file)
+        return _upload_via_web_server(config, network_devices, binary)
 
     return _upload_via_native_api(config, network_devices, args)
 
@@ -1361,16 +1364,6 @@ def _esphome_ota_conf(config: ConfigType) -> ConfigType | None:
     return next(iter(esphome_ota_items(config)), None)
 
 
-def _app_ota_image(args: ArgsProtocol) -> Path:
-    """The file an app update sends: ``--file`` or the build output, as the
-    target platform's ota_image hook picks it."""
-    binary = CORE.firmware_bin
-    if (file := getattr(args, "file", None)) is not None:
-        binary = Path(file)
-    ota_image = platform_hooks.get_platform_hook(CORE.target_platform, "ota_image")
-    return ota_image(binary) if ota_image is not None else binary
-
-
 def _upload_via_native_api(
     config: ConfigType, network_devices: list[str], args: ArgsProtocol
 ) -> tuple[int, str | None]:
@@ -1420,27 +1413,27 @@ def _upload_via_native_api(
                 f"flash a build with the option enabled, and then retry {option_string}."
             )
 
+    binary = CORE.firmware_bin
     ota_type = espota2.OTA_TYPE_UPDATE_APP
-    file = getattr(args, "file", None)
     if getattr(args, "partition_table", False):
         # Fail fast if the resolved ESPHome OTA config does not enable allow_partition_access.
         # The device-side handshake also rejects this with "Device only supports app updates",
         # but checking here surfaces the misconfiguration before opening a network connection.
         check_partition_access("--partition-table")
-        binary = Path(file) if file is not None else CORE.partition_table_bin
+        binary = CORE.partition_table_bin
         ota_type = espota2.OTA_TYPE_UPDATE_PARTITION_TABLE
     elif getattr(args, "bootloader", False):
         check_partition_access("--bootloader")
         if (
-            file is None
+            getattr(args, "file", None) is None
             and (native := native_backend())
             and (hint := getattr(native, "missing_image_hint", lambda: None)())
         ):
             raise EsphomeError(hint)
-        binary = Path(file) if file is not None else CORE.bootloader_bin
+        binary = CORE.bootloader_bin
         ota_type = espota2.OTA_TYPE_UPDATE_BOOTLOADER
-    else:
-        binary = _app_ota_image(args)
+    if getattr(args, "file", None) is not None:
+        binary = Path(args.file)
 
     if ota_type == espota2.OTA_TYPE_UPDATE_PARTITION_TABLE:
         _validate_partition_table_binary(binary)
@@ -2240,6 +2233,12 @@ def command_analyze_memory(args: ArgsProtocol, config: ConfigType) -> int:
     return 0
 
 
+def command_rotate_key(args: ArgsProtocol, config: ConfigType) -> int | None:
+    from esphome.cli.rotate_key import command_rotate_key as run
+
+    return run(args, config)
+
+
 def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:
     from esphome.cli.rename import command_rename as run
 
@@ -2266,6 +2265,7 @@ POST_CONFIG_ACTIONS = {
     "clean-mqtt": command_clean_mqtt,
     "idedata": command_idedata,
     "rename": command_rename,
+    "rotate-key": command_rotate_key,
     "discover": command_discover,
     "analyze-memory": command_analyze_memory,
     "bundle": command_bundle,
@@ -2663,6 +2663,37 @@ def parse_args(argv):
     )
     parser_rename.add_argument("name", help="The new name for the device.", type=str)
 
+    parser_rotate_key = subparsers.add_parser(
+        "rotate-key",
+        help=(
+            "Change the OTA encryption key over the air: write a new key to the "
+            "yaml, build with it, upload with the current key, confirm."
+        ),
+        parents=[mqtt_options],
+    )
+    parser_rotate_key.add_argument(
+        "configuration", help="Your YAML configuration file.", nargs=1
+    )
+    parser_rotate_key.add_argument(
+        "--device",
+        action="append",
+        help=ARGUMENT_HELP_DEVICE,
+    )
+    parser_rotate_key.add_argument(
+        "--prompt-new-key",
+        action="store_true",
+        help=(
+            "Ask for the new key instead of generating one; read without echo, "
+            "or as one line from stdin when that is not a terminal."
+        ),
+    )
+    parser_rotate_key.add_argument(
+        "-y",
+        "--yes",
+        action="store_true",
+        help="Do not ask for confirmation when the api encryption key changes too.",
+    )
+
     parser_analyze_memory = subparsers.add_parser(
         "analyze-memory",
         help="Analyze memory usage by component.",

esphome/cli/rotate_key.py

@@ -0,0 +1,289 @@
+"""``esphome rotate-key``."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+import sys
+
+from esphome import espota2
+from esphome.components.noise import (
+    generate_encryption_key,
+    static_encryption_key,
+    validate_encryption_key,
+)
+from esphome.config_validation import Invalid
+from esphome.const import CONF_API, CONF_PASSWORD, CONF_PORT
+from esphome.core import CORE, EsphomeError
+from esphome.espota2 import ProbeResult
+from esphome.log import AnsiFore, color
+from esphome.ota_key_edit import locate_key_edits, old_key_edit, with_sharers
+from esphome.types import ConfigType
+from esphome.upload_targets import PortType, get_port_type
+from esphome.util import (
+    ESPHOME_COMMAND,
+    read_secret_line,
+    run_external_process,
+    safe_input,
+    safe_print,
+)
+from esphome.yaml_edit import RestoreError, Snapshot, apply_line_edits, restore_files
+
+PRECHECK_TIMEOUT = 15.0
+# What to do when the device does not accept the key in the yaml before anything is written
+PRECHECK_ADVICE = {
+    ProbeResult.UNREACHABLE: (
+        "The device could not be reached; check that it is online, then try again."
+    ),
+    ProbeResult.REJECTED: (
+        "The device runs a different key than the yaml. Install the current "
+        "configuration first, presenting its key with --prompt-ota-key, then rotate."
+    ),
+    ProbeResult.NOT_OFFERED: (
+        "The device does not offer encryption: it runs a firmware older than "
+        "ESPHome 2026.9.0 or without a key. Install the current configuration "
+        "first, then rotate."
+    ),
+    ProbeResult.DEVICE_ERROR: (
+        "The device answered with an error (see the warning above); fix that first."
+    ),
+}
+# Why the device did not confirm the new key after the upload
+CONFIRM_FAILURE = {
+    ProbeResult.REJECTED: "The device still answers with the previous key.",
+    ProbeResult.NOT_OFFERED: "The device came back without encryption.",
+    ProbeResult.DEVICE_ERROR: "The device answered with an error (see the warning above).",
+    ProbeResult.UNREACHABLE: "The device did not come back to confirm the new key.",
+}
+
+
+def command_rotate_key(args: argparse.Namespace, config: ConfigType) -> int | None:
+    """Write a new key, build with it, upload with the current one, confirm."""
+    from esphome.__main__ import (
+        Purpose,
+        _esphome_ota_conf,
+        _resolve_network_devices,
+        choose_upload_log_host,
+    )
+
+    def fail(message: str) -> int:
+        safe_print(color(AnsiFore.BOLD_RED, message))
+        return 1
+
+    ota_conf = _esphome_ota_conf(config)
+    if ota_conf is None:
+        return fail("rotate-key needs the esphome OTA platform in the configuration")
+    old_key = static_encryption_key(ota_conf)
+    if old_key is None:
+        return fail(
+            "rotate-key needs 'encryption:' under the esphome OTA platform with a "
+            "key in the yaml; a key provisioned at runtime is rotated from Home "
+            "Assistant"
+        )
+    if CORE.is_host:
+        return fail("rotate-key is for devices updated over the air")
+    old_key = str(old_key)
+    remote_port = int(ota_conf[CONF_PORT])
+
+    devices = choose_upload_log_host(
+        default=args.device, check_default=None, purpose=Purpose.UPLOADING
+    )
+    if get_port_type(devices[0]) in (PortType.SERIAL, PortType.BOOTSEL):
+        return fail(
+            "rotate-key works over the air; on a serial connection change the "
+            "key in the yaml and run 'esphome run' instead"
+        )
+    network_devices = _resolve_network_devices(devices, config, args)
+
+    if args.prompt_new_key:
+        new_key = read_secret_line("New OTA encryption key: ")
+    else:
+        new_key = generate_encryption_key()
+    try:
+        validate_encryption_key(new_key)
+    except Invalid as err:
+        return fail(f"Invalid new key: {err}")
+    if new_key == old_key:
+        return fail("The new key is the same as the current one")
+
+    try:
+        edits = locate_key_edits(old_key, new_key)
+        # The previous key stays in the config as old_key so an install still
+        # reaches the device if it ends up running either key
+        edits += old_key_edit(old_key)
+        unchecked = with_sharers(edits)
+    except EsphomeError as err:
+        return fail(str(err))
+
+    warnings = []
+    # With the recommended shape the ota inherits the api key, so rotating
+    # it changes what Home Assistant connects with
+    if str(static_encryption_key(config.get(CONF_API) or {})) == old_key:
+        warnings.append(
+            "The api encryption key changes too; Home Assistant will ask for "
+            "the new key after the install."
+        )
+    shared: dict[str, set[str]] = {}
+    for edit in edits:
+        what = f"secret '{edit.secret}'" if edit.secret else edit.path.name
+        for path in edit.shared_with:
+            shared.setdefault(
+                str(path.relative_to(CORE.config_dir.resolve())), set()
+            ).add(what)
+    if unchecked:
+        warnings.append(
+            "Could not check every configuration for shared use: "
+            + "; ".join(unchecked)
+        )
+    if shared:
+        warnings.append(
+            "Other configurations share what is rewritten: "
+            + "; ".join(
+                f"{p} ({', '.join(sorted(w))})" for p, w in sorted(shared.items())
+            )
+            + ". Their devices keep the previous key until their next install; "
+            "make sure those configurations carry it as old_key."
+        )
+    if warnings:
+        for warning in warnings:
+            safe_print(color(AnsiFore.BOLD_YELLOW, warning))
+        if not args.yes:
+            if not sys.stdin.isatty():
+                return fail("Confirm with --yes when there is no terminal")
+            if safe_input("Continue? [y/N] ").strip().lower() not in ("y", "yes"):
+                return 1
+
+    safe_print("Checking that the device accepts the current key...")
+    found = espota2.probe_ota_key(
+        network_devices,
+        remote_port,
+        old_key,
+        timeout=PRECHECK_TIMEOUT,
+        retry_rejected=False,
+    )
+    if found is not ProbeResult.ACCEPTED:
+        return fail(PRECHECK_ADVICE[found])
+
+    # Both keys are shown only when a file could not be put back
+    keys = f"Previous key: {old_key}\nNew key: {new_key}"
+    try:
+        originals = apply_line_edits(edits)
+    except RestoreError as err:
+        safe_print(keys)
+        return fail(str(err))
+    except EsphomeError as err:
+        return fail(str(err))
+    except KeyboardInterrupt as err:
+        if notes := getattr(err, "__notes__", None):  # a rollback that failed too
+            safe_print(keys)
+            return fail("Interrupted; " + "; ".join(notes))
+        return fail("Interrupted before the upload")
+    # From here every exit restores or reports, so nothing sits outside the try
+    uploaded = False
+    try:
+        safe_print(
+            "Wrote the new key to "
+            + ", ".join(color(AnsiFore.CYAN, str(p)) for p in originals)
+        )
+        # Global options go before the subcommand; the compile parser is strict
+        cli_args = ["--dashboard"] if CORE.dashboard else []
+        if args.toolchain:
+            cli_args += ["--toolchain", str(args.toolchain)]
+        for key, value in args.substitution or []:
+            cli_args += ["-s", key, value]
+        cli_args += ["compile", str(CORE.config_path)]
+        if run_external_process(*ESPHOME_COMMAND, *cli_args) != 0:
+            return fail("Compiling with the new key failed")
+
+        def image_sent() -> None:
+            # The device accepted the session and may commit from here on;
+            # the edited config reaches it on either key
+            nonlocal uploaded
+            uploaded = True
+
+        rc, _ = espota2.run_ota(
+            network_devices,
+            remote_port,
+            ota_conf.get(CONF_PASSWORD),
+            CORE.firmware_bin,
+            noise_psk=old_key,
+            plaintext_fallback=False,
+            on_upload_start=image_sent,
+        )
+        if rc != 0:
+            return fail(
+                "Uploading with the current key failed. The configuration keeps "
+                "the new key with the previous one as old_key, so the next "
+                "install reaches the device whichever key it runs."
+                if uploaded
+                else "The upload failed before the image was sent (see the error "
+                "above); nothing changed on the device."
+            )
+        safe_print("Waiting for the device to come back with the new key...")
+        found = espota2.probe_ota_key(network_devices, remote_port, new_key)
+        if found is not ProbeResult.ACCEPTED:
+            return fail(
+                CONFIRM_FAILURE[found]
+                + " The configuration keeps the new key, with the previous one as "
+                "old_key, so the next install reaches the device either way."
+            )
+    except KeyboardInterrupt:
+        return fail(
+            "Interrupted; the device's key was not confirmed"
+            if uploaded
+            else "Interrupted before the upload"
+        )
+    finally:
+        # Before the upload nothing changed on the device; after it the device
+        # most likely runs the new key, and old_key covers the other case
+        if uploaded:
+            safe_print(f"New OTA encryption key: {color(AnsiFore.CYAN, new_key)}")
+        else:
+            _restore(originals, keys)
+
+    safe_print(color(AnsiFore.BOLD_GREEN, "SUCCESS"))
+    return 0
+
+
+def _restore(originals: dict[Path, Snapshot], keys: str) -> None:
+    """Put the previous key back, and drop the images a build made with the
+    new key; both keys are shown when a step fails."""
+    files = ", ".join(str(p) for p in originals)
+    try:
+        restore_files(originals)
+    except (EsphomeError, KeyboardInterrupt) as err:
+        # An interrupted restore carries its report as a note
+        why = str(err) or "; ".join(getattr(err, "__notes__", ()))
+        safe_print(color(AnsiFore.BOLD_RED, why or f"Interrupted restoring {files}"))
+        safe_print(keys)
+    else:
+        safe_print(color(AnsiFore.BOLD_YELLOW, f"Restored the previous key in {files}"))
+    build_dir = CORE.firmware_bin.parent
+    failed: list[str] = []
+    images: list[Path] = []
+    try:
+        images = [p for s in (".bin", ".uf2") for p in build_dir.glob(f"*{s}")]
+        for image in images:  # every one is tried, whatever the others did
+            try:
+                image.unlink()
+            except OSError as err:
+                failed.append(f"{image}: {err}")
+    except (OSError, KeyboardInterrupt) as err:
+        failed.append(f"{build_dir}: {str(err) or 'interrupted'}")
+    if failed:
+        safe_print(
+            color(
+                AnsiFore.BOLD_RED,
+                "Could not remove images that may hold the new key: "
+                + "; ".join(failed),
+            )
+        )
+        safe_print(keys)
+    elif images:
+        safe_print(
+            color(
+                AnsiFore.BOLD_YELLOW,
+                f"Removed the images in {build_dir}, which may hold the new key; "
+                "compile again before the next upload.",
+            )
+        )

esphome/components/ade7880/ade7880.cpp

@@ -297,7 +297,9 @@ void ADE7880::init_device_() {
 void ADE7880::reset_device_() {
   if (this->reset_pin_ != nullptr) {
     ESP_LOGD(TAG, "Reset device using RESET pin");
-    pulse_reset(this->reset_pin_, {.low_ms = 1});
+    this->reset_pin_->digital_write(false);
+    delay(1);
+    this->reset_pin_->digital_write(true);
   } else {
     ESP_LOGD(TAG, "Reset device using SWRST command");
     this->write_u16_register16_(CONFIG, CONFIG_SWRST);

esphome/components/axs15231/touchscreen/axs15231_touchscreen.cpp

@@ -16,7 +16,13 @@ constexpr static const uint8_t AXS_READ_TOUCHPAD[11] = {0xb5, 0xab, 0xa5, 0x5a,
   }
 
 void AXS15231Touchscreen::setup() {
-  pulse_reset(this->reset_pin_, {.low_ms = 5, .high_after_ms = 10});
+  if (this->reset_pin_ != nullptr) {
+    this->reset_pin_->setup();
+    this->reset_pin_->digital_write(false);
+    delay(5);
+    this->reset_pin_->digital_write(true);
+    delay(10);
+  }
   if (this->interrupt_pin_ != nullptr) {
     this->interrupt_pin_->pin_mode(gpio::FLAG_INPUT);
     this->interrupt_pin_->setup();

esphome/components/bk72xx/__init__.py

@@ -78,7 +78,3 @@ async def pin_to_code(config: ConfigType) -> MockObj:
 # Called by writer.py; delegates to the shared libretiny implementation.
 def copy_files() -> None:
     libretiny.copy_files()
-
-
-# Called by the OTA upload; the shared libretiny implementation.
-ota_image = libretiny.device_ota_image

esphome/components/cst226/touchscreen/cst226_touchscreen.cpp

@@ -6,7 +6,12 @@ ESPHOME_LOG_TAG(TAG, "cst226.touchscreen");
 
 void CST226Touchscreen::setup() {
   if (this->reset_pin_ != nullptr) {
-    pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 5});
+    this->reset_pin_->setup();
+    this->reset_pin_->digital_write(true);
+    delay(5);
+    this->reset_pin_->digital_write(false);
+    delay(5);
+    this->reset_pin_->digital_write(true);
     this->set_timeout(30, [this] { this->continue_setup_(); });
   } else {
     this->continue_setup_();

esphome/components/cst328/touchscreen/cst328_touchscreen.cpp

@@ -43,7 +43,9 @@ void CST328Touchscreen::setup() {
 }
 
 void CST328Touchscreen::reset_device_() {
-  pulse_reset(this->reset_pin_, {.low_ms = 5});
+  this->reset_pin_->digital_write(false);
+  delay(5);
+  this->reset_pin_->digital_write(true);
   this->set_timeout(CST328_TRANSITION_TIMEOUT, [this] { this->continue_setup_(); });
 }
 

esphome/components/cst816/touchscreen/cst816_touchscreen.cpp

@@ -52,7 +52,12 @@ void CST816Touchscreen::continue_setup_() {
 
 void CST816Touchscreen::setup() {
   if (this->reset_pin_ != nullptr) {
-    pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 5});
+    this->reset_pin_->setup();
+    this->reset_pin_->digital_write(true);
+    delay(5);
+    this->reset_pin_->digital_write(false);
+    delay(5);
+    this->reset_pin_->digital_write(true);
     this->set_timeout(30, [this] { this->continue_setup_(); });
   } else {
     this->continue_setup_();

esphome/components/cst9220/touchscreen/cst9220_touchscreen.cpp

@@ -10,7 +10,14 @@ static constexpr uint32_t REREAD_TIMEOUT_ID = 1;
 static constexpr uint32_t REREAD_DELAY_MS = 20;
 
 void CST9220Touchscreen::setup() {
-  pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 10});
+  if (this->reset_pin_ != nullptr) {
+    this->reset_pin_->setup();
+    this->reset_pin_->digital_write(true);
+    delay(5);
+    this->reset_pin_->digital_write(false);
+    delay(10);
+    this->reset_pin_->digital_write(true);
+  }
   // Wait for the controller to leave its bootloader before talking to it.
   this->set_timeout(30, [this] { this->continue_setup_(); });
 }

esphome/components/dfrobot_sen0395/__init__.py

@@ -42,7 +42,6 @@ CONFIG_SCHEMA = cv.All(
 
 
 async def to_code(config: ConfigType) -> None:
-    cg.add_define("USE_LINE_READER")
     await uart.new_uart_component(config)
 
 

esphome/components/dfrobot_sen0395/dfrobot_sen0395.cpp

@@ -11,10 +11,10 @@
 namespace esphome::dfrobot_sen0395 {
 
 ESPHOME_LOG_TAG(TAG, "dfrobot_sen0395");
+static constexpr char ASCII_CR = 0x0D;
+static constexpr char ASCII_LF = 0x0A;
 static constexpr float DET_RANGE_STEP_M = 0.15f;
 static constexpr float LATENCY_STEP_S = 0.025f;
-// Sent after a reset with no line end
-static constexpr char PROMPT[] PROGMEM = "leapMMW:/>";
 
 // What each CommandType cannot do while the sensor runs; empty for commands that do not need it stopped
 PROGMEM_STRING_TABLE(BusyActions, "", "configure range config", "configure output latency",
@@ -121,9 +121,9 @@ bool DfrobotSen0395Component::run_command_(const Command &cmd, uint32_t now) {
     return false;  // Command not done yet
   }
 
-  if (const char *message = this->read_message_()) {
-    if (ESPHOME_strstr_P(message, ESPHOME_PSTR("is not recognized as a CLI command")) == nullptr) {
-      if (!this->handle_reply_(cmd, message))
+  if (this->read_message_()) {
+    if (ESPHOME_strstr_P(this->read_buffer_, ESPHOME_PSTR("is not recognized as a CLI command")) == nullptr) {
+      if (!this->handle_reply_(cmd))
         return false;         // Command not done yet
       this->read_message_();  // Skip the prompt
       return true;
@@ -146,13 +146,12 @@ bool DfrobotSen0395Component::run_command_(const Command &cmd, uint32_t now) {
 }
 
 void DfrobotSen0395Component::read_state_() {
-  const char *message = this->read_message_();
-  if (message == nullptr)
+  if (!this->read_message_())
     return;
-  if (ESPHOME_strstr_P(message, ESPHOME_PSTR("$JYBSS,0, , , *")) != nullptr) {
+  if (ESPHOME_strstr_P(this->read_buffer_, ESPHOME_PSTR("$JYBSS,0, , , *")) != nullptr) {
     this->set_detected_(false);
     this->set_active(true);
-  } else if (ESPHOME_strstr_P(message, ESPHOME_PSTR("$JYBSS,1, , , *")) != nullptr) {
+  } else if (ESPHOME_strstr_P(this->read_buffer_, ESPHOME_PSTR("$JYBSS,1, , , *")) != nullptr) {
     this->set_detected_(true);
     this->set_active(true);
   }
@@ -207,7 +206,8 @@ void DfrobotSen0395Component::format_command(const Command &cmd, char *buf, size
   }
 }
 
-bool DfrobotSen0395Component::handle_reply_(const Command &cmd, const char *message) {
+bool DfrobotSen0395Component::handle_reply_(const Command &cmd) {
+  const char *message = this->read_buffer_;
   auto is = [message](ESPHOME_PGM_P text) { return ESPHOME_strcmp_P(message, text) == 0; };
 
   // Commands with their own replies
@@ -230,7 +230,7 @@ bool DfrobotSen0395Component::handle_reply_(const Command &cmd, const char *mess
       }
       return true;
     case CommandType::COMMAND_TYPE_RESET_SYSTEM:
-      if (!is(PROMPT))
+      if (!is(ESPHOME_PSTR("leapMMW:/>")))
         return false;
       ESP_LOGI(TAG, "Restarted sensor.");
       return true;
@@ -300,24 +300,34 @@ bool DfrobotSen0395Component::handle_reply_(const Command &cmd, const char *mess
   return true;
 }
 
-const char *DfrobotSen0395Component::read_message_() {
+bool DfrobotSen0395Component::read_message_() {
   while (this->available()) {
     uint8_t byte;
     this->read_byte(&byte);
+
+    if (this->read_pos_ == MMWAVE_READ_BUFFER_LENGTH)
+      this->read_pos_ = 0;
+
+    ESP_LOGVV(TAG, "Buffer pos: %u %d", this->read_pos_, byte);
+
+    if (byte == ASCII_CR)
+      continue;
     if (byte >= 0x7F)
       byte = '?';  // needs to be valid utf8 string for log functions.
-    const char *line = this->line_reader_.feed(byte);
-    if (line == nullptr && byte == '>' && this->line_reader_.length() == sizeof(PROMPT) - 1) {
-      line = this->line_reader_.partial();
-      this->line_reader_.reset();
+    this->read_buffer_[this->read_pos_] = byte;
+
+    if (this->read_pos_ == 9 && byte == '>')
+      this->read_buffer_[++this->read_pos_] = ASCII_LF;
+
+    if (this->read_buffer_[this->read_pos_] == ASCII_LF) {
+      this->read_buffer_[this->read_pos_] = 0;
+      this->read_pos_ = 0;
+      ESP_LOGV(TAG, "Message: %s", this->read_buffer_);
+      return true;  // Full message in buffer
     }
-    if (line != nullptr) {
-      ESP_LOGV(TAG, "Message: %s", line);
-      return line;
-    }
-    this->line_reader_.log_if_overflowed(TAG);
+    this->read_pos_++;
   }
-  return nullptr;
+  return false;  // No full message yet
 }
 
 void DfrobotSen0395Component::set_detected_(bool detected) {

esphome/components/dfrobot_sen0395/dfrobot_sen0395.h

@@ -4,7 +4,6 @@
 #include "esphome/core/automation.h"
 #include "esphome/core/component.h"
 #include "esphome/core/helpers.h"
-#include "esphome/core/line_reader.h"
 
 #ifdef USE_BINARY_SENSOR
 #include "esphome/components/binary_sensor/binary_sensor.h"
@@ -15,7 +14,7 @@
 
 namespace esphome::dfrobot_sen0395 {
 
-static constexpr uint8_t MMWAVE_READ_BUFFER_LENGTH = 253;  // largest LineReader size with a one byte position
+static constexpr uint8_t MMWAVE_READ_BUFFER_LENGTH = 255;
 static constexpr uint8_t COMMAND_QUEUE_SIZE = 20;
 
 // Keep in sync with the BusyActions table in dfrobot_sen0395.cpp
@@ -122,7 +121,8 @@ class DfrobotSen0395Component final : public uart::UARTDevice, public Component
   bool led_active_{false};
   bool uart_presence_active_{false};
   bool start_after_boot_{false};
-  LineReader<MMWAVE_READ_BUFFER_LENGTH> line_reader_;
+  char read_buffer_[MMWAVE_READ_BUFFER_LENGTH];
+  size_t read_pos_{0};
   bool cmd_sent_{false};
   uint8_t retries_left_{2};
   StaticRingBuffer<Command, COMMAND_QUEUE_SIZE> cmd_queue_;
@@ -131,13 +131,12 @@ class DfrobotSen0395Component final : public uart::UARTDevice, public Component
   int16_t det_range_steps_[8]{};
   uint16_t latency_steps_[2]{};
 
-  /// The next complete line, valid until the next call, or nullptr
-  const char *read_message_();
+  bool read_message_();
   void read_state_();
   /// Run the command at the front of the queue; true when it is done
   bool run_command_(const Command &cmd, uint32_t now);
   /// Handle a reply to a sent command; true when the command is done
-  bool handle_reply_(const Command &cmd, const char *message);
+  bool handle_reply_(const Command &cmd);
 
   void set_detected_(bool detected);
 };

esphome/components/ektf2232/touchscreen/ektf2232.cpp

@@ -20,7 +20,9 @@ void EKTF2232Touchscreen::setup() {
 
   this->attach_interrupt_(this->interrupt_pin_, gpio::INTERRUPT_FALLING_EDGE);
 
-  pulse_reset(this->reset_pin_, {.low_ms = 15, .high_after_ms = 15});
+  this->reset_pin_->setup();
+
+  this->hard_reset_();
   if (!this->soft_reset_()) {
     ESP_LOGE(TAG, "Failed to soft reset EKT2232!");
     this->interrupt_pin_->detach_interrupt();
@@ -94,6 +96,13 @@ bool EKTF2232Touchscreen::get_power_state() {
   return (received[1] >> 3) & 1;
 }
 
+void EKTF2232Touchscreen::hard_reset_() {
+  this->reset_pin_->digital_write(false);
+  delay(15);
+  this->reset_pin_->digital_write(true);
+  delay(15);
+}
+
 bool EKTF2232Touchscreen::soft_reset_() {
   auto err = this->write(SOFT_RESET_CMD, 4);
   if (err != i2c::ERROR_OK)

esphome/components/ektf2232/touchscreen/ektf2232.h

@@ -22,6 +22,7 @@ class EKTF2232Touchscreen final : public Touchscreen, public i2c::I2CDevice {
   bool get_power_state();
 
  protected:
+  void hard_reset_();
   bool soft_reset_();
   void update_touches() override;
 

esphome/components/epaper_spi/epaper_spi_jd79660.cpp

@@ -72,7 +72,9 @@ bool EPaperJD79660::reset() {
 
       // As commented on SLEEP_MS_RESET1: Reset pulse must happen within time window.
       // So do not use FSM loop, and avoid other calls/logs during pulse below.
-      pulse_reset(this->reset_pin_, {.low_ms = SLEEP_MS_RESET1});
+      this->reset_pin_->digital_write(false);
+      delay(SLEEP_MS_RESET1);
+      this->reset_pin_->digital_write(true);
 
       this->reset_duration_ = SLEEP_MS_RESET2;
       this->step_ = FSMState::RESET_STEP2_IDLECHECK;

esphome/components/ft63x6/ft63x6.cpp

@@ -34,7 +34,10 @@ void FT63X6Touchscreen::setup() {
     this->attach_interrupt_(this->interrupt_pin_, gpio::INTERRUPT_ANY_EDGE);
   }
 
-  pulse_reset(this->reset_pin_, {.low_ms = 10});
+  if (this->reset_pin_ != nullptr) {
+    this->reset_pin_->setup();
+    this->hard_reset_();
+  }
 
   // Get touch resolution
   if (this->x_raw_max_ == this->x_raw_min_) {
@@ -54,6 +57,14 @@ void FT63X6Touchscreen::setup() {
   this->write_byte(FT6X36_ADDR_TOUCHRATE_ACTIVE, 0x0E);
 }
 
+void FT63X6Touchscreen::hard_reset_() {
+  if (this->reset_pin_ != nullptr) {
+    this->reset_pin_->digital_write(false);
+    delay(10);
+    this->reset_pin_->digital_write(true);
+  }
+}
+
 void FT63X6Touchscreen::dump_config() {
   ESP_LOGCONFIG(TAG, "FT63X6 Touchscreen:");
   LOG_I2C_DEVICE(this);

esphome/components/ft63x6/ft63x6.h

@@ -27,6 +27,7 @@ class FT63X6Touchscreen final : public Touchscreen, public i2c::I2CDevice {
   void set_threshold(uint8_t threshold) { this->threshold_ = threshold; }
 
  protected:
+  void hard_reset_();
   void update_touches() override;
 
   InternalGPIOPin *interrupt_pin_{nullptr};

esphome/components/gsl3670/gsl3670_touchscreen.cpp

@@ -83,7 +83,13 @@ void GSL3670Touchscreen::update_touches() {
 void GSL3670Touchscreen::clear_reg_() {
   ESP_LOGD(TAG, "clear_reg");
 
-  pulse_reset(this->reset_pin_, {.low_ms = 1, .high_after_ms = 5});
+  // GPIO reset pulse
+  if (this->reset_pin_ != nullptr) {
+    this->reset_pin_->digital_write(false);
+    delay(1);
+    this->reset_pin_->digital_write(true);
+    delay(5);
+  }
 
   this->write_reg8_(0x88, 0x01);
   // delay(5);
@@ -100,7 +106,12 @@ void GSL3670Touchscreen::clear_reg_() {
 void GSL3670Touchscreen::reset_() {
   ESP_LOGD(TAG, "reset");
 
-  pulse_reset(this->reset_pin_, {.low_ms = 1, .high_after_ms = 5});
+  if (this->reset_pin_ != nullptr) {
+    this->reset_pin_->digital_write(false);
+    delay(1);
+    this->reset_pin_->digital_write(true);
+    delay(5);
+  }
 
   this->write_reg8_(0xe4, 0x04);
 

esphome/components/icnt86/icnt86.cpp

@@ -19,7 +19,12 @@ void ICNT86Touchscreen::setup() {
   }
 
   // Perform reset if necessary
-  pulse_reset(this->reset_pin_, {.low_ms = 10});
+  if (this->reset_pin_ != nullptr) {
+    this->reset_pin_->setup();
+    this->reset_pin_->digital_write(false);
+    delay(10);
+    this->reset_pin_->digital_write(true);
+  }
 
   if (this->x_raw_max_ == this->x_raw_min_) {
     this->x_raw_max_ = this->display_->get_native_width();

esphome/components/ir_rf_base/__init__.py

@@ -1,7 +1,8 @@
 """Shared base for the infrared and radio_frequency entity components."""
 
 import esphome.codegen as cg
-from esphome.const import CONF_API
+from esphome.components import remote_base
+from esphome.components.api import DOMAIN as API_DOMAIN
 from esphome.core import CORE
 from esphome.core.entity_helpers import queue_entity_register
 from esphome.types import ConfigType
@@ -23,10 +24,8 @@ async def attach_transmitter(var: cg.MockObj, config: ConfigType, key: str) -> N
     does. Without this call anywhere in the build a request is answered as
     soon as the frame is handed over.
     """
-    from esphome.components import remote_base
-
     await remote_base.register_transmittable(var, config, key)
-    if CONF_API in CORE.config:
+    if API_DOMAIN in CORE.config:
         cg.add_define("USE_IR_RF_TRANSMIT_COMPLETE")
 
 

esphome/components/libretiny/__init__.py

@@ -26,7 +26,7 @@ from esphome.const import (
 )
 from esphome.core import CORE
 from esphome.core.config import BOARD_MAX_LENGTH
-from esphome.helpers import copy_file_if_changed, write_file
+from esphome.helpers import copy_file_if_changed
 from esphome.platformio.toolchain import copy_ccache_script
 from esphome.storage_json import StorageJSON
 
@@ -656,26 +656,6 @@ async def component_to_code(config):
     await cg.register_component(var, config)
 
 
-# Called by the OTA upload through the platforms' ota_image hook
-def device_ota_image(firmware: Path) -> Path:
-    """The file to send for an app update: the UF2 without the images the
-    device skips, written next to the build output, or ``firmware`` itself
-    when there is nothing to drop."""
-    from .ota_uf2 import device_ota_uf2
-
-    data = firmware.read_bytes()
-    if (trimmed := device_ota_uf2(data)) is None:
-        return firmware
-    image = CORE.relative_pioenvs_path(CORE.name, "firmware.ota.uf2")
-    write_file(image, trimmed)
-    _LOGGER.info(
-        "Sending the images this device writes: %s of %s bytes",
-        len(trimmed),
-        len(data),
-    )
-    return image
-
-
 # Called by writer.py
 def copy_files() -> None:
     script_dir = Path(__file__).parent

esphome/components/libretiny/generate_components.py

@@ -91,10 +91,6 @@ async def pin_to_code(config: ConfigType) -> MockObj:
 # Called by writer.py; delegates to the shared libretiny implementation.
 def copy_files() -> None:
     libretiny.copy_files()
-
-
-# Called by the OTA upload; the shared libretiny implementation.
-ota_image = libretiny.device_ota_image
 '''
 
 BASE_CODE_BOARDS = '''

esphome/components/libretiny/ota_uf2.py

@@ -1,98 +0,0 @@
-"""Trim a LibreTiny UF2 package to the images a device writes over OTA.
-
-A LibreTiny UF2 holds one image per OTA scheme: the device schemes (single,
-dual slot 1, dual slot 2) and the flasher schemes used over UART. The device
-skips every block of an image whose scheme it does not run (uf2ota), so an
-image no device scheme writes is only sent to be thrown away. On single OTA
-chips (BK72xx, LN882H) that is the whole flasher image, more than half the
-package. The legacy OTA format names the same device partitions, so devices
-that read it write the same images.
-"""
-
-import struct
-
-UF2_BLOCK_SIZE = 512
-UF2_MAGIC_HEAD = struct.pack("<2I", 0x0A324655, 0x9E5D5157)
-UF2_MAGIC_TAIL = struct.pack("<I", 0x0AB16F30)
-UF2_FLAG_HAS_TAGS = 0x8000
-# Header words after the magic: flags, target address, payload length, block
-# number, block count
-_HEADER_WORDS = struct.Struct("<5I")
-_SEQ_OFFSET = 20
-_DATA_START = 32
-_DATA_END = UF2_BLOCK_SIZE - len(UF2_MAGIC_TAIL)
-
-# One partition index nibble per scheme, high nibble first: device single,
-# dual 1 and dual 2 are the first three, the flasher schemes the next three
-UF2_TAG_OTA_PART_INFO = 0xC0EE0C
-# Tags uf2ota acts on beyond the image they sit in; one in an image that
-# would be dropped leaves the package untrimmed
-_KEEP_TAGS = frozenset(
-    {
-        0x5D57D0,  # OTA_FORMAT_1
-        0x6C8492,  # OTA_FORMAT_2
-        0x6EC68A,  # OTA_PART_LIST
-        0xB948DE,  # BINPATCH
-        0x8288ED,  # FAL_PTABLE
-    }
-)
-
-
-def _tags(block: bytes, length: int) -> dict[int, bytes] | None:
-    """The extension tags after a block's payload, or None if malformed."""
-    tags: dict[int, bytes] = {}
-    pos = _DATA_START + length
-    while pos + 4 <= _DATA_END and (size := block[pos]):
-        if size < 4 or pos + size > _DATA_END:
-            return None
-        tags[int.from_bytes(block[pos + 1 : pos + 4], "little")] = block[
-            pos + 4 : pos + size
-        ]
-        pos += (size + 3) & ~3
-    return tags
-
-
-def device_ota_uf2(data: bytes) -> bytes | None:
-    """``data`` without the images no device scheme writes.
-
-    None when there is nothing to drop, or when ``data`` is not a well
-    formed UF2, so the caller sends the file as it is.
-    """
-    if len(data) % UF2_BLOCK_SIZE:
-        return None
-    count = len(data) // UF2_BLOCK_SIZE
-    kept: list[bytes] = []
-    keep = True
-    for seq in range(count):
-        block = data[seq * UF2_BLOCK_SIZE : (seq + 1) * UF2_BLOCK_SIZE]
-        flags, _, length, block_seq, block_count = _HEADER_WORDS.unpack_from(block, 8)
-        # A missing or repeated block must stay visible to the device, which
-        # takes the update size from the block count
-        if (
-            block[:8] != UF2_MAGIC_HEAD
-            or block[_DATA_END:] != UF2_MAGIC_TAIL
-            or _DATA_START + length > _DATA_END
-            or (block_seq, block_count) != (seq, count)
-        ):
-            return None
-        tags = _tags(block, length) if flags & UF2_FLAG_HAS_TAGS else {}
-        if tags is None:
-            return None
-        if (part_info := tags.get(UF2_TAG_OTA_PART_INFO)) is not None:
-            # uf2ota rejects a selector without the whole scheme map
-            if len(part_info) < 3:
-                return None
-            # Any of the three device scheme nibbles names a partition
-            keep = int.from_bytes(part_info[:2], "big") >> 4 != 0
-        if keep:
-            kept.append(block)
-        elif not _KEEP_TAGS.isdisjoint(tags):
-            # Without its image's OTA_PART_INFO the device would write this
-            # block to the partition of the image before it
-            return None
-    if len(kept) == count:
-        return None
-    out = bytearray(b"".join(kept))
-    for seq in range(len(kept)):
-        struct.pack_into("<2I", out, seq * UF2_BLOCK_SIZE + _SEQ_OFFSET, seq, len(kept))
-    return bytes(out)

esphome/components/ln882x/__init__.py

@@ -77,7 +77,3 @@ async def pin_to_code(config: ConfigType) -> MockObj:
 # Called by writer.py; delegates to the shared libretiny implementation.
 def copy_files() -> None:
     libretiny.copy_files()
-
-
-# Called by the OTA upload; the shared libretiny implementation.
-ota_image = libretiny.device_ota_image

esphome/components/logger/__init__.py

@@ -607,10 +607,9 @@ async def _late_logger_init(config: ConfigType) -> None:
         # esp_rom_printf instead of pulling ~1.2KB esp_rom_vprintf into IRAM.
         # S2/C2 have a ROM vprintf, so IDF forces CONSTRAINED_ENV_SAFE=1 and
         # cache-off DRAM logs would reach the flash-resident wrapper; keep V1.
-        if (
-            CORE.using_toolchain_esp_idf
-            and idf_version() >= cv.Version(6, 1, 0)
-            and get_esp32_variant() not in (VARIANT_ESP32C2, VARIANT_ESP32S2)
+        if idf_version() >= cv.Version(6, 1, 0) and get_esp32_variant() not in (
+            VARIANT_ESP32C2,
+            VARIANT_ESP32S2,
         ):
             add_idf_sdkconfig_option("CONFIG_LOG_VERSION_1", False)
             add_idf_sdkconfig_option("CONFIG_LOG_VERSION_2", True)

esphome/components/media_player/__init__.py

@@ -2,6 +2,7 @@ from collections.abc import Callable
 
 from esphome import automation
 import esphome.codegen as cg
+from esphome.components import audio
 from esphome.components.const import CONF_BITRATE
 import esphome.config_validation as cv
 from esphome.const import (
@@ -115,8 +116,6 @@ def validate_preferred_format(
     """
 
     def validator(config: ConfigType) -> ConfigType:
-        from esphome.components import audio
-
         # Inherit settings from audio device if not manually set
         inherit_property_from(CONF_NUM_CHANNELS, audio_device_key)(config)
         inherit_property_from(CONF_SAMPLE_RATE, audio_device_key)(config)
@@ -156,8 +155,6 @@ def request_codecs_for_format_configs(
 
     Public API for external components. Do not remove.
     """
-    from esphome.components import audio
-
     needed_formats: set[str] = set()
     need_all = False
 

esphome/components/mipi_dsi/mipi_dsi.cpp

@@ -107,7 +107,12 @@ void MipiDsi::setup() {
   }
 #endif
   if (this->reset_pin_ != nullptr) {
-    pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 5});
+    this->reset_pin_->setup();
+    this->reset_pin_->digital_write(true);
+    delay(5);
+    this->reset_pin_->digital_write(false);
+    delay(5);
+    this->reset_pin_->digital_write(true);
   } else {
     esp_lcd_panel_io_tx_param(this->io_handle_, SW_RESET_CMD, nullptr, 0);
   }

esphome/components/mipi_rgb/mipi_rgb.cpp

@@ -33,7 +33,14 @@ void MipiRgb::setup_enables_() {
     }
     delay(10);
   }
-  pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 5});
+  if (this->reset_pin_ != nullptr) {
+    this->reset_pin_->setup();
+    this->reset_pin_->digital_write(true);
+    delay(5);
+    this->reset_pin_->digital_write(false);
+    delay(5);
+    this->reset_pin_->digital_write(true);
+  }
 }
 
 #ifdef USE_SPI

esphome/components/mipi_spi/mipi_spi.h

@@ -151,8 +151,15 @@ class MipiSpi : public display::Display,
       pin->setup();
       pin->digital_write(true);
     }
-    // required delay after reset is already in the init sequence, don't duplicate
-    pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 5});
+    if (this->reset_pin_ != nullptr) {
+      this->reset_pin_->setup();
+      this->reset_pin_->digital_write(true);
+      delay(5);
+      this->reset_pin_->digital_write(false);
+      delay(5);
+      this->reset_pin_->digital_write(true);
+      // required delay after reset is already in the init sequence, don't duplicate
+    }
 
     // need to know when the display is ready for SLPOUT command - will be 120ms after reset
     auto when = millis() + 120;

esphome/components/openthread/__init__.py

@@ -226,12 +226,6 @@ def set_sdkconfig_options(config: ConfigType) -> None:
     add_idf_sdkconfig_option("CONFIG_OPENTHREAD_DNS64_CLIENT", not border_router)
     add_idf_sdkconfig_option("CONFIG_OPENTHREAD_SRP_CLIENT", not border_router)
     if not border_router:
-        # The DNS64 hook breaks IPv6 literal lookups; see openthread_esp.cpp.
-        # Fixed upstream on ESP-IDF master, remove this once a release includes
-        # "fix(openthread): bypass DNS64 synthesis for IPv6 literals",
-        # https://github.com/espressif/esp-idf/commit/87b133ee1a7e91c1ee95176b4733b7f72236e5da
-        # (merged in https://github.com/espressif/esp-idf/commit/83239509832bf5d427935cd365f4b62a320e8765)
-        cg.add_build_flag("-Wl,--wrap=lwip_hook_dns_external_resolve")
         add_idf_sdkconfig_option("CONFIG_OPENTHREAD_SRP_CLIENT_MAX_SERVICES", 5)
 
     if border_router:

esphome/components/openthread/openthread_esp.cpp

@@ -35,10 +35,6 @@
 #include "esp_vfs_eventfd.h"
 #include "freertos/FreeRTOS.h"
 #include "freertos/task.h"
-#ifndef USE_OPENTHREAD_BORDER_ROUTER
-#include "lwip/dns.h"
-#include "lwip/ip_addr.h"
-#endif
 
 ESPHOME_LOG_TAG(TAG, "openthread");
 
@@ -493,26 +489,4 @@ InstanceLock::~InstanceLock() {
 }
 
 }  // namespace esphome::openthread
-
-#ifndef USE_OPENTHREAD_BORDER_ROUTER
-// ESP-IDF's DNS64 hook runs before lwIP's literal address check and retries every lookup as IPv4,
-// so an IPv6 literal never resolves. Let lwIP handle those; linked with --wrap from __init__.py.
-// Fixed upstream on ESP-IDF master, remove this once a release includes
-// "fix(openthread): bypass DNS64 synthesis for IPv6 literals",
-// https://github.com/espressif/esp-idf/commit/87b133ee1a7e91c1ee95176b4733b7f72236e5da
-// (merged in https://github.com/espressif/esp-idf/commit/83239509832bf5d427935cd365f4b62a320e8765)
-// NOLINTBEGIN(bugprone-reserved-identifier,cert-dcl37-c,cert-dcl51-cpp,readability-identifier-naming)
-extern "C" {
-int __real_lwip_hook_dns_external_resolve(const char *name, ip_addr_t *addr, dns_found_callback found,
-                                          void *callback_arg, u8_t addrtype, err_t *err);
-int __wrap_lwip_hook_dns_external_resolve(const char *name, ip_addr_t *addr, dns_found_callback found,
-                                          void *callback_arg, u8_t addrtype, err_t *err) {
-  ip_addr_t literal;
-  if (ipaddr_aton(name, &literal) && IP_IS_V6_VAL(literal))
-    return 0;
-  return __real_lwip_hook_dns_external_resolve(name, addr, found, callback_arg, addrtype, err);
-}
-}  // extern "C"
-// NOLINTEND(bugprone-reserved-identifier,cert-dcl37-c,cert-dcl51-cpp,readability-identifier-naming)
-#endif  // USE_OPENTHREAD_BORDER_ROUTER
 #endif

esphome/components/pcd8544/pcd_8544.cpp

@@ -9,10 +9,23 @@ ESPHOME_LOG_TAG(TAG, "pcd_8544");
 
 void PCD8544::setup_pins_() {
   this->spi_setup();
-  pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10});
+  this->init_reset_();
   this->dc_pin_->setup();
 }
 
+void PCD8544::init_reset_() {
+  if (this->reset_pin_ != nullptr) {
+    this->reset_pin_->setup();
+    this->reset_pin_->digital_write(true);
+    delay(1);
+    // Trigger Reset
+    this->reset_pin_->digital_write(false);
+    delay(10);
+    // Wake up
+    this->reset_pin_->digital_write(true);
+  }
+}
+
 void PCD8544::initialize() {
   this->init_internal_(this->get_buffer_length_());
 

esphome/components/pcd8544/pcd_8544.h

@@ -57,6 +57,8 @@ class PCD8544 final : public display::DisplayBuffer,
 
   void setup_pins_();
 
+  void init_reset_();
+
   size_t get_buffer_length_();
 
   void start_command_();

esphome/components/rtl87xx/__init__.py

@@ -77,7 +77,3 @@ async def pin_to_code(config: ConfigType) -> MockObj:
 # Called by writer.py; delegates to the shared libretiny implementation.
 def copy_files() -> None:
     libretiny.copy_files()
-
-
-# Called by the OTA upload; the shared libretiny implementation.
-ota_image = libretiny.device_ota_image

esphome/components/sh1122_base/sh1122_base.cpp

@@ -122,7 +122,18 @@ void SH1122::fill(Color color) {
   const uint8_t fill = (color4 & SH1122_COLORMASK) | ((color4 & SH1122_COLORMASK) << SH1122_COLORSHIFT);
   memset(this->buffer_, fill, this->get_buffer_length_());
 }
-void SH1122::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
+void SH1122::init_reset_() {
+  if (this->reset_pin_ != nullptr) {
+    this->reset_pin_->setup();
+    this->reset_pin_->digital_write(true);
+    delay(1);
+    // Trigger Reset
+    this->reset_pin_->digital_write(false);
+    delay(10);
+    // Wake up
+    this->reset_pin_->digital_write(true);
+  }
+}
 const LogString *SH1122::model_str_() {
   switch (this->model_) {
     case SH1122_MODEL_256_64:

esphome/components/ssd1306_base/ssd1306_base.cpp

@@ -355,7 +355,18 @@ void SSD1306::fill(Color color) {
   for (uint32_t i = 0; i < this->get_buffer_length_(); i++)
     this->buffer_[i] = fill;
 }
-void SSD1306::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
+void SSD1306::init_reset_() {
+  if (this->reset_pin_ != nullptr) {
+    this->reset_pin_->setup();
+    this->reset_pin_->digital_write(true);
+    delay(1);
+    // Trigger Reset
+    this->reset_pin_->digital_write(false);
+    delay(10);
+    // Wake up
+    this->reset_pin_->digital_write(true);
+  }
+}
 const LogString *SSD1306::model_str_() {
   return ModelStrings::get_log_str(static_cast<uint8_t>(this->model_), ModelStrings::LAST_INDEX);
 }

esphome/components/ssd1322_base/ssd1322_base.cpp

@@ -184,7 +184,18 @@ void SSD1322::fill(Color color) {
   for (uint32_t i = 0; i < this->get_buffer_length_(); i++)
     this->buffer_[i] = fill;
 }
-void SSD1322::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
+void SSD1322::init_reset_() {
+  if (this->reset_pin_ != nullptr) {
+    this->reset_pin_->setup();
+    this->reset_pin_->digital_write(true);
+    delay(1);
+    // Trigger Reset
+    this->reset_pin_->digital_write(false);
+    delay(10);
+    // Wake up
+    this->reset_pin_->digital_write(true);
+  }
+}
 const char *SSD1322::model_str_() {
   switch (this->model_) {
     case SSD1322_MODEL_256_64:

esphome/components/ssd1325_base/ssd1325_base.cpp

@@ -211,7 +211,18 @@ void SSD1325::fill(Color color) {
   for (uint32_t i = 0; i < this->get_buffer_length_(); i++)
     this->buffer_[i] = fill;
 }
-void SSD1325::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
+void SSD1325::init_reset_() {
+  if (this->reset_pin_ != nullptr) {
+    this->reset_pin_->setup();
+    this->reset_pin_->digital_write(true);
+    delay(1);
+    // Trigger Reset
+    this->reset_pin_->digital_write(false);
+    delay(10);
+    // Wake up
+    this->reset_pin_->digital_write(true);
+  }
+}
 const char *SSD1325::model_str_() {
   switch (this->model_) {
     case SSD1325_MODEL_128_32:

esphome/components/ssd1327_base/ssd1327_base.cpp

@@ -160,7 +160,18 @@ void SSD1327::fill(Color color) {
   for (uint32_t i = 0; i < this->get_buffer_length_(); i++)
     this->buffer_[i] = fill;
 }
-void SSD1327::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
+void SSD1327::init_reset_() {
+  if (this->reset_pin_ != nullptr) {
+    this->reset_pin_->setup();
+    this->reset_pin_->digital_write(true);
+    delay(1);
+    // Trigger Reset
+    this->reset_pin_->digital_write(false);
+    delay(10);
+    // Wake up
+    this->reset_pin_->digital_write(true);
+  }
+}
 const char *SSD1327::model_str_() {
   switch (this->model_) {
     case SSD1327_MODEL_128_128:

esphome/components/ssd1331_base/ssd1331_base.cpp

@@ -142,6 +142,17 @@ void SSD1331::fill(Color color) {
     }
   }
 }
-void SSD1331::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
+void SSD1331::init_reset_() {
+  if (this->reset_pin_ != nullptr) {
+    this->reset_pin_->setup();
+    this->reset_pin_->digital_write(true);
+    delay(1);
+    // Trigger Reset
+    this->reset_pin_->digital_write(false);
+    delay(10);
+    // Wake up
+    this->reset_pin_->digital_write(true);
+  }
+}
 
 }  // namespace esphome::ssd1331_base

esphome/components/ssd1351_base/ssd1351_base.cpp

@@ -174,7 +174,18 @@ void SSD1351::fill(Color color) {
     }
   }
 }
-void SSD1351::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
+void SSD1351::init_reset_() {
+  if (this->reset_pin_ != nullptr) {
+    this->reset_pin_->setup();
+    this->reset_pin_->digital_write(true);
+    delay(1);
+    // Trigger Reset
+    this->reset_pin_->digital_write(false);
+    delay(10);
+    // Wake up
+    this->reset_pin_->digital_write(true);
+  }
+}
 const char *SSD1351::model_str_() {
   switch (this->model_) {
     case SSD1351_MODEL_128_96:

esphome/components/st7123/touchscreen/st7123_touchscreen.cpp

@@ -9,8 +9,12 @@ ESPHOME_LOG_TAG(TAG, "st7123.touchscreen");
 
 void ST7123Touchscreen::setup() {
   if (this->reset_pin_ != nullptr) {
-    // TP_RESX is active low, assert for at least tRSTW (2ms)
-    pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 5});
+    this->reset_pin_->setup();
+    this->reset_pin_->digital_write(true);
+    delay(5);
+    this->reset_pin_->digital_write(false);  // TP_RESX is active low, assert for at least tRSTW (2ms)
+    delay(5);
+    this->reset_pin_->digital_write(true);
     // The controller needs up to 20ms to initialize after reset before it can be accessed.
     this->setup_time_ = millis() + 30;
   }

esphome/components/st7567_base/st7567_base.cpp

@@ -137,7 +137,18 @@ void ST7567::fill(Color color) {
   memset(buffer_, fill, this->get_buffer_length_());
 }
 
-void ST7567::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
+void ST7567::init_reset_() {
+  if (this->reset_pin_ != nullptr) {
+    this->reset_pin_->setup();
+    this->reset_pin_->digital_write(true);
+    delay(1);
+    // Trigger Reset
+    this->reset_pin_->digital_write(false);
+    delay(10);
+    // Wake up
+    this->reset_pin_->digital_write(true);
+  }
+}
 
 const char *ST7567::model_str_() { return "ST7567 128x64"; }
 

esphome/components/tt21100/touchscreen/tt21100.cpp

@@ -54,7 +54,10 @@ void TT21100Touchscreen::setup() {
   }
 
   // Perform reset if necessary
-  pulse_reset(this->reset_pin_, {.low_ms = 10, .high_after_ms = 10});
+  if (this->reset_pin_ != nullptr) {
+    this->reset_pin_->setup();
+    this->reset_();
+  }
 
   // Update display dimensions if they were updated during display setup
   if (this->display_ != nullptr) {
@@ -119,6 +122,15 @@ void TT21100Touchscreen::update_touches() {
   }
 }
 
+void TT21100Touchscreen::reset_() {
+  if (this->reset_pin_ != nullptr) {
+    this->reset_pin_->digital_write(false);
+    delay(10);
+    this->reset_pin_->digital_write(true);
+    delay(10);
+  }
+}
+
 void TT21100Touchscreen::dump_config() {
   ESP_LOGCONFIG(TAG, "TT21100 Touchscreen:");
   LOG_I2C_DEVICE(this);

esphome/components/tt21100/touchscreen/tt21100.h

@@ -28,6 +28,8 @@ class TT21100Touchscreen final : public Touchscreen, public i2c::I2CDevice {
   void register_button_listener(TT21100ButtonListener *listener) { this->button_listeners_.push_back(listener); }
 
  protected:
+  void reset_();
+
   void update_touches() override;
 
   InternalGPIOPin *interrupt_pin_;

esphome/components/wifi/__init__.py

@@ -245,22 +245,6 @@ WiFiConfigureAction = wifi_ns.class_(
 )
 
 
-# The supplicant parses the bootstrapping key as a DER encoded P-256 private key in hex; these wrap the
-# 32 byte key the way the ESP-IDF enrollee example does
-DPP_KEY_DER_PREFIX = "30310201010420"
-DPP_KEY_DER_SUFFIX = "a00a06082a8648ce3d030107"
-
-
-def _validate_dpp_private_key(value: object) -> str:
-    """The key is 32 bytes written as 64 hexadecimal characters."""
-    value = cv.string_strict(value)
-    if len(value) != 64 or any(c not in "0123456789abcdefABCDEF" for c in value):
-        raise cv.Invalid(
-            "private_key must be 32 bytes written as 64 hexadecimal characters"
-        )
-    return value.lower()
-
-
 def _validate_dpp_device_info(value: object) -> str:
     """The supplicant puts the text on a space separated command line and reads it up to the next space."""
     value = cv.string_strict(value)
@@ -338,7 +322,7 @@ EAP_AUTH_SCHEMA = cv.All(
 # Wi-Fi Easy Connect (DPP) enrollee; the ESP-IDF supplicant supports the QR code method
 DPP_SCHEMA = cv.Schema(
     {
-        cv.Optional(CONF_PRIVATE_KEY): cv.sensitive(_validate_dpp_private_key),
+        cv.Optional(CONF_PRIVATE_KEY): cv.sensitive(cv.string_strict),
         cv.Optional(CONF_DEVICE_INFO): _validate_dpp_device_info,
         cv.Optional(CONF_CHANNELS, default=[6]): cv.ensure_list(
             cv.int_range(min=1, max=14)
@@ -920,7 +904,7 @@ async def to_code(config):
         add_idf_sdkconfig_option("CONFIG_ESP_WIFI_DPP_SUPPORT", True)
         add_idf_sdkconfig_option("CONFIG_ESP_HOSTED_HOST_FEAT_WIFI_EXT_DPP", True)
         if (key := dpp.get(CONF_PRIVATE_KEY)) is not None:
-            cg.add(var.set_dpp_key(f"{DPP_KEY_DER_PREFIX}{key}{DPP_KEY_DER_SUFFIX}"))
+            cg.add(var.set_dpp_key(key))
         if (info := dpp.get(CONF_DEVICE_INFO)) is not None:
             cg.add(var.set_dpp_device_info(info))
         cg.add(var.set_dpp_channels(",".join(str(c) for c in dpp[CONF_CHANNELS])))

esphome/core/gpio.cpp

@@ -1,5 +1,4 @@
 #include "esphome/core/gpio.h"
-#include "esphome/core/hal.h"
 #include "esphome/core/log.h"
 
 namespace esphome {
@@ -19,18 +18,4 @@ void log_pin(const char *tag, const char *prefix, GPIOPin *pin) {
 }
 #endif
 
-namespace internal {
-void pulse_reset_ms(GPIOPin *pin, uint32_t high_before_ms, uint32_t low_ms, uint32_t high_after_ms) {
-  if (pin == nullptr)
-    return;
-  pin->setup();
-  const uint32_t waits[] = {high_before_ms, low_ms, high_after_ms};
-  // Level for each step is high, low, high; a zero high_before_ms skips the first step
-  for (size_t i = high_before_ms == 0; i < 3; i++) {
-    pin->digital_write(i != 1);
-    delay(waits[i]);
-  }
-}
-}  // namespace internal
-
 }  // namespace esphome

esphome/core/gpio_pin.h

@@ -144,24 +144,4 @@ void log_pin(const char *tag, const __FlashStringHelper *prefix, GPIOPin *pin);
 void log_pin(const char *tag, const char *prefix, GPIOPin *pin);
 #endif
 
-/// Waveform for pulse_reset(): high, low, then high again. Name the fields at the call site, e.g.
-/// pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}).
-struct ResetPulse {
-  uint32_t high_before_ms{0};  ///< Drive high this long first; skipped when 0
-  uint32_t low_ms{0};          ///< Hold the pin low (in reset) this long
-  uint32_t high_after_ms{0};   ///< Wait this long after releasing the pin
-};
-
-namespace internal {
-/// Out-of-line body of pulse_reset(); call that instead.
-void pulse_reset_ms(GPIOPin *pin, uint32_t high_before_ms, uint32_t low_ms, uint32_t high_after_ms);
-}  // namespace internal
-
-/// Set up an active-low reset pin and drive it through \p pulse. Blocks for the whole sequence;
-/// does nothing when pin is null. Inline so the fields reach pulse_reset_ms() in registers
-/// (RISC-V would otherwise pass the struct by reference).
-ESPHOME_ALWAYS_INLINE inline void pulse_reset(GPIOPin *pin, ResetPulse pulse) {
-  internal::pulse_reset_ms(pin, pulse.high_before_ms, pulse.low_ms, pulse.high_after_ms);
-}
-
 }  // namespace esphome

esphome/ota_key_edit.py

@@ -0,0 +1,385 @@
+"""Rewrite the OTA encryption key in place: the `key:` line the loader read
+it from, or the secrets.yaml line a `!secret` on it points to."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+import logging
+import os
+from pathlib import Path
+import re
+
+from esphome.components.noise import static_encryption_key
+from esphome.const import CONF_API, CONF_ENCRYPTION, CONF_KEY
+from esphome.core import CORE, EsphomeError
+from esphome.espota2 import CONF_OLD_KEY, esphome_ota_items
+from esphome.types import ConfigType
+from esphome.yaml_edit import (
+    LineEdit,
+    editable_file,
+    field_line_re,
+    own_documents,
+    read_text,
+    rewrite,
+    source_line,
+    source_of,
+)
+from esphome.yaml_util import secrets_path_for
+
+_LOGGER = logging.getLogger(__name__)
+_DOMAIN = "ota_key_edit"
+
+
+@dataclass
+class KeyEdit(LineEdit):
+    """A line edit with the secret it holds and the other configs using it."""
+
+    secret: str | None = None
+    shared_with: list[Path] = field(default_factory=list)
+
+
+def _api_encryption() -> ConfigType:
+    return ((CORE.raw_config or {}).get(CONF_API) or {}).get(CONF_ENCRYPTION) or {}
+
+
+def _key_blocks(raw: ConfigType, key: str) -> list[ConfigType]:
+    """The api and esphome ota ``encryption:`` mappings whose key is ``key``."""
+    blocks = [raw.get(CONF_API) or {}, *esphome_ota_items(raw)]
+    found = []
+    for block in blocks:
+        if (node := static_encryption_key(block)) is None or str(node) != key:
+            continue
+        encryption = block[CONF_ENCRYPTION]
+        if source_of(encryption, CONF_KEY) is not None:
+            found.append(encryption)
+        elif hasattr(next(k for k in encryption if k == CONF_KEY), "esp_range"):
+            # Read from a file but not on a line of its own, as with a merge
+            # key; only a key validation added, which has no range at all,
+            # is passed over (a bare ota block takes the api key)
+            raise EsphomeError(
+                f"The '{CONF_KEY}' of an '{CONF_ENCRYPTION}:' block is not on a "
+                "line of its own; edit the key by hand"
+            )
+    return found
+
+
+_INDENT_RE = re.compile(r"\s*")
+_KEY_FIELD_RE = re.compile(rf"^\s*{CONF_KEY}\s*:")
+_BARE_BLOCK_RE = re.compile(rf"^\s*{CONF_ENCRYPTION}\s*:\s*(#.*)?$")
+_YAML_SUFFIXES = (".yaml", ".yml")
+_INCLUDE_DIR_RE = re.compile(r"!include_dir_\w+\s+(?P<dir>[^\s#]+)")
+# An include or package file whose path holds a substitution
+_SUBSTITUTED_INCLUDE_RE = re.compile(
+    r"(?:!include\w*\s+|\bfile:\s*)[\"']?(?P<target>[^\s#\"']*\$[^\s#\"']*)"
+)
+
+
+def _secret_ref(name: str = r"[^\s\"']+") -> str:
+    """`!secret name` as the whole scalar: `!secret ota key` is not `ota`, and
+    a comment needs whitespace before its `#`, so `a#b` is one name."""
+    return rf"!secret\s+(?P<q>[\"']?)(?P<name>{name})(?P=q)(?=\s+#|\s*$)"
+
+
+def _secret_field_re(name: str) -> re.Pattern[str]:
+    """``name: !secret <any>``."""
+    return re.compile(rf"^\s*{name}\s*:\s*{_secret_ref()}")
+
+
+def _secret_use_re(name: str) -> re.Pattern[str]:
+    """A use of the secret ``name`` anywhere on a line."""
+    return re.compile(_secret_ref(re.escape(name)))
+
+
+_KEY_SECRET_RE = _secret_field_re(CONF_KEY)
+_OLD_KEY_SECRET_RE = _secret_field_re(CONF_OLD_KEY)
+
+
+def _indent(text: str) -> str:
+    return _INDENT_RE.match(text).group()
+
+
+def _without_comment(text: str) -> str:
+    """The line up to its comment: a `#` after whitespace that is not inside
+    a quoted scalar, so `!include "shared # ota.yaml"` keeps its name."""
+    quote = ""
+    for i, char in enumerate(text):
+        if quote:
+            if char == quote:
+                quote = ""
+        elif char in "\"'":
+            quote = char
+        elif char == "#" and (i == 0 or text[i - 1].isspace()):
+            return text[:i]
+    return text
+
+
+def _file_use_re(file_name: str) -> re.Pattern[str]:
+    """The file name at a path boundary; `wifi_api.yaml` is not `api.yaml`."""
+    return re.compile(rf"(?<![\w.-]){re.escape(file_name)}(?![\w.])")
+
+
+def _root_indent(lines: list[str]) -> str:
+    """The indent of the first content line; a root mapping may be indented."""
+    content = (
+        t for t in lines if t.strip() and not t.strip().startswith(("#", "---", "%"))
+    )
+    return next((_indent(t) for t in content), "")
+
+
+def _config_texts() -> tuple[list[tuple[Path, str]], list[str]]:
+    """Every yaml file under the config dir with its text, and what could
+    not be read (a symlinked directory is not walked); walked once per run."""
+    if (found := CORE.data.get(_DOMAIN)) is None:
+        found = CORE.data[_DOMAIN] = _walk_config_texts()
+    return found
+
+
+def _walk_config_texts() -> tuple[list[tuple[Path, str]], list[str]]:
+    data_dir = CORE.data_dir.resolve()
+    found = []
+    skipped: list[str] = []
+
+    def skip(what: object) -> None:
+        skipped.append(str(what))
+        _LOGGER.debug("Skipping %s", what)
+
+    for dirpath, dirnames, filenames in os.walk(
+        CORE.config_dir.resolve(), onerror=skip
+    ):
+        for name in list(dirnames):
+            path = Path(dirpath, name)
+            if path == data_dir:
+                dirnames.remove(name)
+            elif path.is_symlink():
+                dirnames.remove(name)
+                skip(f"{path} is a link")
+        for filename in filenames:
+            path = Path(dirpath, filename)
+            if path.suffix not in _YAML_SUFFIXES:
+                continue
+            try:
+                text = read_text(path)
+            except EsphomeError as err:
+                skip(err)
+                continue
+            # Searched, never edited: comments do not count
+            found.append((path, "\n".join(map(_without_comment, text.splitlines()))))
+    return sorted(found), skipped
+
+
+def with_sharers(edits: list[KeyEdit]) -> list[str]:
+    """Fill in ``shared_with``: the other configs that use each rewritten
+    line's secret or include, directly or through further includes. Returns
+    what the scan could not follow (unreadable files, directory and
+    substituted includes)."""
+    main = CORE.config_path.resolve()
+    own = own_documents()
+    # This config's own includes stay in: another device may reach a line
+    # through one of them
+    texts, skipped = _config_texts()
+    texts = [(path, text) for path, text in texts if path.resolve() != main]
+    unchecked = [*skipped]
+    for path, text in texts:
+        if match := _INCLUDE_DIR_RE.search(text):
+            unchecked.append(f"{path} includes the directory {match['dir']}")
+        if match := _SUBSTITUTED_INCLUDE_RE.search(text):
+            unchecked.append(
+                f"{path} includes {match['target']} through a substitution"
+            )
+    for edit in edits:
+        if edit.insert_after and edit.secret:
+            continue  # a new secrets line has no users yet
+        ref = (
+            _secret_use_re(edit.secret) if edit.secret else _file_use_re(edit.path.name)
+        )
+        users = {path for path, text in texts if ref.search(text)}
+        frontier = users
+        while frontier:
+            frontier = {
+                path
+                for name in {p.name for p in frontier}
+                for path, text in texts
+                if path not in users and _file_use_re(name).search(text)
+            }
+            users |= frontier
+        edit.shared_with = sorted(p for p in users if p.resolve() not in own)
+    return unchecked
+
+
+def _secret_file(doc: Path) -> Path:
+    return editable_file(secrets_path_for(doc))
+
+
+def _secret_line(
+    secrets_path: Path, name: str, value: str | None = None
+) -> tuple[int, re.Match[str]] | None:
+    """The root level ``name:`` line, None when absent."""
+    lines = read_text(secrets_path).splitlines()
+    line_re = field_line_re(name, value, re.escape(_root_indent(lines)))
+    return next(
+        ((i, m) for i, text in enumerate(lines) if (m := line_re.match(text))), None
+    )
+
+
+def _secret_rewrite(
+    secrets_path: Path, name: str, value: str, expect: str | None = None
+) -> KeyEdit | None:
+    """Set the root level ``name:`` line to ``value``; None when it is not a
+    plain scalar holding ``expect``."""
+    if (hit := _secret_line(secrets_path, name, expect)) is None:
+        return None
+    i, m = hit
+    return KeyEdit(secrets_path, i, m.string, rewrite(m, value), secret=name)
+
+
+def _secret_insert(secrets_path: Path, after: str, name: str, value: str) -> KeyEdit:
+    """Add ``name: value`` under the root level ``after`` line."""
+    if (hit := _secret_line(secrets_path, after)) is None:
+        raise EsphomeError(
+            f"No plain '{after}:' line in {secrets_path}; edit it by hand"
+        )
+    i, m = hit
+    return KeyEdit(
+        secrets_path,
+        i,
+        m.string,
+        f'{_indent(m.string)}{name}: "{value}"',
+        insert_after=True,
+        secret=name,
+    )
+
+
+def locate_key_edits(old_key: str, new_key: str) -> list[KeyEdit]:
+    """The edits that replace the current key; refuses a substitution, flow
+    mapping or remote package."""
+    literal_re = field_line_re(CONF_KEY, old_key)
+    blocks = _key_blocks(CORE.raw_config or {}, old_key)
+    if not blocks:
+        raise EsphomeError("The current key was not found on a line of the yaml")
+    edits: list[KeyEdit] = []
+    key_lines: set[tuple[Path, int]] = set()
+    for block in blocks:
+        doc, line_no, text = source_line(block, CONF_KEY)
+        key_lines.add((doc.resolve(), line_no))
+        if match := _KEY_SECRET_RE.match(text):
+            name, secrets_path = match["name"], _secret_file(doc)
+            edit = _secret_rewrite(secrets_path, name, new_key, expect=old_key)
+            if edit is None:
+                raise EsphomeError(
+                    f"No '{name}:' line with the current key in {secrets_path}"
+                )
+        elif match := literal_re.match(text):
+            edit = KeyEdit(doc, line_no, text, rewrite(match, new_key))
+        else:
+            raise EsphomeError(
+                f"{doc}:{line_no + 1} does not hold the key as a plain value or "
+                "a !secret; edit the key by hand"
+            )
+        # The api and ota blocks may share one !secret line
+        if edit not in edits:
+            edits.append(edit)
+    for edit in edits:
+        if edit.secret:
+            _refuse_other_own_uses(edit.secret, key_lines)
+    return edits
+
+
+def _refuse_other_own_uses(name: str, allowed: set[tuple[Path, int]]) -> None:
+    """The secret may only feed the ``allowed`` lines; a wifi password on it
+    would change too."""
+    ref = _secret_use_re(name)
+    for doc in own_documents():
+        for i, text in enumerate(read_text(doc).splitlines()):
+            if ref.search(_without_comment(text)) and (doc, i) not in allowed:
+                raise EsphomeError(
+                    f"'{name}' is also used at {doc}:{i + 1}; a rotation would "
+                    "change that value too, edit the key by hand"
+                )
+
+
+def _secret_name(block: ConfigType) -> str | None:
+    """The secrets file name a ``key: !secret <name>`` line of ``block`` holds."""
+    if source_of(block, CONF_KEY) is None:
+        return None
+    match = _KEY_SECRET_RE.match(source_line(block, CONF_KEY)[2])
+    return match["name"] if match else None
+
+
+def old_key_edit(old_key: str) -> list[KeyEdit]:
+    """Set ``old_key:`` on the esphome ota block, rewriting or adding it. A
+    key from secrets.yaml stays there as ``<name>_old``. Of several entries
+    the one with its own old_key or key line gets it."""
+    items = [
+        (item, item[CONF_ENCRYPTION] or {})
+        for item in esphome_ota_items(CORE.raw_config or {})
+        if CONF_ENCRYPTION in item
+    ]
+    if not items:
+        raise EsphomeError("The esphome OTA platform has no 'encryption:' block")
+    item, encryption = next(
+        (
+            pair
+            for name in (CONF_OLD_KEY, CONF_KEY)
+            for pair in items
+            if source_of(pair[1], name) is not None
+        ),
+        items[0],
+    )
+    if (existing := encryption.get(CONF_OLD_KEY)) is not None:
+        doc, line_no, text = source_line(encryption, CONF_OLD_KEY)
+        if match := _OLD_KEY_SECRET_RE.match(text):
+            name, secrets_path = match["name"], _secret_file(doc)
+            if (edit := _secret_rewrite(secrets_path, name, old_key)) is None:
+                raise EsphomeError(
+                    f"No plain '{name}:' line in {secrets_path}; edit it by hand"
+                )
+            _refuse_other_own_uses(name, {(doc.resolve(), line_no)})
+            return [edit]
+        match = field_line_re(CONF_OLD_KEY, str(existing)).match(text)
+        if match is None:
+            raise EsphomeError(
+                f"{doc}:{line_no + 1} does not hold '{CONF_OLD_KEY}' as a plain "
+                "value; edit it by hand"
+            )
+        return [KeyEdit(doc, line_no, text, rewrite(match, old_key))]
+    # Indent from the block's own key line, else one level under `encryption:`
+    if source_of(encryption, CONF_KEY) is not None:
+        mapping, anchor, anchor_re, step = encryption, CONF_KEY, _KEY_FIELD_RE, ""
+    else:
+        mapping, anchor, anchor_re, step = item, CONF_ENCRYPTION, _BARE_BLOCK_RE, "  "
+    doc, line_no, text = source_line(mapping, anchor)
+    if not anchor_re.match(text):
+        raise EsphomeError(
+            f"{doc}:{line_no + 1} does not hold '{anchor}:' as a block line; "
+            "edit it by hand"
+        )
+    # The block's own key line decides; a bare block takes after the api key
+    keyed = encryption if mapping is encryption else _api_encryption()
+    if (secret := _secret_name(keyed)) is None:
+        rendered = f'{CONF_OLD_KEY}: "{old_key}"'
+        extra = []
+    else:
+        rendered = f"{CONF_OLD_KEY}: !secret {secret}_old"
+        secrets_path = _secret_file(doc)  # the file that gets old_key
+        # An existing `<secret>_old` line is reused only when nothing uses it
+        if (
+            kept := _secret_rewrite(secrets_path, f"{secret}_old", old_key)
+        ) is not None:
+            ref = _secret_use_re(f"{secret}_old")
+            texts, skipped = _config_texts()
+            if skipped:  # an unread file might use the line
+                raise EsphomeError(
+                    "Could not read every configuration to check the secret: "
+                    + skipped[0]
+                )
+            if users := [p for p, t in texts if ref.search(t)]:
+                raise EsphomeError(
+                    f"'{secret}_old:' in {secrets_path} is used by "
+                    + ", ".join(str(p) for p in users)
+                    + f"; rename it or add '{CONF_OLD_KEY}' by hand"
+                )
+        extra = [kept or _secret_insert(secrets_path, secret, f"{secret}_old", old_key)]
+    edit = KeyEdit(
+        doc, line_no, text, f"{_indent(text)}{step}{rendered}", insert_after=True
+    )
+    return [edit, *extra]

esphome/platform_hooks.py

@@ -26,20 +26,17 @@ from esphome.const import (
     PLATFORM_BK72XX,
     PLATFORM_ESP32,
     PLATFORM_ESP8266,
-    PLATFORM_LN882X,
     PLATFORM_NRF52,
     PLATFORM_RP2,
-    PLATFORM_RTL87XX,
     Platform,
 )
 
 _LOGGER = logging.getLogger(__name__)
 
-# Hooks whose loss only degrades diagnostics or upload size; skipping one
-# of these is logged at debug, while skipping a hook that changes what the
-# CLI does (upload method, log transport) warns. A new hook is loud by
-# default.
-COSMETIC_HOOKS: Final = frozenset({"process_stacktrace", "ota_image"})
+# Hooks whose loss only degrades diagnostics; skipping one of these is
+# logged at debug, while skipping a hook that changes what the CLI does
+# (upload method, log transport) warns. A new hook is loud by default.
+COSMETIC_HOOKS: Final = frozenset({"process_stacktrace"})
 
 # Platforms whose firmware prints the crash handler report share one gate.
 _CRASH_HANDLER_GATE = r"0x[0-9a-fA-F]{3,}\b|CRASH DETECTED ON PREVIOUS BOOT"
@@ -74,8 +71,6 @@ PLATFORM_HOOKS: Final[dict[str, frozenset[str]]] = {
     "show_logs": frozenset({PLATFORM_NRF52}),
     "upload_program": frozenset({PLATFORM_NRF52}),
     "process_stacktrace": frozenset(STACKTRACE_GATES),
-    # The file an app update sends, when it is not the build output itself
-    "ota_image": frozenset({PLATFORM_BK72XX, PLATFORM_LN882X, PLATFORM_RTL87XX}),
 }
 
 

esphome/yaml_util.py

@@ -966,9 +966,7 @@ class ESPHomeLoaderMixin:
             if self.name == CORE.config_path:
                 raise e
             try:
-                main_config_dir = CORE.config_path.parent
-                main_secret_yml = main_config_dir / SECRET_YAML
-                secrets = self.yaml_loader(main_secret_yml)
+                secrets = self.yaml_loader(CORE.config_path.parent / SECRET_YAML)
             except EsphomeError as er:
                 raise EsphomeError(f"{e}\n{er}") from er
 
@@ -1357,6 +1355,20 @@ def find_files(directory: Path, pattern: str) -> Iterator[Path]:
                 yield filename
 
 
+def secrets_path_for(document: Path) -> Path:
+    """The secrets.yaml a ``!secret`` in ``document`` resolves against: the
+    one beside the document when it loads, else the main config's, the
+    way construct_secret falls back."""
+    beside = document.parent / SECRET_YAML
+    if document == CORE.config_path:
+        return beside
+    try:
+        load_yaml(beside, clear_secrets=False, track_document_range=False)
+    except EsphomeError:
+        return CORE.config_path.parent / SECRET_YAML
+    return beside
+
+
 def is_secret(value):
     try:
         return _SECRET_VALUES[str(value)]

tests/component_tests/logger/test_logger.py

@@ -138,26 +138,3 @@ def test_task_log_buffer_size_unaligned_allowed_on_libretiny(
 ) -> None:
     set_core_config(PlatformFramework.BK72XX_ARDUINO)
     assert validate_task_log_buffer_alignment(641) == 641
-
-
-@pytest.mark.parametrize(
-    ("config", "log_v2"),
-    [
-        pytest.param("test_logger_idf6_esp_idf.yaml", True, id="native"),
-        pytest.param("test_logger_idf6_platformio.yaml", False, id="platformio"),
-    ],
-)
-def test_log_v2_only_on_native_toolchain(
-    generate_main: Callable[[str | Path], str],
-    config: str,
-    log_v2: bool,
-) -> None:
-    """PlatformIO links the bootloader with the app's build flags, so the wrap has no target there."""
-    from esphome.components.esp32 import KEY_ESP32, KEY_SDKCONFIG_OPTIONS
-
-    generate_main(f"tests/component_tests/logger/{config}")
-
-    assert ("-Wl,--wrap=esp_log_format" in CORE.build_flags) is log_v2
-    assert ("USE_ESP32_LOG_V2" in {d.name for d in CORE.defines}) is log_v2
-    sdkconfig = CORE.data[KEY_ESP32][KEY_SDKCONFIG_OPTIONS]
-    assert ("CONFIG_LOG_VERSION_2" in sdkconfig) is log_v2

tests/component_tests/logger/test_logger_idf6_esp_idf.yaml

@@ -1,11 +0,0 @@
-esphome:
-  name: test
-
-esp32:
-  board: esp32dev
-  toolchain: esp-idf
-  framework:
-    type: esp-idf
-    version: 6.1.0
-
-logger:

tests/component_tests/logger/test_logger_idf6_platformio.yaml

@@ -1,12 +0,0 @@
-esphome:
-  name: test
-
-esp32:
-  board: esp32dev
-  toolchain: platformio
-  framework:
-    type: esp-idf
-    version: 6.1.0
-    platform_version: 61.4.0-RC1
-
-logger:

tests/component_tests/wifi/config/dpp_key.yaml

@@ -1,12 +0,0 @@
----
-esphome:
-  name: test
-
-esp32:
-  board: esp32dev
-  framework:
-    type: esp-idf
-
-wifi:
-  dpp:
-    private_key: 00112233445566778899AABBCCDDEEFF00112233445566778899aabbccddeeff

tests/component_tests/wifi/test_dpp_key.py

@@ -1,42 +0,0 @@
-"""The DPP bootstrapping key is given as 32 raw bytes in hex and handed to the supplicant DER encoded."""
-
-from collections.abc import Callable
-from pathlib import Path
-
-import pytest
-
-from esphome.components import wifi
-import esphome.config_validation as cv
-
-RAW_KEY = "00112233445566778899aabbccddeeff00112233445566778899aabbccddeeff"
-
-
-@pytest.mark.parametrize(
-    "value",
-    [
-        RAW_KEY[:-2],
-        RAW_KEY + "00",
-        "zz" + RAW_KEY[2:],
-        wifi.DPP_KEY_DER_PREFIX + RAW_KEY + wifi.DPP_KEY_DER_SUFFIX,
-    ],
-)
-def test_key_must_be_64_hex_digits(value: str) -> None:
-    with pytest.raises(cv.Invalid, match="64 hexadecimal characters"):
-        wifi._validate_dpp_private_key(value)
-
-
-def test_key_is_lower_cased() -> None:
-    assert wifi._validate_dpp_private_key(RAW_KEY.upper()) == RAW_KEY
-
-
-def test_generated_key_is_der_encoded(
-    generate_main: Callable[[str | Path], str],
-    component_config_path: Callable[[str], Path],
-) -> None:
-    """The ESP-IDF supplicant expects the P-256 DER header around the raw key."""
-    main_cpp = generate_main(component_config_path("dpp_key.yaml"))
-
-    assert (
-        f'set_dpp_key("{wifi.DPP_KEY_DER_PREFIX}{RAW_KEY}{wifi.DPP_KEY_DER_SUFFIX}")'
-        in main_cpp
-    )

tests/components/core/test_pulse_reset.cpp

@@ -1,51 +0,0 @@
-#include <gtest/gtest.h>
-
-#include <string>
-#include <vector>
-
-#include "esphome/core/gpio.h"
-#include "esphome/core/hal.h"
-
-namespace esphome::core::testing {
-
-class RecordingPin : public GPIOPin {
- public:
-  void setup() override { this->events += 'S'; }
-  void pin_mode(gpio::Flags /*flags*/) override {}
-  gpio::Flags get_flags() const override { return gpio::FLAG_OUTPUT; }
-  bool digital_read() override { return false; }
-  void digital_write(bool value) override {
-    this->events += value ? 'H' : 'L';
-    this->write_ms.push_back(millis());
-  }
-
-  std::string events;
-  std::vector<uint32_t> write_ms;
-};
-
-TEST(PulseReset, NullPinIsIgnored) { pulse_reset(nullptr, {.low_ms = 1}); }
-
-TEST(PulseReset, LowThenHigh) {
-  RecordingPin pin;
-  pulse_reset(&pin, {.low_ms = 1});
-  EXPECT_EQ(pin.events, "SLH");
-}
-
-TEST(PulseReset, PreHighDrivesHighFirst) {
-  RecordingPin pin;
-  pulse_reset(&pin, {.high_before_ms = 1, .low_ms = 1, .high_after_ms = 1});
-  EXPECT_EQ(pin.events, "SHLH");
-}
-
-// Distinct waits, so dropping or swapping one shows up; only lower bounds, a busy host can run long
-TEST(PulseReset, HoldsEachLevelForItsTime) {
-  RecordingPin pin;
-  pulse_reset(&pin, {.high_before_ms = 10, .low_ms = 20, .high_after_ms = 30});
-  const uint32_t end = millis();
-  ASSERT_EQ(pin.events, "SHLH");
-  EXPECT_GE(pin.write_ms[1] - pin.write_ms[0], 10u);
-  EXPECT_GE(pin.write_ms[2] - pin.write_ms[1], 20u);
-  EXPECT_GE(end - pin.write_ms[2], 30u);
-}
-
-}  // namespace esphome::core::testing

tests/components/dfrobot_sen0395/__init__.py

@@ -1,14 +0,0 @@
-import esphome.codegen as cg
-from esphome.types import ConfigType
-from tests.testing_helpers import ComponentManifestOverride
-
-
-def override_manifest(manifest: ComponentManifestOverride) -> None:
-    # One empty config entry so to_code runs and sets the define that keeps
-    # core/line_reader.cpp in the test build.
-    manifest.multi_conf = False
-
-    async def to_code(config: ConfigType) -> None:
-        cg.add_define("USE_LINE_READER")
-
-    manifest.to_code = to_code

tests/components/wifi/test-dpp.esp32-idf.yaml

@@ -2,7 +2,6 @@ wifi:
   dpp:
     channels: [1, 6, 11]
     device_info: ESPHome
-    private_key: 00112233445566778899aabbccddeeff00112233445566778899aabbccddeeff
     timeout: 30s
   on_dpp_uri:
     - logger.log:

tests/integration/fixtures/uart_mock_dfrobot_sen0395.yaml

@@ -1,70 +0,0 @@
-esphome:
-  name: uart-mock-dfrobot-sen0395
-
-host:
-api:
-  batch_delay: 0ms  # Disable batching to receive all state updates
-logger:
-  level: VERBOSE
-
-external_components:
-  - source:
-      type: local
-      path: EXTERNAL_COMPONENT_PATH
-
-# Dummy uart entry to satisfy DEPENDENCIES = ["uart"]
-uart:
-  baud_rate: 115200
-  port: /dev/null
-
-uart_mock:
-  id: mock_uart
-  baud_rate: 115200
-  # A small model of the module's command line: echo, answer and the prompt (no line end)
-  on_tx:
-    - then:
-        - lambda: |-
-            std::string cmd(data.begin(), data.end());
-            std::string reply = cmd + "\r\n";
-            if (cmd == "resetSystem") {
-              // The restart only answers with the prompt
-            } else {
-              reply += "Done\r\n";
-            }
-            reply += "leapMMW:/>";
-            id(mock_uart).inject_to_rx_buffer(std::vector<uint8_t>(reply.begin(), reply.end()));
-
-dfrobot_sen0395:
-  id: sen0395
-  uart_id: mock_uart
-
-binary_sensor:
-  - platform: dfrobot_sen0395
-    name: Detected
-
-switch:
-  - platform: dfrobot_sen0395
-    type: sensor_active
-    name: Sensor Active
-
-button:
-  - platform: template
-    name: Inject Present
-    on_press:
-      - lambda: |-
-          // '$' is kept apart so it is not read as a substitution
-          std::string frame = std::string(1, '$') + "JYBSS,1, , , *\r\n";
-          id(mock_uart).inject_to_rx_buffer(std::vector<uint8_t>(frame.begin(), frame.end()));
-  - platform: template
-    name: Inject Overlong Then Absent
-    on_press:
-      - lambda: |-
-          // An overlong line is dropped up to its line end; the next frame still parses
-          std::string data(300, 'x');
-          data += "\r\n" + std::string(1, '$') + "JYBSS,0, , , *\r\n";
-          id(mock_uart).inject_to_rx_buffer(std::vector<uint8_t>(data.begin(), data.end()));
-  - platform: template
-    name: Reset Sensor
-    on_press:
-      - dfrobot_sen0395.reset:
-          id: sen0395

tests/integration/fixtures/uart_mock_ld2410s.yaml

@@ -74,22 +74,6 @@ uart_mock:
       inject_rx: [0xFD, 0xFC, 0xFB, 0xFA, 0x04, 0x00, 0x70, 0x01, 0x00, 0x00, 0x04, 0x03, 0x02, 0x01]
     - expect_tx: [0xFD, 0xFC, 0xFB, 0xFA, 0x02, 0x00, 0xFE, 0x00, 0x04, 0x03, 0x02, 0x01]
       inject_rx: [0xFD, 0xFC, 0xFB, 0xFA, 0x04, 0x00, 0xFE, 0x01, 0x00, 0x00, 0x04, 0x03, 0x02, 0x01]
-    # Then the response speed Fast: the same values with speed 10, acknowledged
-    - expect_tx:
-        [
-          0xFD, 0xFC, 0xFB, 0xFA, 0x26, 0x00, 0x70, 0x00, 0x05, 0x00, 0x0A, 0x00, 0x00, 0x00, 0x0A, 0x00,
-          0x00, 0x00, 0x00, 0x00, 0x06, 0x00, 0x0A, 0x00, 0x00, 0x00, 0x02, 0x00, 0x50, 0x00, 0x00, 0x00,
-          0x0C, 0x00, 0x50, 0x00, 0x00, 0x00, 0x0B, 0x00, 0x0A, 0x00, 0x00, 0x00, 0x04, 0x03, 0x02, 0x01,
-        ]
-      inject_rx: [0xFD, 0xFC, 0xFB, 0xFA, 0x04, 0x00, 0x70, 0x01, 0x00, 0x00, 0x04, 0x03, 0x02, 0x01]
-    # Then no delay 20 s, rejected with ack 1: the component reads the module's values back
-    - expect_tx:
-        [
-          0xFD, 0xFC, 0xFB, 0xFA, 0x26, 0x00, 0x70, 0x00, 0x05, 0x00, 0x0A, 0x00, 0x00, 0x00, 0x0A, 0x00,
-          0x00, 0x00, 0x00, 0x00, 0x06, 0x00, 0x14, 0x00, 0x00, 0x00, 0x02, 0x00, 0x50, 0x00, 0x00, 0x00,
-          0x0C, 0x00, 0x50, 0x00, 0x00, 0x00, 0x0B, 0x00, 0x0A, 0x00, 0x00, 0x00, 0x04, 0x03, 0x02, 0x01,
-        ]
-      inject_rx: [0xFD, 0xFC, 0xFB, 0xFA, 0x04, 0x00, 0x70, 0x01, 0x01, 0x00, 0x04, 0x03, 0x02, 0x01]
   injections:
     # Short frame, state 2 (target), distance 300 -> presence on
     - delay: 100ms
@@ -153,14 +137,6 @@ number:
       name: "Max Distance"
     no_delay:
       name: "No Delay"
-    status_reporting_frequency:
-      name: "Status Reporting Frequency"
-
-select:
-  - platform: ld2410s
-    ld2410s_id: ld2410s_dev
-    response_speed:
-      name: "Response Speed"
 
 button:
   - platform: template

tests/integration/fixtures/uart_mock_ld2410s_recovery.yaml

@@ -92,19 +92,6 @@ binary_sensor:
     has_target:
       name: "Has Target"
 
-# Commanded before the module answered: the changes are ignored and nothing is published
-number:
-  - platform: ld2410s
-    ld2410s_id: ld2410s_dev
-    max_distance:
-      name: "Max Distance"
-
-select:
-  - platform: ld2410s
-    ld2410s_id: ld2410s_dev
-    response_speed:
-      name: "Response Speed"
-
 button:
   - platform: template
     name: "Start Scenario"

tests/integration/test_uart_mock_dfrobot_sen0395.py

@@ -1,93 +0,0 @@
-"""Integration test for the DFRobot SEN0395 component against a model of the module's command line.
-
-Checks presence frames, recovery after an overlong line, a command answered by "Done" and the
-reset command, which the module answers only with its "leapMMW:/>" prompt (no line end).
-"""
-
-from __future__ import annotations
-
-import asyncio
-from collections.abc import Callable
-from pathlib import Path
-
-from aioesphomeapi import (
-    BinarySensorInfo,
-    ButtonInfo,
-    EntityInfo,
-    EntityState,
-    SwitchInfo,
-)
-import pytest
-
-from .state_utils import require_entity
-from .types import APIClientConnectedFactory, RunCompiledFunction
-
-
-@pytest.mark.asyncio
-async def test_uart_mock_dfrobot_sen0395(
-    yaml_config: str,
-    run_compiled: RunCompiledFunction,
-    api_client_connected: APIClientConnectedFactory,
-) -> None:
-    """Frames parse, an overlong line is dropped and the prompt completes the reset."""
-    external_components_path = str(
-        Path(__file__).parent / "fixtures" / "external_components"
-    )
-    yaml_config = yaml_config.replace(
-        "EXTERNAL_COMPONENT_PATH", external_components_path
-    )
-
-    loop = asyncio.get_running_loop()
-    dropped = loop.create_future()
-    restarted = loop.create_future()
-
-    def line_callback(line: str) -> None:
-        if "Line over 252 chars dropped" in line and not dropped.done():
-            dropped.set_result(True)
-        if "Restarted sensor." in line and not restarted.done():
-            restarted.set_result(True)
-
-    states: dict[int, EntityState] = {}
-
-    async def wait_for(condition: Callable[[], bool], what: str) -> None:
-        try:
-            async with asyncio.timeout(10.0):
-                while not condition():
-                    await asyncio.sleep(0.05)
-        except TimeoutError:
-            pytest.fail(f"Timeout waiting for {what}; states: {states}")
-
-    async with (
-        run_compiled(yaml_config, line_callback=line_callback),
-        api_client_connected() as client,
-    ):
-        entities, _ = await client.list_entities_services()
-        detected = require_entity(entities, "detected", BinarySensorInfo)
-        active = require_entity(entities, "sensor_active", SwitchInfo)
-        present = require_entity(entities, "inject_present", ButtonInfo)
-        overlong = require_entity(entities, "inject_overlong_then_absent", ButtonInfo)
-        reset = require_entity(entities, "reset_sensor", ButtonInfo)
-
-        client.subscribe_states(lambda state: states.__setitem__(state.key, state))
-
-        def state_of(entity: EntityInfo) -> object | None:
-            state = states.get(entity.key)
-            return None if state is None or state.missing_state else state.state
-
-        client.button_command(present.key)
-        await wait_for(lambda: state_of(detected) is True, "presence")
-
-        client.button_command(overlong.key)
-        await wait_for(
-            lambda: state_of(detected) is False, "absence after overlong line"
-        )
-        await asyncio.wait_for(dropped, timeout=5.0)
-
-        client.switch_command(active.key, False)
-        await wait_for(lambda: state_of(active) is False, "sensor stopped")
-
-        client.button_command(reset.key)
-        try:
-            await asyncio.wait_for(restarted, timeout=10.0)
-        except TimeoutError:
-            pytest.fail("The reset was not completed by the prompt")

tests/integration/test_uart_mock_ld2410s.py

@@ -5,15 +5,9 @@ test_uart_mock_ld2410s:
      switch and config-mode end go out in order, each after the previous one is
      acknowledged, then the component logs "Setup done" and clears its warning;
      the version and parameter answers are parsed and logged
-  1a. The numbers show the values read at init in their own units (gates of 0.7 m,
-     seconds, tenths of a hertz) and the response speed select shows Normal
-  1b. Setting the max distance number to 7.3 m writes all six parameters (gate 10
+  1b. Setting the max distance number to 7 m writes all six parameters (gate 10
      plus the values read at init) inside a configuration-mode session, and the
-     number reports the 7 m the module got, not 7.3 m
-  1c. Selecting the Fast response speed writes speed 10 with the same values and
-     the select reports Fast
-  1d. Setting no delay to 20 s reports 20 s at once; the module rejects the write,
-     so the values are read back and every entity shows the module's own value again
+     number reports 7 m
   2. Presence from short frames (states 0/1 off, 2/3 on) and from a standard
      data frame
   3. Resync - a noise byte or a partial header directly before a short frame
@@ -22,8 +16,6 @@ test_uart_mock_ld2410s:
      dropped without publishing, and the short frame after each still publishes
 
 test_uart_mock_ld2410s_recovery (the mock never answers the init commands):
-  0. A number or select changed before the module answered logs that the change
-     is ignored and publishes nothing
   1. Config-mode start is resent after the acknowledgement timeout
   2. Once the retries run out, config-mode end is sent and the component warns
      that it is starting over
@@ -44,19 +36,9 @@ test_uart_mock_ld2410s_link_lost:
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Callable
 import re
 
-from aioesphomeapi import (
-    APIClient,
-    ButtonInfo,
-    EntityInfo,
-    EntityState,
-    NumberInfo,
-    NumberState,
-    SelectInfo,
-    SelectState,
-)
+from aioesphomeapi import APIClient, ButtonInfo, EntityState, NumberInfo, NumberState
 import pytest
 
 from .log_utils import LineWaiter
@@ -84,20 +66,12 @@ INIT_TX = [
     OUTPUT_MODE_TX,
     CONFIG_END_TX,
 ]
-
-
-def _parameters_write_tx(no_delay: int, speed: int) -> str:
-    """The 0x0070 write with max gate 10, min gate 0 and both frequencies 80."""
-    values = [(0x05, 10), (0x0A, 0), (0x06, no_delay), (0x02, 80), (0x0C, 80)]
-    body = "".join(f"{key:02X}:00:{value:02X}:00:00:00:" for key, value in values)
-    return f"FD:FC:FB:FA:26:00:70:00:{body}0B:00:{speed:02X}:00:00:00:04:03:02:01"
-
-
-WRITE_TX = [CONFIG_START_TX, _parameters_write_tx(10, 5), CONFIG_END_TX]
-# The same values with response speed 10 (Fast)
-SPEED_WRITE_TX = [CONFIG_START_TX, _parameters_write_tx(10, 10), CONFIG_END_TX]
-# No delay 20 s; the fixture rejects it
-REJECTED_WRITE_TX = [CONFIG_START_TX, _parameters_write_tx(20, 10), CONFIG_END_TX]
+# All six parameters: max gate 10, min gate 0, no delay 10, both frequencies 80, speed 5
+PARAMETERS_WRITE_TX = (
+    "FD:FC:FB:FA:26:00:70:00:05:00:0A:00:00:00:0A:00:00:00:00:00:06:00:0A:00:00:00:02:00:50:00:00:00:"
+    "0C:00:50:00:00:00:0B:00:05:00:00:00:04:03:02:01"
+)
+WRITE_TX = [CONFIG_START_TX, PARAMETERS_WRITE_TX, CONFIG_END_TX]
 # Logged once the version and parameter answers are parsed
 PARSED_ANSWERS = [
     "Firmware 1.2.3",
@@ -143,20 +117,13 @@ def _tx_frame(line: str) -> str | None:
     return ANSI_ESCAPE.sub("", line.split("bytes: ", 1)[1]).strip()
 
 
-async def _start_scenario(
-    client: APIClient,
-    collector: SensorStateCollector,
-    on_state: Callable[[EntityState], None] | None = None,
-) -> list[EntityInfo]:
-    """Subscribe past the initial states, then start the mock scenario; returns the entities."""
+async def _start_scenario(client: APIClient, collector: SensorStateCollector) -> None:
+    """Subscribe the collector past the initial states, then start the mock scenario."""
     entities, _ = await client.list_entities_services()
     collector.build_key_mapping(entities)
-    await subscribe_after_initial_states(
-        client, entities, on_state=on_state or collector.on_state
-    )
+    await subscribe_after_initial_states(client, entities, on_state=collector.on_state)
     start_btn = require_entity(entities, "start_scenario", ButtonInfo)
     client.button_command(start_btn.key)
-    return entities
 
 
 async def _wait_for_presence(
@@ -189,15 +156,12 @@ async def test_uart_mock_ld2410s(
     collector = SensorStateCollector(
         sensor_names=["distance"], binary_sensor_names=["has_target"]
     )
-    number_states: dict[int, list[float]] = {}
-    select_states: list[str] = []
+    number_states: list[float] = []
 
     def on_state(state: EntityState) -> None:
         collector.on_state(state)
-        if isinstance(state, NumberState):
-            number_states.setdefault(state.key, []).append(state.state)
-        if isinstance(state, SelectState):
-            select_states.append(state.state)
+        if isinstance(state, NumberState) and state.key == max_distance_key:
+            number_states.append(state.state)
 
     all_presence = collector.add_waiter(
         lambda: len(collector.binary_states["has_target"]) >= len(EXPECTED_PRESENCE)
@@ -227,28 +191,16 @@ async def test_uart_mock_ld2410s(
         collector.build_key_mapping(entities)
         max_distance_key = require_entity(entities, "max_distance", NumberInfo).key
         sub = await subscribe_after_initial_states(client, entities, on_state=on_state)
-        # The values read from the module: gate 16, 10 s, 80 tenths of a hertz, speed 5
+        # The values read from the module: gate 16 and 10 s
         no_delay_key = require_entity(entities, "no_delay", NumberInfo).key
-        frequency_key = require_entity(
-            entities, "status_reporting_frequency", NumberInfo
-        ).key
-        speed_key = require_entity(entities, "response_speed", SelectInfo).key
         initial = sub.initial_states
         assert initial[max_distance_key].state == pytest.approx(11.2)
         assert initial[no_delay_key].state == pytest.approx(10)
-        assert initial[frequency_key].state == pytest.approx(8.0)
-        assert initial[speed_key].state == "Normal"
 
-        # 7.3 m is between gates 10 and 11: gate 10 is written and 7 m reported
-        client.number_command(max_distance_key, 7.3)
+        client.number_command(max_distance_key, 7.0)
         await waiter.wait_count(TX_LINE, 9, 5.0, "Parameter write not sent")
         assert tx_frames()[6:9] == WRITE_TX, tx_frames()
-        assert number_states[max_distance_key] == [pytest.approx(7.0)]
-
-        client.select_command(speed_key, "Fast")
-        await waiter.wait_count(TX_LINE, 12, 5.0, "Response speed write not sent")
-        assert tx_frames()[9:12] == SPEED_WRITE_TX, tx_frames()
-        assert select_states == ["Fast"]
+        assert number_states == [pytest.approx(7.0)]
 
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
         client.button_command(start_btn.key)
@@ -266,29 +218,7 @@ async def test_uart_mock_ld2410s(
         # A target 300 cm away, then none: the distance follows the presence
         assert collector.sensor_states["distance"][:2] == [300, 0]
         # Data frames must not restart the init sequence
-        assert len(tx_frames()) == 12, tx_frames()
-
-        # A rejected write leaves configuration mode and reads the module's values back
-        read_back = waiter.expect("Parameters: gates 0 to 16, no delay 10 s")
-        client.number_command(no_delay_key, 20)
-        await wait_or_fail(read_back, 5.0, "Values not read back", tx_frames=tx_frames)
-
-        async def republished() -> None:
-            while (
-                len(number_states.get(no_delay_key, ())) < 2
-                or len(number_states.get(max_distance_key, ())) < 2
-                or len(select_states) < 2
-            ):
-                await asyncio.sleep(0.01)
-
-        await wait_or_fail(republished(), 5.0, "Read-back values not published")
-        assert tx_frames()[12:15] == REJECTED_WRITE_TX, tx_frames()
-        assert number_states[no_delay_key] == [pytest.approx(20), pytest.approx(10)]
-        assert number_states[max_distance_key] == [
-            pytest.approx(7.0),
-            pytest.approx(11.2),
-        ]
-        assert select_states == ["Fast", "Normal"]
+        assert len(tx_frames()) == 9, tx_frames()
 
 
 @pytest.mark.asyncio
@@ -304,14 +234,10 @@ async def test_uart_mock_ld2410s_recovery(
     gave_up = loop.create_future()
     oversized_dropped = loop.create_future()
     footer_mismatches: list[str] = []
-    ignored_changes = 0
 
     def line_callback(line: str) -> None:
-        nonlocal ignored_changes
         if (frame := _tx_frame(line)) is not None:
             tx_frames.append((loop.time(), frame))
-        if "Parameters not read yet, change ignored" in line:
-            ignored_changes += 1
         if "No acknowledgement for command 00FF, resending" in line and not (
             resent.done()
         ):
@@ -341,25 +267,12 @@ async def test_uart_mock_ld2410s_recovery(
             >= len(EXPECTED_RECOVERY_PRESENCE)
         )
     )
-    setting_states: list[EntityState] = []
-
-    def on_state(state: EntityState) -> None:
-        collector.on_state(state)
-        if isinstance(state, (NumberState, SelectState)):
-            setting_states.append(state)
 
     async with (
         run_compiled(yaml_config, line_callback=line_callback),
         api_client_connected() as client,
     ):
-        entities = await _start_scenario(client, collector, on_state)
-        # The module never answered, so both changes are refused and nothing is published
-        client.number_command(
-            require_entity(entities, "max_distance", NumberInfo).key, 7.0
-        )
-        client.select_command(
-            require_entity(entities, "response_speed", SelectInfo).key, "Fast"
-        )
+        await _start_scenario(client, collector)
         await _wait_for_presence(collector, all_presence)
         await wait_or_fail(
             asyncio.gather(resent, gave_up),
@@ -373,8 +286,6 @@ async def test_uart_mock_ld2410s_recovery(
         # Recovery shows in the published states alone; the footer log below is
         # only a cross-check that needs the fixture's VERBOSE logger level
         assert collector.binary_states["has_target"] == EXPECTED_RECOVERY_PRESENCE
-        assert ignored_changes == 2
-        assert setting_states == []
         assert oversized_dropped.done(), "Oversized length field was not dropped"
         assert footer_mismatches == [TRUNCATED_FRAME_HEX, BAD_COMMAND_FRAME_HEX]
 

tests/unit_tests/cli/test_rotate_key.py

@@ -0,0 +1,494 @@
+"""Tests for ``esphome rotate-key``."""
+
+from __future__ import annotations
+
+from collections.abc import Generator
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+from unittest.mock import ANY, Mock, patch
+
+import pytest
+
+from esphome import yaml_util
+from esphome.cli.rotate_key import PRECHECK_ADVICE, command_rotate_key
+from esphome.const import (
+    CONF_API,
+    CONF_ENCRYPTION,
+    CONF_ESPHOME,
+    CONF_KEY,
+    CONF_OTA,
+    CONF_PLATFORM,
+    CONF_PORT,
+    CONF_WEB_SERVER,
+    KEY_CORE,
+    KEY_TARGET_PLATFORM,
+    PLATFORM_ESP32,
+)
+from esphome.core import CORE, EsphomeError, Toolchain
+from esphome.espota2 import ProbeResult
+from esphome.ota_key_edit import KeyEdit
+from esphome.upload_targets import PortType
+from esphome.util import ESPHOME_COMMAND
+from esphome.yaml_edit import RestoreError
+
+
+@dataclass
+class MockArgs:
+    device: list[str] | None = None
+    prompt_new_key: bool = False
+    yes: bool = False
+    substitution: list[list[str]] | None = None
+    toolchain: Toolchain | None = None
+
+
+def _interrupt_after(writes: int) -> Any:
+    """A write_keeping_mode that lets ``writes`` calls through, then Ctrl-C."""
+    from esphome.yaml_edit import write_keeping_mode
+
+    calls = 0
+
+    def write(*args: Any, **kwargs: Any) -> None:
+        nonlocal calls
+        calls += 1
+        if calls > writes:
+            raise KeyboardInterrupt
+        write_keeping_mode(*args, **kwargs)
+
+    return write
+
+
+def setup_core(tmp_path: Path) -> None:
+    CORE.reset()
+    CORE.config_path = tmp_path / "test.yaml"
+    CORE.name = "test"
+    CORE.build_path = tmp_path / ".esphome" / "build" / "test"
+    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: PLATFORM_ESP32}
+
+
+OLD_KEY = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+NEW_KEY = "AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA="
+API_YAML = f"""esphome:
+  name: test
+
+api:
+  encryption:
+    key: "{OLD_KEY}"
+
+ota:
+  - platform: esphome
+    encryption:
+"""
+OTA_CONF = {
+    CONF_PLATFORM: CONF_ESPHOME,
+    CONF_PORT: 3232,
+    CONF_ENCRYPTION: {CONF_KEY: OLD_KEY},
+}
+
+
+@pytest.fixture
+def env(tmp_path: Path) -> Generator[dict[str, Mock]]:
+    """A config with an inline api key, loaded the way read_config does so
+    the key node carries its source range, and every network step mocked."""
+    setup_core(tmp_path)
+    CORE.config_path.write_text(API_YAML, encoding="utf-8")
+    CORE.raw_config = yaml_util.load_yaml(CORE.config_path)
+    CORE.config = {
+        CONF_API: {CONF_ENCRYPTION: {CONF_KEY: OLD_KEY}},
+        CONF_OTA: [OTA_CONF],
+    }
+    with (
+        patch("esphome.__main__.choose_upload_log_host", return_value=["dev.local"]),
+        patch("esphome.__main__._resolve_network_devices", return_value=["dev.local"]),
+        patch("esphome.cli.rotate_key.get_port_type", return_value=PortType.NETWORK),
+        patch(
+            "esphome.espota2.probe_ota_key", return_value=ProbeResult.ACCEPTED
+        ) as probe,
+        patch(
+            "esphome.cli.rotate_key.run_external_process", return_value=0
+        ) as compile_,
+        patch("esphome.espota2.run_ota") as upload,
+        patch("esphome.cli.rotate_key.safe_input", return_value="y") as confirm,
+        patch("esphome.cli.rotate_key.sys.stdin") as stdin,
+        patch(
+            "esphome.components.noise.secrets.token_bytes",
+            return_value=bytes(range(1, 33)),
+        ),
+    ):
+        stdin.isatty.return_value = True
+        # The real upload reports the connection before its result
+        upload.return_value = (0, "dev.local")
+
+        def upload_after_connecting(
+            *_args: Any, **kwargs: Any
+        ) -> tuple[int, str | None]:
+            kwargs["on_upload_start"]()
+            return upload.return_value
+
+        upload.side_effect = upload_after_connecting
+        yield {
+            "probe": probe,
+            "compile": compile_,
+            "upload": upload,
+            "confirm": confirm,
+            "stdin": stdin,
+        }
+
+
+def test_success(env: dict[str, Mock], capfd: pytest.CaptureFixture[str]) -> None:
+    """Generate, write, compile in a child, upload with the old key, confirm
+    with the new one, print it."""
+    args = MockArgs()
+    assert command_rotate_key(args, CORE.config) == 0
+
+    assert (
+        CORE.config_path.read_text()
+        == API_YAML.replace(OLD_KEY, NEW_KEY) + f'      old_key: "{OLD_KEY}"\n'
+    )
+    env["confirm"].assert_called_once()
+    precheck, confirm = env["probe"].call_args_list
+    assert precheck.args[2] == OLD_KEY
+    assert precheck.kwargs["retry_rejected"] is False
+    assert confirm.args[2] == NEW_KEY
+    compile_args = env["compile"].call_args.args
+    assert compile_args[-2:] == ("compile", str(CORE.config_path))
+    env["upload"].assert_called_once_with(
+        ["dev.local"],
+        3232,
+        None,
+        CORE.firmware_bin,
+        noise_psk=OLD_KEY,
+        plaintext_fallback=False,
+        on_upload_start=ANY,
+    )
+    out = capfd.readouterr().out
+    assert NEW_KEY in out
+    assert "SUCCESS" in out
+
+
+def test_yes_skips_confirmation(
+    env: dict[str, Mock],
+) -> None:
+    assert command_rotate_key(MockArgs(yes=True), CORE.config) == 0
+    env["confirm"].assert_not_called()
+
+
+def test_declined(env: dict[str, Mock]) -> None:
+    env["confirm"].return_value = "n"
+    assert command_rotate_key(MockArgs(), CORE.config) == 1
+    assert CORE.config_path.read_text() == API_YAML
+    env["probe"].assert_not_called()
+
+
+def test_warns_about_a_shared_secret(
+    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
+) -> None:
+    """Other configurations using the rewritten secret are named before the
+    confirmation."""
+
+    def scan(edits: list[KeyEdit]) -> list[str]:
+        edits[0].secret = "device_key"
+        edits[0].shared_with = [CORE.config_dir / "b.yaml"]
+        return ["c.yaml is a link"]
+
+    with patch("esphome.cli.rotate_key.with_sharers", side_effect=scan):
+        env["confirm"].return_value = "n"
+        assert command_rotate_key(MockArgs(), CORE.config) == 1
+    out = capfd.readouterr().out
+    assert "b.yaml (secret 'device_key')" in out
+    assert "Could not check every configuration for shared use: c.yaml is a link" in out
+
+
+def test_no_terminal_needs_yes(
+    env: dict[str, Mock],
+) -> None:
+    env["stdin"].isatty.return_value = False
+    assert command_rotate_key(MockArgs(), CORE.config) == 1
+    assert CORE.config_path.read_text() == API_YAML
+
+
+def test_own_ota_key_asks_nothing(
+    env: dict[str, Mock],
+) -> None:
+    """Without an api key nothing changes for Home Assistant."""
+    config = {CONF_OTA: [OTA_CONF]}
+    assert command_rotate_key(MockArgs(), config) == 0
+    env["confirm"].assert_not_called()
+    assert NEW_KEY in CORE.config_path.read_text()
+
+
+def test_prompted_new_key(env: dict[str, Mock]) -> None:
+    key = "AgMEBQYHCAkKCwwNDg8QERITFBUWFxgZGhscHR4fICE="
+    with patch("esphome.cli.rotate_key.read_secret_line", return_value=key):
+        assert command_rotate_key(MockArgs(prompt_new_key=True), CORE.config) == 0
+    assert key in CORE.config_path.read_text()
+    assert env["probe"].call_args_list[1].args[2] == key
+
+
+@pytest.mark.parametrize("key", ["not-base64", OLD_KEY], ids=["invalid", "same"])
+def test_rejects_bad_new_key(env: dict[str, Mock], key: str) -> None:
+    with patch("esphome.cli.rotate_key.read_secret_line", return_value=key):
+        assert command_rotate_key(MockArgs(prompt_new_key=True), CORE.config) == 1
+    assert CORE.config_path.read_text() == API_YAML
+    env["probe"].assert_not_called()
+
+
+def test_precheck_fails_writes_nothing(
+    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
+) -> None:
+    env["probe"].return_value = ProbeResult.REJECTED
+    assert command_rotate_key(MockArgs(), CORE.config) == 1
+    assert CORE.config_path.read_text() == API_YAML
+    env["compile"].assert_not_called()
+    assert "runs a different key than the yaml" in capfd.readouterr().out
+    for found in (
+        ProbeResult.UNREACHABLE,
+        ProbeResult.NOT_OFFERED,
+        ProbeResult.DEVICE_ERROR,
+    ):
+        env["probe"].return_value = found
+        assert command_rotate_key(MockArgs(), CORE.config) == 1
+        assert PRECHECK_ADVICE[found] in capfd.readouterr().out
+
+
+@pytest.mark.parametrize("step", ["compile", "interrupt"])
+def test_restores_before_the_upload(
+    env: dict[str, Mock], capfd: pytest.CaptureFixture[str], step: str
+) -> None:
+    if step == "compile":
+        env["compile"].return_value = 1
+    else:
+        env["compile"].side_effect = KeyboardInterrupt
+    image = CORE.firmware_bin
+    image.parent.mkdir(parents=True)
+    factory = image.with_name("firmware.factory.bin")
+    for built in (image, factory):
+        built.write_bytes(b"built with the new key")
+    assert command_rotate_key(MockArgs(), CORE.config) == 1
+    assert CORE.config_path.read_text() == API_YAML
+    assert env["probe"].call_count == 1
+    out = capfd.readouterr().out
+    assert ("Interrupted before the upload" in out) is (step == "interrupt")
+    # The images may hold the new key, so no upload can install them
+    assert not image.exists() and not factory.exists()
+    assert "compile again before the next upload" in out
+    assert NEW_KEY not in out
+
+
+def test_a_build_that_cannot_be_removed_shows_the_new_key(
+    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
+) -> None:
+    """Every image is tried; the ones that stay are all named."""
+    env["compile"].return_value = 1
+    image = CORE.firmware_bin
+    image.mkdir(parents=True)  # unlink fails on a directory
+    stuck = image.with_name("firmware.factory.bin")
+    stuck.mkdir()
+    removable = image.with_name("firmware.ota.bin")
+    removable.write_bytes(b"")
+    assert command_rotate_key(MockArgs(), CORE.config) == 1
+    out = capfd.readouterr().out
+    assert str(image) in out and str(stuck) in out
+    assert not removable.exists()
+    assert OLD_KEY in out and NEW_KEY in out
+
+
+def test_an_interrupt_during_the_image_cleanup_shows_the_keys(
+    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
+) -> None:
+    env["compile"].return_value = 1
+    CORE.firmware_bin.parent.mkdir(parents=True)
+    CORE.firmware_bin.write_bytes(b"")
+    with patch("pathlib.Path.glob", side_effect=KeyboardInterrupt):
+        assert command_rotate_key(MockArgs(), CORE.config) == 1
+    out = capfd.readouterr().out
+    assert f"{CORE.firmware_bin.parent}: interrupted" in out
+    assert OLD_KEY in out and NEW_KEY in out
+
+
+def test_a_build_from_before_the_rotation_is_kept(
+    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
+) -> None:
+    env["probe"].return_value = ProbeResult.UNREACHABLE
+    image = CORE.firmware_bin
+    image.parent.mkdir(parents=True)
+    image.write_bytes(b"built with the current key")
+    assert command_rotate_key(MockArgs(), CORE.config) == 1
+    assert image.exists()
+    assert "compile again" not in capfd.readouterr().out
+
+
+@pytest.mark.parametrize("rollback", ["succeeded", "failed"])
+def test_an_interrupt_while_writing_reports_a_failed_rollback(
+    env: dict[str, Mock], capfd: pytest.CaptureFixture[str], rollback: str
+) -> None:
+    """apply_line_edits keeps an interrupt an interrupt and notes a rollback
+    that failed too; then a file still holds the new key, so both are shown."""
+    interrupt = KeyboardInterrupt()
+    if rollback == "failed":
+        interrupt.add_note("Could not restore test.yaml: disk")
+    with patch("esphome.cli.rotate_key.apply_line_edits", side_effect=interrupt):
+        assert command_rotate_key(MockArgs(), CORE.config) == 1
+    out = capfd.readouterr().out
+    assert ("Could not restore test.yaml: disk" in out) is (rollback == "failed")
+    assert (NEW_KEY in out) is (rollback == "failed")
+    assert ("Interrupted before the upload" in out) is (rollback == "succeeded")
+
+
+def test_a_second_interrupt_during_the_restore_shows_the_keys(
+    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
+) -> None:
+    env["compile"].return_value = 1
+    with patch("esphome.yaml_edit.write_keeping_mode", side_effect=_interrupt_after(1)):
+        assert command_rotate_key(MockArgs(), CORE.config) == 1
+    out = capfd.readouterr().out
+    assert f"Could not restore {CORE.config_path}: interrupted" in out
+    assert OLD_KEY in out and NEW_KEY in out
+
+
+def test_restores_when_the_device_was_never_reached(
+    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
+) -> None:
+    """An upload that fails before the image is sent cannot have committed."""
+    env["upload"].side_effect = None
+    env["upload"].return_value = (1, None)
+    assert command_rotate_key(MockArgs(), CORE.config) == 1
+    assert CORE.config_path.read_text() == API_YAML
+    out = capfd.readouterr().out
+    assert "failed before the image was sent" in out
+    assert "put back" not in out
+    assert "Restored the previous key" in out
+
+
+def test_keeps_the_edit_after_a_failed_upload(
+    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
+) -> None:
+    """An attempted upload may have committed, so the config keeps both keys
+    and the new one is printed."""
+    env["upload"].return_value = (1, None)
+    assert command_rotate_key(MockArgs(), CORE.config) == 1
+    text = CORE.config_path.read_text()
+    assert NEW_KEY in text and f'old_key: "{OLD_KEY}"' in text
+    assert NEW_KEY in capfd.readouterr().out
+    assert env["probe"].call_count == 1
+
+
+def test_shows_both_keys_only_when_a_file_stayed_rewritten(
+    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
+) -> None:
+    with patch(
+        "esphome.cli.rotate_key.apply_line_edits", side_effect=RestoreError("stuck")
+    ):
+        assert command_rotate_key(MockArgs(), CORE.config) == 1
+    out = capfd.readouterr().out
+    assert "stuck" in out and OLD_KEY in out and NEW_KEY in out
+
+
+def test_reports_a_failed_restore(
+    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
+) -> None:
+    env["compile"].return_value = 1
+    with patch(
+        "esphome.cli.rotate_key.restore_files",
+        side_effect=RestoreError("Could not restore x"),
+    ):
+        assert command_rotate_key(MockArgs(), CORE.config) == 1
+    out = capfd.readouterr().out
+    assert "Could not restore x" in out
+    assert OLD_KEY in out and NEW_KEY in out
+
+
+@pytest.mark.parametrize(
+    "outcome", ["unreachable", "rejected", "not_offered", "device_error", "interrupted"]
+)
+def test_keeps_the_new_key_after_the_upload(
+    env: dict[str, Mock], capfd: pytest.CaptureFixture[str], outcome: str
+) -> None:
+    """Once the upload went through the device most likely runs the new key
+    and old_key covers the other case, so the files stay and the key is
+    printed."""
+    env["probe"].side_effect = [
+        ProbeResult.ACCEPTED,
+        {
+            "interrupted": KeyboardInterrupt,
+            "unreachable": ProbeResult.UNREACHABLE,
+            "rejected": ProbeResult.REJECTED,
+            "not_offered": ProbeResult.NOT_OFFERED,
+            "device_error": ProbeResult.DEVICE_ERROR,
+        }[outcome],
+    ]
+    assert command_rotate_key(MockArgs(), CORE.config) == 1
+    assert NEW_KEY in CORE.config_path.read_text()
+    assert f'old_key: "{OLD_KEY}"' in CORE.config_path.read_text()
+    out = capfd.readouterr().out
+    assert NEW_KEY in out
+    assert ("Interrupted" in out) is (outcome == "interrupted")
+    assert ("still answers with the previous key" in out) is (outcome == "rejected")
+    assert ("did not come back" in out) is (outcome == "unreachable")
+    assert ("came back without encryption" in out) is (outcome == "not_offered")
+    assert ("answered with an error" in out) is (outcome == "device_error")
+
+
+def test_child_compile_gets_global_options_first(
+    env: dict[str, Mock],
+) -> None:
+    """The compile parser is strict, so -s and --toolchain precede it."""
+    args = MockArgs(substitution=[["name", "kitchen"]])
+    CORE.dashboard = True
+    args.toolchain = Toolchain.PLATFORMIO
+    assert command_rotate_key(args, CORE.config) == 0
+    assert env["compile"].call_args.args[len(ESPHOME_COMMAND) :] == (
+        "--dashboard",
+        "--toolchain",
+        "platformio",
+        "-s",
+        "name",
+        "kitchen",
+        "compile",
+        str(CORE.config_path),
+    )
+
+
+@pytest.mark.parametrize(
+    ("config", "port_type", "match"),
+    [
+        (
+            {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3232}]},
+            "NETWORK",
+            "needs 'encryption:'",
+        ),
+        ({CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}]}, "NETWORK", "esphome OTA"),
+        ({CONF_OTA: [OTA_CONF]}, PortType.SERIAL, "over the air"),
+    ],
+    ids=["no_encryption", "web_server_only", "serial"],
+)
+def test_refuses(
+    env: dict[str, Mock],
+    capfd: pytest.CaptureFixture[str],
+    config: dict[str, Any],
+    port_type: str,
+    match: str,
+) -> None:
+    with patch("esphome.cli.rotate_key.get_port_type", return_value=port_type):
+        assert command_rotate_key(MockArgs(), config) == 1
+    assert match in capfd.readouterr().out
+    env["probe"].assert_not_called()
+
+
+def test_refuses_host(env: dict[str, Mock]) -> None:
+    CORE.data[KEY_CORE][KEY_TARGET_PLATFORM] = "host"
+    assert command_rotate_key(MockArgs(), CORE.config) == 1
+    env["probe"].assert_not_called()
+
+
+@pytest.mark.parametrize("helper", ["locate_key_edits", "apply_line_edits"])
+def test_reports_edit_errors(
+    env: dict[str, Mock], capfd: pytest.CaptureFixture[str], helper: str
+) -> None:
+    with patch(f"esphome.cli.rotate_key.{helper}", side_effect=EsphomeError("nope")):
+        assert command_rotate_key(MockArgs(), CORE.config) == 1
+    out = capfd.readouterr().out
+    assert "nope" in out
+    # A refusal, or an edit rolled back cleanly, shows no key
+    assert NEW_KEY not in out
+    assert CORE.config_path.read_text() == API_YAML

tests/unit_tests/components/test_libretiny_ota_uf2.py

@@ -1,125 +0,0 @@
-"""Tests for trimming a LibreTiny UF2 to the images a device writes over OTA."""
-
-from pathlib import Path
-import struct
-
-import pytest
-
-from esphome.components.libretiny import device_ota_image
-from esphome.components.libretiny.ota_uf2 import UF2_TAG_OTA_PART_INFO, device_ota_uf2
-from esphome.core import CORE
-
-# Written from the UF2 and LibreTiny formats, not from the module under test
-MAGIC_1, MAGIC_2, MAGIC_3 = 0x0A324655, 0x9E5D5157, 0x0AB16F30
-FLAG_FAMILY, FLAG_TAGS = 0x2000, 0x8000
-FAMILY = 0x7B3EF230
-TAG_OTA_FORMAT_2 = 0x6C8492
-TAG_BINPATCH = 0xB948DE
-# OTA_PART_INFO index nibbles: device single, dual 1, dual 2, flasher single, dual 1, dual 2
-DEVICE_SINGLE = b"\x10\x00\x00"
-FLASHER_SINGLE = b"\x00\x01\x00"
-DEVICE_DUAL_1 = b"\x01\x00\x00"
-DEVICE_DUAL_2 = b"\x00\x10\x00"
-
-
-def _tag(tag: int, value: bytes) -> bytes:
-    size = 4 + len(value)
-    return bytes([size]) + tag.to_bytes(3, "little") + value + bytes(-size % 4)
-
-
-def _block(payload: bytes = b"", tags: dict[int, bytes] | None = None) -> bytes:
-    """A block numbered 0 of 0; _package numbers it."""
-    flags = FLAG_FAMILY | (FLAG_TAGS if tags else 0)
-    header = struct.pack("<8I", MAGIC_1, MAGIC_2, flags, 0, len(payload), 0, 0, FAMILY)
-    body = payload + b"".join(_tag(t, v) for t, v in (tags or {}).items())
-    return header + body + bytes(512 - 32 - 4 - len(body)) + struct.pack("<I", MAGIC_3)
-
-
-def _image(part_info: bytes, blocks: int, fill: int) -> list[bytes]:
-    """An image: its first block names the partitions, the rest carry data."""
-    data = bytes([fill]) * 256
-    first = _block(data, {UF2_TAG_OTA_PART_INFO: part_info + b"part\x00"})
-    return [first] + [_block(data) for _ in range(blocks - 1)]
-
-
-def _package(*blocks: bytes) -> bytes:
-    """The blocks numbered in order, as a UF2 file."""
-    return b"".join(
-        block[:20] + struct.pack("<2I", seq, len(blocks)) + block[28:]
-        for seq, block in enumerate(blocks)
-    )
-
-
-HEADER = _block(tags={TAG_OTA_FORMAT_2: b"\x02"})
-FLASHER = _image(FLASHER_SINGLE, 5, 0xAA)
-DEVICE = _image(DEVICE_SINGLE, 3, 0xBB)
-SLOT_1 = _image(DEVICE_DUAL_1, 2, 0x11)
-SLOT_2 = _image(DEVICE_DUAL_2, 2, 0x22)
-PATCHED = _block(bytes([0xAA]) * 256, {TAG_BINPATCH: b"\x01\x02\x03\x04"})
-
-
-@pytest.mark.parametrize(
-    ("blocks", "expected"),
-    [
-        pytest.param([HEADER, *FLASHER, *DEVICE], [HEADER, *DEVICE], id="single_ota"),
-        pytest.param(
-            [HEADER, *SLOT_1, *FLASHER, *SLOT_2],
-            [HEADER, *SLOT_1, *SLOT_2],
-            id="dual_ota_keeps_both_slots",
-        ),
-        pytest.param(
-            [HEADER, *FLASHER, PATCHED, *DEVICE],
-            None,
-            id="dropped_image_carries_a_tag_the_device_acts_on",
-        ),
-        pytest.param([HEADER, *DEVICE], None, id="nothing_to_drop"),
-    ],
-)
-def test_device_ota_uf2(blocks: list[bytes], expected: list[bytes] | None) -> None:
-    trimmed = device_ota_uf2(_package(*blocks))
-    assert trimmed == (_package(*expected) if expected is not None else None)
-
-
-def test_data_that_is_not_a_well_formed_uf2_is_left_alone() -> None:
-    package = _package(HEADER, *FLASHER, *DEVICE)
-    bad_magic = package[:512] + b"\x00" + package[513:]
-    # A tag that runs past the data area of the first flasher block
-    at = 512 + 32 + 256
-    bad_tag = package[:at] + bytes([250]) + package[at + 1 :]
-
-    # A whole block missing, or repeated, keeps the original numbering
-    missing = package[:1024] + package[1536:]
-    repeated = package[:1536] + package[1024:]
-    # OTA_PART_INFO too short to hold the scheme map, which uf2ota rejects
-    short_selectors = [
-        _package(HEADER, _block(bytes(256), {UF2_TAG_OTA_PART_INFO: bytes(n)}), *DEVICE)
-        for n in range(3)
-    ]
-
-    for data in (
-        b"",
-        bytes(range(256)) * 4,
-        package[:-1],
-        bad_magic,
-        bad_tag,
-        missing,
-        repeated,
-        *short_selectors,
-    ):
-        assert device_ota_uf2(data) is None
-
-
-def test_device_ota_image(setup_core: Path) -> None:
-    tmp_path = setup_core
-    CORE.build_path = tmp_path / "build"
-    CORE.name = "test"
-    firmware = tmp_path / "firmware.uf2"
-
-    firmware.write_bytes(_package(HEADER, *FLASHER, *DEVICE))
-    image = device_ota_image(firmware)
-    assert image == CORE.relative_pioenvs_path("test", "firmware.ota.uf2")
-    assert image.read_bytes() == _package(HEADER, *DEVICE)
-
-    # Nothing to drop: the firmware itself is sent
-    firmware.write_bytes(_package(HEADER, *DEVICE))
-    assert device_ota_image(firmware) == firmware

tests/unit_tests/test_lazy_imports.py

@@ -222,9 +222,15 @@ def test_api_client_does_not_import_heavy_modules() -> None:
 
 # Packages a domain needs for its own schema; cheap and never an integration
 _ENTITY_COMPONENT_DEPENDENCIES = {
-    "infrared": ("esphome.components.ir_rf_base",),
     "light": ("esphome.components.power_supply", "esphome.components.const.css_colors"),
-    "radio_frequency": ("esphome.components.ir_rf_base",),
+}
+
+
+# Entity domains that still import integrations at top level
+_ENTITY_COMPONENT_KNOWN_LEAKS = {
+    "infrared": "imports remote_base and api at top level",
+    "media_player": "audio pulls in the esp32 chain",
+    "radio_frequency": "imports remote_base and api at top level",
 }
 
 
@@ -242,11 +248,26 @@ def _entity_components() -> list[str]:
 def test_entity_component_tables_match_discovery() -> None:
     """Catch stale table entries and a discovery that silently finds nothing."""
     found = set(_entity_components())
-    stale = _ENTITY_COMPONENT_DEPENDENCIES.keys() - found
+    stale = (
+        _ENTITY_COMPONENT_KNOWN_LEAKS.keys() | _ENTITY_COMPONENT_DEPENDENCIES.keys()
+    ) - found
     assert not stale, f"not discovered as entity components: {sorted(stale)}"
 
 
-@pytest.mark.parametrize("component", _entity_components())
+@pytest.mark.parametrize(
+    "component",
+    [
+        pytest.param(
+            c,
+            marks=pytest.mark.xfail(
+                reason=_ENTITY_COMPONENT_KNOWN_LEAKS[c], strict=True
+            ),
+        )
+        if c in _ENTITY_COMPONENT_KNOWN_LEAKS
+        else c
+        for c in _entity_components()
+    ],
+)
 def test_entity_component_does_not_import_integrations(component: str) -> None:
     """An entity component must not drag in its optional integrations.
 

tests/unit_tests/test_main.py

@@ -2135,118 +2135,6 @@ def test_upload_program_ota_success(
     )
 
 
-@pytest.mark.parametrize(
-    ("config", "uploader", "image_arg"),
-    [
-        pytest.param(
-            {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 8892}]},
-            "mock_run_ota",
-            3,
-            id="native",
-        ),
-        pytest.param(
-            {
-                CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}],
-                CONF_WEB_SERVER: {CONF_PORT: 80},
-            },
-            "mock_run_web_server_ota",
-            4,
-            id="web_server",
-        ),
-    ],
-)
-def test_upload_program_ota_sends_the_platform_ota_image(
-    config: ConfigType,
-    uploader: str,
-    image_arg: int,
-    mock_get_port_type: Mock,
-    tmp_path: Path,
-    request: pytest.FixtureRequest,
-) -> None:
-    """An app update sends the file the platform's ota_image hook picks."""
-    setup_core(platform=PLATFORM_BK72XX, tmp_path=tmp_path)
-    mock_get_port_type.return_value = "NETWORK"
-    mock_upload = request.getfixturevalue(uploader)
-    mock_upload.return_value = (0, "192.168.1.100")
-    image = tmp_path / "firmware.ota.uf2"
-
-    with patch(
-        "esphome.components.bk72xx.ota_image", return_value=image
-    ) as mock_ota_image:
-        exit_code, _ = upload_program(config, MockArgs(), ["192.168.1.100"])
-
-    assert exit_code == 0
-    mock_ota_image.assert_called_once_with(CORE.firmware_bin)
-    assert mock_upload.call_args.args[image_arg] == image
-
-
-def test_upload_program_ota_file_goes_through_the_ota_image_hook(
-    mock_run_ota: Mock,
-    mock_get_port_type: Mock,
-    tmp_path: Path,
-) -> None:
-    """--file is the firmware the platform's ota_image hook starts from."""
-    setup_core(platform=PLATFORM_BK72XX, tmp_path=tmp_path)
-    mock_get_port_type.return_value = "NETWORK"
-    mock_run_ota.return_value = (0, "192.168.1.100")
-    other = tmp_path / "other.uf2"
-    image = tmp_path / "firmware.ota.uf2"
-
-    config = {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 8892}]}
-    with patch(
-        "esphome.components.bk72xx.ota_image", return_value=image
-    ) as mock_ota_image:
-        upload_program(config, MockArgs(file=str(other)), ["192.168.1.100"])
-
-    mock_ota_image.assert_called_once_with(other)
-    assert mock_run_ota.call_args.args[3] == image
-
-
-def test_upload_program_ota_partition_table_skips_the_ota_image_hook(
-    mock_run_ota: Mock,
-    mock_get_port_type: Mock,
-    tmp_path: Path,
-) -> None:
-    """Only app updates go through the ota_image hook."""
-    setup_core(platform=PLATFORM_BK72XX, tmp_path=tmp_path)
-    mock_get_port_type.return_value = "NETWORK"
-    mock_run_ota.return_value = (0, "192.168.1.100")
-    partition_file = tmp_path / "partitions.bin"
-    partition_file.write_bytes(_make_partition_table_bytes())
-
-    config = {
-        CONF_OTA: [
-            {
-                CONF_PLATFORM: CONF_ESPHOME,
-                CONF_PORT: 8892,
-                "allow_partition_access": True,
-            }
-        ]
-    }
-    args = MockArgs(file=str(partition_file), partition_table=True)
-    with patch("esphome.components.bk72xx.ota_image") as mock_ota_image:
-        upload_program(config, args, ["192.168.1.100"])
-
-    mock_ota_image.assert_not_called()
-    assert mock_run_ota.call_args.args[3] == partition_file
-
-
-def test_upload_program_ota_without_a_hook_sends_the_firmware(
-    mock_run_ota: Mock,
-    mock_get_port_type: Mock,
-    tmp_path: Path,
-) -> None:
-    """A platform without an ota_image hook sends its build output."""
-    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
-    mock_get_port_type.return_value = "NETWORK"
-    mock_run_ota.return_value = (0, "192.168.1.100")
-
-    config = {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3232}]}
-    upload_program(config, MockArgs(), ["192.168.1.100"])
-
-    assert mock_run_ota.call_args.args[3].name == "firmware.bin"
-
-
 @pytest.mark.parametrize(
     "old_key",
     [None, "AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA="],
@@ -8248,3 +8136,32 @@ def test_command_rename_is_dispatched_to_the_cli_module() -> None:
     with patch("esphome.cli.rename.command_rename", return_value=7) as run:
         assert main.command_rename(args, {}) == 7
     run.assert_called_once_with(args, {})
+
+
+def test_command_rotate_key_is_dispatched_to_the_cli_module() -> None:
+    args = MockArgs()
+    with patch("esphome.cli.rotate_key.command_rotate_key", return_value=7) as run:
+        assert main.command_rotate_key(args, {}) == 7
+    run.assert_called_once_with(args, {})
+
+
+def test_parse_args_rotate_key() -> None:
+    args = parse_args(
+        [
+            "esphome",
+            "rotate-key",
+            "--device",
+            "dev.local",
+            "--prompt-new-key",
+            "-y",
+            "--username",
+            "mqtt-user",
+            "device.yaml",
+        ]
+    )
+    assert args.command == "rotate-key"
+    assert args.device == ["dev.local"]
+    assert args.prompt_new_key is True
+    assert args.yes is True
+    # MQTT device resolution reads these like upload does
+    assert args.username == "mqtt-user"