home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2026-04-20 00:19:02 +02:00
Code Issues Packages Projects Releases Wiki Activity

Fix Lutron Caseta shade stuttering and improve stop functionality (#152207)

Browse Source 
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
J. Nick Koston
2025-09-14 12:35:18 -05:00
committed by GitHub
parent d2b255ba92
commit d877d6d93f
4 changed files with 363 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
homeassistant/components/lutron_caseta/cover.py
View File

@@ -1,5 +1,6 @@
"""Support for Lutron Caseta shades."""
from enum import Enum
from typing import Any
from homeassistant.components.cover import (
@@ -17,6 +18,14 @@ from .entity import LutronCasetaUpdatableEntity
from .models import LutronCasetaConfigEntry
class ShadeMovementDirection(Enum):
"""Enum for shade movement direction."""
OPENING = "opening"
CLOSING = "closing"
STOPPED = "stopped"
class LutronCasetaShade(LutronCasetaUpdatableEntity, CoverEntity):
"""Representation of a Lutron shade with open/close functionality."""
@@ -27,6 +36,8 @@ class LutronCasetaShade(LutronCasetaUpdatableEntity, CoverEntity):
| CoverEntityFeature.SET_POSITION
)
_attr_device_class = CoverDeviceClass.SHADE
_previous_position: int | None = None
_movement_direction: ShadeMovementDirection | None = None
@property
def is_closed(self) -> bool:
@@ -38,19 +49,50 @@ class LutronCasetaShade(LutronCasetaUpdatableEntity, CoverEntity):
"""Return the current position of cover."""
return self._device["current_state"]
def _handle_bridge_update(self) -> None:
"""Handle updated data from the bridge and track movement direction."""
current_position = self.current_cover_position
# Track movement direction based on position changes or endpoint status
if self._previous_position is not None:
if current_position > self._previous_position or current_position >= 100:
# Moving up or at fully open
self._movement_direction = ShadeMovementDirection.OPENING
elif current_position < self._previous_position or current_position <= 0:
# Moving down or at fully closed
self._movement_direction = ShadeMovementDirection.CLOSING
else:
# Stopped
self._movement_direction = ShadeMovementDirection.STOPPED
self._previous_position = current_position
super()._handle_bridge_update()
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close the cover."""
await self._smartbridge.lower_cover(self.device_id)
# Use set_value to avoid the stuttering issue
await self._smartbridge.set_value(self.device_id, 0)
await self.async_update()
self.async_write_ha_state()
async def async_stop_cover(self, **kwargs: Any) -> None:
"""Stop the cover."""
# Send appropriate directional command before stop to ensure it works correctly
# Use tracked direction if moving, otherwise use position-based heuristic
if self._movement_direction == ShadeMovementDirection.OPENING or (
self._movement_direction in (ShadeMovementDirection.STOPPED, None)
and self.current_cover_position >= 50
):
await self._smartbridge.raise_cover(self.device_id)
else:
await self._smartbridge.lower_cover(self.device_id)
await self._smartbridge.stop_cover(self.device_id)
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open the cover."""
await self._smartbridge.raise_cover(self.device_id)
# Use set_value to avoid the stuttering issue
await self._smartbridge.set_value(self.device_id, 100)
await self.async_update()
self.async_write_ha_state()

6
homeassistant/components/lutron_caseta/entity.py
View File

@@ -65,7 +65,11 @@ class LutronCasetaEntity(Entity):
async def async_added_to_hass(self) -> None:
"""Register callbacks."""
self._smartbridge.add_subscriber(self.device_id, self.async_write_ha_state)
self._smartbridge.add_subscriber(self.device_id, self._handle_bridge_update)
def _handle_bridge_update(self) -> None:
"""Handle updated data from the bridge."""
self.async_write_ha_state()
def _handle_none_serial(self, serial: str | int | None) -> str | int:
"""Handle None serial returned by RA3 and QSX processors."""

28
tests/components/lutron_caseta/__init__.py
View File

@@ -100,6 +100,7 @@ class MockBridge:
self.scenes = self.get_scenes()
self.devices = self.load_devices()
self.buttons = self.load_buttons()
self._subscribers: dict[str, list] = {}
async def connect(self):
"""Connect the mock bridge."""
@@ -110,10 +111,23 @@ class MockBridge:
def add_subscriber(self, device_id: str, callback_):
"""Mock a listener to be notified of state changes."""
if device_id not in self._subscribers:
self._subscribers[device_id] = []
self._subscribers[device_id].append(callback_)
def add_button_subscriber(self, button_id: str, callback_):
"""Mock a listener for button presses."""
def call_subscribers(self, device_id: str):
"""Notify subscribers of a device state change."""
if device_id in self._subscribers:
for callback in self._subscribers[device_id]:
callback()
def get_device_by_id(self, device_id: str):
"""Get a device by its ID."""
return self.devices.get(device_id)
def is_connected(self):
"""Return whether the mock bridge is connected."""
return self.is_currently_connected
@@ -309,6 +323,20 @@ class MockBridge:
def tap_button(self, button_id: str):
"""Mock a button press and release message for the given button ID."""
async def set_value(self, device_id: str, value: int) -> None:
"""Mock setting a device value."""
if device_id in self.devices:
self.devices[device_id]["current_state"] = value
async def raise_cover(self, device_id: str) -> None:
"""Mock raising a cover."""
async def lower_cover(self, device_id: str) -> None:
"""Mock lowering a cover."""
async def stop_cover(self, device_id: str) -> None:
"""Mock stopping a cover."""
async def close(self):
"""Close the mock bridge connection."""
self.is_currently_connected = False

287
tests/components/lutron_caseta/test_cover.py
View File

@@ -1,18 +1,303 @@
"""Tests for the Lutron Caseta integration."""
from typing import Any
from unittest.mock import AsyncMock
import pytest
from homeassistant.components.cover import (
DOMAIN as COVER_DOMAIN,
SERVICE_CLOSE_COVER,
SERVICE_OPEN_COVER,
SERVICE_STOP_COVER,
)
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import MockBridge, async_setup_integration
@pytest.fixture
async def mock_bridge_with_cover_mocks(hass: HomeAssistant) -> MockBridge:
"""Set up mock bridge with all cover methods mocked for testing."""
instance = MockBridge()
def factory(*args: Any, **kwargs: Any) -> MockBridge:
"""Return the mock bridge instance."""
return instance
# Patch all cover methods on the instance with AsyncMocks
instance.set_value = AsyncMock()
instance.raise_cover = AsyncMock()
instance.lower_cover = AsyncMock()
instance.stop_cover = AsyncMock()
await async_setup_integration(hass, factory)
await hass.async_block_till_done()
return instance
async def test_cover_unique_id(
hass: HomeAssistant, entity_registry: er.EntityRegistry
) -> None:
"""Test a light unique id."""
"""Test a cover unique ID."""
await async_setup_integration(hass, MockBridge)
cover_entity_id = "cover.basement_bedroom_left_shade"
# Assert that Caseta covers will have the bridge serial hash and the zone id as the uniqueID
assert entity_registry.async_get(cover_entity_id).unique_id == "000004d2_802"
async def test_cover_open_close_using_set_value(
hass: HomeAssistant, mock_bridge_with_cover_mocks: MockBridge
) -> None:
"""Test that open/close commands use set_value to avoid stuttering."""
mock_instance = mock_bridge_with_cover_mocks
cover_entity_id = "cover.basement_bedroom_left_shade"
# Test opening the cover
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: cover_entity_id},
blocking=True,
)
# Should use set_value(100) instead of raise_cover
mock_instance.set_value.assert_called_with("802", 100)
mock_instance.raise_cover.assert_not_called()
mock_instance.set_value.reset_mock()
mock_instance.lower_cover.reset_mock()
# Test closing the cover
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: cover_entity_id},
blocking=True,
)
# Should use set_value(0) instead of lower_cover
mock_instance.set_value.assert_called_with("802", 0)
mock_instance.lower_cover.assert_not_called()
async def test_cover_stop_with_direction_tracking(
hass: HomeAssistant, mock_bridge_with_cover_mocks: MockBridge
) -> None:
"""Test that stop command sends appropriate directional command first."""
mock_instance = mock_bridge_with_cover_mocks
cover_entity_id = "cover.basement_bedroom_left_shade"
# Simulate shade moving up (opening)
mock_instance.devices["802"]["current_state"] = 30
mock_instance.call_subscribers("802")
await hass.async_block_till_done()
mock_instance.devices["802"]["current_state"] = 60
mock_instance.call_subscribers("802")
await hass.async_block_till_done()
# Now stop while opening
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: cover_entity_id},
blocking=True,
)
# Should send raise_cover before stop_cover when opening
mock_instance.raise_cover.assert_called_with("802")
mock_instance.stop_cover.assert_called_with("802")
mock_instance.lower_cover.assert_not_called()
mock_instance.raise_cover.reset_mock()
mock_instance.lower_cover.reset_mock()
mock_instance.stop_cover.reset_mock()
# Simulate shade moving down (closing)
mock_instance.devices["802"]["current_state"] = 40
mock_instance.call_subscribers("802")
await hass.async_block_till_done()
mock_instance.devices["802"]["current_state"] = 20
mock_instance.call_subscribers("802")
await hass.async_block_till_done()
# Now stop while closing
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: cover_entity_id},
blocking=True,
)
# Should send lower_cover before stop_cover when closing
mock_instance.lower_cover.assert_called_with("802")
mock_instance.stop_cover.assert_called_with("802")
mock_instance.raise_cover.assert_not_called()
async def test_cover_stop_at_endpoints(
hass: HomeAssistant, mock_bridge_with_cover_mocks: MockBridge
) -> None:
"""Test stop command behavior when shade is at fully open or closed."""
mock_instance = mock_bridge_with_cover_mocks
cover_entity_id = "cover.basement_bedroom_left_shade"
# Test stop at fully open (100) - should infer it was opening
mock_instance.devices["802"]["current_state"] = 100
mock_instance.call_subscribers("802")
await hass.async_block_till_done()
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: cover_entity_id},
blocking=True,
)
# At fully open, should send raise_cover before stop
mock_instance.raise_cover.assert_called_with("802")
mock_instance.stop_cover.assert_called_with("802")
mock_instance.raise_cover.reset_mock()
mock_instance.lower_cover.reset_mock()
mock_instance.stop_cover.reset_mock()
# Test stop at fully closed (0) - should infer it was closing
mock_instance.devices["802"]["current_state"] = 0
mock_instance.call_subscribers("802")
await hass.async_block_till_done()
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: cover_entity_id},
blocking=True,
)
# At fully closed, should send lower_cover before stop
mock_instance.lower_cover.assert_called_with("802")
mock_instance.stop_cover.assert_called_with("802")
async def test_cover_position_heuristic_fallback(
hass: HomeAssistant, mock_bridge_with_cover_mocks: MockBridge
) -> None:
"""Test stop command uses position heuristic when movement direction is unknown."""
mock_instance = mock_bridge_with_cover_mocks
cover_entity_id = "cover.basement_bedroom_left_shade"
# Test stop at position < 50 with no movement
# Update the device data directly in the bridge's devices dict
mock_instance.devices["802"]["current_state"] = 30
mock_instance.call_subscribers("802")
await hass.async_block_till_done()
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: cover_entity_id},
blocking=True,
)
# Position < 50, should send lower_cover
mock_instance.lower_cover.assert_called_with("802")
mock_instance.stop_cover.assert_called_with("802")
mock_instance.raise_cover.reset_mock()
mock_instance.lower_cover.reset_mock()
mock_instance.stop_cover.reset_mock()
# Test stop at position >= 50 with no movement
mock_instance.devices["802"]["current_state"] = 70
mock_instance.call_subscribers("802")
await hass.async_block_till_done()
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: cover_entity_id},
blocking=True,
)
# Position >= 50, should send raise_cover
mock_instance.raise_cover.assert_called_with("802")
mock_instance.stop_cover.assert_called_with("802")
async def test_cover_stopped_movement_detection(
hass: HomeAssistant, mock_bridge_with_cover_mocks: MockBridge
) -> None:
"""Test that movement direction is set to STOPPED when position doesn't change."""
mock_instance = mock_bridge_with_cover_mocks
cover_entity_id = "cover.basement_bedroom_left_shade"
# Set initial position
mock_instance.devices["802"]["current_state"] = 50
mock_instance.call_subscribers("802")
await hass.async_block_till_done()
# Send same position again - should detect as stopped
mock_instance.devices["802"]["current_state"] = 50
mock_instance.call_subscribers("802")
await hass.async_block_till_done()
# Now stop command should use position heuristic (>= 50)
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: cover_entity_id},
blocking=True,
)
# Position >= 50 with STOPPED direction, should send raise_cover
mock_instance.raise_cover.assert_called_with("802")
mock_instance.stop_cover.assert_called_with("802")
async def test_cover_startup_with_shade_in_motion(
hass: HomeAssistant, mock_bridge_with_cover_mocks: MockBridge
) -> None:
"""Test stop command when HA starts with shade already in motion."""
mock_instance = mock_bridge_with_cover_mocks
cover_entity_id = "cover.basement_bedroom_left_shade"
# Shade starts at position 50 (simulating HA startup with shade in motion)
# First stop without seeing movement should use position heuristic
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: cover_entity_id},
blocking=True,
)
# Should have used position heuristic since we haven't seen movement yet
# Initial position is 100 from MockBridge, so >= 50, should send raise_cover
mock_instance.raise_cover.assert_called_with("802")
mock_instance.stop_cover.assert_called_with("802")
mock_instance.raise_cover.reset_mock()
mock_instance.stop_cover.reset_mock()
# Now simulate shade moving down (shade was actually in motion)
mock_instance.devices["802"]["current_state"] = 45
mock_instance.call_subscribers("802")
await hass.async_block_till_done()
# Now we've detected downward movement
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: cover_entity_id},
blocking=True,
)
# Should now correctly send lower_cover since we detected downward movement
mock_instance.lower_cover.assert_called_with("802")
mock_instance.stop_cover.assert_called_with("802")