Add shutter group to HomematicIP Cloud (#31916)

homeassistant/components/homematicip_cloud/cover.py

@@ -7,6 +7,7 @@ from homematicip.aio.device import (
     AsyncFullFlushShutter,
     AsyncGarageDoorModuleTormatic,
 )
+from homematicip.aio.group import AsyncExtendedLinkedShutterGroup
 from homematicip.base.enums import DoorCommand, DoorState
 
 from homeassistant.components.cover import (
@@ -18,6 +19,7 @@ from homeassistant.config_entries import ConfigEntry
 from homeassistant.helpers.typing import HomeAssistantType
 
 from . import DOMAIN as HMIPC_DOMAIN, HomematicipGenericDevice
+from .hap import HomematicipHAP
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -41,6 +43,10 @@ async def async_setup_entry(
         elif isinstance(device, AsyncGarageDoorModuleTormatic):
             entities.append(HomematicipGarageDoorModuleTormatic(hap, device))
 
+    for group in hap.home.groups:
+        if isinstance(group, AsyncExtendedLinkedShutterGroup):
+            entities.append(HomematicipCoverShutterGroup(hap, group))
+
     if entities:
         async_add_entities(entities)
 
@@ -142,3 +148,12 @@ class HomematicipGarageDoorModuleTormatic(HomematicipGenericDevice, CoverDevice)
     async def async_stop_cover(self, **kwargs) -> None:
         """Stop the cover."""
         await self._device.send_door_command(DoorCommand.STOP)
+
+
+class HomematicipCoverShutterGroup(HomematicipCoverSlats, CoverDevice):
+    """Representation of a HomematicIP Cloud cover shutter group."""
+
+    def __init__(self, hap: HomematicipHAP, device, post: str = "ShutterGroup") -> None:
+        """Initialize switching group."""
+        device.modelType = f"HmIP-{post}"
+        super().__init__(hap, device, post)

tests/components/homematicip_cloud/test_cover.py

@@ -205,3 +205,139 @@ async def test_hmip_garage_door_tormatic(hass, default_mock_hap_factory):
     assert len(hmip_device.mock_calls) == service_call_counter + 5
     assert hmip_device.mock_calls[-1][0] == "send_door_command"
     assert hmip_device.mock_calls[-1][1] == (DoorCommand.STOP,)
+
+
+async def test_hmip_cover_shutter_group(hass, default_mock_hap_factory):
+    """Test HomematicipCoverShutteGroup."""
+    entity_id = "cover.rollos_shuttergroup"
+    entity_name = "Rollos ShutterGroup"
+    device_model = None
+    mock_hap = await default_mock_hap_factory.async_get_mock_hap(test_groups=["Rollos"])
+
+    ha_state, hmip_device = get_and_check_entity_basics(
+        hass, mock_hap, entity_id, entity_name, device_model
+    )
+
+    assert ha_state.state == "closed"
+    assert ha_state.attributes[ATTR_CURRENT_POSITION] == 0
+    service_call_counter = len(hmip_device.mock_calls)
+
+    await hass.services.async_call(
+        "cover", "open_cover", {"entity_id": entity_id}, blocking=True
+    )
+    assert len(hmip_device.mock_calls) == service_call_counter + 1
+    assert hmip_device.mock_calls[-1][0] == "set_shutter_level"
+    assert hmip_device.mock_calls[-1][1] == (0,)
+    await async_manipulate_test_data(hass, hmip_device, "shutterLevel", 0)
+    ha_state = hass.states.get(entity_id)
+    assert ha_state.state == STATE_OPEN
+    assert ha_state.attributes[ATTR_CURRENT_POSITION] == 100
+
+    await hass.services.async_call(
+        "cover",
+        "set_cover_position",
+        {"entity_id": entity_id, "position": "50"},
+        blocking=True,
+    )
+    assert len(hmip_device.mock_calls) == service_call_counter + 3
+    assert hmip_device.mock_calls[-1][0] == "set_shutter_level"
+    assert hmip_device.mock_calls[-1][1] == (0.5,)
+    await async_manipulate_test_data(hass, hmip_device, "shutterLevel", 0.5)
+    ha_state = hass.states.get(entity_id)
+    assert ha_state.state == STATE_OPEN
+    assert ha_state.attributes[ATTR_CURRENT_POSITION] == 50
+
+    await hass.services.async_call(
+        "cover", "close_cover", {"entity_id": entity_id}, blocking=True
+    )
+    assert len(hmip_device.mock_calls) == service_call_counter + 5
+    assert hmip_device.mock_calls[-1][0] == "set_shutter_level"
+    assert hmip_device.mock_calls[-1][1] == (1,)
+    await async_manipulate_test_data(hass, hmip_device, "shutterLevel", 1)
+    ha_state = hass.states.get(entity_id)
+    assert ha_state.state == STATE_CLOSED
+    assert ha_state.attributes[ATTR_CURRENT_POSITION] == 0
+
+    await hass.services.async_call(
+        "cover", "stop_cover", {"entity_id": entity_id}, blocking=True
+    )
+    assert len(hmip_device.mock_calls) == service_call_counter + 7
+    assert hmip_device.mock_calls[-1][0] == "set_shutter_stop"
+    assert hmip_device.mock_calls[-1][1] == ()
+
+    await async_manipulate_test_data(hass, hmip_device, "shutterLevel", None)
+    ha_state = hass.states.get(entity_id)
+    assert ha_state.state == STATE_UNKNOWN
+
+
+async def test_hmip_cover_slats_group(hass, default_mock_hap_factory):
+    """Test slats with HomematicipCoverShutteGroup."""
+    entity_id = "cover.rollos_shuttergroup"
+    entity_name = "Rollos ShutterGroup"
+    device_model = None
+    mock_hap = await default_mock_hap_factory.async_get_mock_hap(test_groups=["Rollos"])
+
+    ha_state, hmip_device = get_and_check_entity_basics(
+        hass, mock_hap, entity_id, entity_name, device_model
+    )
+    await async_manipulate_test_data(hass, hmip_device, "slatsLevel", 1)
+    ha_state = hass.states.get(entity_id)
+
+    assert ha_state.state == STATE_CLOSED
+    assert ha_state.attributes[ATTR_CURRENT_POSITION] == 0
+    assert ha_state.attributes[ATTR_CURRENT_TILT_POSITION] == 0
+    service_call_counter = len(hmip_device.mock_calls)
+
+    await hass.services.async_call(
+        "cover",
+        "set_cover_position",
+        {"entity_id": entity_id, "position": "50"},
+        blocking=True,
+    )
+    await hass.services.async_call(
+        "cover", "open_cover_tilt", {"entity_id": entity_id}, blocking=True
+    )
+
+    assert len(hmip_device.mock_calls) == service_call_counter + 2
+    assert hmip_device.mock_calls[-1][0] == "set_slats_level"
+    assert hmip_device.mock_calls[-1][1] == (0,)
+    await async_manipulate_test_data(hass, hmip_device, "shutterLevel", 0.5)
+    await async_manipulate_test_data(hass, hmip_device, "slatsLevel", 0)
+    ha_state = hass.states.get(entity_id)
+    assert ha_state.state == STATE_OPEN
+    assert ha_state.attributes[ATTR_CURRENT_POSITION] == 50
+    assert ha_state.attributes[ATTR_CURRENT_TILT_POSITION] == 100
+
+    await hass.services.async_call(
+        "cover",
+        "set_cover_tilt_position",
+        {"entity_id": entity_id, "tilt_position": "50"},
+        blocking=True,
+    )
+    assert len(hmip_device.mock_calls) == service_call_counter + 5
+    assert hmip_device.mock_calls[-1][0] == "set_slats_level"
+    assert hmip_device.mock_calls[-1][1] == (0.5,)
+    await async_manipulate_test_data(hass, hmip_device, "slatsLevel", 0.5)
+    ha_state = hass.states.get(entity_id)
+    assert ha_state.state == STATE_OPEN
+    assert ha_state.attributes[ATTR_CURRENT_POSITION] == 50
+    assert ha_state.attributes[ATTR_CURRENT_TILT_POSITION] == 50
+
+    await hass.services.async_call(
+        "cover", "close_cover_tilt", {"entity_id": entity_id}, blocking=True
+    )
+    assert len(hmip_device.mock_calls) == service_call_counter + 7
+    assert hmip_device.mock_calls[-1][0] == "set_slats_level"
+    assert hmip_device.mock_calls[-1][1] == (1,)
+    await async_manipulate_test_data(hass, hmip_device, "slatsLevel", 1)
+    ha_state = hass.states.get(entity_id)
+    assert ha_state.state == STATE_OPEN
+    assert ha_state.attributes[ATTR_CURRENT_POSITION] == 50
+    assert ha_state.attributes[ATTR_CURRENT_TILT_POSITION] == 0
+
+    await hass.services.async_call(
+        "cover", "stop_cover_tilt", {"entity_id": entity_id}, blocking=True
+    )
+    assert len(hmip_device.mock_calls) == service_call_counter + 9
+    assert hmip_device.mock_calls[-1][0] == "set_shutter_stop"
+    assert hmip_device.mock_calls[-1][1] == ()