Source code for soco.snapshot

# -*- coding: utf-8 -*-
# pylint: disable=too-many-instance-attributes

"""Functionality to support saving and restoring the current Sonos state.

This is useful for scenarios such as when you want to switch to radio
and then back again to what was playing previously.

[docs]class Snapshot(object): """A snapshot of the current state. Note: This does not change anything to do with the configuration such as which group the speaker is in, just settings that impact what is playing, or how it is played. List of sources that may be playing using root of media_uri: | ``x-rincon-queue``: playing from Queue | ``x-sonosapi-stream``: playing a stream (eg radio) | ``x-file-cifs``: playing file | ``x-rincon``: slave zone (only change volume etc. rest from coordinator) """ def __init__(self, device, snapshot_queue=False): """ Args: device (SoCo): The device to snapshot snapshot_queue (bool): Whether the queue should be snapshotted. Defaults to `False`. Warning: It is strongly advised that you do not snapshot the queue unless you really need to as it takes a very long time to restore large queues as it is done one track at a time. """ # The device that will be snapshotted self.device = device # The values that will be stored # For all zones: self.media_uri = None self.is_coordinator = False self.is_playing_queue = False self.volume = None self.mute = None self.bass = None self.treble = None self.loudness = None # For coordinator zone playing from Queue: self.play_mode = None self.cross_fade = None self.playlist_position = None self.track_position = None # For coordinator zone playing a Stream: self.media_metadata = None # For all coordinator zones self.transport_state = None self.queue = None # Only set the queue as a list if we are going to save it if snapshot_queue: self.queue = []
[docs] def snapshot(self): """Record and store the current state of a device. Returns: bool: `True` if the device is a coordinator, `False` otherwise. Useful for determining whether playing an alert on a device will ungroup it. """ # Get information about the currently playing media media_info = self.device.avTransport.GetMediaInfo([('InstanceID', 0)]) self.media_uri = media_info['CurrentURI'] # extract source from media uri if self.media_uri.split(':')[0] != 'x-rincon': self.is_coordinator = True if self.media_uri.split(':')[0] == 'x-rincon-queue': self.is_playing_queue = True # Save the volume, mute and other sound settings self.volume = self.device.volume self.mute = self.device.mute self.bass = self.device.bass self.treble = self.device.treble self.loudness = self.device.loudness # get details required for what's playing: if self.is_playing_queue: # playing from queue - save repeat, random, cross fade, track, etc. self.play_mode = self.device.play_mode self.cross_fade = self.device.cross_fade # Get information about the currently playing track track_info = self.device.get_current_track_info() if track_info is not None: position = track_info['playlist_position'] if position != "": # save as integer self.playlist_position = int(position) self.track_position = track_info['position'] else: # playing from a stream - save media metadata self.media_metadata = media_info['CurrentURIMetaData'] # Work out what the playing state is - if a coordinator if self.is_coordinator: transport_info = self.device.get_current_transport_info() if transport_info is not None: self.transport_state = transport_info[ 'current_transport_state'] # Save of the current queue if we need to self._save_queue() # return if device is a coordinator (helps usage) return self.is_coordinator
# pylint: disable=too-many-branches
[docs] def restore(self, fade=False): """Restore the state of a device to that which was previously saved. For coordinator devices restore everything. For slave devices only restore volume etc., not transport info (transport info comes from the slave's coordinator). Args: fade (bool): Whether volume should be faded up on restore. """ if self.is_coordinator: # Start by ensuring that the speaker is paused as we don't want # things all rolling back when we are changing them, as this could # include things like audio transport_info = self.device.get_current_transport_info() if transport_info is not None: if transport_info['current_transport_state'] == 'PLAYING': self.device.pause() # Check if the queue should be restored self._restore_queue() # Reinstate what was playing if self.is_playing_queue and self.playlist_position > 0: # was playing from playlist if self.playlist_position is not None: # The position in the playlist returned by # get_current_track_info starts at 1, but when # playing from playlist, the index starts at 0 # if position > 0: self.playlist_position -= 1 self.device.play_from_queue(self.playlist_position, False) if self.track_position is not None: if self.track_position != "": # reinstate track, position, play mode, cross fade # Need to make sure there is a proper track selected first self.device.play_mode = self.play_mode self.device.cross_fade = self.cross_fade else: # was playing a stream (radio station, file, or nothing) # reinstate uri and meta data if self.media_uri != "": self.device.play_uri( self.media_uri, self.media_metadata, start=False) # For all devices: # Reinstate all the properties that are pretty easy to do self.device.mute = self.mute self.device.bass = self.bass self.device.treble = self.treble self.device.loudness = self.loudness # Reinstate volume # Can only change volume on device with fixed volume set to False # otherwise get uPnP error, so check first. Before issuing a network # command to check, fixed volume always has volume set to 100. # So only checked fixed volume if volume is 100. if self.volume == 100: fixed_vol = self.device.renderingControl.GetOutputFixed( [('InstanceID', 0)])['CurrentFixed'] else: fixed_vol = False # now set volume if not fixed if not fixed_vol: if fade: # if fade requested in restore # set volume to 0 then fade up to saved volume (non blocking) self.device.volume = 0 self.device.renderingControl.RampToVolume( [('InstanceID', 0), ('Channel', 'Master'), ('RampType', 'SLEEP_TIMER_RAMP_TYPE'), ('DesiredVolume', self.volume), ('ResetVolumeAfter', False), ('ProgramURI', '')]) else: # set volume self.device.volume = self.volume # Now everything is set, see if we need to be playing, stopped # or paused ( only for coordinators) if self.is_coordinator: if self.transport_state == 'PLAYING': elif self.transport_state == 'STOPPED': self.device.stop()
def _save_queue(self): """Save the current state of the queue.""" if self.queue is not None: # Maximum batch is 486, anything larger will still only # return 486 batch_size = 400 total = 0 num_return = batch_size # Need to get all the tracks in batches, but Only get the next # batch if all the items requested were in the last batch while num_return == batch_size: queue_items = self.device.get_queue(total, batch_size) # Check how many entries were returned num_return = len(queue_items) # Make sure the queue is not empty if num_return > 0: self.queue.append(queue_items) # Update the total that have been processed total = total + num_return def _restore_queue(self): """Restore the previous state of the queue. Note: The restore currently adds the items back into the queue using the URI, for items the Sonos system already knows about this is OK, but for other items, they may be missing some of their metadata as it will not be automatically picked up. """ if self.queue is not None: # Clear the queue so that it can be reset self.device.clear_queue() # Now loop around all the queue entries adding them for queue_group in self.queue: for queue_item in queue_group: self.device.add_uri_to_queue(queue_item.uri)