diff --git a/CODEOWNERS b/CODEOWNERS index 12aff01e730..e72b1647614 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -132,6 +132,7 @@ esphome/components/dashboard_import/* @esphome/core esphome/components/datetime/* @jesserockz @rfdarter esphome/components/debug/* @esphome/core esphome/components/delonghi/* @grob6000 +esphome/components/dew_point/* @CFlix esphome/components/dfplayer/* @glmnet esphome/components/dfrobot_sen0395/* @niklasweber esphome/components/dht/* @OttoWinter diff --git a/esphome/__main__.py b/esphome/__main__.py index 0433c09e160..3f0da85a694 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -92,7 +92,6 @@ _RP2040_UDEV_HINT = ( "/blob/master/udev/60-picotool.rules" ) - # Special non-component keys that appear in configs _NON_COMPONENT_KEYS = frozenset( { @@ -706,12 +705,12 @@ def _make_crystal_freq_callback( configured_freq: int, ) -> Callable[[str], str | None]: """Create a callback that checks esptool crystal frequency output.""" - crystal_re = re.compile(r"Crystal frequency:\s+(\d+(?:\.\d+)?)\s*MHz") + crystal_re = re.compile(r"Crystal frequency:\s+(\d+)\s*MHz") def check_crystal_line(line: str) -> str | None: if not (match := crystal_re.search(line)): return None - detected = int(float(match.group(1))) + detected = int(match.group(1)) if detected == configured_freq: return None return ( @@ -724,9 +723,7 @@ def _make_crystal_freq_callback( f" esp32:\n" f" framework:\n" f" sdkconfig_options:\n" - f" CONFIG_XTAL_FREQ_{detected}: 'y'\n" - f" CONFIG_XTAL_FREQ_{configured_freq}: 'n'\n" - f' CONFIG_XTAL_FREQ: "{detected}"\033[0m\n\n' + f" CONFIG_XTAL_FREQ_{detected}: 'y'\033[0m\n\n" ) return check_crystal_line @@ -761,7 +758,11 @@ def upload_using_esptool( mcu = get_esp32_variant().lower() line_callbacks: list[Callable[[str], str | None]] = [] - if CORE.is_esp32 and (configured_freq := _get_configured_xtal_freq()) is not None: + if ( + CORE.is_esp32 + and file is None + and (configured_freq := _get_configured_xtal_freq()) is not None + ): line_callbacks.append(_make_crystal_freq_callback(configured_freq)) def run_esptool(baud_rate): diff --git a/esphome/components/dew_point/__init__.py b/esphome/components/dew_point/__init__.py new file mode 100644 index 00000000000..3b852436c3c --- /dev/null +++ b/esphome/components/dew_point/__init__.py @@ -0,0 +1 @@ +CODEOWNERS = ["@CFlix"] diff --git a/esphome/components/dew_point/dew_point.cpp b/esphome/components/dew_point/dew_point.cpp new file mode 100644 index 00000000000..04ac305e3d2 --- /dev/null +++ b/esphome/components/dew_point/dew_point.cpp @@ -0,0 +1,82 @@ + +#include "dew_point.h" + +namespace esphome::dew_point { + +static const char *const TAG = "dew_point.sensor"; + +void DewPointComponent::setup() { + // Register callbacks for sensor updates + if (this->temperature_sensor_ != nullptr) { + this->temperature_sensor_->add_on_state_callback([this](float state) { + this->temperature_value_ = state; + this->enable_loop(); + }); + // Get initial value + if (this->temperature_sensor_->has_state()) { + this->temperature_value_ = this->temperature_sensor_->get_state(); + } + } + + if (this->humidity_sensor_ != nullptr) { + this->humidity_sensor_->add_on_state_callback([this](float state) { + this->humidity_value_ = state; + this->enable_loop(); + }); + // Get initial value + if (this->humidity_sensor_->has_state()) { + this->humidity_value_ = this->humidity_sensor_->get_state(); + } + } +} + +void DewPointComponent::dump_config() { + LOG_SENSOR("", "Dew Point", this); + ESP_LOGCONFIG(TAG, + "Sources\n" + " Temperature: '%s'\n" + " Humidity: '%s'", + this->temperature_sensor_->get_name().c_str(), this->humidity_sensor_->get_name().c_str()); +} + +float DewPointComponent::get_setup_priority() const { return setup_priority::DATA; } + +void DewPointComponent::loop() { + // Only run once + this->disable_loop(); + + // Check if we have valid values for both sensors + if (std::isnan(this->temperature_value_) || std::isnan(this->humidity_value_)) { + ESP_LOGW(TAG, "Temperature or humidity value is NaN, skipping calculation"); + this->publish_state(NAN); + return; + } + + // Check for valid humidity range + if (this->humidity_value_ <= 0.0f || this->humidity_value_ > 100.0f) { + ESP_LOGW(TAG, "Humidity value out of range (0-100): %.2f", this->humidity_value_); + this->publish_state(NAN); + return; + } + + // Magnus formula constants + const float a{17.625f}; + const float b{243.04f}; + + // Calculate dew point using Magnus formula + // Td = (b * alpha) / (a - alpha) + // where alpha = ln(RH/100) + (a * T) / (b + T) + + const float alpha{std::log(this->humidity_value_ / 100.0f) + + (a * this->temperature_value_) / (b + this->temperature_value_)}; + + const float dew_point{(b * alpha) / (a - alpha)}; + + // Publish the calculated dew point + this->publish_state(dew_point); + + ESP_LOGD(TAG, "'%s' >> %.1f°C (T: %.1f°C, RH: %.1f%%)", this->get_name().c_str(), dew_point, this->temperature_value_, + this->humidity_value_); +} + +} // namespace esphome::dew_point diff --git a/esphome/components/dew_point/dew_point.h b/esphome/components/dew_point/dew_point.h new file mode 100644 index 00000000000..833c50fba25 --- /dev/null +++ b/esphome/components/dew_point/dew_point.h @@ -0,0 +1,26 @@ +#pragma once + +#include "esphome/core/component.h" +#include "esphome/components/sensor/sensor.h" + +namespace esphome::dew_point { + +class DewPointComponent : public Component, public sensor::Sensor { + public: + void set_temperature_sensor(sensor::Sensor *temperature_sensor) { this->temperature_sensor_ = temperature_sensor; } + void set_humidity_sensor(sensor::Sensor *humidity_sensor) { this->humidity_sensor_ = humidity_sensor; } + + void setup() override; + void dump_config() override; + void loop() override; + + float get_setup_priority() const override; + + protected: + sensor::Sensor *temperature_sensor_{nullptr}; + sensor::Sensor *humidity_sensor_{nullptr}; + float temperature_value_{NAN}; + float humidity_value_{NAN}; +}; + +} // namespace esphome::dew_point diff --git a/esphome/components/dew_point/sensor.py b/esphome/components/dew_point/sensor.py new file mode 100644 index 00000000000..4fee0956021 --- /dev/null +++ b/esphome/components/dew_point/sensor.py @@ -0,0 +1,46 @@ +import esphome.codegen as cg +from esphome.components import sensor +import esphome.config_validation as cv +from esphome.const import ( + CONF_HUMIDITY, + CONF_TEMPERATURE, + DEVICE_CLASS_TEMPERATURE, + STATE_CLASS_MEASUREMENT, + UNIT_CELSIUS, +) + +DEPENDENCIES = ["sensor"] + +dew_point_ns = cg.esphome_ns.namespace("dew_point") +DewPointComponent = dew_point_ns.class_( + "DewPointComponent", cg.Component, sensor.Sensor +) + +CONFIG_SCHEMA = ( + sensor.sensor_schema( + DewPointComponent, + unit_of_measurement=UNIT_CELSIUS, + accuracy_decimals=1, + device_class=DEVICE_CLASS_TEMPERATURE, + state_class=STATE_CLASS_MEASUREMENT, + icon="mdi:weather-rainy", + ) + .extend( + { + cv.Required(CONF_TEMPERATURE): cv.use_id(sensor.Sensor), + cv.Required(CONF_HUMIDITY): cv.use_id(sensor.Sensor), + } + ) + .extend(cv.COMPONENT_SCHEMA) +) + + +async def to_code(config): + var = await sensor.new_sensor(config) + await cg.register_component(var, config) + + temperature_sensor = await cg.get_variable(config[CONF_TEMPERATURE]) + cg.add(var.set_temperature_sensor(temperature_sensor)) + + humidity_sensor = await cg.get_variable(config[CONF_HUMIDITY]) + cg.add(var.set_humidity_sensor(humidity_sensor)) diff --git a/esphome/components/esp32/gpio.py b/esphome/components/esp32/gpio.py index c0803f40a83..a7180cbcd7e 100644 --- a/esphome/components/esp32/gpio.py +++ b/esphome/components/esp32/gpio.py @@ -88,8 +88,8 @@ def _translate_pin(value): @dataclass class ESP32ValidationFunctions: - pin_validation: Callable[[Any], Any] - usage_validation: Callable[[Any], Any] + pin_validation: Callable[[int], int] + usage_validation: Callable[[dict[str, Any]], dict[str, Any]] _esp32_validations = { diff --git a/esphome/components/esp32/gpio_esp32.py b/esphome/components/esp32/gpio_esp32.py index 973d2dc0ef3..b3166cf822e 100644 --- a/esphome/components/esp32/gpio_esp32.py +++ b/esphome/components/esp32/gpio_esp32.py @@ -1,4 +1,5 @@ import logging +from typing import Any import esphome.config_validation as cv from esphome.const import ( @@ -22,7 +23,7 @@ _ESP32_STRAPPING_PINS = {0, 2, 5, 12, 15} _LOGGER = logging.getLogger(__name__) -def esp32_validate_gpio_pin(value): +def esp32_validate_gpio_pin(value: int) -> int: if value < 0 or value > 39: raise cv.Invalid(f"Invalid pin number: {value} (must be 0-39)") if value in _ESP_SDIO_PINS: @@ -41,7 +42,7 @@ def esp32_validate_gpio_pin(value): return value -def esp32_validate_supports(value): +def esp32_validate_supports(value: dict[str, Any]) -> dict[str, Any]: num = value[CONF_NUMBER] mode = value[CONF_MODE] is_input = mode[CONF_INPUT] diff --git a/esphome/components/esp32/gpio_esp32_c2.py b/esphome/components/esp32/gpio_esp32_c2.py index 32a24050ca7..2d6e3a4a4ed 100644 --- a/esphome/components/esp32/gpio_esp32_c2.py +++ b/esphome/components/esp32/gpio_esp32_c2.py @@ -1,4 +1,5 @@ import logging +from typing import Any import esphome.config_validation as cv from esphome.const import CONF_INPUT, CONF_MODE, CONF_NUMBER @@ -9,14 +10,14 @@ _ESP32C2_STRAPPING_PINS = {8, 9} _LOGGER = logging.getLogger(__name__) -def esp32_c2_validate_gpio_pin(value): +def esp32_c2_validate_gpio_pin(value: int) -> int: if value < 0 or value > 20: raise cv.Invalid(f"Invalid pin number: {value} (must be 0-20)") return value -def esp32_c2_validate_supports(value): +def esp32_c2_validate_supports(value: dict[str, Any]) -> dict[str, Any]: num = value[CONF_NUMBER] mode = value[CONF_MODE] is_input = mode[CONF_INPUT] diff --git a/esphome/components/esp32/gpio_esp32_c3.py b/esphome/components/esp32/gpio_esp32_c3.py index c1427cc02aa..93e0b970934 100644 --- a/esphome/components/esp32/gpio_esp32_c3.py +++ b/esphome/components/esp32/gpio_esp32_c3.py @@ -1,4 +1,5 @@ import logging +from typing import Any import esphome.config_validation as cv from esphome.const import CONF_INPUT, CONF_MODE, CONF_NUMBER @@ -18,7 +19,7 @@ _ESP32C3_STRAPPING_PINS = {2, 8, 9} _LOGGER = logging.getLogger(__name__) -def esp32_c3_validate_gpio_pin(value): +def esp32_c3_validate_gpio_pin(value: int) -> int: if value < 0 or value > 21: raise cv.Invalid(f"Invalid pin number: {value} (must be 0-21)") if value in _ESP32C3_SPI_PSRAM_PINS: @@ -29,7 +30,7 @@ def esp32_c3_validate_gpio_pin(value): return value -def esp32_c3_validate_supports(value): +def esp32_c3_validate_supports(value: dict[str, Any]) -> dict[str, Any]: num = value[CONF_NUMBER] mode = value[CONF_MODE] is_input = mode[CONF_INPUT] diff --git a/esphome/components/esp32/gpio_esp32_c5.py b/esphome/components/esp32/gpio_esp32_c5.py index fa2ce1a689c..639ed64c9ea 100644 --- a/esphome/components/esp32/gpio_esp32_c5.py +++ b/esphome/components/esp32/gpio_esp32_c5.py @@ -1,4 +1,5 @@ import logging +from typing import Any import esphome.config_validation as cv from esphome.const import CONF_INPUT, CONF_MODE, CONF_NUMBER, CONF_SCL, CONF_SDA @@ -22,7 +23,7 @@ _ESP32C5_STRAPPING_PINS = {2, 7, 27, 28} _LOGGER = logging.getLogger(__name__) -def esp32_c5_validate_gpio_pin(value): +def esp32_c5_validate_gpio_pin(value: int) -> int: if value < 0 or value > 28: raise cv.Invalid(f"Invalid pin number: {value} (must be 0-28)") if value in _ESP32C5_SPI_PSRAM_PINS: @@ -33,7 +34,7 @@ def esp32_c5_validate_gpio_pin(value): return value -def esp32_c5_validate_supports(value): +def esp32_c5_validate_supports(value: dict[str, Any]) -> dict[str, Any]: num = value[CONF_NUMBER] mode = value[CONF_MODE] is_input = mode[CONF_INPUT] diff --git a/esphome/components/esp32/gpio_esp32_c6.py b/esphome/components/esp32/gpio_esp32_c6.py index 5d679dede25..cfd3bca8334 100644 --- a/esphome/components/esp32/gpio_esp32_c6.py +++ b/esphome/components/esp32/gpio_esp32_c6.py @@ -1,4 +1,5 @@ import logging +from typing import Any import esphome.config_validation as cv from esphome.const import CONF_INPUT, CONF_MODE, CONF_NUMBER, CONF_SCL, CONF_SDA @@ -22,7 +23,7 @@ _ESP32C6_STRAPPING_PINS = {8, 9, 15} _LOGGER = logging.getLogger(__name__) -def esp32_c6_validate_gpio_pin(value): +def esp32_c6_validate_gpio_pin(value: int) -> int: if value < 0 or value > 23: raise cv.Invalid(f"Invalid pin number: {value} (must be 0-23)") if value in _ESP32C6_SPI_PSRAM_PINS: @@ -33,7 +34,7 @@ def esp32_c6_validate_gpio_pin(value): return value -def esp32_c6_validate_supports(value): +def esp32_c6_validate_supports(value: dict[str, Any]) -> dict[str, Any]: num = value[CONF_NUMBER] mode = value[CONF_MODE] is_input = mode[CONF_INPUT] diff --git a/esphome/components/esp32/gpio_esp32_c61.py b/esphome/components/esp32/gpio_esp32_c61.py index 77be42db3e7..2f3abe6a0f9 100644 --- a/esphome/components/esp32/gpio_esp32_c61.py +++ b/esphome/components/esp32/gpio_esp32_c61.py @@ -1,4 +1,5 @@ import logging +from typing import Any import esphome.config_validation as cv from esphome.const import CONF_INPUT, CONF_MODE, CONF_NUMBER @@ -20,7 +21,7 @@ _ESP32C61_STRAPPING_PINS = {8, 9} _LOGGER = logging.getLogger(__name__) -def esp32_c61_validate_gpio_pin(value): +def esp32_c61_validate_gpio_pin(value: int) -> int: if value < 0 or value > 29: raise cv.Invalid(f"Invalid pin number: {value} (must be 0-29)") if value in _ESP32C61_SPI_PSRAM_PINS: @@ -31,7 +32,7 @@ def esp32_c61_validate_gpio_pin(value): return value -def esp32_c61_validate_supports(value): +def esp32_c61_validate_supports(value: dict[str, Any]) -> dict[str, Any]: num = value[CONF_NUMBER] mode = value[CONF_MODE] is_input = mode[CONF_INPUT] diff --git a/esphome/components/esp32/gpio_esp32_h2.py b/esphome/components/esp32/gpio_esp32_h2.py index f37297764b8..5e7a6158f9b 100644 --- a/esphome/components/esp32/gpio_esp32_h2.py +++ b/esphome/components/esp32/gpio_esp32_h2.py @@ -1,4 +1,5 @@ import logging +from typing import Any import esphome.config_validation as cv from esphome.const import CONF_INPUT, CONF_MODE, CONF_NUMBER @@ -13,7 +14,7 @@ _ESP32H2_STRAPPING_PINS = {2, 3, 8, 9, 25} _LOGGER = logging.getLogger(__name__) -def esp32_h2_validate_gpio_pin(value): +def esp32_h2_validate_gpio_pin(value: int) -> int: if value < 0 or value > 27: raise cv.Invalid(f"Invalid pin number: {value} (must be 0-27)") if value in _ESP32H2_SPI_FLASH_PINS: @@ -33,7 +34,7 @@ def esp32_h2_validate_gpio_pin(value): return value -def esp32_h2_validate_supports(value): +def esp32_h2_validate_supports(value: dict[str, Any]) -> dict[str, Any]: num = value[CONF_NUMBER] mode = value[CONF_MODE] is_input = mode[CONF_INPUT] diff --git a/esphome/components/esp32/gpio_esp32_p4.py b/esphome/components/esp32/gpio_esp32_p4.py index 2726c5932fa..865db926524 100644 --- a/esphome/components/esp32/gpio_esp32_p4.py +++ b/esphome/components/esp32/gpio_esp32_p4.py @@ -1,4 +1,5 @@ import logging +from typing import Any import esphome.config_validation as cv from esphome.const import CONF_INPUT, CONF_MODE, CONF_NUMBER, CONF_SCL, CONF_SDA @@ -14,7 +15,7 @@ _ESP32P4_STRAPPING_PINS = {34, 35, 36, 37, 38} _LOGGER = logging.getLogger(__name__) -def esp32_p4_validate_gpio_pin(value): +def esp32_p4_validate_gpio_pin(value: int) -> int: if value < 0 or value > 54: raise cv.Invalid(f"Invalid pin number: {value} (must be 0-54)") if value in _ESP32P4_USB_JTAG_PINS: @@ -27,7 +28,7 @@ def esp32_p4_validate_gpio_pin(value): return value -def esp32_p4_validate_supports(value): +def esp32_p4_validate_supports(value: dict[str, Any]) -> dict[str, Any]: num = value[CONF_NUMBER] mode = value[CONF_MODE] is_input = mode[CONF_INPUT] diff --git a/esphome/components/esp32/gpio_esp32_s2.py b/esphome/components/esp32/gpio_esp32_s2.py index 331aeb9d94b..4978f48a1c2 100644 --- a/esphome/components/esp32/gpio_esp32_s2.py +++ b/esphome/components/esp32/gpio_esp32_s2.py @@ -1,4 +1,5 @@ import logging +from typing import Any import esphome.config_validation as cv from esphome.const import ( @@ -26,7 +27,7 @@ _ESP32S2_STRAPPING_PINS = {0, 45, 46} _LOGGER = logging.getLogger(__name__) -def esp32_s2_validate_gpio_pin(value): +def esp32_s2_validate_gpio_pin(value: int) -> int: if value < 0 or value > 46: raise cv.Invalid(f"Invalid pin number: {value} (must be 0-46)") @@ -43,7 +44,7 @@ def esp32_s2_validate_gpio_pin(value): return value -def esp32_s2_validate_supports(value): +def esp32_s2_validate_supports(value: dict[str, Any]) -> dict[str, Any]: num = value[CONF_NUMBER] mode = value[CONF_MODE] is_input = mode[CONF_INPUT] diff --git a/esphome/components/esp32/gpio_esp32_s3.py b/esphome/components/esp32/gpio_esp32_s3.py index aea378f499d..cb0eb8178c3 100644 --- a/esphome/components/esp32/gpio_esp32_s3.py +++ b/esphome/components/esp32/gpio_esp32_s3.py @@ -1,4 +1,5 @@ import logging +from typing import Any import esphome.config_validation as cv from esphome.const import CONF_INPUT, CONF_MODE, CONF_NUMBER @@ -27,7 +28,7 @@ _ESP_32S3_STRAPPING_PINS = {0, 3, 45, 46} _LOGGER = logging.getLogger(__name__) -def esp32_s3_validate_gpio_pin(value): +def esp32_s3_validate_gpio_pin(value: int) -> int: if value < 0 or value > 48: raise cv.Invalid(f"Invalid pin number: {value} (must be 0-48)") @@ -49,7 +50,7 @@ def esp32_s3_validate_gpio_pin(value): return value -def esp32_s3_validate_supports(value): +def esp32_s3_validate_supports(value: dict[str, Any]) -> dict[str, Any]: num = value[CONF_NUMBER] mode = value[CONF_MODE] is_input = mode[CONF_INPUT] diff --git a/esphome/util.py b/esphome/util.py index 754e39afabf..73cc3aa5ab6 100644 --- a/esphome/util.py +++ b/esphome/util.py @@ -209,8 +209,8 @@ def run_external_command( func, *cmd, capture_stdout: bool = False, - filter_lines: str = None, - line_callbacks: list | None = None, + filter_lines: list[str] | None = None, + line_callbacks: list[Callable[[str], str | None]] | None = None, ) -> int | str: """ Run a function from an external package that acts like a main method. @@ -220,7 +220,9 @@ def run_external_command( :param func: Function to execute :param cmd: Command to run as (eg first element of sys.argv) :param capture_stdout: Capture text from stdout and return that. - :param filter_lines: Regular expression used to filter captured output. + Note: line_callbacks are not invoked when capture_stdout is True. + :param filter_lines: Regular expressions used to filter captured output. + :param line_callbacks: Callbacks invoked per line; non-None returns are written to output. :return: str if `capture_stdout` is set else int exit code. """ diff --git a/tests/components/dew_point/common.yaml b/tests/components/dew_point/common.yaml new file mode 100644 index 00000000000..527eeb2f84b --- /dev/null +++ b/tests/components/dew_point/common.yaml @@ -0,0 +1,19 @@ +sensor: + - platform: dew_point + name: Dew Point + temperature: template_temperature + humidity: template_humidity + - platform: template + id: template_humidity + lambda: |- + if (millis() > 10000) { + return 0.6; + } + return 0.0; + - platform: template + id: template_temperature + lambda: |- + if (millis() > 10000) { + return 42.0; + } + return 0.0; diff --git a/tests/components/dew_point/test.esp32-idf.yaml b/tests/components/dew_point/test.esp32-idf.yaml new file mode 100644 index 00000000000..dade44d145b --- /dev/null +++ b/tests/components/dew_point/test.esp32-idf.yaml @@ -0,0 +1 @@ +<<: !include common.yaml diff --git a/tests/components/dew_point/test.esp8266-ard.yaml b/tests/components/dew_point/test.esp8266-ard.yaml new file mode 100644 index 00000000000..dade44d145b --- /dev/null +++ b/tests/components/dew_point/test.esp8266-ard.yaml @@ -0,0 +1 @@ +<<: !include common.yaml diff --git a/tests/components/dew_point/test.rp2040-ard.yaml b/tests/components/dew_point/test.rp2040-ard.yaml new file mode 100644 index 00000000000..dade44d145b --- /dev/null +++ b/tests/components/dew_point/test.rp2040-ard.yaml @@ -0,0 +1 @@ +<<: !include common.yaml diff --git a/tests/unit_tests/test_main.py b/tests/unit_tests/test_main.py index 4c7f86e2441..b8534611517 100644 --- a/tests/unit_tests/test_main.py +++ b/tests/unit_tests/test_main.py @@ -3722,38 +3722,38 @@ esp32: assert "secrets.yaml" not in summary_section -def test_get_configured_xtal_freq_reads_sdkconfig(setup_core: Path) -> None: +def test_get_configured_xtal_freq_reads_sdkconfig(tmp_path: Path) -> None: """Test reading XTAL_FREQ from sdkconfig.""" CORE.name = "test-device" - CORE.build_path = setup_core - sdkconfig = setup_core / "sdkconfig.test-device" + CORE.build_path = tmp_path + sdkconfig = tmp_path / "sdkconfig.test-device" sdkconfig.write_text( "CONFIG_SOC_XTAL_SUPPORT_26M=y\nCONFIG_XTAL_FREQ=26\nCONFIG_XTAL_FREQ_26=y\n" ) assert _get_configured_xtal_freq() == 26 -def test_get_configured_xtal_freq_default_40(setup_core: Path) -> None: +def test_get_configured_xtal_freq_default_40(tmp_path: Path) -> None: """Test reading default 40MHz XTAL_FREQ from sdkconfig.""" CORE.name = "test-device" - CORE.build_path = setup_core - sdkconfig = setup_core / "sdkconfig.test-device" + CORE.build_path = tmp_path + sdkconfig = tmp_path / "sdkconfig.test-device" sdkconfig.write_text("CONFIG_XTAL_FREQ=40\nCONFIG_XTAL_FREQ_40=y\n") assert _get_configured_xtal_freq() == 40 -def test_get_configured_xtal_freq_missing_file(setup_core: Path) -> None: +def test_get_configured_xtal_freq_missing_file(tmp_path: Path) -> None: """Test that missing sdkconfig returns None.""" CORE.name = "test-device" - CORE.build_path = setup_core + CORE.build_path = tmp_path assert _get_configured_xtal_freq() is None -def test_get_configured_xtal_freq_no_xtal_line(setup_core: Path) -> None: +def test_get_configured_xtal_freq_no_xtal_line(tmp_path: Path) -> None: """Test that sdkconfig without XTAL_FREQ returns None.""" CORE.name = "test-device" - CORE.build_path = setup_core - sdkconfig = setup_core / "sdkconfig.test-device" + CORE.build_path = tmp_path + sdkconfig = tmp_path / "sdkconfig.test-device" sdkconfig.write_text("CONFIG_OTHER=123\n") assert _get_configured_xtal_freq() is None diff --git a/tests/unit_tests/test_util.py b/tests/unit_tests/test_util.py index ca3fd9b78ab..ff58fb13944 100644 --- a/tests/unit_tests/test_util.py +++ b/tests/unit_tests/test_util.py @@ -2,9 +2,12 @@ from __future__ import annotations +from collections.abc import Callable +import io from pathlib import Path import subprocess import sys +from typing import Any from unittest.mock import MagicMock, patch import pytest @@ -407,6 +410,178 @@ def test_shlex_quote_edge_cases() -> None: assert util.shlex_quote(" ") == "' '" +def _make_redirect( + line_callbacks: list[Callable[[str], str | None]] | None = None, + filter_lines: list[str] | None = None, +) -> tuple[util.RedirectText, io.StringIO]: + """Create a RedirectText that writes to a StringIO buffer.""" + buf = io.StringIO() + redirect = util.RedirectText( + buf, filter_lines=filter_lines, line_callbacks=line_callbacks + ) + return redirect, buf + + +def test_redirect_text_callback_called_on_matching_line() -> None: + """Test that a line callback is called and its output is written.""" + results: list[str] = [] + + def callback(line: str) -> str | None: + results.append(line) + if "target" in line: + return "CALLBACK OUTPUT\n" + return None + + redirect, buf = _make_redirect(line_callbacks=[callback]) + redirect.write("some target line\n") + + assert "some target line" in buf.getvalue() + assert "CALLBACK OUTPUT" in buf.getvalue() + assert len(results) == 1 + + +def test_redirect_text_callback_not_triggered_on_non_matching_line() -> None: + """Test that callback returns None for non-matching lines.""" + + def callback(line: str) -> str | None: + if "target" in line: + return "FOUND\n" + return None + + redirect, buf = _make_redirect(line_callbacks=[callback]) + redirect.write("no match here\n") + + assert "no match here" in buf.getvalue() + assert "FOUND" not in buf.getvalue() + + +def test_redirect_text_callback_works_without_filter_pattern() -> None: + """Test that callbacks fire even when no filter_lines is set.""" + + def callback(line: str) -> str | None: + if "Crystal" in line: + return "WARNING: mismatch\n" + return None + + redirect, buf = _make_redirect(line_callbacks=[callback]) + redirect.write("Crystal frequency: 26MHz\n") + + assert "Crystal frequency: 26MHz" in buf.getvalue() + assert "WARNING: mismatch" in buf.getvalue() + + +def test_redirect_text_callback_works_with_filter_pattern() -> None: + """Test that callbacks fire alongside filter patterns.""" + + def callback(line: str) -> str | None: + if "important" in line: + return "NOTED\n" + return None + + redirect, buf = _make_redirect( + line_callbacks=[callback], + filter_lines=[r"^skip this.*"], + ) + redirect.write("skip this line\n") + redirect.write("important line\n") + + assert "skip this" not in buf.getvalue() + assert "important line" in buf.getvalue() + assert "NOTED" in buf.getvalue() + + +def test_redirect_text_multiple_callbacks() -> None: + """Test that multiple callbacks are all invoked.""" + + def callback_a(line: str) -> str | None: + if "test" in line: + return "FROM A\n" + return None + + def callback_b(line: str) -> str | None: + if "test" in line: + return "FROM B\n" + return None + + redirect, buf = _make_redirect(line_callbacks=[callback_a, callback_b]) + redirect.write("test line\n") + + output = buf.getvalue() + assert "FROM A" in output + assert "FROM B" in output + + +def test_redirect_text_incomplete_line_buffered() -> None: + """Test that incomplete lines are buffered until newline.""" + results: list[str] = [] + + def callback(line: str) -> str | None: + results.append(line) + return None + + redirect, buf = _make_redirect(line_callbacks=[callback]) + redirect.write("partial") + assert len(results) == 0 + + redirect.write(" line\n") + assert len(results) == 1 + assert results[0] == "partial line" + + +def test_run_external_command_line_callbacks(capsys: pytest.CaptureFixture) -> None: + """Test that run_external_command passes line_callbacks to RedirectText.""" + results: list[str] = [] + + def callback(line: str) -> str | None: + results.append(line) + if "hello" in line: + return "CALLBACK FIRED\n" + return None + + def fake_main() -> int: + print("hello world") + return 0 + + rc = util.run_external_command(fake_main, "fake", line_callbacks=[callback]) + + assert rc == 0 + assert len(results) == 1 + assert "hello world" in results[0] + captured = capsys.readouterr() + assert "CALLBACK FIRED" in captured.out + + +def test_run_external_process_line_callbacks() -> None: + """Test that run_external_process passes line_callbacks to RedirectText.""" + results: list[str] = [] + + def callback(line: str) -> str | None: + results.append(line) + if "from subprocess" in line: + return "PROCESS CALLBACK\n" + return None + + with patch("esphome.util.subprocess.run") as mock_run: + + def run_side_effect(*args: Any, **kwargs: Any) -> MagicMock: + # Simulate subprocess writing to the stdout RedirectText + stdout = kwargs.get("stdout") + if stdout is not None and isinstance(stdout, util.RedirectText): + stdout.write("from subprocess\n") + return MagicMock(returncode=0) + + mock_run.side_effect = run_side_effect + + rc = util.run_external_process( + "echo", + "test", + line_callbacks=[callback], + ) + + assert rc == 0 + assert any("from subprocess" in r for r in results) + + def test_get_picotool_path_found(tmp_path: Path) -> None: """Test picotool path derivation from cc_path.""" # Create the expected directory structure