Add support for more cover devices in Fibaro (#146486)

This commit is contained in:
rappenze 2025-06-11 19:56:38 +02:00 committed by GitHub
parent 864e440685
commit 59aba339d8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 329 additions and 48 deletions

View File

@ -28,45 +28,36 @@ async def async_setup_entry(
) -> None: ) -> None:
"""Set up the Fibaro covers.""" """Set up the Fibaro covers."""
controller = entry.runtime_data controller = entry.runtime_data
async_add_entities(
[FibaroCover(device) for device in controller.fibaro_devices[Platform.COVER]], entities: list[FibaroEntity] = []
True, for device in controller.fibaro_devices[Platform.COVER]:
) # Positionable covers report the position over value
if device.value.has_value:
entities.append(PositionableFibaroCover(device))
else:
entities.append(FibaroCover(device))
async_add_entities(entities, True)
class FibaroCover(FibaroEntity, CoverEntity): class PositionableFibaroCover(FibaroEntity, CoverEntity):
"""Representation a Fibaro Cover.""" """Representation of a fibaro cover which supports positioning."""
def __init__(self, fibaro_device: DeviceModel) -> None: def __init__(self, fibaro_device: DeviceModel) -> None:
"""Initialize the Vera device.""" """Initialize the device."""
super().__init__(fibaro_device) super().__init__(fibaro_device)
self.entity_id = ENTITY_ID_FORMAT.format(self.ha_id) self.entity_id = ENTITY_ID_FORMAT.format(self.ha_id)
if self._is_open_close_only():
self._attr_supported_features = (
CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE
)
if "stop" in self.fibaro_device.actions:
self._attr_supported_features |= CoverEntityFeature.STOP
@staticmethod @staticmethod
def bound(position): def bound(position: int | None) -> int | None:
"""Normalize the position.""" """Normalize the position."""
if position is None: if position is None:
return None return None
position = int(position)
if position <= 5: if position <= 5:
return 0 return 0
if position >= 95: if position >= 95:
return 100 return 100
return position return position
def _is_open_close_only(self) -> bool:
"""Return if only open / close is supported."""
# Normally positionable devices report the position over value,
# so if it is missing we have a device which supports open / close only
return not self.fibaro_device.value.has_value
def update(self) -> None: def update(self) -> None:
"""Update the state.""" """Update the state."""
super().update() super().update()
@ -74,20 +65,15 @@ class FibaroCover(FibaroEntity, CoverEntity):
self._attr_current_cover_position = self.bound(self.level) self._attr_current_cover_position = self.bound(self.level)
self._attr_current_cover_tilt_position = self.bound(self.level2) self._attr_current_cover_tilt_position = self.bound(self.level2)
device_state = self.fibaro_device.state
# Be aware that opening and closing is only available for some modern # Be aware that opening and closing is only available for some modern
# devices. # devices.
# For example the Fibaro Roller Shutter 4 reports this correctly. # For example the Fibaro Roller Shutter 4 reports this correctly.
if device_state.has_value: device_state = self.fibaro_device.state.str_value(default="").lower()
self._attr_is_opening = device_state.str_value().lower() == "opening" self._attr_is_opening = device_state == "opening"
self._attr_is_closing = device_state.str_value().lower() == "closing" self._attr_is_closing = device_state == "closing"
closed: bool | None = None closed: bool | None = None
if self._is_open_close_only(): if self.current_cover_position is not None:
if device_state.has_value and device_state.str_value().lower() != "unknown":
closed = device_state.str_value().lower() == "closed"
elif self.current_cover_position is not None:
closed = self.current_cover_position == 0 closed = self.current_cover_position == 0
self._attr_is_closed = closed self._attr_is_closed = closed
@ -96,7 +82,7 @@ class FibaroCover(FibaroEntity, CoverEntity):
self.set_level(cast(int, kwargs.get(ATTR_POSITION))) self.set_level(cast(int, kwargs.get(ATTR_POSITION)))
def set_cover_tilt_position(self, **kwargs: Any) -> None: def set_cover_tilt_position(self, **kwargs: Any) -> None:
"""Move the cover to a specific position.""" """Move the slats to a specific position."""
self.set_level2(cast(int, kwargs.get(ATTR_TILT_POSITION))) self.set_level2(cast(int, kwargs.get(ATTR_TILT_POSITION)))
def open_cover(self, **kwargs: Any) -> None: def open_cover(self, **kwargs: Any) -> None:
@ -118,3 +104,62 @@ class FibaroCover(FibaroEntity, CoverEntity):
def stop_cover(self, **kwargs: Any) -> None: def stop_cover(self, **kwargs: Any) -> None:
"""Stop the cover.""" """Stop the cover."""
self.action("stop") self.action("stop")
class FibaroCover(FibaroEntity, CoverEntity):
"""Representation of a fibaro cover which supports only open / close commands."""
def __init__(self, fibaro_device: DeviceModel) -> None:
"""Initialize the device."""
super().__init__(fibaro_device)
self.entity_id = ENTITY_ID_FORMAT.format(self.ha_id)
self._attr_supported_features = (
CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE
)
if "stop" in self.fibaro_device.actions:
self._attr_supported_features |= CoverEntityFeature.STOP
if "rotateSlatsUp" in self.fibaro_device.actions:
self._attr_supported_features |= CoverEntityFeature.OPEN_TILT
if "rotateSlatsDown" in self.fibaro_device.actions:
self._attr_supported_features |= CoverEntityFeature.CLOSE_TILT
if "stopSlats" in self.fibaro_device.actions:
self._attr_supported_features |= CoverEntityFeature.STOP_TILT
def update(self) -> None:
"""Update the state."""
super().update()
device_state = self.fibaro_device.state.str_value(default="").lower()
self._attr_is_opening = device_state == "opening"
self._attr_is_closing = device_state == "closing"
closed: bool | None = None
if device_state not in {"", "unknown"}:
closed = device_state == "closed"
self._attr_is_closed = closed
def open_cover(self, **kwargs: Any) -> None:
"""Open the cover."""
self.action("open")
def close_cover(self, **kwargs: Any) -> None:
"""Close the cover."""
self.action("close")
def stop_cover(self, **kwargs: Any) -> None:
"""Stop the cover."""
self.action("stop")
def open_cover_tilt(self, **kwargs: Any) -> None:
"""Open the cover slats."""
self.action("rotateSlatsUp")
def close_cover_tilt(self, **kwargs: Any) -> None:
"""Close the cover slats."""
self.action("rotateSlatsDown")
def stop_cover_tilt(self, **kwargs: Any) -> None:
"""Stop the cover slats turning."""
self.action("stopSlats")

