github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-07-10 08:55:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

[zigbee] Allow to combine and merge endpoints on esp32 (#17402)

Browse Source
This commit is contained in:
luar123
2026-07-08 20:41:36 +02:00
committed by GitHub
parent ce46895270
commit 84f4fbeaa8
8 changed files with 190 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
esphome/components/zigbee/__init__.py
+17 -4
View File
@@ -18,10 +18,13 @@ from esphome.core import CORE, CoroPriority, coroutine_with_priority
from esphome.types import ConfigType
from .const import (
CONF_ENDPOINT,
CONF_MAX_EP_NUMBER,
CONF_ON_JOIN,
CONF_POWER_SOURCE,
CONF_REPORT,
CONF_ROUTER,
CONF_USE_DEVICE_TYPE,
CONF_WIPE_ON_BOOT,
KEY_ZIGBEE,
POWER_SOURCE,
@@ -31,7 +34,7 @@ from .const import (
)
from .const_zephyr import (
CONF_IEEE802154_VENDOR_OUI,
CONF_MAX_EP_NUMBER,
CONF_MAX_EP_NUMBER_ZEPHYR,
CONF_SLEEPY,
CONF_ZIGBEE_ID,
KEY_EP_NUMBER,
@@ -71,7 +74,17 @@ BASE_SCHEMA = cv.Schema(
cv.requires_component("esp32"),
_check_report_deprecation,
cv.enum(REPORT, lower=True),
)
),
cv.Optional(CONF_ENDPOINT): cv.All(
cv.requires_component("zigbee"),
cv.requires_component("esp32"),
cv.int_range(1, CONF_MAX_EP_NUMBER),
),
cv.Optional(CONF_USE_DEVICE_TYPE): cv.All(
cv.requires_component("zigbee"),
cv.requires_component("esp32"),
cv.boolean,
),
}
)
BINARY_SENSOR_SCHEMA = cv.Schema({}).extend(BASE_SCHEMA).extend(zephyr_binary_sensor)
@@ -148,8 +161,8 @@ def validate_number_of_ep(config: ConfigType) -> ConfigType:
_LOGGER.warning(
"Single endpoint requires ZHA or at leatst Zigbee2MQTT 2.8.0. For older versions of Zigbee2MQTT use multiple endpoints"
)
if count > CONF_MAX_EP_NUMBER and not CORE.testing_mode:
raise cv.Invalid(f"Maximum number of end points is {CONF_MAX_EP_NUMBER}")
if count > CONF_MAX_EP_NUMBER_ZEPHYR and not CORE.testing_mode:
raise cv.Invalid(f"Maximum number of end points is {CONF_MAX_EP_NUMBER_ZEPHYR}")
return config
esphome/components/zigbee/const.py
+3
View File
@@ -58,11 +58,14 @@ REPORT = {
"default": report.ZIGBEE_REPORT_DEFAULT,
}
CONF_ENDPOINT = "endpoint"
CONF_MAX_EP_NUMBER = 239
CONF_ON_JOIN = "on_join"
CONF_WIPE_ON_BOOT = "wipe_on_boot"
CONF_REPORT = "report"
CONF_ROUTER = "router"
CONF_POWER_SOURCE = "power_source"
CONF_USE_DEVICE_TYPE = "use_device_type"
POWER_SOURCE = {
"UNKNOWN": 0x00, # ZB_ZCL_BASIC_POWER_SOURCE_UNKNOWN
"MAINS_SINGLE_PHASE": 0x01, # ZB_ZCL_BASIC_POWER_SOURCE_MAINS_SINGLE_PHASE
esphome/components/zigbee/const_esp32.py
+2 -5
View File
@@ -2,16 +2,13 @@ import esphome.codegen as cg
DEVICE_TYPE = "device_type"
ROLE = "role"
CONF_MAX_EP_NUMBER = 239
CONF_NUM = "num"
CONF_CLUSTERS = "clusters"
CONF_ATTRIBUTES = "attributes"
CONF_ENDPOINT = "endpoint"
CONF_CLUSTER = "cluster"
SCALE = "scale"
CONF_ATTRIBUTE_ID = "attribute_id"
KEY_BS_EP = "binary_sensor_ep"
KEY_SENSOR_EP = "sensor_ep"
KEY_ZIGBEE_EP = "zigbee_ep"
KEY_ZIGBEE_EP_NO_NUM = "zigbee_ep_no_num"
DEVICE_ID = {
"RANGE_EXTENDER": cg.RawExpression("EZB_ZHA_RANGE_EXTENDER_DEVICE_ID"),
esphome/components/zigbee/const_zephyr.py
+1 -1
View File
@@ -1,4 +1,4 @@
CONF_MAX_EP_NUMBER = 8
CONF_MAX_EP_NUMBER_ZEPHYR = 8
CONF_ZIGBEE_ID = "zigbee_id"
CONF_ZIGBEE_BINARY_SENSOR = "zigbee_binary_sensor"
CONF_ZIGBEE_SENSOR = "zigbee_sensor"
esphome/components/zigbee/zigbee_ep_esp32.py
+133 -24
View File
@@ -2,16 +2,22 @@ from typing import Any
import esphome.config_validation as cv
from esphome.const import CONF_DEVICE, CONF_ID, CONF_TYPE
from esphome.core import CORE
from .const import CONF_REPORT, REPORT
from .const import (
CONF_MAX_EP_NUMBER,
CONF_REPORT,
CONF_USE_DEVICE_TYPE,
KEY_ZIGBEE,
REPORT,
)
from .const_esp32 import (
CLUSTER_ROLE,
CONF_ATTRIBUTE_ID,
CONF_ATTRIBUTES,
CONF_CLUSTERS,
CONF_MAX_EP_NUMBER,
CONF_NUM,
DEVICE_TYPE,
KEY_ZIGBEE_EP,
KEY_ZIGBEE_EP_NO_NUM,
ROLE,
)
@@ -22,12 +28,12 @@ ep_configs: dict[str, dict[str, Any]] = {
CONF_CLUSTERS: [
{
CONF_ID: "BINARY_INPUT",
ROLE: CLUSTER_ROLE["SERVER"],
ROLE: "SERVER",
CONF_ATTRIBUTES: [
{
CONF_ATTRIBUTE_ID: 0x55,
CONF_TYPE: "BOOL",
CONF_REPORT: REPORT["default"],
CONF_REPORT: cv.enum(REPORT, lower=True)("default"),
CONF_DEVICE: None,
},
{
@@ -47,16 +53,15 @@ ep_configs: dict[str, dict[str, Any]] = {
],
},
"analog_input": {
DEVICE_TYPE: "CUSTOM_ATTR",
CONF_CLUSTERS: [
{
CONF_ID: "ANALOG_INPUT",
ROLE: CLUSTER_ROLE["SERVER"],
ROLE: "SERVER",
CONF_ATTRIBUTES: [
{
CONF_ATTRIBUTE_ID: 0x55,
CONF_TYPE: "SINGLE",
CONF_REPORT: REPORT["default"],
CONF_REPORT: cv.enum(REPORT, lower=True)("default"),
CONF_DEVICE: None,
},
{
@@ -78,22 +83,126 @@ ep_configs: dict[str, dict[str, Any]] = {
}
def create_ep(ep_list: list[dict[str, Any]], router: bool) -> list[dict[str, Any]]:
def get_next_ep_num(eps: list[int]) -> int:
try:
ep_num = [i for i in range(1, CONF_MAX_EP_NUMBER + 1) if i not in eps][0]
eps.append(ep_num)
except IndexError as e:
raise cv.Invalid(
f"Too many devices. Zigbee can define only {CONF_MAX_EP_NUMBER} endpoints."
) from e
return ep_num
def merge_endpoint(
existing_ep: dict[str, Any],
ep_num: int | None,
ep: dict[str, Any],
use_type: bool | 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:
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 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]
):
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):
existing_ep[DEVICE_TYPE] = ep[DEVICE_TYPE]
existing_ep[CONF_CLUSTERS].extend(ep[CONF_CLUSTERS])
return True
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, [])
# create dummy endpoint if list is empty
if not ep_list:
if not ep_dict and not ep_list:
ep_type = "CUSTOM_ATTR"
if router:
ep_type = "RANGE_EXTENDER"
ep_list = [
{
DEVICE_TYPE: ep_type,
}
]
# enumerate endpoints
for i, ep in enumerate(ep_list, 1):
ep[CONF_NUM] = i
if len(ep_list) > CONF_MAX_EP_NUMBER:
raise cv.Invalid(
f"Too many devices. Zigbee can define only {CONF_MAX_EP_NUMBER} endpoints."
)
return ep_list
ep_dict[1] = {DEVICE_TYPE: ep_type}
if ep_list:
# merge endpoint with different clusters
ep_list_new: list[dict] = []
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
):
added = True
break
if not added:
ep_list_new.append(ep)
# Add endpoints with no number to the endpoint dict with a new number
eps = list(ep_dict.keys())
for ep in ep_list_new:
ep_num = get_next_ep_num(eps)
ep_dict[ep_num] = ep
# clear list so that it is not processed again
del zb_data[KEY_ZIGBEE_EP_NO_NUM]
# Add default device type to endpoints that have none
for ep in ep_dict.values():
if not ep.get(DEVICE_TYPE):
ep[DEVICE_TYPE] = "CUSTOM_ATTR"
def add_ep(ep: dict[str, Any], ep_num: int | None, use_type: bool | None) -> None:
zb_data = CORE.data.setdefault(KEY_ZIGBEE, {})
if ep_num is None:
if use_type:
ep[CONF_USE_DEVICE_TYPE] = use_type
ep_list: list[dict] = zb_data.setdefault(KEY_ZIGBEE_EP_NO_NUM, [])
ep_list.append(ep)
else:
ep_dict: dict[int, dict] = zb_data.setdefault(KEY_ZIGBEE_EP, {})
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)
else:
if use_type is not None:
ep[CONF_USE_DEVICE_TYPE] = use_type
ep_dict[ep_num] = ep
esphome/components/zigbee/zigbee_esp32.py
+16 -20
View File
@@ -35,9 +35,11 @@ from .const import (
ANALOG_INPUT_APPTYPE,
BACNET_UNIT_NO_UNITS,
BACNET_UNITS,
CONF_ENDPOINT,
CONF_POWER_SOURCE,
CONF_REPORT,
CONF_ROUTER,
CONF_USE_DEVICE_TYPE,
KEY_ZIGBEE,
POWER_SOURCE,
ZigbeeAttribute,
@@ -45,18 +47,17 @@ from .const import (
from .const_esp32 import (
ATTR_TYPE,
CLUSTER_ID,
CLUSTER_ROLE,
CONF_ATTRIBUTE_ID,
CONF_ATTRIBUTES,
CONF_CLUSTERS,
CONF_NUM,
DEVICE_ID,
DEVICE_TYPE,
KEY_BS_EP,
KEY_SENSOR_EP,
KEY_ZIGBEE_EP,
ROLE,
SCALE,
)
from .zigbee_ep_esp32 import create_ep, ep_configs
from .zigbee_ep_esp32 import add_ep, create_ep, ep_configs
_LOGGER = logging.getLogger(__name__)
@@ -146,6 +147,7 @@ def final_validate_esp32(config: ConfigType) -> ConfigType:
raise cv.Invalid(
f"Partition '{partition}' in your custom partition table has wrong format. It should be: '{partition}, {types['type']}, {types['subtype']}, , {types['size']},'"
)
create_ep(config.get(CONF_ROUTER))
return config
@@ -199,18 +201,14 @@ def validate_sensor_esp32(config: ConfigType) -> ConfigType:
},
)
setup_attributes(config, ep[CONF_CLUSTERS])
zb_data = CORE.data.setdefault(KEY_ZIGBEE, {})
sensor_ep: list[dict] = zb_data.setdefault(KEY_SENSOR_EP, [])
sensor_ep.append(ep)
add_ep(ep, config.get(CONF_ENDPOINT), config.get(CONF_USE_DEVICE_TYPE))
return config
def validate_binary_sensor_esp32(config: ConfigType) -> ConfigType:
ep = copy.deepcopy(ep_configs["binary_input"])
setup_attributes(config, ep[CONF_CLUSTERS])
zb_data = CORE.data.setdefault(KEY_ZIGBEE, {})
binary_sensor_ep: list[dict] = zb_data.setdefault(KEY_BS_EP, [])
binary_sensor_ep.append(ep)
add_ep(ep, config.get(CONF_ENDPOINT), config.get(CONF_USE_DEVICE_TYPE))
return config
@@ -243,7 +241,7 @@ async def attributes_to_code(
var.add_attr(
ep_num,
CLUSTER_ID.get(cl[CONF_ID], cl[CONF_ID]),
cl[ROLE],
CLUSTER_ROLE[cl[ROLE]],
attr[CONF_ATTRIBUTE_ID],
attr.get(CONF_MAX_LENGTH, 0),
attr[CONF_VALUE],
@@ -255,7 +253,7 @@ async def attributes_to_code(
var,
ep_num,
CLUSTER_ID.get(cl[CONF_ID], cl[CONF_ID]),
cl[ROLE],
CLUSTER_ROLE[cl[ROLE]],
attr[CONF_ATTRIBUTE_ID],
ATTR_TYPE[attr[CONF_TYPE]],
attr.get(SCALE, 1),
@@ -287,9 +285,7 @@ async def esp32_to_code(config: ConfigType) -> "MockObj":
# create endpoints
zb_data = CORE.data.get(KEY_ZIGBEE, {})
sensor_ep: list[dict] = zb_data.get(KEY_SENSOR_EP, [])
binary_sensor_ep: list[dict] = zb_data.get(KEY_BS_EP, [])
ep_list = create_ep(sensor_ep + binary_sensor_ep, config.get(CONF_ROUTER))
ep_dict: dict[int, dict] = zb_data.get(KEY_ZIGBEE_EP, {})
# setup zigbee components
var = cg.new_Pvariable(config[CONF_ID])
@@ -301,15 +297,15 @@ async def esp32_to_code(config: ConfigType) -> "MockObj":
POWER_SOURCE[config[CONF_POWER_SOURCE]],
)
)
for ep in ep_list:
cg.add(var.create_default_cluster(ep[CONF_NUM], DEVICE_ID[ep[DEVICE_TYPE]]))
for ep_num, ep in ep_dict.items():
cg.add(var.create_default_cluster(ep_num, DEVICE_ID[ep[DEVICE_TYPE]]))
for cl in ep.get(CONF_CLUSTERS, []):
cg.add(
var.add_cluster(
ep[CONF_NUM],
ep_num,
CLUSTER_ID.get(cl[CONF_ID], cl[CONF_ID]),
cl[ROLE],
CLUSTER_ROLE[cl[ROLE]],
)
)
await attributes_to_code(var, ep[CONF_NUM], cl)
await attributes_to_code(var, ep_num, cl)
return var
tests/components/zigbee/common_esp32.yaml
+11 -1
View File
@@ -8,10 +8,20 @@ binary_sensor:
- platform: template
name: "Garage Door Open 12"
report: "force"
endpoint: 1
sensor:
- platform: template
name: "Temperature Sensor"
lambda: return 10.0;
device_class: temperature
unit_of_measurement: "°C"
endpoint: 1
use_device_type: true
zigbee:
model: zigbee_test
router: true
router: false
power_source: MAINS_SINGLE_PHASE
on_join:
then:
tests/components/zigbee/test-router.esp32-c6-idf.yaml
+7
View File
@@ -0,0 +1,7 @@
zigbee:
model: zigbee_test
router: true
power_source: MAINS_SINGLE_PHASE
on_join:
then:
- logger.log: "Joined network"