239 lines
8.2 KiB
Python
239 lines
8.2 KiB
Python
"""Support for media metadata handling."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
from typing import Any
|
|
|
|
from soco.core import (
|
|
MUSIC_SRC_AIRPLAY,
|
|
MUSIC_SRC_LINE_IN,
|
|
MUSIC_SRC_RADIO,
|
|
MUSIC_SRC_SPOTIFY_CONNECT,
|
|
MUSIC_SRC_TV,
|
|
SoCo,
|
|
)
|
|
from soco.data_structures import DidlAudioBroadcast, DidlPlaylistContainer
|
|
from soco.music_library import MusicLibrary
|
|
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers.config_validation import time_period_str
|
|
from homeassistant.helpers.dispatcher import dispatcher_send
|
|
from homeassistant.util import dt as dt_util
|
|
|
|
from .const import (
|
|
SONOS_MEDIA_UPDATED,
|
|
SONOS_STATE_PLAYING,
|
|
SONOS_STATE_TRANSITIONING,
|
|
SOURCE_AIRPLAY,
|
|
SOURCE_LINEIN,
|
|
SOURCE_SPOTIFY_CONNECT,
|
|
SOURCE_TV,
|
|
)
|
|
from .helpers import soco_error
|
|
|
|
LINEIN_SOURCES = (MUSIC_SRC_TV, MUSIC_SRC_LINE_IN)
|
|
SOURCE_MAPPING = {
|
|
MUSIC_SRC_AIRPLAY: SOURCE_AIRPLAY,
|
|
MUSIC_SRC_TV: SOURCE_TV,
|
|
MUSIC_SRC_LINE_IN: SOURCE_LINEIN,
|
|
MUSIC_SRC_SPOTIFY_CONNECT: SOURCE_SPOTIFY_CONNECT,
|
|
}
|
|
UNAVAILABLE_VALUES = {"", "NOT_IMPLEMENTED", None}
|
|
DURATION_SECONDS = "duration_in_s"
|
|
POSITION_SECONDS = "position_in_s"
|
|
|
|
|
|
def _timespan_secs(timespan: str | None) -> int | None:
|
|
"""Parse a time-span into number of seconds."""
|
|
if timespan in UNAVAILABLE_VALUES:
|
|
return None
|
|
return int(time_period_str(timespan).total_seconds()) # type: ignore[arg-type]
|
|
|
|
|
|
class SonosMedia:
|
|
"""Representation of the current Sonos media."""
|
|
|
|
def __init__(self, hass: HomeAssistant, soco: SoCo) -> None:
|
|
"""Initialize a SonosMedia."""
|
|
self.hass = hass
|
|
self.soco = soco
|
|
self.play_mode: str | None = None
|
|
self.playback_status: str | None = None
|
|
|
|
# This block is reset with clear()
|
|
self.album_name: str | None = None
|
|
self.artist: str | None = None
|
|
self.channel: str | None = None
|
|
self.duration: float | None = None
|
|
self.image_url: str | None = None
|
|
self.queue_position: int | None = None
|
|
self.queue_size: int | None = None
|
|
self.playlist_name: str | None = None
|
|
self.source_name: str | None = None
|
|
self.title: str | None = None
|
|
self.uri: str | None = None
|
|
|
|
self.position: int | None = None
|
|
self.position_updated_at: datetime.datetime | None = None
|
|
|
|
def clear(self) -> None:
|
|
"""Clear basic media info."""
|
|
self.album_name = None
|
|
self.artist = None
|
|
self.channel = None
|
|
self.duration = None
|
|
self.image_url = None
|
|
self.playlist_name = None
|
|
self.queue_position = None
|
|
self.queue_size = None
|
|
self.source_name = None
|
|
self.title = None
|
|
self.uri = None
|
|
|
|
def clear_position(self) -> None:
|
|
"""Clear the position attributes."""
|
|
self.position = None
|
|
self.position_updated_at = None
|
|
|
|
@property
|
|
def library(self) -> MusicLibrary:
|
|
"""Return the soco MusicLibrary instance."""
|
|
return self.soco.music_library
|
|
|
|
@soco_error()
|
|
def poll_track_info(self) -> dict[str, Any]:
|
|
"""Poll the speaker for current track info, add converted position values."""
|
|
track_info: dict[str, Any] = self.soco.get_current_track_info()
|
|
track_info[DURATION_SECONDS] = _timespan_secs(track_info.get("duration"))
|
|
track_info[POSITION_SECONDS] = _timespan_secs(track_info.get("position"))
|
|
return track_info
|
|
|
|
def write_media_player_states(self) -> None:
|
|
"""Send a signal to media player(s) to write new states."""
|
|
dispatcher_send(self.hass, SONOS_MEDIA_UPDATED, self.soco.uid)
|
|
|
|
def set_basic_track_info(self, update_position: bool = False) -> None:
|
|
"""Query the speaker to update media metadata and position info."""
|
|
self.clear()
|
|
|
|
track_info = self.poll_track_info()
|
|
if not track_info["uri"]:
|
|
return
|
|
self.uri = track_info["uri"]
|
|
|
|
audio_source = self.soco.music_source_from_uri(self.uri)
|
|
if source := SOURCE_MAPPING.get(audio_source):
|
|
self.source_name = source
|
|
if audio_source in LINEIN_SOURCES:
|
|
self.clear_position()
|
|
self.title = source
|
|
return
|
|
|
|
self.artist = track_info.get("artist")
|
|
self.album_name = track_info.get("album")
|
|
self.title = track_info.get("title")
|
|
self.image_url = track_info.get("album_art")
|
|
|
|
playlist_position = int(track_info.get("playlist_position", -1))
|
|
if playlist_position > 0:
|
|
self.queue_position = playlist_position
|
|
|
|
self.update_media_position(track_info, force_update=update_position)
|
|
|
|
def update_media_from_event(self, evars: dict[str, Any]) -> None:
|
|
"""Update information about currently playing media using an event payload."""
|
|
new_status = evars["transport_state"]
|
|
state_changed = new_status != self.playback_status
|
|
|
|
self.play_mode = evars["current_play_mode"]
|
|
self.playback_status = new_status
|
|
|
|
track_uri = evars["enqueued_transport_uri"] or evars["current_track_uri"]
|
|
audio_source = self.soco.music_source_from_uri(track_uri)
|
|
|
|
self.set_basic_track_info(update_position=state_changed)
|
|
|
|
if ct_md := evars["current_track_meta_data"]:
|
|
if not self.image_url:
|
|
if album_art_uri := getattr(ct_md, "album_art_uri", None):
|
|
self.image_url = self.library.build_album_art_full_uri(
|
|
album_art_uri
|
|
)
|
|
|
|
et_uri_md = evars["enqueued_transport_uri_meta_data"]
|
|
if isinstance(et_uri_md, DidlPlaylistContainer):
|
|
self.playlist_name = et_uri_md.title
|
|
|
|
if queue_size := evars.get("number_of_tracks", 0):
|
|
self.queue_size = int(queue_size)
|
|
|
|
if audio_source == MUSIC_SRC_RADIO:
|
|
if et_uri_md:
|
|
self.channel = et_uri_md.title
|
|
|
|
# Extra guards for S1 compatibility
|
|
if ct_md and hasattr(ct_md, "radio_show") and ct_md.radio_show:
|
|
radio_show = ct_md.radio_show.split(",")[0]
|
|
self.channel = " • ".join(filter(None, [self.channel, radio_show]))
|
|
|
|
if isinstance(et_uri_md, DidlAudioBroadcast):
|
|
self.title = self.title or self.channel
|
|
|
|
self.write_media_player_states()
|
|
|
|
@soco_error()
|
|
def poll_media(self) -> None:
|
|
"""Poll information about currently playing media."""
|
|
transport_info = self.soco.get_current_transport_info()
|
|
new_status = transport_info["current_transport_state"]
|
|
|
|
if new_status == SONOS_STATE_TRANSITIONING:
|
|
return
|
|
|
|
update_position = new_status != self.playback_status
|
|
self.playback_status = new_status
|
|
self.play_mode = self.soco.play_mode
|
|
|
|
self.set_basic_track_info(update_position=update_position)
|
|
|
|
self.write_media_player_states()
|
|
|
|
def update_media_position(
|
|
self, position_info: dict[str, int], force_update: bool = False
|
|
) -> None:
|
|
"""Update state when playing music tracks."""
|
|
duration = position_info.get(DURATION_SECONDS)
|
|
current_position = position_info.get(POSITION_SECONDS)
|
|
|
|
if not (duration or current_position):
|
|
self.clear_position()
|
|
return
|
|
|
|
should_update = force_update
|
|
self.duration = duration
|
|
|
|
# player started reporting position?
|
|
if current_position is not None and self.position is None:
|
|
should_update = True
|
|
|
|
# position jumped?
|
|
if current_position is not None and self.position is not None:
|
|
if self.playback_status == SONOS_STATE_PLAYING:
|
|
assert self.position_updated_at is not None
|
|
time_delta = dt_util.utcnow() - self.position_updated_at
|
|
time_diff = time_delta.total_seconds()
|
|
else:
|
|
time_diff = 0
|
|
|
|
calculated_position = self.position + time_diff
|
|
|
|
if abs(calculated_position - current_position) > 1.5:
|
|
should_update = True
|
|
|
|
if current_position is None:
|
|
self.clear_position()
|
|
elif should_update:
|
|
self.position = current_position
|
|
self.position_updated_at = dt_util.utcnow()
|