Add support for cover tilt for Shelly 2PM Gen3 (#125717)

* Add support for tilt

* Fix config

* Add test

* Increase test coverage
pull/125819/head
Maciej Bieniek 2024-09-12 09:28:36 +02:00 committed by GitHub
parent e89b258970
commit da401cafdf
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 111 additions and 0 deletions

View File

@ -9,6 +9,7 @@ from aioshelly.const import RPC_GENERATIONS
from homeassistant.components.cover import (
ATTR_POSITION,
ATTR_TILT_POSITION,
CoverDeviceClass,
CoverEntity,
CoverEntityFeature,
@ -157,6 +158,13 @@ class RpcShellyCover(ShellyRpcEntity, CoverEntity):
self._id = id_
if self.status["pos_control"]:
self._attr_supported_features |= CoverEntityFeature.SET_POSITION
if coordinator.device.config[f"cover:{id_}"].get("slat", {}).get("enable"):
self._attr_supported_features |= (
CoverEntityFeature.OPEN_TILT
| CoverEntityFeature.CLOSE_TILT
| CoverEntityFeature.STOP_TILT
| CoverEntityFeature.SET_TILT_POSITION
)
@property
def is_closed(self) -> bool | None:
@ -171,6 +179,14 @@ class RpcShellyCover(ShellyRpcEntity, CoverEntity):
return cast(int, self.status["current_pos"])
@property
def current_cover_tilt_position(self) -> int | None:
"""Return current position of cover tilt."""
if "slat_pos" not in self.status:
return None
return cast(int, self.status["slat_pos"])
@property
def is_closing(self) -> bool:
"""Return if the cover is closing."""
@ -198,3 +214,22 @@ class RpcShellyCover(ShellyRpcEntity, CoverEntity):
async def async_stop_cover(self, **_kwargs: Any) -> None:
"""Stop the cover."""
await self.call_rpc("Cover.Stop", {"id": self._id})
async def async_open_cover_tilt(self, **kwargs: Any) -> None:
"""Open the cover tilt."""
await self.call_rpc("Cover.GoToPosition", {"id": self._id, "slat_pos": 100})
async def async_close_cover_tilt(self, **kwargs: Any) -> None:
"""Close the cover tilt."""
await self.call_rpc("Cover.GoToPosition", {"id": self._id, "slat_pos": 0})
async def async_set_cover_tilt_position(self, **kwargs: Any) -> None:
"""Move the cover tilt to a specific position."""
await self.call_rpc(
"Cover.GoToPosition",
{"id": self._id, "slat_pos": kwargs[ATTR_TILT_POSITION]},
)
async def async_stop_cover_tilt(self, **kwargs: Any) -> None:
"""Stop the cover."""
await self.call_rpc("Cover.Stop", {"id": self._id})

View File

@ -1,17 +1,24 @@
"""Tests for Shelly cover platform."""
from copy import deepcopy
from unittest.mock import Mock
import pytest
from homeassistant.components.cover import (
ATTR_CURRENT_POSITION,
ATTR_CURRENT_TILT_POSITION,
ATTR_POSITION,
ATTR_TILT_POSITION,
DOMAIN as COVER_DOMAIN,
SERVICE_CLOSE_COVER,
SERVICE_CLOSE_COVER_TILT,
SERVICE_OPEN_COVER,
SERVICE_OPEN_COVER_TILT,
SERVICE_SET_COVER_POSITION,
SERVICE_SET_COVER_TILT_POSITION,
SERVICE_STOP_COVER,
SERVICE_STOP_COVER_TILT,
STATE_CLOSED,
STATE_CLOSING,
STATE_OPEN,
@ -187,3 +194,72 @@ async def test_rpc_device_no_position_control(
)
await init_integration(hass, 2)
assert hass.states.get("cover.test_cover_0").state == STATE_OPEN
async def test_rpc_cover_tilt(
hass: HomeAssistant,
mock_rpc_device: Mock,
monkeypatch: pytest.MonkeyPatch,
entity_registry: EntityRegistry,
) -> None:
"""Test RPC cover that supports tilt."""
entity_id = "cover.test_cover_0"
config = deepcopy(mock_rpc_device.config)
config["cover:0"]["slat"] = {"enable": True}
monkeypatch.setattr(mock_rpc_device, "config", config)
status = deepcopy(mock_rpc_device.status)
status["cover:0"]["slat_pos"] = 0
monkeypatch.setattr(mock_rpc_device, "status", status)
await init_integration(hass, 3)
state = hass.states.get(entity_id)
assert state.attributes[ATTR_CURRENT_TILT_POSITION] == 0
entry = entity_registry.async_get(entity_id)
assert entry
assert entry.unique_id == "123456789ABC-cover:0"
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_SET_COVER_TILT_POSITION,
{ATTR_ENTITY_ID: entity_id, ATTR_TILT_POSITION: 50},
blocking=True,
)
mutate_rpc_device_status(monkeypatch, mock_rpc_device, "cover:0", "slat_pos", 50)
mock_rpc_device.mock_update()
state = hass.states.get(entity_id)
assert state.attributes[ATTR_CURRENT_TILT_POSITION] == 50
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_OPEN_COVER_TILT,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mutate_rpc_device_status(monkeypatch, mock_rpc_device, "cover:0", "slat_pos", 100)
mock_rpc_device.mock_update()
state = hass.states.get(entity_id)
assert state.attributes[ATTR_CURRENT_TILT_POSITION] == 100
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_CLOSE_COVER_TILT,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_STOP_COVER_TILT,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mutate_rpc_device_status(monkeypatch, mock_rpc_device, "cover:0", "slat_pos", 10)
mock_rpc_device.mock_update()
state = hass.states.get(entity_id)
assert state.attributes[ATTR_CURRENT_TILT_POSITION] == 10