mirror of
https://github.com/home-assistant/core.git
synced 2026-08-24 10:13:52 -05:00
131 lines
4.6 KiB
Python
131 lines
4.6 KiB
Python
"""Class representing Sonos alarms."""
|
|
|
|
from collections.abc import Iterator
|
|
import logging
|
|
from typing import TYPE_CHECKING, Any, override
|
|
|
|
from soco import SoCo, SoCoException
|
|
from soco.alarms import Alarm, Alarms
|
|
from soco.events_base import Event as SonosEvent
|
|
|
|
from homeassistant.helpers.dispatcher import async_dispatcher_send
|
|
|
|
from .const import SONOS_ALARMS_UPDATED, SONOS_CREATE_ALARM
|
|
from .helpers import soco_error
|
|
from .household_coordinator import SonosHouseholdCoordinator
|
|
|
|
if TYPE_CHECKING:
|
|
from .speaker import SonosSpeaker
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class SonosAlarms(SonosHouseholdCoordinator):
|
|
"""Coordinator class for Sonos alarms."""
|
|
|
|
def __init__(self, *args: Any) -> None:
|
|
"""Initialize the data."""
|
|
super().__init__(*args)
|
|
self.alarms: Alarms = Alarms()
|
|
self.created_alarm_ids: set[str] = set()
|
|
self._household_mismatch_logged = False
|
|
|
|
def __iter__(self) -> Iterator:
|
|
"""Return an iterator for the known alarms."""
|
|
return iter(self.alarms)
|
|
|
|
def get(self, alarm_id: str) -> Alarm | None:
|
|
"""Get an Alarm instance."""
|
|
return self.alarms.get(alarm_id)
|
|
|
|
@override
|
|
async def async_update_entities(
|
|
self, soco: SoCo, update_id: int | None = None
|
|
) -> None:
|
|
"""Create and update alarms entities, return success."""
|
|
updated = await self.hass.async_add_executor_job(
|
|
self.update_cache, soco, update_id
|
|
)
|
|
if not updated:
|
|
return
|
|
|
|
for alarm_id, alarm in self.alarms.alarms.items():
|
|
if alarm_id in self.created_alarm_ids:
|
|
continue
|
|
speaker = self.config_entry.runtime_data.discovered.get(alarm.zone.uid)
|
|
if speaker:
|
|
async_dispatcher_send(
|
|
self.hass, SONOS_CREATE_ALARM, speaker, [alarm_id]
|
|
)
|
|
async_dispatcher_send(self.hass, f"{SONOS_ALARMS_UPDATED}-{self.household_id}")
|
|
|
|
async def async_process_event(
|
|
self, event: SonosEvent, speaker: SonosSpeaker
|
|
) -> None:
|
|
"""Process the event payload in an async lock and update entities."""
|
|
event_id = event.variables["alarm_list_version"].split(":")[-1]
|
|
event_id = int(event_id)
|
|
async with self.cache_update_lock:
|
|
if (
|
|
self.last_processed_event_id
|
|
and event_id <= self.last_processed_event_id
|
|
):
|
|
# Skip updates if this event_id has already been seen
|
|
return
|
|
speaker.event_stats.process(event)
|
|
await self.async_update_entities(speaker.soco, event_id)
|
|
|
|
@soco_error()
|
|
@override
|
|
def update_cache(
|
|
self,
|
|
soco: SoCo,
|
|
update_id: int | None = None,
|
|
) -> bool:
|
|
"""Update cache of known alarms and return whether any were seen."""
|
|
try:
|
|
self.alarms.update(soco)
|
|
except SoCoException as err:
|
|
err_msg = str(err)
|
|
# Only catch the specific household mismatch error
|
|
if "Alarm list UID" in err_msg and "does not match" in err_msg:
|
|
if not self._household_mismatch_logged:
|
|
_LOGGER.warning(
|
|
"Sonos alarms for %s cannot be updated"
|
|
" due to a household mismatch. "
|
|
"This is a known limitation in setups"
|
|
" with multiple households. "
|
|
"You can safely ignore this warning,"
|
|
" or to silence it, remove the "
|
|
"affected household from your Sonos"
|
|
" system. Error: %s",
|
|
soco.player_name,
|
|
err_msg,
|
|
)
|
|
self._household_mismatch_logged = True
|
|
return False
|
|
# Let all other exceptions bubble up to be handled by @soco_error()
|
|
raise
|
|
|
|
if update_id and self.alarms.last_id < update_id:
|
|
# Skip updates if latest query result is outdated or lagging
|
|
return False
|
|
if (
|
|
self.last_processed_event_id
|
|
and self.alarms.last_id <= self.last_processed_event_id
|
|
):
|
|
return False
|
|
_LOGGER.debug(
|
|
"Updating processed event %s from %s (was %s)",
|
|
self.alarms.last_id,
|
|
soco,
|
|
self.last_processed_event_id,
|
|
)
|
|
self.last_processed_event_id = self.alarms.last_id
|
|
return True
|
|
|
|
@override
|
|
def add_speaker(self, soco: SoCo) -> None:
|
|
"""Update any skipped alarms when speaker is added."""
|
|
self.alarms.update_skipped(soco)
|