Allow creating ZHA groups with specific IDs (#50781)

homeassistant/components/zha/api.py

@@ -363,6 +363,7 @@ def cv_group_member(value: Any) -> GroupMember:
     {
         vol.Required(TYPE): "zha/group/add",
         vol.Required(GROUP_NAME): cv.string,
+        vol.Optional(GROUP_ID): cv.positive_int,
         vol.Optional(ATTR_MEMBERS): vol.All(cv.ensure_list, [cv_group_member]),
     }
 )
@@ -371,7 +372,8 @@ async def websocket_add_group(hass, connection, msg):
     zha_gateway = hass.data[DATA_ZHA][DATA_ZHA_GATEWAY]
     group_name = msg[GROUP_NAME]
     members = msg.get(ATTR_MEMBERS)
-    group = await zha_gateway.async_create_zigpy_group(group_name, members)
+    group_id = msg.get(GROUP_ID)
+    group = await zha_gateway.async_create_zigpy_group(group_name, members, group_id)
     connection.send_result(msg[ID], group.group_info)
 
 

homeassistant/components/zha/core/gateway.py

@@ -616,13 +616,15 @@ class ZHAGateway:
         zha_device.update_available(True)
 
     async def async_create_zigpy_group(
-        self, name: str, members: list[GroupMember]
+        self, name: str, members: list[GroupMember], group_id: int = None
     ) -> ZhaGroupType:
         """Create a new Zigpy Zigbee group."""
         # we start with two to fill any gaps from a user removing existing groups
-        group_id = 2
-        while group_id in self.groups:
-            group_id += 1
+
+        if group_id is None:
+            group_id = 2
+            while group_id in self.groups:
+                group_id += 1
 
         # guard against group already existing
         if self.async_get_group_by_name(name) is None:

tests/components/zha/test_gateway.py

@@ -176,6 +176,24 @@ async def test_gateway_group_methods(hass, device_light_1, device_light_2, coord
             assert member.device.ieee in [device_light_1.ieee]
 
 
+async def test_gateway_create_group_with_id(hass, device_light_1, coordinator):
+    """Test creating a group with a specific ID."""
+    zha_gateway = get_zha_gateway(hass)
+    assert zha_gateway is not None
+    zha_gateway.coordinator_zha_device = coordinator
+    coordinator._zha_gateway = zha_gateway
+    device_light_1._zha_gateway = zha_gateway
+
+    zha_group = await zha_gateway.async_create_zigpy_group(
+        "Test Group", [GroupMember(device_light_1.ieee, 1)], group_id=0x1234
+    )
+    await hass.async_block_till_done()
+
+    assert len(zha_group.members) == 1
+    assert zha_group.members[0].device is device_light_1
+    assert zha_group.group_id == 0x1234
+
+
 async def test_updating_device_store(hass, zigpy_dev_basic, zha_dev_basic):
     """Test saving data after a delay."""
     zha_gateway = get_zha_gateway(hass)