media_player.py

Go to the documentation of this file.

"""Support for interfacing to iTunes API."""
 
from __future__ import annotations
 
from typing import Any
 
import requests
import voluptuous as vol
 
from homeassistant.components.media_player import (
    PLATFORM_SCHEMA as MEDIA_PLAYER_PLATFORM_SCHEMA,
    MediaPlayerEntity,
    MediaPlayerEntityFeature,
    MediaPlayerState,
    MediaType,
)
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT, CONF_SSL
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
 
DEFAULT_NAME = "iTunes"
DEFAULT_PORT = 8181
DEFAULT_SSL = False
DEFAULT_TIMEOUT = 10
DOMAIN = "itunes"
 
 
PLATFORM_SCHEMA = MEDIA_PLAYER_PLATFORM_SCHEMA.extend(
    {
        vol.Required(CONF_HOST): cv.string,
        vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
        vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
        vol.Optional(CONF_SSL, default=DEFAULT_SSL): cv.boolean,
    }
)
 
 
class Itunes:
    """The iTunes API client."""
 
    def __init__(self, host, port, use_ssl):
        """Initialize the iTunes device."""
        self.hosthost = host
        self.portport = port
        self.use_ssluse_ssl = use_ssl
 
    @property
    def _base_url(self):
        """Return the base URL for endpoints."""
        if self.use_ssluse_ssl:
            uri_scheme = "https://"
        else:
            uri_scheme = "http://"
 
        if self.portport:
            return f"{uri_scheme}{self.host}:{self.port}"
 
        return f"{uri_scheme}{self.host}"
 
    def _request(self, method, path, params=None):
        """Make the actual request and return the parsed response."""
        url = f"{self._base_url}{path}"
 
        try:
            if method == "GET":
                response = requests.get(url, timeout=DEFAULT_TIMEOUT)
            elif method in ("POST", "PUT"):
                response = requests.put(url, params, timeout=DEFAULT_TIMEOUT)
            elif method == "DELETE":
                response = requests.delete(url, timeout=DEFAULT_TIMEOUT)
 
            return response.json()
        except requests.exceptions.HTTPError:
            return {"player_state": "error"}
        except requests.exceptions.RequestException:
            return {"player_state": "offline"}
 
    def _command(self, named_command):
        """Make a request for a controlling command."""
        return self._request_request("PUT", f"/{named_command}")
 
    def now_playing(self):
        """Return the current state."""
        return self._request_request("GET", "/now_playing")
 
    def set_volume(self, level):
        """Set the volume and returns the current state, level 0-100."""
        return self._request_request("PUT", "/volume", {"level": level})
 
    def set_muted(self, muted):
        """Mute and returns the current state, muted True or False."""
        return self._request_request("PUT", "/mute", {"muted": muted})
 
    def set_shuffle(self, shuffle):
        """Set the shuffle mode, shuffle True or False."""
        return self._request_request(
            "PUT", "/shuffle", {"mode": ("songs" if shuffle else "off")}
        )
 
    def play(self):
        """Set playback to play and returns the current state."""
        return self._command_command("play")
 
    def pause(self):
        """Set playback to paused and returns the current state."""
        return self._command_command("pause")
 
    def next(self):
        """Skip to the next track and returns the current state."""
        return self._command_command("next")
 
    def previous(self):
        """Skip back and returns the current state."""
        return self._command_command("previous")
 
    def stop(self):
        """Stop playback and return the current state."""
        return self._command_command("stop")
 
    def play_playlist(self, playlist_id_or_name):
        """Set a playlist to be current and returns the current state."""
        response = self._request_request("GET", "/playlists")
        playlists = response.get("playlists", [])
 
        found_playlists = [
            playlist
            for playlist in playlists
            if (playlist_id_or_name in [playlist["name"], playlist["id"]])
        ]
 
        if found_playlists:
            playlist = found_playlists[0]
            path = f"/playlists/{playlist['id']}/play"
            return self._request_request("PUT", path)
 
        raise ValueError(f"Playlist {playlist_id_or_name} not found")
 
    def artwork_url(self):
        """Return a URL of the current track's album art."""
        return f"{self._base_url}/artwork"
 
    def airplay_devices(self):
        """Return a list of AirPlay devices."""
        return self._request_request("GET", "/airplay_devices")
 
    def airplay_device(self, device_id):
        """Return an AirPlay device."""
        return self._request_request("GET", f"/airplay_devices/{device_id}")
 
    def toggle_airplay_device(self, device_id, toggle):
        """Toggle airplay device on or off, id, toggle True or False."""
        command = "on" if toggle else "off"
        path = f"/airplay_devices/{device_id}/{command}"
        return self._request_request("PUT", path)
 
    def set_volume_airplay_device(self, device_id, level):
        """Set volume, returns current state of device, id,level 0-100."""
        path = f"/airplay_devices/{device_id}/volume"
        return self._request_request("PUT", path, {"level": level})
 
 
def setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None,
) -> None:
    """Set up the iTunes platform."""
    add_entities(
        [
            ItunesDevice(
                config.get(CONF_NAME),
                config.get(CONF_HOST),
                config.get(CONF_PORT),
                config[CONF_SSL],
                add_entities,
            )
        ]
    )
 
 
class ItunesDevice(MediaPlayerEntity):
    """Representation of an iTunes API instance."""
 
    _attr_media_content_type = MediaType.MUSIC
    _attr_supported_features = (
        MediaPlayerEntityFeature.PAUSE
        | MediaPlayerEntityFeature.VOLUME_SET
        | MediaPlayerEntityFeature.VOLUME_MUTE
        | MediaPlayerEntityFeature.PREVIOUS_TRACK
        | MediaPlayerEntityFeature.NEXT_TRACK
        | MediaPlayerEntityFeature.SEEK
        | MediaPlayerEntityFeature.PLAY_MEDIA
        | MediaPlayerEntityFeature.PLAY
        | MediaPlayerEntityFeature.TURN_OFF
        | MediaPlayerEntityFeature.SHUFFLE_SET
    )
 
    def __init__(self, name, host, port, use_ssl, add_entities):
        """Initialize the iTunes device."""
        self._name_name = name
        self._host_host = host
        self._port_port = port
        self._use_ssl_use_ssl = use_ssl
        self._add_entities_add_entities = add_entities
 
        self.clientclient = Itunes(self._host_host, self._port_port, self._use_ssl_use_ssl)
 
        self.current_volumecurrent_volume = None
        self.mutedmuted = None
        self.shuffledshuffled = None
        self.current_titlecurrent_title = None
        self.current_albumcurrent_album = None
        self.current_artistcurrent_artist = None
        self.current_playlistcurrent_playlist = None
        self.content_idcontent_id = None
 
        self.player_stateplayer_state = None
 
        self.airplay_devicesairplay_devices = {}
 
        self.updateupdate()
 
    def update_state(self, state_hash):
        """Update all the state properties with the passed in dictionary."""
        self.player_stateplayer_state = state_hash.get("player_state", None)
 
        self.current_volumecurrent_volume = state_hash.get("volume", 0)
        self.mutedmuted = state_hash.get("muted", None)
        self.current_titlecurrent_title = state_hash.get("name", None)
        self.current_albumcurrent_album = state_hash.get("album", None)
        self.current_artistcurrent_artist = state_hash.get("artist", None)
        self.current_playlistcurrent_playlist = state_hash.get("playlist", None)
        self.content_idcontent_id = state_hash.get("id", None)
 
        _shuffle = state_hash.get("shuffle", None)
        self.shuffledshuffled = _shuffle == "songs"
 
    @property
    def name(self):
        """Return the name of the device."""
        return self._name_name
 
    @property
    def state(self):
        """Return the state of the device."""
        if self.player_stateplayer_state == "offline" or self.player_stateplayer_state is None:
            return "offline"
 
        if self.player_stateplayer_state == "error":
            return "error"
 
        if self.player_stateplayer_state == "stopped":
            return MediaPlayerState.IDLE
 
        if self.player_stateplayer_state == "paused":
            return MediaPlayerState.PAUSED
 
        return MediaPlayerState.PLAYING
 
    def update(self) -> None:
        """Retrieve latest state."""
        now_playing = self.clientclient.now_playing()
        self.update_stateupdate_state(now_playing)
 
        found_devices = self.clientclient.airplay_devices()
        found_devices = found_devices.get("airplay_devices", [])
 
        new_devices = []
 
        for device_data in found_devices:
            device_id = device_data.get("id")
 
            if self.airplay_devicesairplay_devices.get(device_id):
                # update it
                airplay_device = self.airplay_devicesairplay_devices.get(device_id)
                airplay_device.update_state(device_data)
            else:
                # add it
                airplay_device = AirPlayDevice(device_id, self.clientclient)
                airplay_device.update_state(device_data)
                self.airplay_devicesairplay_devices[device_id] = airplay_device
                new_devices.append(airplay_device)
 
        if new_devices:
            self._add_entities_add_entities(new_devices)
 
    @property
    def is_volume_muted(self):
        """Boolean if volume is currently muted."""
        return self.mutedmuted
 
    @property
    def volume_level(self):
        """Volume level of the media player (0..1)."""
        return self.current_volumecurrent_volume / 100.0
 
    @property
    def media_content_id(self):
        """Content ID of current playing media."""
        return self.content_idcontent_id
 
    @property
    def media_image_url(self):
        """Image url of current playing media."""
        if (
            self.player_stateplayer_state
            in {
                MediaPlayerState.PLAYING,
                MediaPlayerState.IDLE,
                MediaPlayerState.PAUSED,
            }
            and self.current_titlecurrent_title is not None
        ):
            return f"{self.client.artwork_url()}?id={self.content_id}"
 
        return (
            "https://cloud.githubusercontent.com/assets/260/9829355"
            "/33fab972-58cf-11e5-8ea2-2ca74bdaae40.png"
        )
 
    @property
    def media_title(self):
        """Title of current playing media."""
        return self.current_titlecurrent_title
 
    @property
    def media_artist(self):
        """Artist of current playing media (Music track only)."""
        return self.current_artistcurrent_artist
 
    @property
    def media_album_name(self):
        """Album of current playing media (Music track only)."""
        return self.current_albumcurrent_album
 
    @property
    def media_playlist(self):
        """Title of the currently playing playlist."""
        return self.current_playlistcurrent_playlist
 
    @property
    def shuffle(self):
        """Boolean if shuffle is enabled."""
        return self.shuffledshuffled
 
    def set_volume_level(self, volume: float) -> None:
        """Set volume level, range 0..1."""
        response = self.clientclient.set_volume(int(volume * 100))
        self.update_stateupdate_state(response)
 
    def mute_volume(self, mute: bool) -> None:
        """Mute (true) or unmute (false) media player."""
        response = self.clientclient.set_muted(mute)
        self.update_stateupdate_state(response)
 
    def set_shuffle(self, shuffle: bool) -> None:
        """Shuffle (true) or no shuffle (false) media player."""
        response = self.clientclient.set_shuffle(shuffle)
        self.update_stateupdate_state(response)
 
    def media_play(self) -> None:
        """Send media_play command to media player."""
        response = self.clientclient.play()
        self.update_stateupdate_state(response)
 
    def media_pause(self) -> None:
        """Send media_pause command to media player."""
        response = self.clientclient.pause()
        self.update_stateupdate_state(response)
 
    def media_next_track(self) -> None:
        """Send media_next command to media player."""
        response = self.clientclient.next()
        self.update_stateupdate_state(response)
 
    def media_previous_track(self) -> None:
        """Send media_previous command media player."""
        response = self.clientclient.previous()
        self.update_stateupdate_state(response)
 
    def play_media(
        self, media_type: MediaType | str, media_id: str, **kwargs: Any
    ) -> None:
        """Send the play_media command to the media player."""
        if media_type == MediaType.PLAYLIST:
            response = self.clientclient.play_playlist(media_id)
            self.update_stateupdate_state(response)
 
    def turn_off(self) -> None:
        """Turn the media player off."""
        response = self.clientclient.stop()
        self.update_stateupdate_state(response)
 
 
