Squeezebox add alarms support - switch platform. Part 1 (#141055)

* initial

* remove dupe name definition

* snapshot update

* name def updates

* test update for new entity name

* remove attributes

* icon translations

* merge fixes

* Snapshot update post merge

* update to class initialisation

* move entity delete to coordinator

* remove some comments

* move known_alarms to coordinator

* test_switch update for syrupy change

* listener and sets

* check self.available

* remove refresh from conftest

* test update

* test tweak

* move listener to switch platform

* updates revew

* SWITCH_DOMAIN
This commit is contained in:
peteS-UK 2025-05-26 16:41:28 +01:00 committed by GitHub
parent 3dc7b75e4b
commit 8623d96deb
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 505 additions and 3 deletions

View File

@ -61,6 +61,7 @@ PLATFORMS = [
Platform.BUTTON,
Platform.MEDIA_PLAYER,
Platform.SENSOR,
Platform.SWITCH,
Platform.UPDATE,
]

View File

@ -44,5 +44,13 @@ DEFAULT_VOLUME_STEP = 5
ATTR_ANNOUNCE_VOLUME = "announce_volume"
ATTR_ANNOUNCE_TIMEOUT = "announce_timeout"
UNPLAYABLE_TYPES = ("text", "actions")
ATTR_ALARM_ID = "alarm_id"
ATTR_DAYS_OF_WEEK = "dow"
ATTR_ENABLED = "enabled"
ATTR_REPEAT = "repeat"
ATTR_SCHEDULED_TODAY = "scheduled_today"
ATTR_TIME = "time"
ATTR_VOLUME = "volume"
ATTR_URL = "url"
UPDATE_PLUGINS_RELEASE_SUMMARY = "update_plugins_release_summary"
UPDATE_RELEASE_SUMMARY = "update_release_summary"

View File

@ -9,8 +9,10 @@ import logging
from typing import TYPE_CHECKING, Any
from pysqueezebox import Player, Server
from pysqueezebox.player import Alarm
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import format_mac
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
@ -98,11 +100,13 @@ class SqueezeBoxPlayerUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
)
self.player = player
self.available = True
self.known_alarms: set[str] = set()
self._remove_dispatcher: Callable | None = None
self.player_uuid = format_mac(player.player_id)
self.server_uuid = server_uuid
async def _async_update_data(self) -> dict[str, Any]:
"""Update Player if available, or listen for rediscovery if not."""
"""Update the Player() object if available, or listen for rediscovery if not."""
if self.available:
# Only update players available at last update, unavailable players are rediscovered instead
await self.player.async_update()
@ -115,7 +119,14 @@ class SqueezeBoxPlayerUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
self._remove_dispatcher = async_dispatcher_connect(
self.hass, SIGNAL_PLAYER_REDISCOVERED, self.rediscovered
)
return {}
alarm_dict: dict[str, Alarm] = (
{alarm["id"]: alarm for alarm in self.player.alarms}
if self.player.alarms
else {}
)
return {"alarms": alarm_dict}
@callback
def rediscovered(self, unique_id: str, connected: bool) -> None:

View File

