diff --git a/esphome/components/image/__init__.py b/esphome/components/image/__init__.py index a7b788bf91b..3f8d9098241 100644 --- a/esphome/components/image/__init__.py +++ b/esphome/components/image/__init__.py @@ -1,6 +1,7 @@ from __future__ import annotations import contextlib +from dataclasses import dataclass import hashlib import io import logging @@ -37,11 +38,21 @@ image_ns = cg.esphome_ns.namespace("image") ImageType = image_ns.enum("ImageType") + +@dataclass(frozen=True) +class ImageMetaData: + width: int + height: int + image_type: str + transparency: str + + CONF_OPAQUE = "opaque" CONF_CHROMA_KEY = "chroma_key" CONF_ALPHA_CHANNEL = "alpha_channel" CONF_INVERT_ALPHA = "invert_alpha" CONF_IMAGES = "images" +KEY_METADATA = "metadata" TRANSPARENCY_TYPES = ( CONF_OPAQUE, @@ -723,10 +734,38 @@ async def write_image(config, all_frames=False): return prog_arr, width, height, image_type, trans_value, frame_count +async def _image_to_code(entry): + """ + Convert a single image entry to code and return its metadata. + :param entry: The config entry for the image. + :return: An ImageMetaData object + """ + prog_arr, width, height, image_type, trans_value, _ = await write_image(entry) + cg.new_Pvariable(entry[CONF_ID], prog_arr, width, height, image_type, trans_value) + return ImageMetaData( + width, + height, + entry[CONF_TYPE], + entry[CONF_TRANSPARENCY], + ) + + async def to_code(config): - # By now the config should be a simple list. - for entry in config: - prog_arr, width, height, image_type, trans_value, _ = await write_image(entry) - cg.new_Pvariable( - entry[CONF_ID], prog_arr, width, height, image_type, trans_value - ) + cg.add_define("USE_IMAGE") + # By now the config will be a simple list. + # Use a subkey to allow for other data in the future + CORE.data[DOMAIN] = { + KEY_METADATA: { + entry[CONF_ID].id: await _image_to_code(entry) for entry in config + } + } + + +def get_all_image_metadata() -> dict[str, ImageMetaData]: + """Get all image metadata.""" + return CORE.data.get(DOMAIN, {}).get(KEY_METADATA, {}) + + +def get_image_metadata(image_id: str) -> ImageMetaData | None: + """Get image metadata by ID for use by other components.""" + return get_all_image_metadata().get(image_id) diff --git a/esphome/components/wifi/wifi_component.cpp b/esphome/components/wifi/wifi_component.cpp index 95d89dfc4f8..f646c858476 100644 --- a/esphome/components/wifi/wifi_component.cpp +++ b/esphome/components/wifi/wifi_component.cpp @@ -151,48 +151,51 @@ static const char *const TAG = "wifi"; /// │ Purpose: Handle AP reboot or power loss scenarios where device │ /// │ connects to suboptimal AP and never switches back │ /// │ │ -/// │ Loop call site: roaming enabled && attempts < 3 && 5 min elapsed │ -/// │ ↓ │ -/// │ ┌─────────────────┐ Hidden? ┌──────────────────────────┐ │ -/// │ │ check_roaming_ ├───────────→│ attempts = MAX, stop │ │ -/// │ └────────┬────────┘ └──────────────────────────┘ │ -/// │ ↓ │ -/// │ attempts++, update last_check │ -/// │ ↓ │ -/// │ RSSI > -49 dBm? ────Yes────→ Skip scan (excellent signal)─┐ │ -/// │ ↓ No │ │ -/// │ ┌─────────────────┐ │ │ -/// │ │ Start scan │ │ │ -/// │ └────────┬────────┘ │ │ -/// │ ↓ │ │ -/// │ ┌────────────────────────┐ │ │ -/// │ │ process_roaming_scan_ │ │ │ -/// │ └────────┬───────────────┘ │ │ -/// │ ↓ │ │ -/// │ ┌─────────────────┐ No ┌───────────────┐ │ │ -/// │ │ +10 dB better AP├────────→│ Stay connected│───────────────┤ │ -/// │ └────────┬────────┘ └───────────────┘ │ │ -/// │ │ Yes │ │ -/// │ ↓ │ │ -/// │ ┌─────────────────┐ │ │ -/// │ │ start_connecting│ (roaming_connect_active_ = true) │ │ -/// │ └────────┬────────┘ │ │ -/// │ ↓ │ │ -/// │ ┌────┴────┐ │ │ -/// │ ↓ ↓ │ │ -/// │ ┌───────┐ ┌───────┐ │ │ -/// │ │SUCCESS│ │FAILED │ │ │ -/// │ └───┬───┘ └───┬───┘ │ │ -/// │ ↓ ↓ │ │ -/// │ Keep counter retry_connect() → normal reconnect flow │ │ -/// │ (no reset) (keeps counter, handles retries) │ │ -/// │ │ │ │ │ -/// │ └──────────────┴────────────────────────────────────────┘ │ +/// │ State Machine (RoamingState): │ /// │ │ -/// │ After 3 checks: attempts >= 3, stop checking │ -/// │ Non-roaming disconnect: clear_roaming_state_() resets counter │ -/// │ Roaming success: counter preserved (prevents ping-pong) │ -/// │ Roaming fail: normal flow handles reconnection, counter preserved │ +/// │ ┌─────────────────────────────────────────────────────────────┐ │ +/// │ │ IDLE │ │ +/// │ │ (waiting for 5 min timer, attempts < 3) │ │ +/// │ └─────────────────────────┬───────────────────────────────────┘ │ +/// │ │ 5 min elapsed, RSSI < -49 dBm │ +/// │ ↓ │ +/// │ ┌─────────────────────────────────────────────────────────────┐ │ +/// │ │ SCANNING │ │ +/// │ │ (check_roaming_ starts scan, attempts++) │ │ +/// │ └─────────────────────────┬───────────────────────────────────┘ │ +/// │ │ scan done │ +/// │ ┌──────────────┴──────────────┐ │ +/// │ ↓ ↓ │ +/// │ No better AP found +10 dB better AP found │ +/// │ │ │ │ +/// │ ↓ ↓ │ +/// │ ┌──────────────────┐ ┌─────────────────────────────────────┐ │ +/// │ │ → IDLE │ │ CONNECTING │ │ +/// │ │ (stay connected)│ │ (process_roaming_scan_ connects) │ │ +/// │ └──────────────────┘ └─────────────────────┬───────────────┘ │ +/// │ │ │ +/// │ ┌───────────────────┴───────────────┐ │ +/// │ ↓ ↓ │ +/// │ SUCCESS FAILED │ +/// │ │ │ │ +/// │ ↓ ↓ │ +/// │ ┌──────────────────────────────────┐ ┌─────────────────────────┐ +/// │ │ → IDLE │ │ RECONNECTING │ +/// │ │ (counter preserved, no reset) │ │ (retry_connect called) │ +/// │ └──────────────────────────────────┘ └───────────┬─────────────┘ +/// │ │ │ +/// │ ↓ │ +/// │ ┌───────────────────────┐ │ +/// │ │ → IDLE │ │ +/// │ │ (counter preserved!) │ │ +/// │ └───────────────────────┘ │ +/// │ │ +/// │ Key behaviors: │ +/// │ - After 3 checks: attempts >= 3, stop checking │ +/// │ - Non-roaming disconnect: clear_roaming_state_() resets counter │ +/// │ - Roaming success (CONNECTING→IDLE): counter preserved │ +/// │ - Roaming fail (RECONNECTING→IDLE): counter preserved │ +/// │ - This prevents ping-pong when roam target AP is unreachable │ /// └──────────────────────────────────────────────────────────────────────┘ static const LogString *retry_phase_to_log_string(WiFiRetryPhase phase) { @@ -574,12 +577,12 @@ void WiFiComponent::loop() { // Post-connect roaming: check for better AP if (this->post_connect_roaming_) { - if (this->roaming_scan_active_) { + if (this->roaming_state_ == RoamingState::SCANNING) { if (this->scan_done_) { this->process_roaming_scan_(); } // else: scan in progress, wait - } else if (this->roaming_attempts_ < ROAMING_MAX_ATTEMPTS && + } else if (this->roaming_state_ == RoamingState::IDLE && this->roaming_attempts_ < ROAMING_MAX_ATTEMPTS && now - this->roaming_last_check_ >= ROAMING_CHECK_INTERVAL) { this->check_roaming_(now); } @@ -1303,11 +1306,16 @@ void WiFiComponent::check_connecting_finished(uint32_t now) { // Reset roaming state on successful connection this->roaming_last_check_ = now; // Only reset attempts if this wasn't a roaming-triggered connection - // (prevents ping-pong between APs) - if (!this->roaming_connect_active_) { + // (CONNECTING = roam attempt, RECONNECTING = failed roam, reconnecting) + // This prevents ping-pong between APs when a roam target is unreachable + if (this->roaming_state_ == RoamingState::CONNECTING) { + ESP_LOGD(TAG, "Roam successful"); + } else if (this->roaming_state_ == RoamingState::RECONNECTING) { + ESP_LOGD(TAG, "Reconnected after failed roam (attempt %u/%u)", this->roaming_attempts_, ROAMING_MAX_ATTEMPTS); + } else { this->roaming_attempts_ = 0; } - this->roaming_connect_active_ = false; + this->roaming_state_ = RoamingState::IDLE; // Clear all priority penalties - the next reconnect will happen when an AP disconnects, // which means the landscape has likely changed and previous tracked failures are stale @@ -1734,14 +1742,15 @@ void WiFiComponent::advance_to_next_target_or_increment_retry_() { } void WiFiComponent::retry_connect() { - // If this was a roaming attempt, preserve roaming_attempts_ count - // (so we stop roaming after ROAMING_MAX_ATTEMPTS failures) + // If this was a roaming attempt, transition to RECONNECTING state + // (preserves roaming_attempts_ so we stop roaming after ROAMING_MAX_ATTEMPTS failures) // Otherwise reset all roaming state - if (this->roaming_connect_active_) { - this->roaming_connect_active_ = false; - this->roaming_scan_active_ = false; + if (this->roaming_state_ == RoamingState::CONNECTING) { + ESP_LOGD(TAG, "Roam failed, reconnecting (attempt %u/%u)", this->roaming_attempts_, ROAMING_MAX_ATTEMPTS); + this->roaming_state_ = RoamingState::RECONNECTING; // Keep roaming_attempts_ - will prevent further roaming after max failures - } else { + } else if (this->roaming_state_ != RoamingState::RECONNECTING) { + // Not a roaming-triggered reconnect, reset state this->clear_roaming_state_(); } @@ -1990,8 +1999,7 @@ bool WiFiScanResult::operator==(const WiFiScanResult &rhs) const { return this-> void WiFiComponent::clear_roaming_state_() { this->roaming_attempts_ = 0; this->roaming_last_check_ = 0; - this->roaming_scan_active_ = false; - this->roaming_connect_active_ = false; + this->roaming_state_ = RoamingState::IDLE; } void WiFiComponent::release_scan_results_() { @@ -2019,17 +2027,20 @@ void WiFiComponent::check_roaming_(uint32_t now) { // Guard: skip scan if signal is already good (no meaningful improvement possible) int8_t rssi = this->wifi_rssi(); - if (rssi > ROAMING_GOOD_RSSI) + if (rssi > ROAMING_GOOD_RSSI) { + ESP_LOGV(TAG, "Roam check skipped, signal good (%d dBm)", rssi); return; + } ESP_LOGD(TAG, "Roam scan (%d dBm)", rssi); - this->roaming_scan_active_ = true; + this->roaming_state_ = RoamingState::SCANNING; this->wifi_scan_start_(this->passive_scan_); } void WiFiComponent::process_roaming_scan_() { this->scan_done_ = false; - this->roaming_scan_active_ = false; + // Default to IDLE - will be set to CONNECTING if we find a better AP + this->roaming_state_ = RoamingState::IDLE; // Get current connection info int8_t current_rssi = this->wifi_rssi(); @@ -2080,7 +2091,7 @@ void WiFiComponent::process_roaming_scan_() { this->release_scan_results_(); // Mark as roaming attempt - affects retry behavior if connection fails - this->roaming_connect_active_ = true; + this->roaming_state_ = RoamingState::CONNECTING; // Connect directly - wifi_sta_connect_ handles disconnect internally this->error_from_callback_ = false; diff --git a/esphome/components/wifi/wifi_component.h b/esphome/components/wifi/wifi_component.h index 09af3847253..9b606bd692b 100644 --- a/esphome/components/wifi/wifi_component.h +++ b/esphome/components/wifi/wifi_component.h @@ -112,6 +112,18 @@ enum class WiFiRetryPhase : uint8_t { RESTARTING_ADAPTER, }; +/// Tracks post-connect roaming state machine +enum class RoamingState : uint8_t { + /// Not roaming, waiting for next check interval + IDLE, + /// Scanning for better AP + SCANNING, + /// Attempting to connect to better AP found in scan + CONNECTING, + /// Roam connection failed, reconnecting to any available AP + RECONNECTING, +}; + /// Struct for setting static IPs in WiFiComponent. struct ManualIP { network::IPAddress static_ip; @@ -667,8 +679,7 @@ class WiFiComponent : public Component { bool did_scan_this_cycle_{false}; bool skip_cooldown_next_cycle_{false}; bool post_connect_roaming_{true}; // Enabled by default - bool roaming_scan_active_{false}; - bool roaming_connect_active_{false}; // True during roaming connection attempt (preserves roaming_attempts_) + RoamingState roaming_state_{RoamingState::IDLE}; #if defined(USE_ESP32) && defined(USE_WIFI_RUNTIME_POWER_SAVE) WiFiPowerSaveMode configured_power_save_{WIFI_POWER_SAVE_NONE}; bool is_high_performance_mode_{false}; diff --git a/tests/component_tests/image/test_init.py b/tests/component_tests/image/test_init.py index f0b132cef81..930bbac8d1e 100644 --- a/tests/component_tests/image/test_init.py +++ b/tests/component_tests/image/test_init.py @@ -9,8 +9,14 @@ from typing import Any import pytest from esphome import config_validation as cv -from esphome.components.image import CONF_TRANSPARENCY, CONFIG_SCHEMA +from esphome.components.image import ( + CONF_TRANSPARENCY, + CONFIG_SCHEMA, + get_all_image_metadata, + get_image_metadata, +) from esphome.const import CONF_ID, CONF_RAW_DATA_ID, CONF_TYPE +from esphome.core import CORE @pytest.mark.parametrize( @@ -235,3 +241,112 @@ def test_image_generation( "cat_img = new image::Image(uint8_t_id, 32, 24, image::IMAGE_TYPE_RGB565, image::TRANSPARENCY_OPAQUE);" in main_cpp ) + + +def test_image_to_code_defines_and_core_data( + generate_main: Callable[[str | Path], str], + component_config_path: Callable[[str], Path], +) -> None: + """Test that to_code() sets USE_IMAGE define and stores image metadata.""" + # Generate the main cpp which will call to_code + generate_main(component_config_path("image_test.yaml")) + + # Verify USE_IMAGE define was added + assert any(d.name == "USE_IMAGE" for d in CORE.defines), ( + "USE_IMAGE define should be set when images are configured" + ) + + # Use the public API to get image metadata + # The test config has an image with id 'cat_img' + cat_img_metadata = get_image_metadata("cat_img") + + assert cat_img_metadata is not None, ( + "Image metadata should be retrievable via get_image_metadata()" + ) + + # Verify the metadata has the expected attributes + assert hasattr(cat_img_metadata, "width"), "Metadata should have width attribute" + assert hasattr(cat_img_metadata, "height"), "Metadata should have height attribute" + assert hasattr(cat_img_metadata, "image_type"), ( + "Metadata should have image_type attribute" + ) + assert hasattr(cat_img_metadata, "transparency"), ( + "Metadata should have transparency attribute" + ) + + # Verify the values are correct (from the test image) + assert cat_img_metadata.width == 32, "Width should be 32" + assert cat_img_metadata.height == 24, "Height should be 24" + assert cat_img_metadata.image_type == "RGB565", "Type should be RGB565" + assert cat_img_metadata.transparency == "opaque", "Transparency should be opaque" + + +def test_image_to_code_multiple_images( + generate_main: Callable[[str | Path], str], + component_config_path: Callable[[str], Path], +) -> None: + """Test that to_code() stores metadata for multiple images.""" + generate_main(component_config_path("image_test.yaml")) + + # Use the public API to get all image metadata + all_metadata = get_all_image_metadata() + + assert isinstance(all_metadata, dict), ( + "get_all_image_metadata() should return a dictionary" + ) + + # Verify that at least one image is present + assert len(all_metadata) > 0, "Should have at least one image metadata entry" + + # Each image ID should map to an ImageMetaData object + for image_id, metadata in all_metadata.items(): + assert isinstance(image_id, str), "Image IDs should be strings" + + # Verify it's an ImageMetaData object with all required attributes + assert hasattr(metadata, "width"), ( + f"Metadata for '{image_id}' should have width" + ) + assert hasattr(metadata, "height"), ( + f"Metadata for '{image_id}' should have height" + ) + assert hasattr(metadata, "image_type"), ( + f"Metadata for '{image_id}' should have image_type" + ) + assert hasattr(metadata, "transparency"), ( + f"Metadata for '{image_id}' should have transparency" + ) + + # Verify values are valid + assert isinstance(metadata.width, int), ( + f"Width for '{image_id}' should be an integer" + ) + assert isinstance(metadata.height, int), ( + f"Height for '{image_id}' should be an integer" + ) + assert isinstance(metadata.image_type, str), ( + f"Type for '{image_id}' should be a string" + ) + assert isinstance(metadata.transparency, str), ( + f"Transparency for '{image_id}' should be a string" + ) + assert metadata.width > 0, f"Width for '{image_id}' should be positive" + assert metadata.height > 0, f"Height for '{image_id}' should be positive" + + +def test_get_image_metadata_nonexistent() -> None: + """Test that get_image_metadata returns None for non-existent image IDs.""" + # This should return None when no images are configured or ID doesn't exist + metadata = get_image_metadata("nonexistent_image_id") + assert metadata is None, ( + "get_image_metadata should return None for non-existent IDs" + ) + + +def test_get_all_image_metadata_empty() -> None: + """Test that get_all_image_metadata returns empty dict when no images configured.""" + # When CORE hasn't been initialized with images, should return empty dict + all_metadata = get_all_image_metadata() + assert isinstance(all_metadata, dict), ( + "get_all_image_metadata should always return a dict" + ) + # Length could be 0 or more depending on what's in CORE at test time