Normalize unique ID in WLED

homeassistant/components/wled/__init__.py

@@ -2,9 +2,12 @@
 
 from __future__ import annotations
 
+import logging
+
+from homeassistant.config_entries import SOURCE_IGNORE
 from homeassistant.const import Platform
 from homeassistant.core import HomeAssistant
-from homeassistant.helpers import config_validation as cv
+from homeassistant.helpers import config_validation as cv, issue_registry as ir
 from homeassistant.helpers.typing import ConfigType
 from homeassistant.util.hass_dict import HassKey
 
@@ -13,6 +16,7 @@
     WLEDConfigEntry,
     WLEDDataUpdateCoordinator,
     WLEDReleasesDataUpdateCoordinator,
+    normalize_mac_address,
 )
 
 PLATFORMS = (
@@ -27,6 +31,9 @@
 
 WLED_KEY: HassKey[WLEDReleasesDataUpdateCoordinator] = HassKey(DOMAIN)
 CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
+UNIQUE_ID_COLLISION_TITLE_LIMIT = 5
+
+_LOGGER = logging.getLogger(__name__)
 
 
 async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
@@ -41,8 +48,71 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     return True
 
 
+def _find_all_entries_with_duplicated_mac(
+    hass: HomeAssistant, entry: WLEDConfigEntry
+) -> list[WLEDConfigEntry]:
+    """Find all WLED config entries with the same MAC address."""
+    assert entry.unique_id
+
+    normalized_mac = normalize_mac_address(entry.unique_id)
+    return [
+        entry
+        for entry in hass.config_entries.async_entries(DOMAIN)
+        if entry.unique_id
+        and normalize_mac_address(entry.unique_id) == normalized_mac
+        and entry.source != SOURCE_IGNORE
+    ]
+
+
+def _handle_device_conflict(hass: HomeAssistant, entry: WLEDConfigEntry) -> None:
+    assert entry.unique_id
+    duplicated_entries = _find_all_entries_with_duplicated_mac(hass, entry)
+
+    if len(duplicated_entries) > 1:
+        _LOGGER.error(
+            "Found %d WLED configuration entries with the same MAC address: %s",
+            len(duplicated_entries),
+            entry.unique_id,
+        )
+        titles = [f"'{entry.title}'" for entry in duplicated_entries]
+        translation_placeholders = {
+            "configure_url": f"/config/integrations/integration/{DOMAIN}",
+            "unique_id": str(entry.unique_id),
+        }
+        if len(titles) <= UNIQUE_ID_COLLISION_TITLE_LIMIT:
+            translation_key = "config_entry_unique_id_collision"
+            translation_placeholders["titles"] = ", ".join(titles)
+        else:
+            translation_key = "config_entry_unique_id_collision_many"
+            translation_placeholders["number_of_entries"] = str(len(titles))
+            translation_placeholders["titles"] = ", ".join(
+                titles[:UNIQUE_ID_COLLISION_TITLE_LIMIT]
+            )
+            translation_placeholders["title_limit"] = str(
+                UNIQUE_ID_COLLISION_TITLE_LIMIT
+            )
+        ir.async_create_issue(
+            hass,
+            DOMAIN,
+            f"device_conflict_{entry.entry_id}",
+            is_fixable=False,
+            severity=ir.IssueSeverity.WARNING,
+            translation_key=translation_key,
+            translation_placeholders=translation_placeholders,
+            data={"entry_id": entry.entry_id},
+        )
+    else:
+        ir.async_delete_issue(hass, DOMAIN, f"device_conflict_{entry.entry_id}")
+        if entry.unique_id != normalize_mac_address(entry.unique_id):
+            hass.config_entries.async_update_entry(
+                entry, unique_id=normalize_mac_address(entry.unique_id)
+            )
+
+
 async def async_setup_entry(hass: HomeAssistant, entry: WLEDConfigEntry) -> bool:
     """Set up WLED from a config entry."""
+    _handle_device_conflict(hass, entry)
+
     entry.runtime_data = WLEDDataUpdateCoordinator(hass, entry=entry)
     await entry.runtime_data.async_config_entry_first_refresh()
 

homeassistant/components/wled/config_flow.py

@@ -21,7 +21,7 @@
 from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
 
 from .const import CONF_KEEP_MAIN_LIGHT, DEFAULT_KEEP_MAIN_LIGHT, DOMAIN
-from .coordinator import WLEDConfigEntry
+from .coordinator import WLEDConfigEntry, normalize_mac_address
 
 
 class WLEDFlowHandler(ConfigFlow, domain=DOMAIN):
@@ -53,9 +53,8 @@ async def async_step_user(
             except WLEDConnectionError:
                 errors["base"] = "cannot_connect"
             else:
-                await self.async_set_unique_id(
-                    device.info.mac_address, raise_on_progress=False
-                )
+                mac_address = normalize_mac_address(device.info.mac_address)
+                await self.async_set_unique_id(mac_address, raise_on_progress=False)
                 if self.source == SOURCE_RECONFIGURE:
                     entry = self._get_reconfigure_entry()
                     self._abort_if_unique_id_mismatch(
@@ -104,7 +103,7 @@ async def async_step_zeroconf(
         """Handle zeroconf discovery."""
         # Abort quick if the mac address is provided by discovery info
         if mac := discovery_info.properties.get(CONF_MAC):
-            await self.async_set_unique_id(mac)
+            await self.async_set_unique_id(normalize_mac_address(mac))
             self._abort_if_unique_id_configured(
                 updates={CONF_HOST: discovery_info.host}
             )
@@ -117,7 +116,10 @@ async def async_step_zeroconf(
         except WLEDConnectionError:
             return self.async_abort(reason="cannot_connect")
 
-        await self.async_set_unique_id(self.discovered_device.info.mac_address)
+        device_mac_address = normalize_mac_address(
+            self.discovered_device.info.mac_address
+        )
+        await self.async_set_unique_id(device_mac_address)
         self._abort_if_unique_id_configured(updates={CONF_HOST: discovery_info.host})
 
         self.context.update(

homeassistant/components/wled/coordinator.py

@@ -32,6 +32,11 @@
 type WLEDConfigEntry = ConfigEntry[WLEDDataUpdateCoordinator]
 
 
+def normalize_mac_address(mac: str) -> str:
+    """Normalize a MAC address to lowercase without separators."""
+    return mac.lower().replace(":", "").replace("-", "").replace(".", "").strip()
+
+
 class WLEDDataUpdateCoordinator(DataUpdateCoordinator[WLEDDevice]):
     """Class to manage fetching WLED data from single endpoint."""
 
@@ -51,6 +56,9 @@ def __init__(
         self.wled = WLED(entry.data[CONF_HOST], session=async_get_clientsession(hass))
         self.unsub: CALLBACK_TYPE | None = None
 
+        assert entry.unique_id
+        self.config_mac_address = normalize_mac_address(entry.unique_id)
+
         super().__init__(
             hass,
             LOGGER,
@@ -131,7 +139,8 @@ async def _async_update_data(self) -> WLEDDevice:
                 translation_placeholders={"error": str(error)},
             ) from error
 
-        if device.info.mac_address != self.config_entry.unique_id:
+        device_mac_address = normalize_mac_address(device.info.mac_address)
+        if device_mac_address != self.config_mac_address:
             raise ConfigEntryError(
                 translation_domain=DOMAIN,
                 translation_key="mac_address_mismatch",

homeassistant/components/wled/strings.json

@@ -148,6 +148,16 @@
       "message": "The WLED device's firmware version is not supported: {error}"
     }
   },
+  "issues": {
+    "config_entry_unique_id_collision": {
+      "description": "There are multiple WLED config entries with the same device MAC address.\nThe config entries are named {titles}.\n\nTo fix this error, [configure the integration]({configure_url}) and remove all except one of the duplicates.\n\nNote: Another group of duplicates may be revealed after removing these duplicates.",
+      "title": "Multiple WLED config entries with same device MAC address"
+    },
+    "config_entry_unique_id_collision_many": {
+      "description": "There are multiple ({number_of_entries}) WLED config entries with the same device MAC address.\nThe first {title_limit} config entries are named {titles}.\n\nTo fix this error, [configure the integration]({configure_url}) and remove all except one of the duplicates.\n\nNote: Another group of duplicates may be revealed after removing these duplicates.",
+      "title": "[%key:component::wled::issues::config_entry_unique_id_collision::title%]"
+    }
+  },
   "options": {
     "step": {
       "init": {

tests/components/wled/conftest.py

@@ -22,6 +22,7 @@ def mock_config_entry() -> MockConfigEntry:
         domain=DOMAIN,
         data={CONF_HOST: "192.168.1.123"},
         unique_id="aabbccddeeff",
+        title="Main WLED",
     )
 
 

tests/components/wled/test_config_flow.py

@@ -281,12 +281,15 @@ async def test_zeroconf_unsupported_version_error(
 
 
 @pytest.mark.usefixtures("mock_wled")
+@pytest.mark.parametrize("device_mac", ["aabbccddeeff", "AABBCCDDEEFF"])
 async def test_user_device_exists_abort(
     hass: HomeAssistant,
     mock_config_entry: MockConfigEntry,
     mock_wled: MagicMock,
+    device_mac: str,
 ) -> None:
     """Test we abort zeroconf flow if WLED device already configured."""
+    mock_wled.update.return_value.info.mac_address = device_mac
     mock_config_entry.add_to_hass(hass)
     result = await hass.config_entries.flow.async_init(
         DOMAIN,
@@ -323,10 +326,12 @@ async def test_zeroconf_without_mac_device_exists_abort(
     assert result.get("reason") == "already_configured"
 
 
+@pytest.mark.parametrize("device_mac", ["aabbccddeeff", "AABBCCDDEEFF"])
 async def test_zeroconf_with_mac_device_exists_abort(
     hass: HomeAssistant,
     mock_config_entry: MockConfigEntry,
     mock_wled: MagicMock,
+    device_mac: str,
 ) -> None:
     """Test we abort zeroconf flow if WLED device already configured."""
     mock_config_entry.add_to_hass(hass)
@@ -339,7 +344,7 @@ async def test_zeroconf_with_mac_device_exists_abort(
             hostname="example.local.",
             name="mock_name",
             port=None,
-            properties={CONF_MAC: "aabbccddeeff"},
+            properties={CONF_MAC: device_mac},
             type="mock_type",
         ),
     )

tests/components/wled/test_init.py

@@ -7,8 +7,10 @@
 import pytest
 from wled import WLEDConnectionError
 
+from homeassistant.components.wled.const import DOMAIN
 from homeassistant.config_entries import ConfigEntryState
 from homeassistant.core import HomeAssistant
+from homeassistant.helpers import issue_registry as ir
 
 from tests.common import MockConfigEntry
 
@@ -67,3 +69,104 @@ async def test_setting_unique_id(
     """Test we set unique ID if not set yet."""
     assert init_integration.runtime_data
     assert init_integration.unique_id == "aabbccddeeff"
+
+
+@pytest.mark.parametrize(
+    (
+        "duplicated_entries_count",
+        "translation_key",
+    ),
+    [
+        (2, "config_entry_unique_id_collision"),
+        (6, "config_entry_unique_id_collision_many"),
+    ],
+)
+async def test_handle_device_conflict_creates_issue_for_duplicates(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_wled: MagicMock,
+    duplicated_entries_count: int,
+    translation_key: str,
+) -> None:
+    """When there are multiple entries for the same MAC, create an issue."""
+    mock_config_entry.add_to_hass(hass)
+    assert mock_config_entry.unique_id
+
+    duplicated_entries = []
+    for i in range(1, duplicated_entries_count):
+        new_entry = MockConfigEntry(
+            domain=DOMAIN,
+            title=f"Duplicate WLED #{i}",
+            unique_id=mock_config_entry.unique_id.upper(),
+            data=mock_config_entry.data,
+        )
+        new_entry.add_to_hass(hass)
+        duplicated_entries.append(new_entry)
+
+    issue_reg = ir.async_get(hass)
+
+    await hass.config_entries.async_setup(mock_config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    issue_id = f"device_conflict_{mock_config_entry.entry_id}"
+    assert (issue := issue_reg.async_get_issue(DOMAIN, issue_id))
+
+    assert issue.domain == DOMAIN
+    assert issue.is_fixable is False
+    assert issue.severity == ir.IssueSeverity.WARNING
+    assert issue.translation_key == translation_key
+
+    # Check translation placeholders
+    assert (placeholders := issue.translation_placeholders)
+    assert placeholders["configure_url"] == "/config/integrations/integration/wled"
+    assert placeholders["unique_id"] == mock_config_entry.unique_id
+
+    # Two titles should be listed in "titles"
+    titles = placeholders["titles"]
+    assert "'Main WLED'" in titles
+    assert "'Duplicate WLED #1'" in titles
+
+    # When there is a conflict, we do not change the unique_id
+    assert mock_config_entry.unique_id == "aabbccddeeff"
+    assert duplicated_entries[0].unique_id == "AABBCCDDEEFF"
+
+
+async def test_handle_device_conflict_normalizes_unique_id_without_duplicates(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_wled: MagicMock,
+) -> None:
+    """When only one entry exists, normalize MAC and delete any existing issue."""
+    assert mock_config_entry.unique_id
+
+    mock_config_entry.add_to_hass(hass)
+
+    # Create a dummy issue to be removed
+    issue_id = f"device_conflict_{mock_config_entry.entry_id}"
+    ir.async_create_issue(
+        hass,
+        DOMAIN,
+        issue_id,
+        is_fixable=False,
+        severity=ir.IssueSeverity.WARNING,
+        translation_key="config_entry_unique_id_collision",
+        translation_placeholders={
+            "configure_url": f"/config/integrations/integration/{DOMAIN}",
+            "unique_id": mock_config_entry.unique_id,
+            "titles": "'Main WLED', 'Duplicate WLED'",
+        },
+        data={},
+    )
+
+    issue_reg = ir.async_get(hass)
+
+    assert issue_reg.async_get_issue(DOMAIN, issue_id)
+
+    await hass.config_entries.async_setup(mock_config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    # Issue should be removed
+    assert issue_id not in issue_reg.issues
+
+    # Unique_id should be normalized
+    assert mock_config_entry.unique_id == "aabbccddeeff"