github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-08-22 22:26:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

[zigbee] Fix merge endpoint (#17511)

Browse Source
This commit is contained in:
luar123
2026-07-13 09:19:23 +12:00
committed by Jesse Hills
parent 1e5cfe6b0f
commit a4650a2345
2 changed files with 64 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
esphome/components/zigbee/zigbee_ep_esp32.py
+60 -42
View File
@@ -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,
skip_error: bool,
) -> bool:
add = True
) -> tuple[str | int, str] | None:
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:
raise cv.Invalid(
f"Endpoint {ep_num} has more than one cluster with cluster id {cl[0]} and role {cl[1]}."
)
add = False
break
if not add:
return cl
return None
def merge_endpoints(
existing_ep: dict[str, Any],
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)
ep.get(DEVICE_TYPE)
and existing_ep.get(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 (
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]
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]
):
if not skip_error:
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"Endpoint {ep_num} has already a conflicting device type {existing_ep[DEVICE_TYPE]} and use_type is not set for both."
f"There is more than one component with endpoint: {num} and {CONF_USE_DEVICE_TYPE}: True"
)
return False
if ep.get(DEVICE_TYPE):
existing_ep[DEVICE_TYPE] = ep[DEVICE_TYPE]
existing_ep[CONF_CLUSTERS].extend(ep[CONF_CLUSTERS])
return 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(
existing_ep, None, ep, ep.get(CONF_USE_DEVICE_TYPE), True
):
if merge_endpoints(existing_ep, ep, ep.get(CONF_USE_DEVICE_TYPE)):
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:
ep[CONF_USE_DEVICE_TYPE] = use_type
if use_type or ep.get(DEVICE_TYPE):
ep[CONF_USE_DEVICE_TYPE] = {ep.get(DEVICE_TYPE): use_type}
ep_dict[ep_num] = ep
tests/components/zigbee/common_esp32.yaml
+1
View File
@@ -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"