View File

@ -83,8 +83,8 @@ def mock_power_sensor() -> Mock:
@pytest.fixture @pytest.fixture
def mock_cover() -> Mock: def mock_positionable_cover() -> Mock:
"""Fixture for a cover.""" """Fixture for a positionable cover."""
cover = Mock() cover = Mock()
cover.fibaro_id = 3 cover.fibaro_id = 3
cover.parent_fibaro_id = 0 cover.parent_fibaro_id = 0
@ -112,6 +112,42 @@ def mock_cover() -> Mock:
return cover return cover
@pytest.fixture
def mock_cover() -> Mock:
"""Fixture for a cover supporting slats but without positioning."""
cover = Mock()
cover.fibaro_id = 4
cover.parent_fibaro_id = 0
cover.name = "Test cover"
cover.room_id = 1
cover.dead = False
cover.visible = True
cover.enabled = True
cover.type = "com.fibaro.baseShutter"
cover.base_type = "com.fibaro.actor"
cover.properties = {"manufacturer": ""}
cover.actions = {
"open": 0,
"close": 0,
"stop": 0,
"rotateSlatsUp": 0,
"rotateSlatsDown": 0,
"stopSlats": 0,
}
cover.supported_features = {}
value_mock = Mock()
value_mock.has_value = False
cover.value = value_mock
value2_mock = Mock()
value2_mock.has_value = False
cover.value_2 = value2_mock
state_mock = Mock()
state_mock.has_value = True
state_mock.str_value.return_value = "closed"
cover.state = state_mock
return cover
@pytest.fixture @pytest.fixture
def mock_light() -> Mock: def mock_light() -> Mock:
"""Fixture for a dimmmable light.""" """Fixture for a dimmmable light."""

