""" homeassistant.components.media_player.squeezebox ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ Provides an interface to the Logitech SqueezeBox API For more details about this platform, please refer to the documentation at https://home-assistant.io/components/media_player.squeezebox/ """ import logging import telnetlib import urllib.parse from homeassistant.components.media_player import ( DOMAIN, MEDIA_TYPE_MUSIC, SUPPORT_NEXT_TRACK, SUPPORT_PAUSE, SUPPORT_PREVIOUS_TRACK, SUPPORT_SEEK, SUPPORT_TURN_OFF, SUPPORT_TURN_ON, SUPPORT_VOLUME_MUTE, SUPPORT_VOLUME_SET, MediaPlayerDevice) from homeassistant.const import ( CONF_HOST, CONF_PASSWORD, CONF_USERNAME, STATE_IDLE, STATE_OFF, STATE_PAUSED, STATE_PLAYING, STATE_UNKNOWN) _LOGGER = logging.getLogger(__name__) SUPPORT_SQUEEZEBOX = SUPPORT_PAUSE | SUPPORT_VOLUME_SET | \ SUPPORT_VOLUME_MUTE | SUPPORT_PREVIOUS_TRACK | SUPPORT_NEXT_TRACK | \ SUPPORT_SEEK | SUPPORT_TURN_ON | SUPPORT_TURN_OFF def setup_platform(hass, config, add_devices, discovery_info=None): """ Sets up the squeezebox platform. """ if not config.get(CONF_HOST): _LOGGER.error( "Missing required configuration items in %s: %s", DOMAIN, CONF_HOST) return False lms = LogitechMediaServer( config.get(CONF_HOST), config.get('port', '9090'), config.get(CONF_USERNAME), config.get(CONF_PASSWORD)) if not lms.init_success: return False add_devices(lms.create_players()) return True class LogitechMediaServer(object): """ Represents a Logitech media server. """ def __init__(self, host, port, username, password): self.host = host self.port = port self._username = username self._password = password self.http_port = self._get_http_port() self.init_success = True if self.http_port else False def _get_http_port(self): """ Get http port from media server, it is used to get cover art. """ http_port = None try: http_port = self.query('pref', 'httpport', '?') if not http_port: _LOGGER.error( "Unable to read data from server %s:%s", self.host, self.port) return return http_port except ConnectionError as ex: _LOGGER.error( "Failed to connect to server %s:%s - %s", self.host, self.port, ex) return def create_players(self): """ Create a list of SqueezeBoxDevices connected to the LMS. """ players = [] count = self.query('player', 'count', '?') for index in range(0, int(count)): player_id = self.query('player', 'id', str(index), '?') player = SqueezeBoxDevice(self, player_id) players.append(player) return players def query(self, *parameters): """ Send request and await response from server. """ telnet = telnetlib.Telnet(self.host, self.port) if self._username and self._password: telnet.write('login {username} {password}\n'.format( username=self._username, password=self._password).encode('UTF-8')) telnet.read_until(b'\n', timeout=3) message = '{}\n'.format(' '.join(parameters)) telnet.write(message.encode('UTF-8')) response = telnet.read_until(b'\n', timeout=3)\ .decode('UTF-8')\ .split(' ')[-1]\ .strip() telnet.write(b'exit\n') return urllib.parse.unquote(response) def get_player_status(self, player): """ Get ithe status of a player. """ # (title) : Song title # Requested Information # a (artist): Artist name 'artist' # d (duration): Song duration in seconds 'duration' # K (artwork_url): URL to remote artwork tags = 'adK' new_status = {} telnet = telnetlib.Telnet(self.host, self.port) telnet.write('{player} status - 1 tags:{tags}\n'.format( player=player, tags=tags ).encode('UTF-8')) response = telnet.read_until(b'\n', timeout=3)\ .decode('UTF-8')\ .split(' ') telnet.write(b'exit\n') for item in response: parts = urllib.parse.unquote(item).partition(':') new_status[parts[0]] = parts[2] return new_status # pylint: disable=too-many-instance-attributes # pylint: disable=too-many-public-methods class SqueezeBoxDevice(MediaPlayerDevice): """ Represents a SqueezeBox device. """ # pylint: disable=too-many-arguments, abstract-method def __init__(self, lms, player_id): super(SqueezeBoxDevice, self).__init__() self._lms = lms self._id = player_id self._name = self._lms.query(self._id, 'name', '?') self._status = self._lms.get_player_status(self._id) @property def name(self): """ Returns the name of the device. """ return self._name @property def state(self): """ Returns the state of the device. """ if 'power' in self._status and self._status['power'] == '0': return STATE_OFF if 'mode' in self._status: if self._status['mode'] == 'pause': return STATE_PAUSED if self._status['mode'] == 'play': return STATE_PLAYING if self._status['mode'] == 'stop': return STATE_IDLE return STATE_UNKNOWN def update(self): """ Retrieve latest state. """ self._status = self._lms.get_player_status(self._id) @property def volume_level(self): """ Volume level of the media player (0..1). """ if 'mixer volume' in self._status: return int(float(self._status['mixer volume'])) / 100.0 @property def is_volume_muted(self): if 'mixer volume' in self._status: return self._status['mixer volume'].startswith('-') @property def media_content_id(self): """ Content ID of current playing media. """ if 'current_title' in self._status: return self._status['current_title'] @property def media_content_type(self): """ Content type of current playing media. """ return MEDIA_TYPE_MUSIC @property def media_duration(self): """ Duration of current playing media in seconds. """ if 'duration' in self._status: return int(float(self._status['duration'])) @property def media_image_url(self): """ Image url of current playing media. """ if 'artwork_url' in self._status: media_url = self._status['artwork_url'] elif 'id' in self._status: media_url = ('/music/{track_id}/cover.jpg').format( track_id=self._status['id']) else: media_url = ('/music/current/cover.jpg?player={player}').format( player=self._id) base_url = 'http://{server}:{port}/'.format( server=self._lms.host, port=self._lms.http_port) return urllib.parse.urljoin(base_url, media_url) @property def media_title(self): """ Title of current playing media. """ if 'artist' in self._status and 'title' in self._status: return '{artist} - {title}'.format( artist=self._status['artist'], title=self._status['title'] ) if 'current_title' in self._status: return self._status['current_title'] @property def supported_media_commands(self): """ Flags of media commands that are supported. """ return SUPPORT_SQUEEZEBOX def turn_off(self): """ turn_off media player. """ self._lms.query(self._id, 'power', '0') self.update_ha_state() def volume_up(self): """ volume_up media player. """ self._lms.query(self._id, 'mixer', 'volume', '+5') self.update_ha_state() def volume_down(self): """ volume_down media player. """ self._lms.query(self._id, 'mixer', 'volume', '-5') self.update_ha_state() def set_volume_level(self, volume): """ set volume level, range 0..1. """ volume_percent = str(int(volume*100)) self._lms.query(self._id, 'mixer', 'volume', volume_percent) self.update_ha_state() def mute_volume(self, mute): """ mute (true) or unmute (false) media player. """ mute_numeric = '1' if mute else '0' self._lms.query(self._id, 'mixer', 'muting', mute_numeric) self.update_ha_state() def media_play_pause(self): """ media_play_pause media player. """ self._lms.query(self._id, 'pause') self.update_ha_state() def media_play(self): """ media_play media player. """ self._lms.query(self._id, 'play') self.update_ha_state() def media_pause(self): """ media_pause media player. """ self._lms.query(self._id, 'pause', '1') self.update_ha_state() def media_next_track(self): """ Send next track command. """ self._lms.query(self._id, 'playlist', 'index', '+1') self.update_ha_state() def media_previous_track(self): """ Send next track command. """ self._lms.query(self._id, 'playlist', 'index', '-1') self.update_ha_state() def media_seek(self, position): """ Send seek command. """ self._lms.query(self._id, 'time', position) self.update_ha_state() def turn_on(self): """ turn the media player on. """ self._lms.query(self._id, 'power', '1') self.update_ha_state()