media_player.py

Go to the documentation of this file.

"""Play media via gstreamer."""
 
from __future__ import annotations
 
import logging
from typing import Any
 
from gsp import STATE_IDLE, STATE_PAUSED, STATE_PLAYING, GstreamerPlayer
import voluptuous as vol
 
from homeassistant.components import media_source
from homeassistant.components.media_player import (
    PLATFORM_SCHEMA as MEDIA_PLAYER_PLATFORM_SCHEMA,
    BrowseMedia,
    MediaPlayerEntity,
    MediaPlayerEntityFeature,
    MediaPlayerState,
    MediaType,
    async_process_play_media_url,
)
from homeassistant.const import CONF_NAME, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 
_LOGGER = logging.getLogger(__name__)
 
CONF_PIPELINE = "pipeline"
 
DOMAIN = "gstreamer"
 
PLATFORM_SCHEMA = MEDIA_PLAYER_PLATFORM_SCHEMA.extend(
    {vol.Optional(CONF_NAME): cv.string, vol.Optional(CONF_PIPELINE): cv.string}
)
 
GSP_STATE_MAPPING = {
    STATE_IDLE: MediaPlayerState.IDLE,
    STATE_PAUSED: MediaPlayerState.PAUSED,
    STATE_PLAYING: MediaPlayerState.PLAYING,
}
 
 
def setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None,
) -> None:
    """Set up the Gstreamer platform."""
 
    name = config.get(CONF_NAME)
    pipeline = config.get(CONF_PIPELINE)
    player = GstreamerPlayer(pipeline)
 
    def _shutdown(call):
        """Quit the player on shutdown."""
        player.quit()
 
    hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, _shutdown)
    add_entities([GstreamerDevice(player, name)])
 
 
class GstreamerDevice(MediaPlayerEntity):
    """Representation of a Gstreamer device."""
 
    _attr_media_content_type = MediaType.MUSIC
    _attr_supported_features = (
        MediaPlayerEntityFeature.VOLUME_SET
        | MediaPlayerEntityFeature.PLAY
        | MediaPlayerEntityFeature.PAUSE
        | MediaPlayerEntityFeature.PLAY_MEDIA
        | MediaPlayerEntityFeature.NEXT_TRACK
        | MediaPlayerEntityFeature.BROWSE_MEDIA
    )
 
    def __init__(self, player: GstreamerPlayer, name: str | None) -> None:
        """Initialize the Gstreamer device."""
        self._player_player = player
        self._name_name = name or DOMAIN
        self._attr_state_attr_state = MediaPlayerState.IDLE
        self._volume_volume = None
        self._duration_duration = None
        self._uri_uri = None
        self._title_title = None
        self._artist_artist = None
        self._album_album = None
 
    def update(self) -> None:
        """Update properties."""
        self._attr_state_attr_state = GSP_STATE_MAPPING.get(self._player_player.state)
        self._volume_volume = self._player_player.volume
        self._duration_duration = self._player_player.duration
        self._uri_uri = self._player_player.uri
        self._title_title = self._player_player.title
        self._album_album = self._player_player.album
        self._artist_artist = self._player_player.artist
 
    def set_volume_level(self, volume: float) -> None:
        """Set the volume level."""
        self._player_player.volume = volume
 
    async def async_play_media(
        self, media_type: MediaType | str, media_id: str, **kwargs: Any
    ) -> None:
        """Play media."""
        # Handle media_source
        if media_source.is_media_source_id(media_id):
            sourced_media = await media_source.async_resolve_media(
                self.hasshass, media_id, self.entity_identity_id
            )
            media_id = sourced_media.url
 
        elif media_type != MediaType.MUSIC:
            _LOGGER.error("Invalid media type")
            return
 
        media_id = async_process_play_media_url(self.hasshass, media_id)
 
        await self.hasshass.async_add_executor_job(self._player_player.queue, media_id)
 
    def media_play(self) -> None:
        """Play."""
        self._player_player.play()
 
    def media_pause(self) -> None:
        """Pause."""
        self._player_player.pause()
 
    def media_next_track(self) -> None:
        """Next track."""
        self._player_player.next()
 
    @property
    def media_content_id(self):
        """Content ID of currently playing media."""
        return self._uri_uri
 
    @property
    def name(self):
        """Return the name of the device."""
        return self._name_name
 
    @property
    def volume_level(self):
        """Return the volume level."""
        return self._volume_volume
 
    @property
    def media_duration(self):
        """Duration of current playing media in seconds."""
        return self._duration_duration
 
    @property
    def media_title(self):
        """Media title."""
        return self._title_title
 
    @property
    def media_artist(self):
        """Media artist."""
        return self._artist_artist
 
    @property
    def media_album_name(self):
        """Media album."""
        return self._album_album
 
    async def async_browse_media(
        self,
        media_content_type: MediaType | str | None = None,
        media_content_id: str | None = None,
    ) -> BrowseMedia:
        """Implement the websocket media browsing helper."""
        return await media_source.async_browse_media(
            self.hasshass,
            media_content_id,
            content_filter=lambda item: item.media_content_type.startswith("audio/"),
        )