View File

@ -2,7 +2,7 @@
from unittest.mock import Mock, patch from unittest.mock import Mock, patch
from homeassistant.components.cover import CoverState from homeassistant.components.cover import CoverEntityFeature, CoverState
from homeassistant.const import Platform from homeassistant.const import Platform
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er from homeassistant.helpers import entity_registry as er
@ -12,6 +12,98 @@ from .conftest import init_integration
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
async def test_positionable_cover_setup(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_fibaro_client: Mock,
mock_config_entry: MockConfigEntry,
mock_positionable_cover: Mock,
mock_room: Mock,
) -> None:
"""Test that the cover creates an entity."""
# Arrange
mock_fibaro_client.read_rooms.return_value = [mock_room]
mock_fibaro_client.read_devices.return_value = [mock_positionable_cover]
with patch("homeassistant.components.fibaro.PLATFORMS", [Platform.COVER]):
# Act
await init_integration(hass, mock_config_entry)
# Assert
entry = entity_registry.async_get("cover.room_1_test_cover_3")
assert entry
assert entry.supported_features == (
CoverEntityFeature.OPEN
| CoverEntityFeature.CLOSE
| CoverEntityFeature.STOP
| CoverEntityFeature.SET_POSITION
)
assert entry.unique_id == "hc2_111111.3"
assert entry.original_name == "Room 1 Test cover"
async def test_cover_opening(
hass: HomeAssistant,
mock_fibaro_client: Mock,
mock_config_entry: MockConfigEntry,
mock_positionable_cover: Mock,
mock_room: Mock,
) -> None:
"""Test that the cover opening state is reported."""
# Arrange
mock_fibaro_client.read_rooms.return_value = [mock_room]
mock_fibaro_client.read_devices.return_value = [mock_positionable_cover]
with patch("homeassistant.components.fibaro.PLATFORMS", [Platform.COVER]):
# Act
await init_integration(hass, mock_config_entry)
# Assert
assert hass.states.get("cover.room_1_test_cover_3").state == CoverState.OPENING
async def test_cover_opening_closing_none(
hass: HomeAssistant,
mock_fibaro_client: Mock,
mock_config_entry: MockConfigEntry,
mock_positionable_cover: Mock,
mock_room: Mock,
) -> None:
"""Test that the cover opening closing states return None if not available."""
# Arrange
mock_fibaro_client.read_rooms.return_value = [mock_room]
mock_positionable_cover.state.str_value.return_value = ""
mock_fibaro_client.read_devices.return_value = [mock_positionable_cover]
with patch("homeassistant.components.fibaro.PLATFORMS", [Platform.COVER]):
# Act
await init_integration(hass, mock_config_entry)
# Assert
assert hass.states.get("cover.room_1_test_cover_3").state == CoverState.OPEN
async def test_cover_closing(
hass: HomeAssistant,
mock_fibaro_client: Mock,
mock_config_entry: MockConfigEntry,
mock_positionable_cover: Mock,
mock_room: Mock,
) -> None:
"""Test that the cover closing state is reported."""
# Arrange
mock_fibaro_client.read_rooms.return_value = [mock_room]
mock_positionable_cover.state.str_value.return_value = "closing"
mock_fibaro_client.read_devices.return_value = [mock_positionable_cover]
with patch("homeassistant.components.fibaro.PLATFORMS", [Platform.COVER]):
# Act
await init_integration(hass, mock_config_entry)
# Assert
assert hass.states.get("cover.room_1_test_cover_3").state == CoverState.CLOSING
async def test_cover_setup( async def test_cover_setup(
hass: HomeAssistant, hass: HomeAssistant,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
@ -30,20 +122,28 @@ async def test_cover_setup(
# Act # Act
await init_integration(hass, mock_config_entry) await init_integration(hass, mock_config_entry)
# Assert # Assert
entry = entity_registry.async_get("cover.room_1_test_cover_3") entry = entity_registry.async_get("cover.room_1_test_cover_4")
assert entry assert entry
assert entry.unique_id == "hc2_111111.3" assert entry.supported_features == (
CoverEntityFeature.OPEN
| CoverEntityFeature.CLOSE
| CoverEntityFeature.STOP
| CoverEntityFeature.OPEN_TILT
| CoverEntityFeature.CLOSE_TILT
| CoverEntityFeature.STOP_TILT
)
assert entry.unique_id == "hc2_111111.4"
assert entry.original_name == "Room 1 Test cover" assert entry.original_name == "Room 1 Test cover"
async def test_cover_opening( async def test_cover_open_action(
hass: HomeAssistant, hass: HomeAssistant,
mock_fibaro_client: Mock, mock_fibaro_client: Mock,
mock_config_entry: MockConfigEntry, mock_config_entry: MockConfigEntry,
mock_cover: Mock, mock_cover: Mock,
mock_room: Mock, mock_room: Mock,
) -> None: ) -> None:
"""Test that the cover opening state is reported.""" """Test that open_cover works."""
# Arrange # Arrange
mock_fibaro_client.read_rooms.return_value = [mock_room] mock_fibaro_client.read_rooms.return_value = [mock_room]
@ -52,47 +152,147 @@ async def test_cover_opening(
with patch("homeassistant.components.fibaro.PLATFORMS", [Platform.COVER]): with patch("homeassistant.components.fibaro.PLATFORMS", [Platform.COVER]):
# Act # Act
await init_integration(hass, mock_config_entry) await init_integration(hass, mock_config_entry)
await hass.services.async_call(
"cover",
"open_cover",
{"entity_id": "cover.room_1_test_cover_4"},
blocking=True,
)
# Assert # Assert
assert hass.states.get("cover.room_1_test_cover_3").state == CoverState.OPENING mock_cover.execute_action.assert_called_once_with("open", ())
async def test_cover_opening_closing_none( async def test_cover_close_action(
hass: HomeAssistant, hass: HomeAssistant,
mock_fibaro_client: Mock, mock_fibaro_client: Mock,
mock_config_entry: MockConfigEntry, mock_config_entry: MockConfigEntry,
mock_cover: Mock, mock_cover: Mock,
mock_room: Mock, mock_room: Mock,
) -> None: ) -> None:
"""Test that the cover opening closing states return None if not available.""" """Test that close_cover works."""
# Arrange # Arrange
mock_fibaro_client.read_rooms.return_value = [mock_room] mock_fibaro_client.read_rooms.return_value = [mock_room]
mock_cover.state.has_value = False
mock_fibaro_client.read_devices.return_value = [mock_cover] mock_fibaro_client.read_devices.return_value = [mock_cover]
with patch("homeassistant.components.fibaro.PLATFORMS", [Platform.COVER]): with patch("homeassistant.components.fibaro.PLATFORMS", [Platform.COVER]):
# Act # Act
await init_integration(hass, mock_config_entry) await init_integration(hass, mock_config_entry)
await hass.services.async_call(
"cover",
"close_cover",
{"entity_id": "cover.room_1_test_cover_4"},
blocking=True,
)
# Assert # Assert
assert hass.states.get("cover.room_1_test_cover_3").state == CoverState.OPEN mock_cover.execute_action.assert_called_once_with("close", ())
async def test_cover_closing( async def test_cover_stop_action(
hass: HomeAssistant, hass: HomeAssistant,
mock_fibaro_client: Mock, mock_fibaro_client: Mock,
mock_config_entry: MockConfigEntry, mock_config_entry: MockConfigEntry,
mock_cover: Mock, mock_cover: Mock,
mock_room: Mock, mock_room: Mock,
) -> None: ) -> None:
"""Test that the cover closing state is reported.""" """Test that stop_cover works."""
# Arrange # Arrange
mock_fibaro_client.read_rooms.return_value = [mock_room] mock_fibaro_client.read_rooms.return_value = [mock_room]
mock_cover.state.str_value.return_value = "closing"
mock_fibaro_client.read_devices.return_value = [mock_cover] mock_fibaro_client.read_devices.return_value = [mock_cover]
with patch("homeassistant.components.fibaro.PLATFORMS", [Platform.COVER]): with patch("homeassistant.components.fibaro.PLATFORMS", [Platform.COVER]):
# Act # Act
await init_integration(hass, mock_config_entry) await init_integration(hass, mock_config_entry)
await hass.services.async_call(
"cover",
"stop_cover",
{"entity_id": "cover.room_1_test_cover_4"},
blocking=True,
)
# Assert # Assert
assert hass.states.get("cover.room_1_test_cover_3").state == CoverState.CLOSING mock_cover.execute_action.assert_called_once_with("stop", ())
async def test_cover_open_slats_action(
hass: HomeAssistant,
mock_fibaro_client: Mock,
mock_config_entry: MockConfigEntry,
mock_cover: Mock,
mock_room: Mock,
) -> None:
"""Test that open_cover_tilt works."""
# Arrange
mock_fibaro_client.read_rooms.return_value = [mock_room]
mock_fibaro_client.read_devices.return_value = [mock_cover]
with patch("homeassistant.components.fibaro.PLATFORMS", [Platform.COVER]):
# Act
await init_integration(hass, mock_config_entry)
await hass.services.async_call(
"cover",
"open_cover_tilt",
{"entity_id": "cover.room_1_test_cover_4"},
blocking=True,
)
# Assert
mock_cover.execute_action.assert_called_once_with("rotateSlatsUp", ())
async def test_cover_close_tilt_action(
hass: HomeAssistant,
mock_fibaro_client: Mock,
mock_config_entry: MockConfigEntry,
mock_cover: Mock,
mock_room: Mock,
) -> None:
"""Test that close_cover_tilt works."""
# Arrange
mock_fibaro_client.read_rooms.return_value = [mock_room]
mock_fibaro_client.read_devices.return_value = [mock_cover]
with patch("homeassistant.components.fibaro.PLATFORMS", [Platform.COVER]):
# Act
await init_integration(hass, mock_config_entry)
await hass.services.async_call(
"cover",
"close_cover_tilt",
{"entity_id": "cover.room_1_test_cover_4"},
blocking=True,
)
# Assert
mock_cover.execute_action.assert_called_once_with("rotateSlatsDown", ())
async def test_cover_stop_slats_action(
hass: HomeAssistant,
mock_fibaro_client: Mock,
mock_config_entry: MockConfigEntry,
mock_cover: Mock,
mock_room: Mock,
) -> None:
"""Test that stop_cover_tilt works."""
# Arrange
mock_fibaro_client.read_rooms.return_value = [mock_room]
mock_fibaro_client.read_devices.return_value = [mock_cover]
with patch("homeassistant.components.fibaro.PLATFORMS", [Platform.COVER]):
# Act
await init_integration(hass, mock_config_entry)
await hass.services.async_call(
"cover",
"stop_cover_tilt",
{"entity_id": "cover.room_1_test_cover_4"},
blocking=True,
)
# Assert
mock_cover.execute_action.assert_called_once_with("stopSlats", ())