@ -19,6 +19,22 @@
"other_player_count": {
"default": "mdi:folder-play-outline"
}
},
"switch": {
"alarms_enabled": {
"default": "mdi:alarm-check",
"state": {
"on": "mdi:alarm-check",
"off": "mdi:alarm-off"
}
},
"alarm": {
"default": "mdi:alarm",
"state": {
"on": "mdi:alarm",
"off": "mdi:alarm-off"
}
}
}
},
"services": {

View File

@ -133,6 +133,14 @@
"unit_of_measurement": "[%key:component::squeezebox::entity::sensor::player_count::unit_of_measurement%]"
}
},
"switch": {
"alarm": {
"name": "Alarm ({alarm_id})"
},
"alarms_enabled": {
"name": "Alarms enabled"
}
},
"update": {
"newversion": {
"name": "Lyrion Music Server"

View File

@ -0,0 +1,185 @@
"""Switch entity representing a Squeezebox alarm."""
import datetime
import logging
from typing import Any, cast
from pysqueezebox.player import Alarm
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.device_registry import format_mac
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.event import async_track_time_change
from .const import ATTR_ALARM_ID, DOMAIN, SIGNAL_PLAYER_DISCOVERED
from .coordinator import SqueezeBoxPlayerUpdateCoordinator
from .entity import SqueezeboxEntity
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Squeezebox alarm switch."""
async def _player_discovered(
coordinator: SqueezeBoxPlayerUpdateCoordinator,
) -> None:
def _async_listener() -> None:
"""Handle alarm creation and deletion after coordinator data update."""
new_alarms: set[str] = set()
received_alarms: set[str] = set()
if coordinator.data["alarms"] and coordinator.available:
received_alarms = set(coordinator.data["alarms"])
new_alarms = received_alarms - coordinator.known_alarms
removed_alarms = coordinator.known_alarms - received_alarms
if new_alarms:
for new_alarm in new_alarms:
coordinator.known_alarms.add(new_alarm)
_LOGGER.debug(
"Setting up alarm entity for alarm %s on player %s",
new_alarm,
coordinator.player,
)
async_add_entities([SqueezeBoxAlarmEntity(coordinator, new_alarm)])
if removed_alarms and coordinator.available:
for removed_alarm in removed_alarms:
_uid = f"{coordinator.player_uuid}_alarm_{removed_alarm}"
_LOGGER.debug(
"Alarm %s with unique_id %s needs to be deleted",
removed_alarm,
_uid,
)
entity_registry = er.async_get(hass)
_entity_id = entity_registry.async_get_entity_id(
Platform.SWITCH,
DOMAIN,
_uid,
)
if _entity_id:
entity_registry.async_remove(_entity_id)
coordinator.known_alarms.remove(removed_alarm)
_LOGGER.debug(
"Setting up alarm enabled entity for player %s", coordinator.player
)
# Add listener first for future coordinator refresh
coordinator.async_add_listener(_async_listener)
# If coordinator already has alarm data from the initial refresh,
# call the listener immediately to process existing alarms and create alarm entities.
if coordinator.data["alarms"]:
_LOGGER.debug(
"Coordinator has alarm data, calling _async_listener immediately for player %s",
coordinator.player,
)
_async_listener()
async_add_entities([SqueezeBoxAlarmsEnabledEntity(coordinator)])
entry.async_on_unload(
async_dispatcher_connect(hass, SIGNAL_PLAYER_DISCOVERED, _player_discovered)
)
class SqueezeBoxAlarmEntity(SqueezeboxEntity, SwitchEntity):
"""Representation of a Squeezebox alarm switch."""
_attr_translation_key = "alarm"
_attr_entity_category = EntityCategory.CONFIG
def __init__(
self, coordinator: SqueezeBoxPlayerUpdateCoordinator, alarm_id: str
) -> None:
"""Initialize the Squeezebox alarm switch."""
super().__init__(coordinator)
self._alarm_id = alarm_id
self._attr_translation_placeholders = {"alarm_id": self._alarm_id}
self._attr_unique_id: str = (
f"{format_mac(self._player.player_id)}_alarm_{self._alarm_id}"
)
async def async_added_to_hass(self) -> None:
"""Set up alarm switch when added to hass."""
await super().async_added_to_hass()
async def async_write_state_daily(now: datetime.datetime) -> None:
"""Update alarm state attributes each calendar day."""
_LOGGER.debug("Updating state attributes for %s", self.name)
self.async_write_ha_state()
self.async_on_remove(
async_track_time_change(
self.hass, async_write_state_daily, hour=0, minute=0, second=0
)
)
@property
def alarm(self) -> Alarm:
"""Return the alarm object."""
return self.coordinator.data["alarms"][self._alarm_id]
@property
def available(self) -> bool:
"""Return whether the alarm is available."""
return super().available and self._alarm_id in self.coordinator.data["alarms"]
@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return attributes of Squeezebox alarm switch."""
return {ATTR_ALARM_ID: str(self._alarm_id)}
@property
def is_on(self) -> bool:
"""Return the state of the switch."""
return cast(bool, self.alarm["enabled"])
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the switch."""
await self.coordinator.player.async_update_alarm(self._alarm_id, enabled=False)
await self.coordinator.async_request_refresh()
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the switch."""
await self.coordinator.player.async_update_alarm(self._alarm_id, enabled=True)
await self.coordinator.async_request_refresh()
class SqueezeBoxAlarmsEnabledEntity(SqueezeboxEntity, SwitchEntity):
"""Representation of a Squeezebox players alarms enabled master switch."""
_attr_translation_key = "alarms_enabled"
_attr_entity_category = EntityCategory.CONFIG
def __init__(self, coordinator: SqueezeBoxPlayerUpdateCoordinator) -> None:
"""Initialize the Squeezebox alarm switch."""
super().__init__(coordinator)
self._attr_unique_id: str = (
f"{format_mac(self._player.player_id)}_alarms_enabled"
)
@property
def is_on(self) -> bool:
"""Return the state of the switch."""
return cast(bool, self.coordinator.player.alarms_enabled)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the switch."""
await self.coordinator.player.async_set_alarms_enabled(False)
await self.coordinator.async_request_refresh()
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the switch."""
await self.coordinator.player.async_set_alarms_enabled(True)
await self.coordinator.async_request_refresh()

View File

@ -32,7 +32,6 @@ from homeassistant.const import CONF_HOST, CONF_PORT, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import format_mac
# from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
CONF_VOLUME_STEP = "volume_step"
@ -48,6 +47,7 @@ SERVER_UUIDS = [
TEST_MAC = ["aa:bb:cc:dd:ee:ff", "ff:ee:dd:cc:bb:aa"]
TEST_PLAYER_NAME = "Test Player"
TEST_SERVER_NAME = "Test Server"
TEST_ALARM_ID = "1"
FAKE_VALID_ITEM_ID = "1234"
FAKE_INVALID_ITEM_ID = "4321"
@ -294,6 +294,7 @@ def mock_pysqueezebox_player(uuid: str) -> MagicMock:
mock_player.image_url = None
mock_player.model = "SqueezeLite"
mock_player.creator = "Ralph Irving & Adrian Smith"
mock_player.alarms_enabled = True
return mock_player
@ -363,6 +364,47 @@ async def configure_squeezebox_media_player_button_platform(
await hass.async_block_till_done(wait_background_tasks=True)
async def configure_squeezebox_switch_platform(
hass: HomeAssistant,
config_entry: MockConfigEntry,
lms: MagicMock,
) -> None:
"""Configure a squeezebox config entry with appropriate mocks for switch."""
with (
patch(
"homeassistant.components.squeezebox.PLATFORMS",
[Platform.SWITCH],
),
patch("homeassistant.components.squeezebox.Server", return_value=lms),
):
# Set up the switch platform.
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done(wait_background_tasks=True)
@pytest.fixture
async def mock_alarms_player(
hass: HomeAssistant,
config_entry: MockConfigEntry,
lms: MagicMock,
) -> MagicMock:
"""Mock the alarms of a configured player."""
players = await lms.async_get_players()
players[0].alarms = [
{
"id": TEST_ALARM_ID,
"enabled": True,
"time": "07:00",
"dow": [0, 1, 2, 3, 4, 5, 6],
"repeat": False,
"url": "CURRENT_PLAYLIST",
"volume": 50,
},
]
await configure_squeezebox_switch_platform(hass, config_entry, lms)
return players[0]
@pytest.fixture
async def configured_player(
hass: HomeAssistant, config_entry: MockConfigEntry, lms: MagicMock

View File

@ -0,0 +1,96 @@
# serializer version: 1
# name: test_entity_registry[switch.test_player_alarm_1-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'switch.test_player_alarm_1',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Alarm (1)',
'platform': 'squeezebox',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'alarm',
'unique_id': 'aa:bb:cc:dd:ee:ff_alarm_1',
'unit_of_measurement': None,
})
# ---
# name: test_entity_registry[switch.test_player_alarm_1-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'alarm_id': '1',
'friendly_name': 'Test Player Alarm (1)',
}),
'context': <ANY>,
'entity_id': 'switch.test_player_alarm_1',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_entity_registry[switch.test_player_alarms_enabled-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'switch.test_player_alarms_enabled',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Alarms enabled',
'platform': 'squeezebox',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'alarms_enabled',
'unique_id': 'aa:bb:cc:dd:ee:ff_alarms_enabled',
'unit_of_measurement': None,
})
# ---
# name: test_entity_registry[switch.test_player_alarms_enabled-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Test Player Alarms enabled',
}),
'context': <ANY>,
'entity_id': 'switch.test_player_alarms_enabled',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---