class AirPlayDevice(MediaPlayerEntity):
    """Representation an AirPlay device via an iTunes API instance."""
 
    _attr_media_content_type = MediaType.MUSIC
    _attr_supported_features = (
        MediaPlayerEntityFeature.VOLUME_SET
        | MediaPlayerEntityFeature.TURN_ON
        | MediaPlayerEntityFeature.TURN_OFF
    )
 
    def __init__(self, device_id, client):
        """Initialize the AirPlay device."""
        self._id_id = device_id
        self.clientclient = client
        self.device_namedevice_name = "AirPlay"
        self.kindkind = None
        self.activeactive = False
        self.selectedselected = False
        self.volumevolume = 0
        self.supports_audiosupports_audio = False
        self.supports_videosupports_video = False
        self.player_stateplayer_state = None
 
    def update_state(self, state_hash):
        """Update all the state properties with the passed in dictionary."""
        if "player_state" in state_hash:
            self.player_stateplayer_state = state_hash.get("player_state", None)
 
        if "name" in state_hash:
            name = state_hash.get("name", "")
            self.device_namedevice_name = f"{name} AirTunes Speaker".strip()
 
        if "kind" in state_hash:
            self.kindkind = state_hash.get("kind", None)
 
        if "active" in state_hash:
            self.activeactive = state_hash.get("active", None)
 
        if "selected" in state_hash:
            self.selectedselected = state_hash.get("selected", None)
 
        if "sound_volume" in state_hash:
            self.volumevolume = state_hash.get("sound_volume", 0)
 
        if "supports_audio" in state_hash:
            self.supports_audiosupports_audio = state_hash.get("supports_audio", None)
 
        if "supports_video" in state_hash:
            self.supports_videosupports_video = state_hash.get("supports_video", None)
 
    @property
    def name(self):
        """Return the name of the device."""
        return self.device_namedevice_name
 
    @property
    def icon(self):
        """Return the icon to use in the frontend, if any."""
        if self.selectedselected is True:
            return "mdi:volume-high"
 
        return "mdi:volume-off"
 
    @property
    def state(self) -> MediaPlayerState:
        """Return the state of the device."""
        if self.selectedselected is True:
            return MediaPlayerState.ON
 
        return MediaPlayerState.OFF
 
    def update(self) -> None:
        """Retrieve latest state."""
 
    @property
    def volume_level(self):
        """Return the volume."""
        return float(self.volumevolume) / 100.0
 
    def set_volume_level(self, volume: float) -> None:
        """Set volume level, range 0..1."""
        volume = int(volume * 100)
        response = self.clientclient.set_volume_airplay_device(self._id_id, volume)
        self.update_stateupdate_state(response)
 
    def turn_on(self) -> None:
        """Select AirPlay."""
        self.update_stateupdate_state({"selected": True})
        self.schedule_update_ha_stateschedule_update_ha_state()
        response = self.clientclient.toggle_airplay_device(self._id_id, True)
        self.update_stateupdate_state(response)
 
    def turn_off(self) -> None:
        """Deselect AirPlay."""
        self.update_stateupdate_state({"selected": False})
        self.schedule_update_ha_stateschedule_update_ha_state()
        response = self.clientclient.toggle_airplay_device(self._id_id, False)
        self.update_stateupdate_state(response)