[zigbee] Fix merge endpoint (#17511)

esphome/components/zigbee/zigbee_ep_esp32.py

@@ -94,66 +94,73 @@ def get_next_ep_num(eps: list[int]) -> int:
     return ep_num
 
 
-def merge_endpoint(
+def compare_clusters(
     existing_ep: dict[str, Any],
-    ep_num: int | None,
     ep: dict[str, Any],
-    use_type: bool | None,
+) -> tuple[str | int, str] | None:
-    skip_error: bool,
-) -> bool:
-    add = True
     existing_clusters = [(cl[CONF_ID], cl[ROLE]) for cl in existing_ep[CONF_CLUSTERS]]
     for cl in [(cl[CONF_ID], cl[ROLE]) for cl in ep[CONF_CLUSTERS]]:
         if cl in existing_clusters:
-            if not skip_error:
+            return cl
-                raise cv.Invalid(
+    return None
-                    f"Endpoint {ep_num} has more than one cluster with cluster id {cl[0]} and role {cl[1]}."
+
-                )
+
-            add = False
+def merge_endpoints(
-            break
+    existing_ep: dict[str, Any],
-    if not add:
+    ep: dict[str, Any],
+    use_type: bool | None,
+) -> bool:
+    if compare_clusters(existing_ep, ep):
         return False
-    if (
-        use_type
-        and existing_ep.get(CONF_USE_DEVICE_TYPE)
-        and ep.get(DEVICE_TYPE) != existing_ep.get(DEVICE_TYPE)
-    ):
-        if not skip_error:
-            raise cv.Invalid(
-                f"Endpoint {ep_num} has a conflicting device type {existing_ep.get(DEVICE_TYPE, 'CUSTOM_ATTR')} and use_type is set for both."
-            )
-        return False
-    if use_type:
-        existing_ep[CONF_USE_DEVICE_TYPE] = use_type
-        if ep.get(DEVICE_TYPE):
-            existing_ep[DEVICE_TYPE] = ep[DEVICE_TYPE]
-        else:
-            existing_ep.pop(DEVICE_TYPE, None)
-        existing_ep[CONF_CLUSTERS].extend(ep[CONF_CLUSTERS])
-        return True
-    if existing_ep.get(CONF_USE_DEVICE_TYPE):
-        existing_ep[CONF_CLUSTERS].extend(ep[CONF_CLUSTERS])
-        return True
     if (
         ep.get(DEVICE_TYPE)
         and existing_ep.get(DEVICE_TYPE)
-        and ep[DEVICE_TYPE] != existing_ep[DEVICE_TYPE]
+        and ep.get(DEVICE_TYPE) != existing_ep.get(DEVICE_TYPE)
     ):
-        if not skip_error:
-            raise cv.Invalid(
-                f"Endpoint {ep_num} has already a conflicting device type {existing_ep[DEVICE_TYPE]} and use_type is not set for both."
-            )
         return False
+    if (
+        ep.get(DEVICE_TYPE)
+        and not existing_ep.get(DEVICE_TYPE)
+        and existing_ep.get(CONF_USE_DEVICE_TYPE)
+    ):
+        return False
+    if existing_ep.get(DEVICE_TYPE) and not ep.get(DEVICE_TYPE) and use_type:
+        return False
+    if use_type:
+        existing_ep[CONF_USE_DEVICE_TYPE] = use_type
     if ep.get(DEVICE_TYPE):
         existing_ep[DEVICE_TYPE] = ep[DEVICE_TYPE]
     existing_ep[CONF_CLUSTERS].extend(ep[CONF_CLUSTERS])
     return True
 
 
+def validate_endpoints(ep_dict: dict[int, dict]) -> None:
+    for num, ep in ep_dict.items():
+        types_dict = ep.get(CONF_USE_DEVICE_TYPE)
+        if not types_dict:
+            continue
+        if len(types_dict) == 1:
+            ep[DEVICE_TYPE] = list(types_dict.keys())[0]
+            del ep[CONF_USE_DEVICE_TYPE]
+            continue
+        types_list = [t[0] for t in types_dict.items() if t[1]]
+        if len(types_list) > 1:
+            raise cv.Invalid(
+                f"There is more than one component with endpoint: {num} and {CONF_USE_DEVICE_TYPE}: True"
+            )
+        if not types_list:
+            raise cv.Invalid(
+                f"Multiple device types on endpoint: {num}. Set {CONF_USE_DEVICE_TYPE}: True on one component."
+            )
+        ep[DEVICE_TYPE] = types_list[0]
+        del ep[CONF_USE_DEVICE_TYPE]
+
+
 def create_ep(router: bool) -> None:
     zb_data = CORE.data.setdefault(KEY_ZIGBEE, {})
     ep_dict: dict[int, dict] = zb_data.setdefault(KEY_ZIGBEE_EP, {})
     ep_list: list[dict] = zb_data.setdefault(KEY_ZIGBEE_EP_NO_NUM, [])
+    validate_endpoints(ep_dict)
     # create dummy endpoint if list is empty
     if not ep_dict and not ep_list:
         ep_type = "CUSTOM_ATTR"
@@ -166,9 +173,7 @@ def create_ep(router: bool) -> None:
         for ep in ep_list:
             added = False
             for existing_ep in ep_list_new:
-                if merge_endpoint(
+                if merge_endpoints(existing_ep, ep, ep.get(CONF_USE_DEVICE_TYPE)):
-                    existing_ep, None, ep, ep.get(CONF_USE_DEVICE_TYPE), True
-                ):
                     added = True
                     break
             if not added:
@@ -191,6 +196,8 @@ def create_ep(router: bool) -> None:
 
 def add_ep(ep: dict[str, Any], ep_num: int | None, use_type: bool | None) -> None:
     zb_data = CORE.data.setdefault(KEY_ZIGBEE, {})
+    if use_type is False:
+        ep.pop(DEVICE_TYPE, None)
     if ep_num is None:
         if use_type:
             ep[CONF_USE_DEVICE_TYPE] = use_type
@@ -201,8 +208,19 @@ def add_ep(ep: dict[str, Any], ep_num: int | None, use_type: bool | None) -> Non
         if ep_num in ep_dict:
             # check if the existing endpoint has same clusters
             existing_ep = ep_dict[ep_num]
-            merge_endpoint(existing_ep, ep_num, ep, use_type, False)
+            if cl := compare_clusters(
+                existing_ep,
+                ep,
+            ):
+                raise cv.Invalid(
+                    f"Endpoint {ep_num} has more than one cluster with cluster id {cl[0]} and role {cl[1]}."
+                )
+            if ep.get(DEVICE_TYPE) or use_type:
+                types_dict = existing_ep.setdefault(CONF_USE_DEVICE_TYPE, {})
+                if not types_dict.get(ep.get(DEVICE_TYPE)) or use_type:
+                    types_dict[ep.get(DEVICE_TYPE)] = use_type
+            existing_ep[CONF_CLUSTERS].extend(ep[CONF_CLUSTERS])
         else:
-            if use_type is not None:
+            if use_type or ep.get(DEVICE_TYPE):
-                ep[CONF_USE_DEVICE_TYPE] = use_type
+                ep[CONF_USE_DEVICE_TYPE] = {ep.get(DEVICE_TYPE): use_type}
             ep_dict[ep_num] = ep

tests/components/zigbee/common_esp32.yaml

@@ -5,6 +5,7 @@ binary_sensor:
   - platform: template
     name: "Garage Door Open 10"
     report: "default"
+    use_device_type: false
   - platform: template
     name: "Garage Door Open 12"
     report: "force"