View File

@ -0,0 +1,135 @@
"""Tests for the Squeezebox alarm switch platform."""
from datetime import timedelta
from unittest.mock import MagicMock
from freezegun.api import FrozenDateTimeFactory
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.squeezebox.const import PLAYER_UPDATE_INTERVAL
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import CONF_ENTITY_ID, SERVICE_TURN_OFF, SERVICE_TURN_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_registry import EntityRegistry
from .conftest import TEST_ALARM_ID
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
async def test_entity_registry(
hass: HomeAssistant,
entity_registry: EntityRegistry,
mock_alarms_player: MagicMock,
snapshot: SnapshotAssertion,
config_entry: MockConfigEntry,
) -> None:
"""Test squeezebox media_player entity registered in the entity registry."""
await snapshot_platform(hass, entity_registry, snapshot, config_entry.entry_id)
async def test_switch_state(
hass: HomeAssistant,
mock_alarms_player: MagicMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test the state of the switch."""
assert hass.states.get(f"switch.test_player_alarm_{TEST_ALARM_ID}").state == "on"
mock_alarms_player.alarms[0]["enabled"] = False
freezer.tick(timedelta(seconds=PLAYER_UPDATE_INTERVAL))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get(f"switch.test_player_alarm_{TEST_ALARM_ID}").state == "off"
async def test_switch_deleted(
hass: HomeAssistant,
mock_alarms_player: MagicMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test detecting switch deleted."""
assert hass.states.get(f"switch.test_player_alarm_{TEST_ALARM_ID}").state == "on"
mock_alarms_player.alarms = []
freezer.tick(timedelta(seconds=PLAYER_UPDATE_INTERVAL))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get(f"switch.test_player_alarm_{TEST_ALARM_ID}") is None
async def test_turn_on(
hass: HomeAssistant,
mock_alarms_player: MagicMock,
) -> None:
"""Test turning on the switch."""
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{CONF_ENTITY_ID: f"switch.test_player_alarm_{TEST_ALARM_ID}"},
blocking=True,
)
mock_alarms_player.async_update_alarm.assert_called_once_with(
TEST_ALARM_ID, enabled=True
)
async def test_turn_off(
hass: HomeAssistant,
mock_alarms_player: MagicMock,
) -> None:
"""Test turning on the switch."""
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{CONF_ENTITY_ID: f"switch.test_player_alarm_{TEST_ALARM_ID}"},
blocking=True,
)
mock_alarms_player.async_update_alarm.assert_called_once_with(
TEST_ALARM_ID, enabled=False
)
async def test_alarms_enabled_state(
hass: HomeAssistant,
mock_alarms_player: MagicMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test the alarms enabled switch."""
assert hass.states.get("switch.test_player_alarms_enabled").state == "on"
mock_alarms_player.alarms_enabled = False
freezer.tick(timedelta(seconds=PLAYER_UPDATE_INTERVAL))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("switch.test_player_alarms_enabled").state == "off"
async def test_alarms_enabled_turn_on(
hass: HomeAssistant,
mock_alarms_player: MagicMock,
) -> None:
"""Test turning on the alarms enabled switch."""
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{CONF_ENTITY_ID: "switch.test_player_alarms_enabled"},
blocking=True,
)
mock_alarms_player.async_set_alarms_enabled.assert_called_once_with(True)
async def test_alarms_enabled_turn_off(
hass: HomeAssistant,
mock_alarms_player: MagicMock,
) -> None:
"""Test turning off the alarms enabled switch."""
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{CONF_ENTITY_ID: "switch.test_player_alarms_enabled"},
blocking=True,
)
mock_alarms_player.async_set_alarms_enabled.assert_called_once_with(False)