mirror of
https://github.com/natekspencer/hacs-oasis_mini.git
synced 2025-12-06 18:44:14 -05:00
296 lines
10 KiB
Python
296 lines
10 KiB
Python
"""Oasis device media player entity."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from homeassistant.components.media_player import (
|
|
MediaPlayerEnqueue,
|
|
MediaPlayerEntity,
|
|
MediaPlayerEntityDescription,
|
|
MediaPlayerEntityFeature,
|
|
MediaPlayerState,
|
|
MediaType,
|
|
RepeatMode,
|
|
)
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import ServiceValidationError
|
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
|
|
|
from . import OasisDeviceConfigEntry, setup_platform_from_coordinator
|
|
from .const import DOMAIN
|
|
from .entity import OasisDeviceEntity
|
|
from .helpers import get_track_id
|
|
from .pyoasiscontrol import OasisDevice
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
entry: OasisDeviceConfigEntry,
|
|
async_add_entities: AddEntitiesCallback,
|
|
) -> None:
|
|
"""Set up Oasis device media_players using config entry."""
|
|
|
|
def make_entities(new_devices: list[OasisDevice]):
|
|
"""
|
|
Create media player entities for the given Oasis devices.
|
|
|
|
Parameters:
|
|
new_devices (list[OasisDevice]): Devices to wrap as media player entities.
|
|
|
|
Returns:
|
|
list[OasisDeviceMediaPlayerEntity]: Media player entities corresponding to each device.
|
|
"""
|
|
return [
|
|
OasisDeviceMediaPlayerEntity(entry.runtime_data, device, DESCRIPTOR)
|
|
for device in new_devices
|
|
]
|
|
|
|
setup_platform_from_coordinator(entry, async_add_entities, make_entities)
|
|
|
|
|
|
DESCRIPTOR = MediaPlayerEntityDescription(key="oasis_mini", name=None)
|
|
|
|
|
|
class OasisDeviceMediaPlayerEntity(OasisDeviceEntity, MediaPlayerEntity):
|
|
"""Oasis device media player entity."""
|
|
|
|
_attr_media_image_remotely_accessible = True
|
|
_attr_supported_features = (
|
|
MediaPlayerEntityFeature.PAUSE
|
|
| MediaPlayerEntityFeature.PLAY
|
|
| MediaPlayerEntityFeature.STOP
|
|
| MediaPlayerEntityFeature.PREVIOUS_TRACK
|
|
| MediaPlayerEntityFeature.NEXT_TRACK
|
|
| MediaPlayerEntityFeature.PLAY_MEDIA
|
|
| MediaPlayerEntityFeature.MEDIA_ENQUEUE
|
|
| MediaPlayerEntityFeature.CLEAR_PLAYLIST
|
|
| MediaPlayerEntityFeature.REPEAT_SET
|
|
)
|
|
|
|
@property
|
|
def media_content_type(self) -> MediaType:
|
|
"""Content type of current playing media."""
|
|
return MediaType.IMAGE
|
|
|
|
@property
|
|
def media_duration(self) -> int | None:
|
|
"""Duration of current playing media in seconds."""
|
|
if (track := self.device.track) and "reduced_svg_content_new" in track:
|
|
return track["reduced_svg_content_new"]
|
|
return None
|
|
|
|
@property
|
|
def media_image_url(self) -> str | None:
|
|
"""
|
|
URL of the image representing the currently playing media.
|
|
|
|
Returns:
|
|
The image URL as a string, or `None` if no image is available.
|
|
"""
|
|
return self.device.track_image_url
|
|
|
|
@property
|
|
def media_position(self) -> int:
|
|
"""
|
|
Playback position of the current media in seconds.
|
|
|
|
Returns:
|
|
int: Position in seconds of the currently playing media.
|
|
"""
|
|
return self.device.progress
|
|
|
|
@property
|
|
def media_position_updated_at(self) -> datetime | None:
|
|
"""When was the position of the current playing media valid."""
|
|
return self.coordinator.last_updated
|
|
|
|
@property
|
|
def media_title(self) -> str | None:
|
|
"""
|
|
Provide the title of the currently playing track.
|
|
|
|
Returns:
|
|
str | None: The track title, or None if no title is available.
|
|
"""
|
|
return self.device.track_name
|
|
|
|
@property
|
|
def repeat(self) -> RepeatMode:
|
|
"""
|
|
Get the current repeat mode for the device.
|
|
|
|
Returns:
|
|
`RepeatMode.ALL` if the device is configured to repeat the playlist, `RepeatMode.OFF` otherwise.
|
|
"""
|
|
return RepeatMode.ALL if self.device.repeat_playlist else RepeatMode.OFF
|
|
|
|
@property
|
|
def state(self) -> MediaPlayerState:
|
|
"""State of the player."""
|
|
status_code = self.device.status_code
|
|
if self.device.error or status_code in (9, 11):
|
|
return MediaPlayerState.OFF
|
|
if status_code == 2:
|
|
return MediaPlayerState.IDLE
|
|
if status_code in (3, 13):
|
|
return MediaPlayerState.BUFFERING
|
|
if status_code == 4:
|
|
return MediaPlayerState.PLAYING
|
|
if status_code == 5:
|
|
return MediaPlayerState.PAUSED
|
|
if status_code == 15:
|
|
return MediaPlayerState.ON
|
|
return MediaPlayerState.IDLE
|
|
|
|
def abort_if_busy(self) -> None:
|
|
"""Abort if the device is currently busy."""
|
|
if self.device.busy:
|
|
raise ServiceValidationError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="device_busy",
|
|
translation_placeholders={"name": self._friendly_name_internal()},
|
|
)
|
|
|
|
async def async_media_pause(self) -> None:
|
|
"""
|
|
Pause playback on the device.
|
|
|
|
Raises:
|
|
ServiceValidationError: If the device is busy and cannot accept commands.
|
|
"""
|
|
self.abort_if_busy()
|
|
await self.device.async_pause()
|
|
|
|
async def async_media_play(self) -> None:
|
|
"""
|
|
Start playback on the device.
|
|
|
|
Raises:
|
|
ServiceValidationError: If the device is currently busy.
|
|
"""
|
|
self.abort_if_busy()
|
|
await self.device.async_play()
|
|
|
|
async def async_media_stop(self) -> None:
|
|
"""
|
|
Stop playback on the Oasis device.
|
|
|
|
Raises:
|
|
ServiceValidationError: If the device is currently busy.
|
|
"""
|
|
self.abort_if_busy()
|
|
await self.device.async_stop()
|
|
|
|
async def async_set_repeat(self, repeat: RepeatMode) -> None:
|
|
"""
|
|
Set the device playlist repeat behavior.
|
|
|
|
Enables or disables looping of the playlist according to the provided RepeatMode:
|
|
- RepeatMode.OFF disables playlist repeat.
|
|
- RepeatMode.ALL enables playlist repeat for the entire playlist.
|
|
- RepeatMode.ONE enables single-track repeat, except when the device is currently repeating the entire playlist; in that case the playlist repeat is disabled to preserve single-track semantics.
|
|
|
|
Parameters:
|
|
repeat (RepeatMode): The desired repeat mode to apply to the device playlist.
|
|
"""
|
|
await self.device.async_set_repeat_playlist(
|
|
repeat != RepeatMode.OFF
|
|
and not (repeat == RepeatMode.ONE and self.repeat == RepeatMode.ALL)
|
|
)
|
|
|
|
async def async_media_previous_track(self) -> None:
|
|
"""
|
|
Move playback to the previous track in the device's playlist, wrapping to the last track when currently at the first.
|
|
|
|
Raises:
|
|
ServiceValidationError: If the device is busy.
|
|
"""
|
|
self.abort_if_busy()
|
|
if (index := self.device.playlist_index - 1) < 0:
|
|
index = len(self.device.playlist) - 1
|
|
await self.device.async_change_track(index)
|
|
|
|
async def async_media_next_track(self) -> None:
|
|
"""
|
|
Advance the device to the next track in its playlist, wrapping to the first track when at the end.
|
|
|
|
Raises:
|
|
ServiceValidationError: if the device is busy.
|
|
"""
|
|
self.abort_if_busy()
|
|
if (index := self.device.playlist_index + 1) >= len(self.device.playlist):
|
|
index = 0
|
|
await self.device.async_change_track(index)
|
|
|
|
async def async_play_media(
|
|
self,
|
|
media_type: MediaType | str,
|
|
media_id: str,
|
|
enqueue: MediaPlayerEnqueue | None = None,
|
|
**kwargs: Any,
|
|
) -> None:
|
|
"""
|
|
Play or enqueue one or more Oasis tracks on the device.
|
|
|
|
Validates the media type and parses one or more track identifiers from `media_id`, then updates the device playlist according to `enqueue`. Depending on the enqueue mode the method can replace the playlist, append tracks, move appended tracks to the next play position, and optionally start playback.
|
|
|
|
Parameters:
|
|
media_type (MediaType | str): The media type being requested.
|
|
media_id (str): A comma-separated string of track identifiers.
|
|
enqueue (MediaPlayerEnqueue | None): How to insert the tracks into the playlist; if omitted defaults to NEXT.
|
|
|
|
Raises:
|
|
ServiceValidationError: If the device is busy, if `media_type` is a playlist (playlists are unsupported), or if `media_id` does not contain any valid track identifiers.
|
|
"""
|
|
self.abort_if_busy()
|
|
if media_type == MediaType.PLAYLIST:
|
|
raise ServiceValidationError(
|
|
translation_domain=DOMAIN, translation_key="playlists_unsupported"
|
|
)
|
|
else:
|
|
track = list(filter(None, map(get_track_id, media_id.split(","))))
|
|
if not track:
|
|
raise ServiceValidationError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="invalid_media",
|
|
translation_placeholders={"media": media_id},
|
|
)
|
|
|
|
device = self.device
|
|
enqueue = MediaPlayerEnqueue.NEXT if not enqueue else enqueue
|
|
if enqueue == MediaPlayerEnqueue.REPLACE:
|
|
await device.async_set_playlist(track)
|
|
else:
|
|
await device.async_add_track_to_playlist(track)
|
|
|
|
if enqueue in (MediaPlayerEnqueue.NEXT, MediaPlayerEnqueue.PLAY):
|
|
# Move track to next item in the playlist
|
|
new_tracks = 1 if isinstance(track, int) else len(track)
|
|
if (index := (len(device.playlist) - new_tracks)) != device.playlist_index:
|
|
if index != (
|
|
_next := min(
|
|
device.playlist_index + 1, len(device.playlist) - new_tracks
|
|
)
|
|
):
|
|
await device.async_move_track(index, _next)
|
|
if enqueue == MediaPlayerEnqueue.PLAY:
|
|
await device.async_change_track(_next)
|
|
|
|
if (
|
|
enqueue in (MediaPlayerEnqueue.PLAY, MediaPlayerEnqueue.REPLACE)
|
|
and device.status_code != 4
|
|
):
|
|
await device.async_play()
|
|
|
|
async def async_clear_playlist(self) -> None:
|
|
"""
|
|
Clear the device's playlist.
|
|
|
|
Raises:
|
|
ServiceValidationError: If the device is busy and cannot accept commands.
|
|
"""
|
|
self.abort_if_busy()
|
|
await self.device.async_clear_playlist()
|