Compare commits

...
Author SHA1 Message Date
J. Nick Koston a8e2a04052 Merge remote-tracking branch 'origin/bq25186-switch-tests' into bq25186-apply-switch 2026-10-26 07:28:07 -10:00
J. Nick Koston 83d2e2efed Merge remote-tracking branch 'origin/apply-publish-on-success' into bq25186-apply-switch 2026-10-26 07:28:07 -10:00
J. Nick Koston c04cdea1a0 [core] Build the on_success condition from the call expression and list the publish modes once 2026-10-26 07:27:55 -10:00
J. Nick Koston b1a2f7aacf [bq25186] Publish the PG/GPO switch with publish="on_success" 2026-10-26 07:27:32 -10:00
J. Nick Koston 2c11cfd1fa [bq25186] Start the refused-write test from on and cover an inverted setup without restore 2026-10-26 07:27:30 -10:00
J. Nick Koston 2c80263ddc Merge remote-tracking branch 'origin/bq25186-switch-tests' into bq25186-apply-switch 2026-10-26 07:26:00 -10:00
J. Nick Koston 050a5a2de0 Merge remote-tracking branch 'origin/apply-publish-on-success' into bq25186-apply-switch 2026-10-26 07:26:00 -10:00
J. Nick Koston 46d8c3ec02 [core] Add publish="on_success" for Apply entities whose hub call can fail 2026-10-26 07:25:16 -10:00
J. Nick Koston 3259e72d9d [bq25186] Name the test helpers for clang-tidy 2026-10-26 07:23:31 -10:00
6cb622417a [tas58xx] Add fault_output and modulation options (#21242)
Co-authored-by: J. Nick Koston <nick@koston.org>
Co-authored-by: J. Nick Koston <nick@home-assistant.io>
2026-10-26 17:18:18 +00:00
Jonathan Swoboda 960316abd6 [core] Percent-encode "/" in git credentials (#21653) 2026-10-26 07:17:44 -10:00
J. Nick Koston ee59cf8974 [bq25186] Use an Apply switch for the PG/GPO switch 2026-10-26 07:14:38 -10:00
J. Nick Koston 29adc6f7e7 [bq25186] Test the PG/GPO switch writes, failed writes, inversion and restore 2026-10-26 07:13:54 -10:00
Jonathan Swoboda 87341dc136 [logger] Enable Log V2 only on the native ESP-IDF toolchain (#21652) 2026-10-26 13:03:15 -04:00
J. Nick Koston cfe5809654 [wifi] Hand the DPP bootstrapping key to the supplicant DER encoded (#20882) 2026-10-26 16:37:25 +00:00
J. Nick Koston 9c54f96520 [libretiny] Send only the UF2 images the device writes over OTA (#21642) 2026-10-26 06:28:54 -10:00
J. Nick Kostonandpre-commit-ci-lite[bot] 0cc5b372fb [core] Defer integration imports in ir_rf_base and media_player (#21601)
Co-authored-by: pre-commit-ci-lite[bot] <117423508+pre-commit-ci-lite[bot]@users.noreply.github.com>
2026-10-26 06:26:00 -10:00
J. Nick Koston f7c8cba8b7 [components] Use pulse_reset for blocking reset pin pulses (#21472) 2026-10-26 06:22:11 -10:00
J. Nick Koston 79a22e9a9a [core] Add pulse_reset for active-low reset pins (#21470) 2026-10-26 06:22:10 -10:00
J. Nick Koston 103023737d [openthread] Resolve IPv6 literal addresses when the DNS64 client is enabled (#19269) 2026-10-26 16:18:12 +00:00
J. Nick Koston 6b6b70a8b6 [dfrobot_sen0395] Collect lines with LineReader (#20979) 2026-10-26 12:16:51 -04:00
J. Nick Koston d312bf1b73 [core] Let the yaml editor insert lines, write with a rollback and restore (#20602) 2026-10-26 06:16:26 -10:00
84 changed files with 1988 additions and 401 deletions

No files matched your search

+2 -1
View File
@@ -443,7 +443,8 @@ file does, and it is the authority when they disagree. The most useful starting
- **Hub forwarding entities need no C++ class:** a button, switch, number or select that only calls its hub
uses `button.ApplyButton` (`switch.ApplySwitch`, `number.ApplyNumber`, `select.ApplySelect`) in its schema
and `apply=` on the `sub_*` call. The body is a call on the hub with `entity` and `state`, `value` or
`index`/`option` in scope; on a switch, number or select `publish="before"` or `"after"` publishes the state.
`index`/`option` in scope; on a switch, number or select `publish="before"` or `"after"` publishes the state,
and `publish="on_success"` publishes only when the hub call returns true (a write the device can refuse).
`publish="momentary"` is only for legacy action switches (restart, factory reset): it publishes off and runs
the body only when turned on. Never use it for a new entity; an action is a button.
```python
+17 -10
View File
@@ -1305,10 +1305,7 @@ 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."
)
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_web_server(config, network_devices, _app_ota_image(args))
return _upload_via_native_api(config, network_devices, args)
@@ -1364,6 +1361,16 @@ 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]:
@@ -1413,27 +1420,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 = CORE.partition_table_bin
binary = Path(file) if file is not None else CORE.partition_table_bin
ota_type = espota2.OTA_TYPE_UPDATE_PARTITION_TABLE
elif getattr(args, "bootloader", False):
check_partition_access("--bootloader")
if (
getattr(args, "file", None) is None
file is None
and (native := native_backend())
and (hint := getattr(native, "missing_image_hint", lambda: None)())
):
raise EsphomeError(hint)
binary = CORE.bootloader_bin
binary = Path(file) if file is not None else CORE.bootloader_bin
ota_type = espota2.OTA_TYPE_UPDATE_BOOTLOADER
if getattr(args, "file", None) is not None:
binary = Path(args.file)
else:
binary = _app_ota_image(args)
if ota_type == espota2.OTA_TYPE_UPDATE_PARTITION_TABLE:
_validate_partition_table_binary(binary)
+27 -1
View File
@@ -33,9 +33,35 @@ _CACHE_VERSION = 1
_LAMBDA_KEY = "__esphome_lambda__"
_CACHE_SUFFIX = ".validated.json"
def compiled_config_path(config_filename: str) -> Path:
"""Path to the cached validated config alongside the storage sidecar."""
return CORE.data_dir / "storage" / f"{config_filename}.validated.json"
return CORE.data_dir / "storage" / f"{config_filename}{_CACHE_SUFFIX}"
def invalidate_compiled_config() -> None:
"""Drop every cache in the storage dir after a change to a file the
mtime check cannot see, such as secrets.yaml or an include; other
configurations may share that file."""
storage = compiled_config_path(CORE.config_filename).parent
try:
caches = [p for p in storage.iterdir() if p.name.endswith(_CACHE_SUFFIX)]
except FileNotFoundError:
return
except OSError as err:
raise EsphomeError(
f"Could not list the validated config caches in {storage}: {err}"
) from err
for path in caches:
try:
path.unlink(missing_ok=True)
except OSError as err:
# A stale cache would present the old key on the next upload
raise EsphomeError(
f"Could not remove the validated config cache {path}: {err}"
) from err
def save_compiled_config(config: ConfigType) -> None:
+1 -3
View File
@@ -297,9 +297,7 @@ void ADE7880::init_device_() {
void ADE7880::reset_device_() {
if (this->reset_pin_ != nullptr) {
ESP_LOGD(TAG, "Reset device using RESET pin");
this->reset_pin_->digital_write(false);
delay(1);
this->reset_pin_->digital_write(true);
pulse_reset(this->reset_pin_, {.low_ms = 1});
} else {
ESP_LOGD(TAG, "Reset device using SWRST command");
this->write_u16_register16_(CONFIG, CONFIG_SWRST);
@@ -16,13 +16,7 @@ constexpr static const uint8_t AXS_READ_TOUCHPAD[11] = {0xb5, 0xab, 0xa5, 0x5a,
}
void AXS15231Touchscreen::setup() {
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);
}
pulse_reset(this->reset_pin_, {.low_ms = 5, .high_after_ms = 10});
if (this->interrupt_pin_ != nullptr) {
this->interrupt_pin_->pin_mode(gpio::FLAG_INPUT);
this->interrupt_pin_->setup();
+4
View File
@@ -78,3 +78,7 @@ 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
@@ -1,21 +0,0 @@
#pragma once
#include "esphome/core/defines.h"
#ifdef USE_BQ25186_PG_GPO
#include "bq25186.h"
#include "esphome/components/switch/switch.h"
namespace esphome::bq25186 {
class BQ25186PgGpoSwitch final : public switch_::Switch, public Parented<BQ25186Component> {
protected:
void write_state(bool state) override {
if (this->parent_->write_pg_gpo_level(state))
this->publish_state(state);
}
};
} // namespace esphome::bq25186
#endif // USE_BQ25186_PG_GPO
+8 -5
View File
@@ -3,18 +3,16 @@ from esphome.components import switch
import esphome.config_validation as cv
from esphome.types import ConfigType
from . import BQ25186_CHILD_SCHEMA, CONF_BQ25186_ID, bq25186_ns
from . import BQ25186_CHILD_SCHEMA, CONF_BQ25186_ID
DEPENDENCIES = ["bq25186"]
CONF_PG_GPO = "pg_gpo"
BQ25186PgGpoSwitch = bq25186_ns.class_("BQ25186PgGpoSwitch", switch.Switch)
CONFIG_SCHEMA = BQ25186_CHILD_SCHEMA.extend(
{
cv.Required(CONF_PG_GPO): switch.switch_schema(
BQ25186PgGpoSwitch, default_restore_mode="DISABLED"
switch.ApplySwitch, default_restore_mode="DISABLED"
),
}
)
@@ -24,4 +22,9 @@ async def to_code(config: ConfigType) -> None:
cg.add_define("USE_BQ25186_PG_GPO")
hub = await cg.get_variable(config[CONF_BQ25186_ID])
switches = switch.sub_switches(config, parent=hub)
await switches(CONF_PG_GPO, hub.set_pg_gpo_switch)
await switches(
CONF_PG_GPO,
hub.set_pg_gpo_switch,
apply="write_pg_gpo_level(state)",
publish="on_success",
)
@@ -6,12 +6,7 @@ ESPHOME_LOG_TAG(TAG, "cst226.touchscreen");
void CST226Touchscreen::setup() {
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);
pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 5});
this->set_timeout(30, [this] { this->continue_setup_(); });
} else {
this->continue_setup_();
@@ -43,9 +43,7 @@ void CST328Touchscreen::setup() {
}
void CST328Touchscreen::reset_device_() {
this->reset_pin_->digital_write(false);
delay(5);
this->reset_pin_->digital_write(true);
pulse_reset(this->reset_pin_, {.low_ms = 5});
this->set_timeout(CST328_TRANSITION_TIMEOUT, [this] { this->continue_setup_(); });
}
@@ -52,12 +52,7 @@ void CST816Touchscreen::continue_setup_() {
void CST816Touchscreen::setup() {
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);
pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 5});
this->set_timeout(30, [this] { this->continue_setup_(); });
} else {
this->continue_setup_();
@@ -10,14 +10,7 @@ static constexpr uint32_t REREAD_TIMEOUT_ID = 1;
static constexpr uint32_t REREAD_DELAY_MS = 20;
void CST9220Touchscreen::setup() {
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);
}
pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 10});
// Wait for the controller to leave its bootloader before talking to it.
this->set_timeout(30, [this] { this->continue_setup_(); });
}
@@ -42,6 +42,7 @@ CONFIG_SCHEMA = cv.All(
async def to_code(config: ConfigType) -> None:
cg.add_define("USE_LINE_READER")
await uart.new_uart_component(config)
@@ -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 (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))
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))
return false; // Command not done yet
this->read_message_(); // Skip the prompt
return true;
@@ -146,12 +146,13 @@ bool DfrobotSen0395Component::run_command_(const Command &cmd, uint32_t now) {
}
void DfrobotSen0395Component::read_state_() {
if (!this->read_message_())
const char *message = this->read_message_();
if (message == nullptr)
return;
if (ESPHOME_strstr_P(this->read_buffer_, ESPHOME_PSTR("$JYBSS,0, , , *")) != nullptr) {
if (ESPHOME_strstr_P(message, ESPHOME_PSTR("$JYBSS,0, , , *")) != nullptr) {
this->set_detected_(false);
this->set_active(true);
} else if (ESPHOME_strstr_P(this->read_buffer_, ESPHOME_PSTR("$JYBSS,1, , , *")) != nullptr) {
} else if (ESPHOME_strstr_P(message, ESPHOME_PSTR("$JYBSS,1, , , *")) != nullptr) {
this->set_detected_(true);
this->set_active(true);
}
@@ -206,8 +207,7 @@ void DfrobotSen0395Component::format_command(const Command &cmd, char *buf, size
}
}
bool DfrobotSen0395Component::handle_reply_(const Command &cmd) {
const char *message = this->read_buffer_;
bool DfrobotSen0395Component::handle_reply_(const Command &cmd, const char *message) {
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) {
}
return true;
case CommandType::COMMAND_TYPE_RESET_SYSTEM:
if (!is(ESPHOME_PSTR("leapMMW:/>")))
if (!is(PROMPT))
return false;
ESP_LOGI(TAG, "Restarted sensor.");
return true;
@@ -300,34 +300,24 @@ bool DfrobotSen0395Component::handle_reply_(const Command &cmd) {
return true;
}
bool DfrobotSen0395Component::read_message_() {
const char *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.
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
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_pos_++;
if (line != nullptr) {
ESP_LOGV(TAG, "Message: %s", line);
return line;
}
this->line_reader_.log_if_overflowed(TAG);
}
return false; // No full message yet
return nullptr;
}
void DfrobotSen0395Component::set_detected_(bool detected) {
@@ -4,6 +4,7 @@
#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"
@@ -14,7 +15,7 @@
namespace esphome::dfrobot_sen0395 {
static constexpr uint8_t MMWAVE_READ_BUFFER_LENGTH = 255;
static constexpr uint8_t MMWAVE_READ_BUFFER_LENGTH = 253; // largest LineReader size with a one byte position
static constexpr uint8_t COMMAND_QUEUE_SIZE = 20;
// Keep in sync with the BusyActions table in dfrobot_sen0395.cpp
@@ -121,8 +122,7 @@ class DfrobotSen0395Component final : public uart::UARTDevice, public Component
bool led_active_{false};
bool uart_presence_active_{false};
bool start_after_boot_{false};
char read_buffer_[MMWAVE_READ_BUFFER_LENGTH];
size_t read_pos_{0};
LineReader<MMWAVE_READ_BUFFER_LENGTH> line_reader_;
bool cmd_sent_{false};
uint8_t retries_left_{2};
StaticRingBuffer<Command, COMMAND_QUEUE_SIZE> cmd_queue_;
@@ -131,12 +131,13 @@ class DfrobotSen0395Component final : public uart::UARTDevice, public Component
int16_t det_range_steps_[8]{};
uint16_t latency_steps_[2]{};
bool read_message_();
/// The next complete line, valid until the next call, or nullptr
const char *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);
bool handle_reply_(const Command &cmd, const char *message);
void set_detected_(bool detected);
};
@@ -20,9 +20,7 @@ void EKTF2232Touchscreen::setup() {
this->attach_interrupt_(this->interrupt_pin_, gpio::INTERRUPT_FALLING_EDGE);
this->reset_pin_->setup();
this->hard_reset_();
pulse_reset(this->reset_pin_, {.low_ms = 15, .high_after_ms = 15});
if (!this->soft_reset_()) {
ESP_LOGE(TAG, "Failed to soft reset EKT2232!");
this->interrupt_pin_->detach_interrupt();
@@ -96,13 +94,6 @@ 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)
@@ -22,7 +22,6 @@ class EKTF2232Touchscreen final : public Touchscreen, public i2c::I2CDevice {
bool get_power_state();
protected:
void hard_reset_();
bool soft_reset_();
void update_touches() override;
@@ -72,9 +72,7 @@ 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.
this->reset_pin_->digital_write(false);
delay(SLEEP_MS_RESET1);
this->reset_pin_->digital_write(true);
pulse_reset(this->reset_pin_, {.low_ms = SLEEP_MS_RESET1});
this->reset_duration_ = SLEEP_MS_RESET2;
this->step_ = FSMState::RESET_STEP2_IDLECHECK;
+1 -12
View File
@@ -34,10 +34,7 @@ void FT63X6Touchscreen::setup() {
this->attach_interrupt_(this->interrupt_pin_, gpio::INTERRUPT_ANY_EDGE);
}
if (this->reset_pin_ != nullptr) {
this->reset_pin_->setup();
this->hard_reset_();
}
pulse_reset(this->reset_pin_, {.low_ms = 10});
// Get touch resolution
if (this->x_raw_max_ == this->x_raw_min_) {
@@ -57,14 +54,6 @@ 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);
-1
View File
@@ -27,7 +27,6 @@ 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};
@@ -83,13 +83,7 @@ void GSL3670Touchscreen::update_touches() {
void GSL3670Touchscreen::clear_reg_() {
ESP_LOGD(TAG, "clear_reg");
// GPIO reset pulse
if (this->reset_pin_ != nullptr) {
this->reset_pin_->digital_write(false);
delay(1);
this->reset_pin_->digital_write(true);
delay(5);
}
pulse_reset(this->reset_pin_, {.low_ms = 1, .high_after_ms = 5});
this->write_reg8_(0x88, 0x01);
// delay(5);
@@ -106,12 +100,7 @@ void GSL3670Touchscreen::clear_reg_() {
void GSL3670Touchscreen::reset_() {
ESP_LOGD(TAG, "reset");
if (this->reset_pin_ != nullptr) {
this->reset_pin_->digital_write(false);
delay(1);
this->reset_pin_->digital_write(true);
delay(5);
}
pulse_reset(this->reset_pin_, {.low_ms = 1, .high_after_ms = 5});
this->write_reg8_(0xe4, 0x04);
+1 -6
View File
@@ -19,12 +19,7 @@ void ICNT86Touchscreen::setup() {
}
// Perform reset if necessary
if (this->reset_pin_ != nullptr) {
this->reset_pin_->setup();
this->reset_pin_->digital_write(false);
delay(10);
this->reset_pin_->digital_write(true);
}
pulse_reset(this->reset_pin_, {.low_ms = 10});
if (this->x_raw_max_ == this->x_raw_min_) {
this->x_raw_max_ = this->display_->get_native_width();
+4 -3
View File
@@ -1,8 +1,7 @@
"""Shared base for the infrared and radio_frequency entity components."""
import esphome.codegen as cg
from esphome.components import remote_base
from esphome.components.api import DOMAIN as API_DOMAIN
from esphome.const import CONF_API
from esphome.core import CORE
from esphome.core.entity_helpers import queue_entity_register
from esphome.types import ConfigType
@@ -24,8 +23,10 @@ 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 API_DOMAIN in CORE.config:
if CONF_API in CORE.config:
cg.add_define("USE_IR_RF_TRANSMIT_COMPLETE")
+21 -1
View File
@@ -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
from esphome.helpers import copy_file_if_changed, write_file
from esphome.platformio.toolchain import copy_ccache_script
from esphome.storage_json import StorageJSON
@@ -656,6 +656,26 @@ 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
@@ -91,6 +91,10 @@ 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 = '''
+98
View File
@@ -0,0 +1,98 @@
"""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)
+4
View File
@@ -77,3 +77,7 @@ 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
+4 -3
View File
@@ -607,9 +607,10 @@ 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 idf_version() >= cv.Version(6, 1, 0) and get_esp32_variant() not in (
VARIANT_ESP32C2,
VARIANT_ESP32S2,
if (
CORE.using_toolchain_esp_idf
and 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)
+4 -1
View File
@@ -2,7 +2,6 @@ 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 (
@@ -116,6 +115,8 @@ 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)
@@ -155,6 +156,8 @@ 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
+1 -6
View File
@@ -107,12 +107,7 @@ void MipiDsi::setup() {
}
#endif
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);
pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 5});
} else {
esp_lcd_panel_io_tx_param(this->io_handle_, SW_RESET_CMD, nullptr, 0);
}
+1 -8
View File
@@ -33,14 +33,7 @@ void MipiRgb::setup_enables_() {
}
delay(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(5);
this->reset_pin_->digital_write(true);
}
pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 5});
}
#ifdef USE_SPI
+2 -9
View File
@@ -151,15 +151,8 @@ class MipiSpi : public display::Display,
pin->setup();
pin->digital_write(true);
}
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
}
// 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});
// need to know when the display is ready for SLPOUT command - will be 120ms after reset
auto when = millis() + 120;
@@ -226,6 +226,12 @@ 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:
@@ -35,6 +35,10 @@
#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");
@@ -489,4 +493,26 @@ 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
+1 -14
View File
@@ -9,23 +9,10 @@ ESPHOME_LOG_TAG(TAG, "pcd_8544");
void PCD8544::setup_pins_() {
this->spi_setup();
this->init_reset_();
pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10});
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_());
-2
View File
@@ -57,8 +57,6 @@ class PCD8544 final : public display::DisplayBuffer,
void setup_pins_();
void init_reset_();
size_t get_buffer_length_();
void start_command_();
+4
View File
@@ -77,3 +77,7 @@ 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
+1 -12
View File
@@ -122,18 +122,7 @@ 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_() {
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 SH1122::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
const LogString *SH1122::model_str_() {
switch (this->model_) {
case SH1122_MODEL_256_64:
@@ -355,18 +355,7 @@ void SSD1306::fill(Color color) {
for (uint32_t i = 0; i < this->get_buffer_length_(); i++)
this->buffer_[i] = fill;
}
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);
}
}
void SSD1306::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
const LogString *SSD1306::model_str_() {
return ModelStrings::get_log_str(static_cast<uint8_t>(this->model_), ModelStrings::LAST_INDEX);
}
@@ -184,18 +184,7 @@ void SSD1322::fill(Color color) {
for (uint32_t i = 0; i < this->get_buffer_length_(); i++)
this->buffer_[i] = fill;
}
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);
}
}
void SSD1322::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
const char *SSD1322::model_str_() {
switch (this->model_) {
case SSD1322_MODEL_256_64:
@@ -211,18 +211,7 @@ void SSD1325::fill(Color color) {
for (uint32_t i = 0; i < this->get_buffer_length_(); i++)
this->buffer_[i] = fill;
}
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);
}
}
void SSD1325::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
const char *SSD1325::model_str_() {
switch (this->model_) {
case SSD1325_MODEL_128_32:
@@ -160,18 +160,7 @@ void SSD1327::fill(Color color) {
for (uint32_t i = 0; i < this->get_buffer_length_(); i++)
this->buffer_[i] = fill;
}
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);
}
}
void SSD1327::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
const char *SSD1327::model_str_() {
switch (this->model_) {
case SSD1327_MODEL_128_128:
@@ -142,17 +142,6 @@ void SSD1331::fill(Color color) {
}
}
}
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);
}
}
void SSD1331::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
} // namespace esphome::ssd1331_base
@@ -174,18 +174,7 @@ void SSD1351::fill(Color color) {
}
}
}
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);
}
}
void SSD1351::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
const char *SSD1351::model_str_() {
switch (this->model_) {
case SSD1351_MODEL_128_96:
@@ -9,12 +9,8 @@ ESPHOME_LOG_TAG(TAG, "st7123.touchscreen");
void ST7123Touchscreen::setup() {
if (this->reset_pin_ != nullptr) {
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);
// TP_RESX is active low, assert for at least tRSTW (2ms)
pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 5});
// The controller needs up to 20ms to initialize after reset before it can be accessed.
this->setup_time_ = millis() + 30;
}
+1 -12
View File
@@ -137,18 +137,7 @@ void ST7567::fill(Color color) {
memset(buffer_, fill, this->get_buffer_length_());
}
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);
}
}
void ST7567::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
const char *ST7567::model_str_() { return "ST7567 128x64"; }
+23
View File
@@ -6,6 +6,7 @@ from esphome.automation import maybe_simple_id
import esphome.codegen as cg
from esphome.components import i2c
from esphome.components.audio_dac import AudioDac
from esphome.components.const import CONF_MODULATION
import esphome.config_validation as cv
from esphome.const import CONF_ENABLE_PIN, CONF_ID, CONF_MODEL
from esphome.cpp_generator import MockObj
@@ -19,6 +20,7 @@ DEPENDENCIES = ["i2c"]
CONF_ANALOG_GAIN = "analog_gain"
CONF_DAC_MODE = "dac_mode"
CONF_FAULT_OUTPUT = "fault_output"
CONF_IGNORE_ENABLE_PIN_WARNING = "ignore_enable_pin_warning"
CONF_MIXER_MODE = "mixer_mode"
CONF_VOLUME_MIN_DB = "volume_min_db"
@@ -39,6 +41,11 @@ MIXER_MODES = MixerMode.options(
"stereo", "stereo_inverse", "mono", "left", "right", prefix="MIXER_MODE_"
)
Modulation = tas58xx_ns.enum("Modulation")
MODULATIONS = Modulation.options("bd_mode", "1spw_mode", prefix="MODULATION_")
FAULT_OUTPUT_NONE = "none"
@dataclass(frozen=True)
class Model:
@@ -49,6 +56,9 @@ class Model:
volume_min_db: float
volume_max_db: float
default_address: int
# Pins that can report FAULTZ: name -> pin number n, selected at 0x61 + n (ModelInfo::fault_output_pins)
fault_outputs: dict[str, int]
default_fault_output: str
MODELS: dict[str, Model] = {
@@ -58,6 +68,8 @@ MODELS: dict[str, Model] = {
volume_min_db=-103.0,
volume_max_db=24.0,
default_address=0x2D,
fault_outputs={"adr": 0},
default_fault_output="adr",
),
DAC_TAS5825M: Model(
model_info=tas58xx_ns.TAS5825M_MODEL,
@@ -65,6 +77,8 @@ MODELS: dict[str, Model] = {
volume_min_db=-103.0,
volume_max_db=24.0,
default_address=0x4C,
fault_outputs={"gpio0": 0, "gpio1": 1, "gpio2": 2},
default_fault_output="gpio1",
),
}
@@ -100,9 +114,15 @@ def _model_schema(model: Model) -> cv.Schema:
cv.Optional(CONF_DAC_MODE, default="btl"): cv.enum(
DAC_MODES, lower=True
),
cv.Optional(
CONF_FAULT_OUTPUT, default=model.default_fault_output
): cv.one_of(FAULT_OUTPUT_NONE, *model.fault_outputs, lower=True),
cv.Optional(CONF_MIXER_MODE, default="stereo"): cv.enum(
MIXER_MODES, lower=True
),
cv.Optional(CONF_MODULATION, default="bd_mode"): cv.enum(
MODULATIONS, lower=True
),
cv.Optional(
CONF_VOLUME_MIN_DB, default=model.volume_min_db
): volume_validator,
@@ -172,7 +192,10 @@ async def to_code(config: ConfigType) -> None:
cg.add(var.set_analog_gain(config[CONF_ANALOG_GAIN]))
cg.add(var.set_dac_mode(config[CONF_DAC_MODE]))
if (fault_output := config[CONF_FAULT_OUTPUT]) != FAULT_OUTPUT_NONE:
cg.add(var.set_fault_output(model.fault_outputs[fault_output]))
cg.add(var.set_mixer_mode(config[CONF_MIXER_MODE]))
cg.add(var.set_modulation(config[CONF_MODULATION]))
cg.add(var.set_volume_min_db(config[CONF_VOLUME_MIN_DB]))
cg.add(var.set_volume_max_db(config[CONF_VOLUME_MAX_DB]))
await cg.add_pin(var.set_enable_pin, config.get(CONF_ENABLE_PIN))
@@ -8,11 +8,8 @@ namespace esphome::tas58xx {
// Registers 0x46, 0x7D, 0x7E and page 1 register 0x51 are not documented in the datasheet.
static constexpr i2c::I2CRegValue STARTUP_SEQUENCE[] PROGMEM = {
{0x03, 0x00}, // DEVICE_CTRL_2: deep sleep
{0x46, 0x01},
{0x46, 0x01}, // PPC 96 kHz process flow value (its 48 kHz FIR flows write 0x11)
{0x03, 0x02}, // DEVICE_CTRL_2: Hi-Z
// The I2C address is latched at power up, after which the ADR pin can report faults
{0x61, 0x0B}, // ADR_PIN_CONFIG: FAULTZ
{0x60, 0x01}, // ADR_PIN_CTRL: output
{0x7D, 0x11},
{0x7E, 0xFF},
{0x00, 0x01}, // Page 1
@@ -25,6 +22,8 @@ static constexpr i2c::I2CRegValue STARTUP_SEQUENCE[] PROGMEM = {
static const LogString *model_name() { return LOG_STR("TAS5805M"); }
static const LogString *fault_output_name(uint8_t) { return LOG_STR("ADR"); }
// An if chain rather than a switch: a switch table would land in rodata, which is RAM on ESP8266.
static const LogString *fault_name(uint8_t index) {
if (index == 0)
@@ -88,6 +87,8 @@ const ModelInfo TAS5805M_MODEL = {
NO_BIT, // FAULT_SENSOR_RIGHT_CHANNEL_CBC_OVER_CURRENT_WARNING
NO_BIT, // FAULT_SENSOR_OVER_TEMP_146C_WARNING
},
.fault_output_pins = 1, // ADR
.fault_output_name = fault_output_name,
};
} // namespace esphome::tas58xx
+12 -5
View File
@@ -6,7 +6,7 @@ namespace esphome::tas58xx {
// Remainder of the startup sequence, run after the reset in TAS58xx::init_(). Register 0x00 selects the page.
// Based on the TI PurePath Console (PPC) export for "TAS5825M 2-Band DRC&AGL 2.0 96k".
// Registers 0x46, 0x7D, 0x7E, page 1 register 0x51 and page 2 registers 0x1D, 0x19 are not documented in datasheet.
// Registers 0x7D, 0x7E, page 1 register 0x51 and page 2 registers 0x1D, 0x19 are not documented in datasheet.
static constexpr i2c::I2CRegValue STARTUP_SEQUENCE[] PROGMEM = {
{0x03, 0x00}, // DEVICE_CTRL_2: deep sleep, the state the PPC sequence starts from
{0x7D, 0x11},
@@ -17,20 +17,25 @@ static constexpr i2c::I2CRegValue STARTUP_SEQUENCE[] PROGMEM = {
{0x1D, 0x00},
{0x19, 0x80},
{0x00, 0x00}, // Page 0
{0x46, 0x11},
{0x46, 0x11}, // DSP_CTRL: 96 kHz processing (bits 4-3 = 10), use default coefficients (bit 0 = 1)
// PPC also writes 0x02 (DEVICE_CTRL_1) and 0x54 (AGAIN) here; init_() writes both from YAML
{0x53, 0x01}, // ANA_CTRL: in phase, 100kHz bandwidth for high audio performance with Fsw=384kHz
{0x03, 0x02}, // DEVICE_CTRL_2: Hi-Z
// Additional register configuration
// GPIO1 is a dedicated pin on TAS5825M (separate from ADR), used as FAULTZ
{0x62, 0x0B}, // GPIO1_SEL: GPIO1 as FAULTZ
{0x60, 0x02}, // GPIO_CTRL: GPIO1 is output
{0x77, 0x07}, // CBC_CONTROL: enable cycle-by-cycle current limit for warnings and faults
};
static const LogString *model_name() { return LOG_STR("TAS5825M"); }
static const LogString *fault_output_name(uint8_t pin) {
if (pin == 0)
return LOG_STR("GPIO0");
if (pin == 1)
return LOG_STR("GPIO1");
return LOG_STR("GPIO2");
}
// An if chain rather than a switch: a switch table would land in rodata, which is RAM on ESP8266.
static const LogString *fault_name(uint8_t index) {
if (index == 0)
@@ -113,6 +118,8 @@ const ModelInfo TAS5825M_MODEL = {
// 25, // FAULT_SENSOR_OVER_TEMP_122C_WARNING // not currently included
// 24, // FAULT_SENSOR_OVER_TEMP_112C_WARNING // not currently included
},
.fault_output_pins = 3, // GPIO0, GPIO1, GPIO2 (ADR is an analog input only on TAS5825M)
.fault_output_name = fault_output_name,
};
} // namespace esphome::tas58xx
+35 -6
View File
@@ -20,6 +20,7 @@ static constexpr uint8_t TAS58XX_RESET_CTRL = 0x01;
static constexpr uint8_t TAS58XX_RESET_CTRL_ALL = 0x11; // Reset DSP/control port and registers
static constexpr uint8_t TAS58XX_DEVICE_CTRL_1 = 0x02;
static constexpr uint8_t TAS58XX_DEVICE_CTRL_1_PBTL = (1 << 2);
static constexpr uint8_t TAS58XX_DEVICE_CTRL_1_1SPW = 0x01; // DAMP_MOD = 01
static constexpr uint8_t TAS58XX_DEVICE_CTRL_2 = 0x03;
static constexpr uint8_t TAS58XX_DEVICE_CTRL_2_MUTE = (1 << 3);
static constexpr uint8_t TAS58XX_CTRL_STATE_MASK = 0x03; // DEVICE_CTRL_2 and POWER_STATE
@@ -33,6 +34,9 @@ static constexpr uint8_t TAS58XX_DIG_VOL_0DB = 0x30;
static constexpr uint8_t TAS58XX_DIG_VOL_MINUS_103DB = 0xFE;
static constexpr uint8_t TAS58XX_AGAIN = 0x54; // 0x00 = 0 dB to 0x1F = -15.5 dB in 0.5 dB steps
static constexpr uint8_t TAS58XX_AGAIN_MINUS_15_5DB = 0x1F;
static constexpr uint8_t TAS58XX_GPIO_CTRL = 0x60; // ADR_PIN_CTRL on TAS5805M: output enable, bit n for pin n
static constexpr uint8_t TAS58XX_GPIO0_SEL = 0x61; // ADR_PIN_CONFIG on TAS5805M: pin n is selected at 0x61 + n
static constexpr uint8_t TAS58XX_GPIO_SEL_FAULTZ = 0x0B; // pin function: FAULTZ output
static constexpr uint8_t TAS58XX_POWER_STATE = 0x68;
static constexpr uint8_t TAS58XX_CHAN_FAULT = 0x70; // GLOBAL_FAULT1, GLOBAL_FAULT2 and OT_WARNING follow
static constexpr uint8_t TAS58XX_FAULT_REGISTER_COUNT = 4;
@@ -98,12 +102,26 @@ bool TAS58xx::init_() {
return false;
}
// Setters are public, so keep the value inside the register range
// The I2C address is latched at power up, so ADR can now be an output. The pin function is written before the
// output enable, so the pin never drives another function.
if (this->fault_output_pin_ != FAULT_OUTPUT_NONE) {
// The setter is public, so only write this model's pins; YAML only allows the pins of the model
if (this->fault_output_pin_ >= this->model_->fault_output_pins) {
ESP_LOGW(TAG, "Invalid fault output index %u", this->fault_output_pin_);
} else if (!this->write_byte(TAS58XX_GPIO0_SEL + this->fault_output_pin_, TAS58XX_GPIO_SEL_FAULTZ) ||
!this->write_byte(TAS58XX_GPIO_CTRL, static_cast<uint8_t>(1 << this->fault_output_pin_))) {
ESP_LOGE(TAG, "I2C write failed during init");
return false;
}
}
// Setters are public, so keep the values inside the register ranges
const uint8_t again =
static_cast<uint8_t>(lroundf(clamp(-this->analog_gain_db_ * 2.0f, 0.0f, float{TAS58XX_AGAIN_MINUS_15_5DB})));
if (!this->write_byte(TAS58XX_DEVICE_CTRL_1, this->dac_mode_ == DAC_MODE_PBTL ? TAS58XX_DEVICE_CTRL_1_PBTL : 0) ||
!this->write_byte(TAS58XX_AGAIN, again) || !this->write_volume_() ||
!this->write_ctrl_state_(TAS58XX_CTRL_STATE_PLAY, this->is_muted_) ||
const uint8_t device_ctrl_1 = (this->dac_mode_ == DAC_MODE_PBTL ? TAS58XX_DEVICE_CTRL_1_PBTL : 0) |
(this->modulation_ == MODULATION_1SPW_MODE ? TAS58XX_DEVICE_CTRL_1_1SPW : 0);
if (!this->write_byte(TAS58XX_DEVICE_CTRL_1, device_ctrl_1) || !this->write_byte(TAS58XX_AGAIN, again) ||
!this->write_volume_() || !this->write_ctrl_state_(TAS58XX_CTRL_STATE_PLAY, this->is_muted_) ||
!this->write_byte(TAS58XX_FAULT_CLEAR, TAS58XX_FAULT_CLEAR_ANALOG)) {
ESP_LOGE(TAG, "I2C write failed during init");
return false;
@@ -263,6 +281,14 @@ void TAS58xx::dump_config() {
} else {
mixer_mode = LOG_STR("Stereo");
}
const LogString *fault_output;
if (this->fault_output_pin_ == FAULT_OUTPUT_NONE) {
fault_output = LOG_STR("None");
} else if (this->fault_output_pin_ >= this->model_->fault_output_pins) {
fault_output = LOG_STR("Invalid");
} else {
fault_output = this->model_->fault_output_name(this->fault_output_pin_);
}
ESP_LOGCONFIG(TAG,
"Audio Amplifier:\n"
" Model: %s",
@@ -273,11 +299,14 @@ void TAS58xx::dump_config() {
ESP_LOGCONFIG(TAG,
" Analog Gain: %.1f dB\n"
" DAC Mode: %s\n"
" Modulation: %s\n"
" Mixer Mode: %s\n"
" Volume Range: %.1f dB to %.1f dB",
" Volume Range: %.1f dB to %.1f dB\n"
" Fault Output: %s",
this->analog_gain_db_,
this->dac_mode_ == DAC_MODE_PBTL ? LOG_STR_LITERAL("PBTL") : LOG_STR_LITERAL("BTL"),
LOG_STR_ARG(mixer_mode), this->volume_min_db_, this->volume_max_db_);
this->modulation_ == MODULATION_1SPW_MODE ? LOG_STR_LITERAL("1SPW") : LOG_STR_LITERAL("BD"),
LOG_STR_ARG(mixer_mode), this->volume_min_db_, this->volume_max_db_, LOG_STR_ARG(fault_output));
#ifdef USE_BINARY_SENSOR
LOG_BINARY_SENSOR(" ", "Any Fault", this->have_fault_binary_sensor_);
for (auto *sensor : this->fault_binary_sensors_) {
+17
View File
@@ -22,6 +22,12 @@ enum DacMode : uint8_t {
DAC_MODE_PBTL = 1, // Parallel bridge tied load, one speaker
};
/// DEVICE_CTRL_1 DAMP_MOD values, the same on all models
enum Modulation : uint8_t {
MODULATION_BD_MODE = 0,
MODULATION_1SPW_MODE = 1,
};
enum MixerMode : uint8_t {
MIXER_MODE_STEREO = 0,
MIXER_MODE_STEREO_INVERSE,
@@ -64,6 +70,8 @@ static_assert(
/// ModelInfo::fault_sensor_bits value for a FaultSensor that the model does not have.
static constexpr uint8_t NO_BIT = 0xFF;
/// fault_output_pin_ value for no fault output
static constexpr uint8_t FAULT_OUTPUT_NONE = 0xFF;
/// Everything that differs between models of the family. One constant instance exists per model, see
/// model_*.cpp, and each TAS58xx instance points to the one for its model.
@@ -91,6 +99,10 @@ struct ModelInfo {
const LogString *(*fault_name)(uint8_t index);
/// Bit index for each FaultSensor, NO_BIT used where the model does not have that fault bit
uint8_t fault_sensor_bits[FAULT_SENSOR_COUNT];
/// Number of pins that can report FAULTZ: pin n is selected at 0x61 + n and enabled by bit n of 0x60
uint8_t fault_output_pins;
/// Name of FAULTZ pin n, for dump_config
const LogString *(*fault_output_name)(uint8_t pin);
};
#ifdef USE_TAS58XX_TAS5805M
@@ -127,7 +139,10 @@ class TAS58xx : public audio_dac::AudioDac, public PollingComponent, public i2c:
void set_enable_pin(GPIOPin *enable_pin) { this->enable_pin_ = enable_pin; }
void set_analog_gain(float analog_gain_db) { this->analog_gain_db_ = analog_gain_db; }
void set_dac_mode(DacMode dac_mode) { this->dac_mode_ = dac_mode; }
/// pin: 0 to 2 for GPIOn on TAS5825M, 0 for ADR on TAS5805M; leave unset for no fault output
void set_fault_output(uint8_t pin) { this->fault_output_pin_ = pin; }
void set_mixer_mode(MixerMode mixer_mode) { this->mixer_mode_ = mixer_mode; }
void set_modulation(Modulation modulation) { this->modulation_ = modulation; }
void set_volume_min_db(float volume_min_db) { this->volume_min_db_ = volume_min_db; }
void set_volume_max_db(float volume_max_db) { this->volume_max_db_ = volume_max_db; }
@@ -171,7 +186,9 @@ class TAS58xx : public audio_dac::AudioDac, public PollingComponent, public i2c:
float volume_max_db_{24.0f};
uint32_t logged_faults_{0}; // Fault bits as last logged, packed like the fault masks
DacMode dac_mode_{DAC_MODE_BTL};
uint8_t fault_output_pin_{FAULT_OUTPUT_NONE}; // No fault output: the pin keeps its power on state
MixerMode mixer_mode_{MIXER_MODE_STEREO};
Modulation modulation_{MODULATION_BD_MODE};
uint8_t ctrl_state_{0};
uint8_t power_state_{0xFF}; // Last POWER_STATE seen by update(), 0xFF until the first read
bool mixer_written_{false}; // Mixer written since the last reset
@@ -54,10 +54,7 @@ void TT21100Touchscreen::setup() {
}
// Perform reset if necessary
if (this->reset_pin_ != nullptr) {
this->reset_pin_->setup();
this->reset_();
}
pulse_reset(this->reset_pin_, {.low_ms = 10, .high_after_ms = 10});
// Update display dimensions if they were updated during display setup
if (this->display_ != nullptr) {
@@ -122,15 +119,6 @@ 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);
@@ -28,8 +28,6 @@ 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_;
+18 -2
View File
@@ -245,6 +245,22 @@ 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)
@@ -322,7 +338,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(cv.string_strict),
cv.Optional(CONF_PRIVATE_KEY): cv.sensitive(_validate_dpp_private_key),
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)
@@ -904,7 +920,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(key))
cg.add(var.set_dpp_key(f"{DPP_KEY_DER_PREFIX}{key}{DPP_KEY_DER_SUFFIX}"))
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])))
+11 -8
View File
@@ -5,7 +5,7 @@ from importlib import import_module
from importlib.util import find_spec
import logging
import re
from typing import Any, Literal
from typing import Any, Literal, get_args
from esphome.automation import apply_static_function, parent_ref
import esphome.codegen as cg
@@ -503,7 +503,8 @@ async def _setup_entity_impl(var: MockObj, config: ConfigType, platform: str) ->
config[_KEY_ICON_IDX] = icon_idx
ApplyPublish = Literal["before", "after", "momentary"] | None
ApplyPublishMode = Literal["before", "after", "momentary", "on_success"]
ApplyPublish = ApplyPublishMode | None
@dataclass(frozen=True, slots=True)
@@ -577,7 +578,8 @@ class SubEntities:
With ``apply`` the entity is the domain's ``Apply*`` class and its control runs
``parent->apply``, or ``apply`` as a namespaced free statement when no parent is bound, with
``publish`` ("before"/"after") placing the domain's publish. ``publish="momentary"`` instead
``publish`` ("before"/"after") placing the domain's publish; ``"on_success"`` publishes only when
``apply`` returns true. ``publish="momentary"`` instead
acknowledges first (a switch publishes off) and runs ``apply`` only when turned on. It is only for
legacy action switches; a new action entity is a button.
"""
@@ -600,10 +602,8 @@ class SubEntities:
)
if publish is not None and (apply is None or self.apply_entity.publish is None):
raise ValueError(f"{entity_id}: publish= needs apply= on a stateful domain")
if publish not in (None, "before", "after", "momentary"):
raise ValueError(
f'{entity_id}: publish= must be "before", "after" or "momentary"'
)
if publish is not None and publish not in (modes := get_args(ApplyPublishMode)):
raise ValueError(f"{entity_id}: publish= must be one of {', '.join(modes)}")
if publish == "momentary" and self.apply_entity.momentary is None:
raise ValueError(f'{entity_id}: publish="momentary" is not supported here')
@@ -626,11 +626,14 @@ class SubEntities:
if isinstance(parent, ID):
parent = await get_variable(parent)
apply_entity = self.apply_entity
call = f"{apply};" if parent is None else f"{parent_ref(parent)}->{apply};"
expr = apply if parent is None else f"{parent_ref(parent)}->{apply}"
call = f"{expr};"
if publish == "before":
body = [apply_entity.publish, call]
elif publish == "after":
body = [call, apply_entity.publish]
elif publish == "on_success":
body = [f"if ({expr}) {{ {apply_entity.publish} }}"]
elif publish == "momentary":
acknowledge, condition = apply_entity.momentary
body = [acknowledge, f"if ({condition}) {{ {call} }}"]
+15
View File
@@ -1,4 +1,5 @@
#include "esphome/core/gpio.h"
#include "esphome/core/hal.h"
#include "esphome/core/log.h"
namespace esphome {
@@ -18,4 +19,18 @@ 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
+20
View File
@@ -144,4 +144,24 @@ 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
+3 -3
View File
@@ -718,9 +718,9 @@ def _clone_or_update_locked(
# compute the same cache key as this one.
original_url = url
if username is not None and password is not None:
url = url.replace(
"://", f"://{urllib.parse.quote(username)}:{urllib.parse.quote(password)}@"
)
user = urllib.parse.quote(username, safe="")
token = urllib.parse.quote(password, safe="")
url = url.replace("://", f"://{user}:{token}@")
hash_dir_name = _compute_destination_path(key, domain).name
repo_dir = _repo_entry_dir(key, domain, subpath)
+9 -4
View File
@@ -26,17 +26,20 @@ 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; 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"})
# 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"})
# 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"
@@ -71,6 +74,8 @@ 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}),
}
+218 -17
View File
@@ -9,7 +9,8 @@ from pathlib import Path
import re
import stat
from esphome.core import EsphomeError
from esphome import compiled_config, yaml_util
from esphome.core import CORE, EsphomeError
from esphome.helpers import write_file
from esphome.types import ConfigType
@@ -20,6 +21,10 @@ PLAIN_SCALAR = r"[^\s#\"'>|&*!%@`\[\]{},]+"
TRAILER = r"(?P<quote>[\"']?){value}(?P=quote)(?P<trail>(?:\s+#.*)?\s*)$"
class RestoreError(EsphomeError):
"""A rollback left at least one file as it was written."""
def read_text(path: Path) -> str:
"""The file as written, line endings included; read_file would fold them."""
try:
@@ -30,12 +35,25 @@ def read_text(path: Path) -> str:
@dataclass
class LineEdit:
"""One line to rewrite; ``old_line`` is what it held when located."""
"""One line to rewrite, or with ``insert_after`` a line to add right
after it; ``old_line`` is what the line held when it was located."""
path: Path
line: int
old_line: str
new_line: str
insert_after: bool = False
@dataclass
class Snapshot:
"""A rewritten file's text before and after, and its mode before; the
restore only puts ``original`` back over ``written``, never over the
user's own edit."""
original: str
written: str
mode: int | None = None
def field_line_re(
@@ -63,17 +81,66 @@ def line_at(doc: Path, line_no: int) -> str:
return lines[line_no]
_DOMAIN = "yaml_edit"
def own_documents() -> set[Path]:
"""Every file the loader read for this configuration, found once per run.
A file behind an include path with a substitution that gave nothing but
substitutions is the one kind this misses."""
if (documents := CORE.data.get(_DOMAIN)) is None:
documents = CORE.data[_DOMAIN] = _own_documents()
return documents
def _own_documents() -> set[Path]:
documents = {str(CORE.config_path)}
def walk(node: object) -> None:
if (rng := getattr(node, "esp_range", None)) is not None:
documents.add(rng.start_mark.document)
if isinstance(node, dict):
for key, value in node.items():
walk(key)
walk(value)
elif isinstance(node, list):
for value in node:
walk(value)
walk(CORE.raw_config or {})
# A file that only gave substitutions leaves no node behind; the tracker
# also lists a secrets file the loader only tried, hence the is_file check
found = yaml_util.discover_user_yaml_files(CORE.config_path, clear_secrets=False)
if found.load_errors: # an incomplete set would misplace a key
raise EsphomeError(
"Could not read every file of the configuration: "
+ "; ".join(found.load_errors)
)
documents.update(map(str, found.files))
return {p.resolve() for d in documents if (p := Path(d)).is_file()}
def source_of(mapping: ConfigType, name: str) -> tuple[Path, int] | None:
"""The file and line ``name:`` was read from, None when validation added
it or a merge key brought it in from an anchor elsewhere. Mapping keys
keep their range through validation, values may not."""
"""The file and line ``name:`` was read from; None when validation added
it or a merge key brought it in from an anchor elsewhere."""
rng = getattr(next((k for k in mapping if k == name), None), "esp_range", None)
if rng is None:
return None
# An included mapping carries the `!include` line of its parent, so only
# a key in the same document can be placed against the mapping
if (own := getattr(mapping, "esp_range", None)) is None:
return None # a mapping built in code, its keys are not on its lines
# merge_config keeps the first mapping's keys with the last one's
# values, so only the value's range names the line that won, and
# only when the key is on it (a block value starts on a line of its own)
rng = getattr(mapping.get(name), "esp_range", None)
if rng is None:
return None
doc, line = Path(rng.start_mark.document), rng.start_mark.line
if doc.resolve() not in own_documents() or not field_line_re(name).match(
line_at(doc, line)
):
return None
return doc, line
# An included mapping carries its parent's `!include` line, so only a
# key in the same document can be checked against the mapping's lines
if (
rng.start_mark.document == own.start_mark.document
and not own.start_mark.line <= rng.start_mark.line <= own.end_mark.line
@@ -82,13 +149,46 @@ def source_of(mapping: ConfigType, name: str) -> tuple[Path, int] | None:
return Path(rng.start_mark.document), rng.start_mark.line
def write_keeping_mode(path: Path, text: str, like: Path | None = None) -> None:
"""Write with the mode of ``like`` (default: the file itself) rather
than write_file's 0644; a 0600 secrets file stays 0600."""
def editable_file(path: Path) -> Path:
"""``path`` resolved, so a symlink's target is edited; refuses a file
outside the config dir or under the build data."""
resolved = path.resolve()
if (
not path.is_file()
or not resolved.is_relative_to(CORE.config_dir.resolve())
or resolved.is_relative_to(CORE.data_dir.resolve())
):
raise EsphomeError(
f"{path} is not an editable file in the configuration directory"
)
return resolved
def source_line(mapping: ConfigType, name: str) -> tuple[Path, int, str]:
"""The file, line number and text ``name:`` was read from."""
if (source := source_of(mapping, name)) is None:
raise EsphomeError(f"'{name}' was not read from a file")
doc, line_no = source
# Returned as the loader saw it: a `!secret` in a symlinked include
# resolves beside the link
editable_file(doc)
return doc, line_no, line_at(doc, line_no)
def mode_of(path: Path) -> int:
try:
mode = stat.S_IMODE((like or path).stat().st_mode)
return stat.S_IMODE(path.stat().st_mode)
except OSError as err:
raise EsphomeError(f"Could not read the mode of {like or path}: {err}") from err
raise EsphomeError(f"Could not read the mode of {path}: {err}") from err
def write_keeping_mode(
path: Path, text: str, like: Path | None = None, mode: int | None = None
) -> None:
"""Write with ``mode``, else the mode of ``like`` (default: the file
itself), rather than write_file's 0644; a 0600 secrets file stays 0600."""
if mode is None:
mode = mode_of(like or path)
try:
write_file(path, text, private=True)
except EsphomeError as err:
@@ -103,12 +203,113 @@ def write_keeping_mode(path: Path, text: str, like: Path | None = None) -> None:
def rewritten_text(original: str, edits: list[LineEdit]) -> str:
"""``original`` with the edits applied; every edit must still find the
line it was located on."""
"""``original`` with the edits applied; a line that changed is refused."""
lines = original.splitlines(keepends=True)
for edit in edits:
file_newline = "\r\n" if "\r\n" in original else "\n"
# Bottom up, so an insertion never shifts a later edit; insertions under
# one line keep their order
order = sorted(
enumerate(edits), key=lambda ie: (ie[1].line, ie[1].insert_after, ie[0])
)
for _, edit in reversed(order):
text = lines[edit.line].rstrip("\r\n") if edit.line < len(lines) else None
if text != edit.old_line:
raise EsphomeError(f"{edit.path}:{edit.line + 1} changed since it was read")
lines[edit.line] = edit.new_line + lines[edit.line][len(text) :]
ending = lines[edit.line][len(text) :]
if edit.insert_after:
# A last line without a newline gets the file's own kind
lines[edit.line] = text + (ending or file_newline)
lines.insert(edit.line + 1, edit.new_line + ending)
else:
lines[edit.line] = edit.new_line + ending
return "".join(lines)
def apply_line_edits(edits: list[LineEdit]) -> dict[Path, Snapshot]:
"""Rewrite the located lines in place; returns each file's text before
and after for restore_files. Nothing is written while any line is stale,
and a file that no longer loads is put back here."""
by_path: dict[Path, list[LineEdit]] = {}
seen_as: dict[Path, Path] = {} # the path the loader read a file by
for edit in edits:
resolved = editable_file(edit.path)
by_path.setdefault(resolved, []).append(edit)
seen_as.setdefault(resolved, edit.path)
originals = {}
for path, own in by_path.items():
original = read_text(path)
originals[path] = Snapshot(
original, rewritten_text(original, own), mode_of(path)
)
try:
for path, snapshot in originals.items():
write_keeping_mode(path, snapshot.written)
# Reloaded as the loader saw them: a relative !include inside a
# symlinked file resolves beside the link
reloads = [seen_as[path] for path in originals]
if CORE.config_path.resolve() not in originals:
# Its !secret tags register a value written elsewhere
reloads.append(CORE.config_path)
for path in reloads:
try:
# The secret registry stays: later output still redacts
yaml_util.load_yaml(
path, clear_secrets=False, track_document_range=False
)
except EsphomeError as err:
raise EsphomeError(f"{path} no longer loads: {err}") from err
compiled_config.invalidate_compiled_config()
except BaseException as err:
try:
restore_files(originals)
except RestoreError as restore_err:
if not isinstance(err, Exception):
err.add_note(str(restore_err)) # an interrupt stays one
raise err from None
raise RestoreError(f"{err}; {restore_err}") from err
raise
return originals
def restore_files(originals: dict[Path, Snapshot]) -> None:
"""Put back what apply_line_edits wrote, mode included; a file that
failed or changed meanwhile is left alone and reported. Every file is
tried whatever went wrong before; an interrupt or exit is raised again
at the end with the report as a note."""
failed: list[str] = []
stop: BaseException | None = None
def report(what: str, err: BaseException) -> None:
nonlocal stop
if isinstance(err, Exception):
why = str(err) or type(err).__name__
else: # an interrupt or exit says what it is, not its code
why = (
"interrupted"
if isinstance(err, KeyboardInterrupt)
else type(err).__name__
)
stop = err
failed.append(f"{what}: {why}")
for path, snapshot in originals.items():
try:
current = read_text(path)
if current == snapshot.original:
continue # never written, or already put back
if current != snapshot.written:
raise EsphomeError("changed since it was written, left as is")
write_keeping_mode(path, snapshot.original, mode=snapshot.mode)
except BaseException as err: # noqa: BLE001 # pylint: disable=broad-except
report(str(path), err)
try:
compiled_config.invalidate_compiled_config()
except BaseException as err: # noqa: BLE001 # pylint: disable=broad-except
report("the validated caches", err)
if not failed:
return
message = "Could not restore " + "; ".join(failed)
if stop is not None:
stop.add_note(message)
raise stop
raise RestoreError(message)
+6 -2
View File
@@ -656,9 +656,13 @@ def named_path(path: Path) -> Path:
return path.parent.resolve() / path.name
def discover_user_yaml_files(config_path: Path) -> DiscoveredYamlFiles:
def discover_user_yaml_files(
config_path: Path, clear_secrets: bool = True
) -> DiscoveredYamlFiles:
"""Fresh-re-parse ``config_path`` and report every file the YAML loader
pulled in, plus which of them came in under a secrets filename.
``clear_secrets`` is passed to the load; False keeps the redaction
registry of a configuration loaded earlier in the run.
Does NOT run schema validation, substitutions, or package resolution — so
component-internal YAML loaded by validators (LVGL helpers, dashboard
@@ -702,7 +706,7 @@ def discover_user_yaml_files(config_path: Path) -> DiscoveredYamlFiles:
load_listeners.append(_on_load)
try:
try:
data = load_yaml(config_path)
data = load_yaml(config_path, clear_secrets=clear_secrets)
except EsphomeError as err:
load_errors = [f"{config_path}: {err}"]
else:
@@ -138,3 +138,26 @@ 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
@@ -0,0 +1,11 @@
esphome:
name: test
esp32:
board: esp32dev
toolchain: esp-idf
framework:
type: esp-idf
version: 6.1.0
logger:
@@ -0,0 +1,12 @@
esphome:
name: test
esp32:
board: esp32dev
toolchain: platformio
framework:
type: esp-idf
version: 6.1.0
platform_version: 61.4.0-RC1
logger:
@@ -0,0 +1,31 @@
"""Tests that the TAS58xx fault_output and modulation defaults keep the existing behaviour."""
from __future__ import annotations
from collections.abc import Callable
from pathlib import Path
def test_tas58xx_defaults_keep_the_fault_pin_and_bd_modulation(
generate_main: Callable[[str | Path], str],
) -> None:
"""Without fault_output or modulation each model keeps its usual fault pin and BD modulation."""
main_cpp = generate_main(
"tests/component_tests/tas58xx/test_tas58xx_fault_output_modulation.yaml"
)
assert "amp_5805_default->set_fault_output(0);" in main_cpp
assert "amp_5825_default->set_fault_output(1);" in main_cpp
for amp in ("amp_5805_default", "amp_5825_default", "amp_5805_none"):
assert f"{amp}->set_modulation(tas58xx::MODULATION_BD_MODE);" in main_cpp
def test_tas58xx_fault_output_none_leaves_the_pin_alone(
generate_main: Callable[[str | Path], str],
) -> None:
"""fault_output: none does not configure a fault pin."""
main_cpp = generate_main(
"tests/component_tests/tas58xx/test_tas58xx_fault_output_modulation.yaml"
)
assert "amp_5805_none->set_fault_output(" not in main_cpp
@@ -0,0 +1,26 @@
esphome:
name: test
esp32:
board: esp32dev
i2c:
id: i2c_bus
sda: 21
scl: 22
audio_dac:
- platform: tas58xx
model: tas5805m
id: amp_5805_default
ignore_enable_pin_warning: true
- platform: tas58xx
model: tas5825m
id: amp_5825_default
ignore_enable_pin_warning: true
- platform: tas58xx
model: tas5805m
id: amp_5805_none
address: 0x2C
ignore_enable_pin_warning: true
fault_output: none
@@ -0,0 +1,12 @@
---
esphome:
name: test
esp32:
board: esp32dev
framework:
type: esp-idf
wifi:
dpp:
private_key: 00112233445566778899AABBCCDDEEFF00112233445566778899aabbccddeeff
@@ -0,0 +1,42 @@
"""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
)
+14
View File
@@ -0,0 +1,14 @@
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 compiles the PG/GPO switch path.
manifest.multi_conf = False
manifest.dependencies = manifest.dependencies + ["i2c", "switch"]
async def to_code(config: ConfigType) -> None:
cg.add_define("USE_BQ25186_PG_GPO")
manifest.to_code = to_code
@@ -0,0 +1,143 @@
#include <gtest/gtest.h>
#include <array>
#include "esphome/components/bq25186/bq25186.h"
#include "esphome/components/switch/switch.h"
namespace esphome::bq25186::testing {
static constexpr uint8_t ADDRESS = 0x6A;
static constexpr uint8_t SYS_REG_OTHER_BITS = 0x0F;
// The chip's registers behind an address pointer that moves on after every byte; writes can be made to fail.
class RegisterBus final : public i2c::I2CBus {
public:
i2c::ErrorCode write_readv(uint8_t address, const uint8_t *data, size_t size, uint8_t *out,
size_t out_size) override {
EXPECT_EQ(address, ADDRESS);
if (size > 1 && this->fail_writes)
return i2c::ERROR_NOT_ACKNOWLEDGED;
if (size != 0) {
this->pointer_ = data[0];
for (size_t i = 1; i < size; i++)
this->registers[this->next_()] = data[i];
}
for (size_t i = 0; i < out_size; i++)
out[i] = this->registers[this->next_()];
return i2c::ERROR_OK;
}
std::array<uint8_t, BQ25186_STATUS_REGISTER_COUNT> registers{};
bool fail_writes{false};
protected:
size_t next_() { return this->pointer_++ % this->registers.size(); }
uint8_t pointer_{0};
};
// The body code generation emits for apply= with publish="on_success"
static BQ25186Component *hub_under_test = nullptr; // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
static void apply_pg_gpo(switch_::Switch *entity, bool state) {
if (hub_under_test->write_pg_gpo_level(state))
entity->publish_state(state);
}
class BQ25186PgGpoTest : public ::testing::Test {
protected:
void SetUp() override {
this->bus_.registers[BQ25186_REG_MASK_ID] = 0x01; // the BQ25186 device id
this->bus_.registers[BQ25186_REG_SYS_REG] = SYS_REG_OTHER_BITS;
this->hub_.set_i2c_bus(&this->bus_);
this->hub_.set_i2c_address(ADDRESS);
hub_under_test = &this->hub_;
this->switch_.set_restore_mode(switch_::SWITCH_RESTORE_DISABLED);
this->hub_.set_pg_gpo_switch(&this->switch_);
}
bool level_() const { return (this->bus_.registers[BQ25186_REG_SYS_REG] & BQ25186_SYS_REG_PG_GPO_LEVEL) != 0; }
uint8_t other_bits_() const { return this->bus_.registers[BQ25186_REG_SYS_REG] & ~BQ25186_SYS_REG_PG_GPO_LEVEL; }
RegisterBus bus_;
BQ25186Component hub_;
switch_::ApplySwitch<apply_pg_gpo> switch_;
};
TEST_F(BQ25186PgGpoTest, TurnOnSetsTheLevelAndPublishes) {
this->switch_.turn_on();
EXPECT_TRUE(this->level_());
EXPECT_EQ(this->other_bits_(), SYS_REG_OTHER_BITS);
EXPECT_TRUE(this->switch_.has_state());
EXPECT_TRUE(this->switch_.state);
}
TEST_F(BQ25186PgGpoTest, TurnOffClearsTheLevelAndPublishes) {
this->bus_.registers[BQ25186_REG_SYS_REG] |= BQ25186_SYS_REG_PG_GPO_LEVEL;
this->switch_.publish_state(true);
this->switch_.turn_off();
EXPECT_FALSE(this->level_());
EXPECT_EQ(this->other_bits_(), SYS_REG_OTHER_BITS);
EXPECT_FALSE(this->switch_.state);
}
TEST_F(BQ25186PgGpoTest, FailedWriteLeavesTheStateUnpublished) {
this->bus_.fail_writes = true;
this->switch_.turn_on();
EXPECT_FALSE(this->level_());
EXPECT_FALSE(this->switch_.has_state());
}
TEST_F(BQ25186PgGpoTest, FailedWriteKeepsThePreviousState) {
// Start from on, so a reset to the default off state would show
this->bus_.registers[BQ25186_REG_SYS_REG] |= BQ25186_SYS_REG_PG_GPO_LEVEL;
this->switch_.publish_state(true);
this->bus_.fail_writes = true;
this->switch_.turn_off();
EXPECT_TRUE(this->level_());
EXPECT_TRUE(this->switch_.state);
}
TEST_F(BQ25186PgGpoTest, InvertedSwitchDrivesTheOppositeLevel) {
this->switch_.set_inverted(true);
this->switch_.turn_on();
EXPECT_FALSE(this->level_());
EXPECT_TRUE(this->switch_.state);
this->switch_.turn_off();
EXPECT_TRUE(this->level_());
EXPECT_FALSE(this->switch_.state);
}
TEST_F(BQ25186PgGpoTest, SetupWritesTheRestoredLevel) {
this->switch_.set_restore_mode(switch_::SWITCH_ALWAYS_ON);
this->hub_.setup();
EXPECT_TRUE(this->level_());
EXPECT_NE(this->bus_.registers[BQ25186_REG_VBAT_CTRL] & BQ25186_VBAT_CTRL_PG_MODE, 0);
EXPECT_TRUE(this->switch_.state);
}
TEST_F(BQ25186PgGpoTest, SetupWritesTheRestoredLevelInverted) {
this->bus_.registers[BQ25186_REG_SYS_REG] |= BQ25186_SYS_REG_PG_GPO_LEVEL;
this->switch_.set_restore_mode(switch_::SWITCH_ALWAYS_ON);
this->switch_.set_inverted(true);
this->hub_.setup();
EXPECT_FALSE(this->level_());
EXPECT_TRUE(this->switch_.state);
}
TEST_F(BQ25186PgGpoTest, SetupWithoutRestoreTakesTheChipLevel) {
this->bus_.registers[BQ25186_REG_SYS_REG] |= BQ25186_SYS_REG_PG_GPO_LEVEL;
this->hub_.setup();
EXPECT_TRUE(this->level_());
EXPECT_TRUE(this->switch_.state);
}
TEST_F(BQ25186PgGpoTest, SetupWithoutRestoreTakesTheChipLevelInverted) {
this->bus_.registers[BQ25186_REG_SYS_REG] |= BQ25186_SYS_REG_PG_GPO_LEVEL;
this->switch_.set_inverted(true);
this->hub_.setup();
EXPECT_TRUE(this->level_());
EXPECT_FALSE(this->switch_.state);
}
} // namespace esphome::bq25186::testing
@@ -0,0 +1,51 @@
#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
@@ -0,0 +1,14 @@
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
+2
View File
@@ -7,7 +7,9 @@ audio_dac:
enable_pin: GPIO12
analog_gain: -15.5dB
dac_mode: btl
fault_output: adr
mixer_mode: stereo
modulation: 1SPW_MODE
volume_min_db: -60dB
volume_max_db: 0dB
update_interval: 1s
@@ -9,3 +9,4 @@ audio_dac:
dac_mode: PBTL
mixer_mode: mono
analog_gain: -3.5dB
fault_output: none
@@ -2,6 +2,7 @@ wifi:
dpp:
channels: [1, 6, 11]
device_info: ESPHome
private_key: 00112233445566778899aabbccddeeff00112233445566778899aabbccddeeff
timeout: 30s
on_dpp_uri:
- logger.log:
@@ -0,0 +1,70 @@
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
@@ -0,0 +1,93 @@
"""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")
@@ -0,0 +1,125 @@
"""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
@@ -1711,6 +1711,14 @@ _SELECT_APPLY = entity_helpers.ApplyEntity(
"after",
"::hub->set_mode(index, options_);\nentity->publish_state(index);",
),
(
"set_mode(option)",
"on_success",
(
"const char *option = entity->option_at(index);\n"
"if (::hub->set_mode(option)) { entity->publish_state(index); }"
),
),
],
)
async def test_sub_entities_apply_locals(
@@ -1883,6 +1891,35 @@ async def test_sub_entities_apply_momentary(
assert apply_functions[0][2] == body
@pytest.mark.asyncio
@pytest.mark.parametrize(
("parent", "apply", "body"),
[
(
_HUB,
"write_level(state)",
"if (::hub->write_level(state)) { entity->publish_state(state); }",
),
(
None,
"chip::write_level(state)",
"if (chip::write_level(state)) { entity->publish_state(state); }",
),
],
)
async def test_sub_entities_apply_on_success(
setup_test_environment: list[str],
apply_functions: list[tuple[str, str, str]],
parent: MockObj | None,
apply: str,
body: str,
) -> None:
"""publish="on_success" publishes only when the hub call returns true."""
sub = _apply_sub([], {"k": _apply_conf("k_sw")}, parent=parent)
await sub("k", None, apply=apply, publish="on_success")
assert apply_functions[0][2] == body
@pytest.mark.asyncio
async def test_sub_entities_apply_without_parent_needs_namespaced_call(
setup_test_environment: list[str],
+44
View File
@@ -18,6 +18,7 @@ from esphome.__main__ import run_esphome
from esphome.compiled_config import (
_LAMBDA_KEY,
compiled_config_path,
invalidate_compiled_config,
load_compiled_config,
save_compiled_config,
save_compiled_config_and_sidecar,
@@ -937,3 +938,46 @@ def test_load_compiled_config_rejects_wizard_only_sidecar(
_set_cache_mtime(cache_path, yaml_path, offset=5)
assert load_compiled_config(yaml_path) is None
def test_invalidate_compiled_config_reports_an_unlink_failure(
tmp_path: Path,
) -> None:
"""A cache that cannot be removed would serve the old key later."""
CORE.config_path = tmp_path / "test.yaml"
cache = compiled_config_path("test.yaml")
cache.parent.mkdir(parents=True)
cache.write_text("{}")
with (
patch("pathlib.Path.unlink", side_effect=OSError("busy")),
pytest.raises(EsphomeError, match="validated config cache"),
):
invalidate_compiled_config()
def test_invalidate_compiled_config_drops_every_cache(tmp_path: Path) -> None:
"""Another configuration sharing the edited secrets.yaml or include has
a cache the mtime check would keep serving."""
CORE.config_path = tmp_path / "test.yaml"
storage = compiled_config_path("test.yaml").parent
storage.mkdir(parents=True)
for name in ("test.yaml", "other.yaml"):
(storage / f"{name}.validated.json").write_text("{}")
(storage / "other.json").write_text("{}")
invalidate_compiled_config()
assert sorted(p.name for p in storage.iterdir()) == ["other.json"]
def test_invalidate_compiled_config_reports_a_storage_dir_it_cannot_list(
tmp_path: Path,
) -> None:
CORE.config_path = tmp_path / "test.yaml"
storage = compiled_config_path("test.yaml").parent
storage.mkdir(parents=True)
with (
patch("pathlib.Path.iterdir", side_effect=OSError("denied")),
pytest.raises(EsphomeError, match="Could not list"),
):
invalidate_compiled_config()
storage.rmdir()
invalidate_compiled_config() # no storage dir yet is nothing to drop
+43
View File
@@ -10,6 +10,7 @@ import threading
import time
from typing import Any
from unittest.mock import Mock, patch
import urllib.parse
from filelock import FileLock
import pytest
@@ -2097,6 +2098,48 @@ def test_recovery_reclone_keeps_credentials_and_cache_key(
assert clone_url.count("@") == 1
def test_clone_credentials_with_slash_are_escaped(
tmp_path: Path, mock_run_git_command: Mock
) -> None:
"""A "/" in the username or password must be percent-encoded, or the
userinfo ends at the "/" and git reports the password in its error."""
CORE.config_path = tmp_path / "test.yaml"
url = "https://github.com/test/repo"
domain = "test"
repo_dir = _compute_repo_dir(url, None, domain)
def git_command_side_effect(
cmd: list[str], cwd: str | None = None, **kwargs: Any
) -> str:
if _get_git_command_type(cmd) == "clone":
_simulate_cloned_repo(repo_dir)
return ""
mock_run_git_command.side_effect = git_command_side_effect
git.clone_or_update(
url=url,
ref=None,
refresh=TimePeriodSeconds(days=1),
domain=domain,
username="team/user",
password="ab/cd@ef",
)
clone_cmds = [
c[0][0]
for c in mock_run_git_command.call_args_list
if _get_git_command_type(c[0][0]) == "clone"
]
assert clone_cmds
clone_url = clone_cmds[0][-2]
assert clone_url == "https://team%2Fuser:ab%2Fcd%40ef@github.com/test/repo"
assert urllib.parse.urlsplit(clone_url).hostname == "github.com"
redacted = git._redact_url_credentials(f"fatal: unable to access '{clone_url}/'")
assert redacted == "fatal: unable to access 'https://***@github.com/test/repo/'"
def test_refresh_submodule_failure_recovers_then_raises(
tmp_path: Path, mock_run_git_command: Mock
) -> None:
+4 -25
View File
@@ -222,15 +222,9 @@ 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"),
}
# 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",
"radio_frequency": ("esphome.components.ir_rf_base",),
}
@@ -248,26 +242,11 @@ 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_KNOWN_LEAKS.keys() | _ENTITY_COMPONENT_DEPENDENCIES.keys()
) - found
stale = _ENTITY_COMPONENT_DEPENDENCIES.keys() - found
assert not stale, f"not discovered as entity components: {sorted(stale)}"
@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()
],
)
@pytest.mark.parametrize("component", _entity_components())
def test_entity_component_does_not_import_integrations(component: str) -> None:
"""An entity component must not drag in its optional integrations.
+112
View File
@@ -2135,6 +2135,118 @@ 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="],
+395 -7
View File
@@ -3,21 +3,32 @@
from __future__ import annotations
from pathlib import Path
import stat
import sys
from typing import Any
from unittest.mock import patch
import pytest
from esphome import yaml_util
from esphome.compiled_config import compiled_config_path
from esphome.config import do_substitution_pass
from esphome.const import CONF_ESPHOME, CONF_NAME
from esphome.core import CORE, EsphomeError
from esphome.yaml_edit import (
LineEdit,
RestoreError,
Snapshot,
apply_line_edits,
editable_file,
field_line_re,
line_at,
own_documents,
read_text,
restore_files,
rewrite,
rewritten_text,
source_line,
source_of,
write_keeping_mode,
)
@@ -98,13 +109,46 @@ def test_a_comment_needs_whitespace_and_a_scalar_is_not_empty() -> None:
assert field_line_re("key", "abc#def").match("key: abc#def") is not None
def test_source_of_is_none_for_a_value_validation_added(tmp_path: Path) -> None:
"""Only a key read from a file carries a range, and only a mapping read
from a file can place its keys; a mapping built in code cannot."""
def test_source_of_is_none_for_a_value_validation_added() -> None:
"""Only a key read from a file carries a range."""
assert source_of({"name": "kitchen"}, "name") is None
def test_a_mapping_built_in_code_places_only_this_configurations_keys(
tmp_path: Path,
) -> None:
"""merge_config rebuilds a mapping without a range; a value read for this
configuration is still placed, one read from elsewhere is not."""
_setup(tmp_path, YAML)
loaded_key = next(iter(CORE.raw_config[CONF_ESPHOME]))
assert source_of({loaded_key: "kitchen"}, CONF_NAME) is None
own_key, own_value = next(iter(CORE.raw_config[CONF_ESPHOME].items()))
assert source_of({own_key: own_value}, CONF_NAME) == (CORE.config_path, 1)
# Only the value names the line that won; without a range it is refused
assert source_of({own_key: "x"}, CONF_NAME) is None
(tmp_path / "elsewhere.yaml").write_bytes(YAML.encode())
other = yaml_util.load_yaml(tmp_path / "elsewhere.yaml")
other_key, other_value = next(iter(other[CONF_ESPHOME].items()))
assert source_of({other_key: other_value}, CONF_NAME) is None
# A value on a line of its own does not name the key's line
_setup(tmp_path, "esphome:\n name:\n kitchen\n")
own_key, own_value = next(iter(CORE.raw_config[CONF_ESPHOME].items()))
assert source_of({own_key: own_value}, CONF_NAME) is None
def test_own_documents_include_a_file_that_only_gives_substitutions(
tmp_path: Path,
) -> None:
(tmp_path / "subs.yaml").write_bytes(b"substitutions:\n x: y\n")
(tmp_path / "sensor.yaml").write_bytes(b"platform: template\nname: x\n")
_setup(
tmp_path,
YAML + "packages:\n subs: !include subs.yaml\nsensor:\n"
" - !include sensor.yaml\n",
)
assert (tmp_path / "sensor.yaml").resolve() in own_documents() # via a list
CORE.reset()
CORE.config_path = tmp_path / "test.yaml"
CORE.raw_config = {} # what is left after substitutions are taken out
assert (tmp_path / "subs.yaml").resolve() in own_documents()
def test_source_of_is_none_for_a_merged_key(tmp_path: Path) -> None:
@@ -152,8 +196,6 @@ def test_write_keeps_the_mode_of_the_file_or_another(tmp_path: Path) -> None:
def test_write_failures_say_which_step_and_why(tmp_path: Path) -> None:
"""A missing mode source, a write that fails, and a mode that cannot be
put back after the write are three different situations."""
from unittest.mock import patch
path = _setup(tmp_path, YAML)
with pytest.raises(EsphomeError, match="Could not read the mode of .*gone.yaml"):
write_keeping_mode(path, YAML, like=tmp_path / "gone.yaml")
@@ -180,3 +222,349 @@ def test_read_text_reports_a_file_it_cannot_decode(tmp_path: Path) -> None:
path.write_bytes(b"caf\xe9: 1\n")
with pytest.raises(EsphomeError, match="Error reading file"):
read_text(path)
def test_source_line_refuses_what_is_not_an_editable_file(tmp_path: Path) -> None:
"""A remote package is checked out under the build data and is not the
user's; a file outside the configuration directory is not either."""
(tmp_path / ".esphome").mkdir()
(tmp_path / ".esphome" / "base.yaml").write_bytes(b"name: kitchen\n")
_setup(tmp_path, "esphome: !include .esphome/base.yaml\n")
with pytest.raises(EsphomeError, match="not an editable file"):
source_line(CORE.raw_config[CONF_ESPHOME], CONF_NAME)
with pytest.raises(EsphomeError, match="was not read from a file"):
source_line({"name": "kitchen"}, CONF_NAME)
_setup(tmp_path, YAML)
assert source_line(CORE.raw_config[CONF_ESPHOME], CONF_NAME) == (
CORE.config_path,
1,
" name: kitchen # the device",
)
outside = tmp_path.parent / "outside.yaml"
outside.write_bytes(b"")
with pytest.raises(EsphomeError, match="not an editable file"):
editable_file(outside)
(tmp_path / "link.yaml").symlink_to(CORE.config_path)
assert editable_file(tmp_path / "link.yaml") == CORE.config_path.resolve()
def test_insert_after_goes_before_the_rewrite_of_the_same_line(
tmp_path: Path,
) -> None:
"""Edits apply from the bottom up so an insertion never moves a later
line; the file's own line ending is used, also on a bare last line."""
text = YAML.replace("\n", "\r\n").rstrip("\r\n")
path = _setup(tmp_path, text)
edit = _name_edit("garage")
added = LineEdit(path, 4, " ssid: kitchen", " password: x", insert_after=True)
more = LineEdit(
path, 4, " ssid: kitchen", " fast_connect: true", insert_after=True
)
assert rewritten_text(read_text(path), [added, more, edit]) == (
text.replace("name: kitchen", "name: garage")
+ "\r\n password: x\r\n fast_connect: true"
)
edit.old_line = "changed"
with pytest.raises(EsphomeError, match="changed since it was read"):
rewritten_text(read_text(path), [added, edit])
def test_own_documents_refuses_an_incomplete_set(tmp_path: Path) -> None:
_setup(tmp_path, YAML)
found = yaml_util.DiscoveredYamlFiles(load_errors=["sub.yaml: broken"])
with (
patch("esphome.yaml_util.discover_user_yaml_files", return_value=found),
pytest.raises(EsphomeError, match="Could not read every file .*sub.yaml"),
):
own_documents()
def test_apply_keeps_the_secret_registry(tmp_path: Path) -> None:
"""The reload must not clear what later output redacts as !secret."""
(tmp_path / "secrets.yaml").write_bytes(b"ssid: hidden\n")
path = _setup(tmp_path, YAML.replace("ssid: kitchen", "ssid: !secret ssid"))
assert yaml_util.is_secret("hidden") == "ssid"
apply_line_edits([_name_edit("garage")])
assert yaml_util.is_secret("hidden") == "ssid"
assert "!secret ssid" in path.read_text()
def test_apply_writes_reloads_drops_the_cache_and_restores(tmp_path: Path) -> None:
path = _setup(tmp_path, YAML)
cache = compiled_config_path(CORE.config_filename)
cache.parent.mkdir(parents=True, exist_ok=True)
cache.write_text("{}")
mode = stat.S_IMODE(path.stat().st_mode)
originals = apply_line_edits([_name_edit("garage")])
assert originals == {
path: Snapshot(YAML, YAML.replace("kitchen #", "garage #"), mode)
}
assert path.read_text() == YAML.replace("kitchen #", "garage #")
assert not cache.exists()
cache.write_text("{}")
restore_files(originals)
assert path.read_text() == YAML
assert not cache.exists()
def test_apply_reloads_a_symlinked_file_by_the_link(tmp_path: Path) -> None:
"""A relative !include inside the file resolves beside the link, as the
loader read it, not beside the target."""
shared = tmp_path / "shared"
shared.mkdir()
(shared / "device.yaml").write_bytes(
b"name: kitchen\nfriendly_name: !include fn.yaml\n"
)
(tmp_path / "fn.yaml").write_bytes(b"Kitchen\n") # beside the link only
(tmp_path / "device.yaml").symlink_to(shared / "device.yaml")
_setup(tmp_path, "esphome: !include device.yaml\n")
doc, line_no = source_of(CORE.raw_config[CONF_ESPHOME], CONF_NAME)
assert doc == tmp_path / "device.yaml"
match = field_line_re(CONF_NAME, "kitchen").match(line_at(doc, line_no))
edit = LineEdit(doc, line_no, match.string, rewrite(match, "garage"))
originals = apply_line_edits([edit])
assert list(originals) == [(shared / "device.yaml").resolve()]
assert (shared / "device.yaml").read_text() == (
"name: garage\nfriendly_name: !include fn.yaml\n"
)
def test_apply_rolls_back_a_rewrite_the_loader_refuses(tmp_path: Path) -> None:
path = _setup(tmp_path, YAML)
edit = _name_edit("garage")
edit.new_line = " name: [unterminated"
with pytest.raises(EsphomeError, match="no longer loads"):
apply_line_edits([edit])
assert path.read_text() == YAML
with (
patch("esphome.yaml_util.load_yaml", side_effect=KeyboardInterrupt),
pytest.raises(KeyboardInterrupt),
):
apply_line_edits([_name_edit("garage")])
assert path.read_text() == YAML
def test_restore_reports_every_file_it_could_not_write(tmp_path: Path) -> None:
"""A missing file, a failed write and a cache that cannot be dropped are
all reported; a file the user changed meanwhile is left alone."""
path = _setup(tmp_path, YAML)
good = tmp_path / "a.yaml"
(tmp_path / "b.yaml").write_bytes(b"")
good.write_bytes(b"")
with (
patch("esphome.yaml_edit.write_file", side_effect=[EsphomeError("disk"), None]),
pytest.raises(EsphomeError, match="Could not restore .*b.yaml: disk"),
):
restore_files({tmp_path / "b.yaml": Snapshot("x", ""), good: Snapshot("y", "")})
with pytest.raises(
EsphomeError, match="Could not restore .*gone.yaml: Error reading file"
):
restore_files({tmp_path / "gone.yaml": Snapshot("x", "x")})
with (
patch(
"esphome.compiled_config.invalidate_compiled_config",
side_effect=EsphomeError("busy"),
),
pytest.raises(
EsphomeError, match="Could not restore the validated caches: busy"
),
):
restore_files({path: Snapshot(YAML, YAML)})
originals = apply_line_edits([_name_edit("garage")])
edited = path.read_text() + "logger:\n"
path.write_bytes(edited.encode())
with pytest.raises(EsphomeError, match="changed since it was written, left as is"):
restore_files(originals)
assert path.read_text() == edited
def test_apply_rolls_back_when_the_cache_cannot_be_dropped(tmp_path: Path) -> None:
path = _setup(tmp_path, YAML)
with (
patch(
"esphome.compiled_config.invalidate_compiled_config",
side_effect=[EsphomeError("busy"), None],
),
pytest.raises(EsphomeError, match="busy"),
):
apply_line_edits([_name_edit("garage")])
assert path.read_text() == YAML
def test_apply_reports_a_rollback_that_also_failed(tmp_path: Path) -> None:
"""The rewrite lands, the reload fails, and the rollback write fails too."""
from esphome.yaml_edit import write_file
_setup(tmp_path, YAML)
writes: list[int] = []
def write_then_fail(*args: Any, **kwargs: Any) -> None:
writes.append(1)
if len(writes) > 1:
raise EsphomeError("disk")
write_file(*args, **kwargs)
with (
patch("esphome.yaml_util.load_yaml", side_effect=EsphomeError("broken")),
patch("esphome.yaml_edit.write_file", side_effect=write_then_fail),
pytest.raises(EsphomeError, match="broken; Could not restore .*disk"),
):
apply_line_edits([_name_edit("garage")])
def test_a_failed_rollback_keeps_an_interrupt_an_interrupt(tmp_path: Path) -> None:
_setup(tmp_path, YAML)
with (
patch("esphome.yaml_util.load_yaml", side_effect=KeyboardInterrupt),
patch("pathlib.Path.chmod", side_effect=[None, OSError("disk")]),
pytest.raises(KeyboardInterrupt) as info,
):
apply_line_edits([_name_edit("garage")])
assert "disk" in "".join(info.value.__notes__)
def test_rollback_after_a_failed_second_write_leaves_the_untouched_file_alone(
tmp_path: Path,
) -> None:
"""The file that was never written is not reported as changed."""
path = _setup(tmp_path, YAML)
other = tmp_path / "other.yaml"
other.write_bytes(b"x: 1\n")
edits = [_name_edit("garage"), LineEdit(other, 0, "x: 1", "x: 2")]
with (
patch(
"esphome.yaml_edit.write_keeping_mode",
side_effect=[None, EsphomeError("disk")],
),
pytest.raises(EsphomeError) as info,
):
apply_line_edits(edits)
assert "changed since it was written" not in str(info.value)
assert path.read_text() == YAML and other.read_text() == "x: 1\n"
def test_restore_goes_on_after_an_interrupt(tmp_path: Path) -> None:
"""A second Ctrl-C must not leave the other files holding the new value."""
path = _setup(tmp_path, YAML)
other = tmp_path / "other.yaml"
other.write_bytes(b"x: 1\n")
calls: list[Path] = []
def interrupt_at(n: int) -> Any:
def write(target: Path, text: str, **kwargs: Any) -> None:
calls.append(target)
if len(calls) == n:
raise KeyboardInterrupt
write_keeping_mode(target, text, **kwargs)
return write
originals = apply_line_edits(
[_name_edit("garage"), LineEdit(other, 0, "x: 1", "x: 2")]
)
with (
patch("esphome.yaml_edit.write_keeping_mode", side_effect=interrupt_at(1)),
pytest.raises(KeyboardInterrupt) as info,
):
restore_files(originals)
assert calls == [path, other] and other.read_text() == "x: 1\n"
assert f"Could not restore {path}: interrupted" in "".join(info.value.__notes__)
# The interrupt stays one inside apply as well
_setup(tmp_path, YAML)
calls.clear()
with (
patch("esphome.yaml_util.load_yaml", side_effect=EsphomeError("broken")),
patch("esphome.yaml_edit.write_keeping_mode", side_effect=interrupt_at(2)),
pytest.raises(KeyboardInterrupt) as info,
):
apply_line_edits([_name_edit("garage")])
assert "Could not restore" in "".join(info.value.__notes__)
def test_restore_finishes_after_any_failure(tmp_path: Path) -> None:
"""A raw OSError still leaves the other files put back, and an interrupt
while dropping the caches carries the report."""
path = _setup(tmp_path, YAML)
other = tmp_path / "other.yaml"
other.write_bytes(b"x: 1\n")
originals = apply_line_edits(
[_name_edit("garage"), LineEdit(other, 0, "x: 1", "x: 2")]
)
writes: list[Path] = []
def fail_first(target: Path, text: str, **kwargs: Any) -> None:
writes.append(target)
if len(writes) == 1:
raise OSError("odd")
write_keeping_mode(target, text, **kwargs)
with (
patch("esphome.yaml_edit.write_keeping_mode", side_effect=fail_first),
pytest.raises(RestoreError) as info,
):
restore_files(originals)
assert f"Could not restore {path}: odd" in str(
info.value
) # a Windows path is no regex
assert "garage" in path.read_text() and other.read_text() == "x: 1\n"
with (
patch(
"esphome.compiled_config.invalidate_compiled_config",
side_effect=KeyboardInterrupt,
),
pytest.raises(KeyboardInterrupt) as info,
):
restore_files(originals)
assert path.read_text() == YAML
assert "the validated caches: interrupted" in "".join(info.value.__notes__)
@pytest.mark.skipif(sys.platform == "win32", reason="posix file modes")
def test_restore_puts_back_the_mode_from_before_the_edit(tmp_path: Path) -> None:
"""A write whose chmod failed leaves the file 0600; the restore does not
copy that, it has the mode from before."""
path = _setup(tmp_path, YAML)
path.chmod(0o644)
originals = apply_line_edits([_name_edit("garage")])
path.chmod(0o600)
restore_files(originals)
assert path.read_text() == YAML
assert path.stat().st_mode & 0o777 == 0o644
def test_restore_finishes_after_an_exit(tmp_path: Path) -> None:
"""A SystemExit from a signal handler still leaves the other files put
back and comes out again with the report."""
path = _setup(tmp_path, YAML)
other = tmp_path / "other.yaml"
other.write_bytes(b"x: 1\n")
originals = apply_line_edits(
[_name_edit("garage"), LineEdit(other, 0, "x: 1", "x: 2")]
)
writes: list[Path] = []
def exit_first(target: Path, text: str, **kwargs: Any) -> None:
writes.append(target)
if len(writes) == 1:
raise SystemExit(1)
write_keeping_mode(target, text, **kwargs)
with (
patch("esphome.yaml_edit.write_keeping_mode", side_effect=exit_first),
pytest.raises(SystemExit) as info,
):
restore_files(originals)
assert other.read_text() == "x: 1\n"
assert f"Could not restore {path}: SystemExit" in "".join(info.value.__notes__)
def test_a_value_written_to_the_secrets_file_is_registered(tmp_path: Path) -> None:
"""Only a resolved !secret registers its value, so the main file is
loaded again after an edit elsewhere; own_documents keeps the registry."""
secrets = tmp_path / "secrets.yaml"
secrets.write_bytes(b"ssid: hidden\n")
_setup(tmp_path, YAML.replace("ssid: kitchen", "ssid: !secret ssid"))
own_documents()
assert yaml_util.is_secret("hidden") == "ssid"
apply_line_edits([LineEdit(secrets, 0, "ssid: hidden", "ssid: hidden2")])
assert yaml_util.is_secret("hidden2") == "ssid"