Source code for mopidy.core.playback

from __future__ import unicode_literals

import logging
import urlparse

from mopidy.audio import PlaybackState

from . import listener


logger = logging.getLogger(__name__)


[docs]class PlaybackController(object): pykka_traversable = True def __init__(self, audio, backends, core): self.audio = audio self.backends = backends self.core = core self._state = PlaybackState.STOPPED self._volume = None self._mute = False def _get_backend(self): if self.current_tl_track is None: return None uri = self.current_tl_track.track.uri uri_scheme = urlparse.urlparse(uri).scheme return self.backends.with_playback.get(uri_scheme, None) ### Properties def get_current_tl_track(self): return self.current_tl_track current_tl_track = None """ The currently playing or selected :class:`mopidy.models.TlTrack`, or :class:`None`. """ def get_current_track(self): return self.current_tl_track and self.current_tl_track.track current_track = property(get_current_track) """ The currently playing or selected :class:`mopidy.models.Track`. Read-only. Extracted from :attr:`current_tl_track` for convenience. """ def get_state(self): return self._state def set_state(self, new_state): (old_state, self._state) = (self.state, new_state) logger.debug('Changing state: %s -> %s', old_state, new_state) self._trigger_playback_state_changed(old_state, new_state) state = property(get_state, set_state) """ The playback state. Must be :attr:`PLAYING`, :attr:`PAUSED`, or :attr:`STOPPED`. Possible states and transitions: .. digraph:: state_transitions "STOPPED" -> "PLAYING" [ label="play" ] "STOPPED" -> "PAUSED" [ label="pause" ] "PLAYING" -> "STOPPED" [ label="stop" ] "PLAYING" -> "PAUSED" [ label="pause" ] "PLAYING" -> "PLAYING" [ label="play" ] "PAUSED" -> "PLAYING" [ label="resume" ] "PAUSED" -> "STOPPED" [ label="stop" ] """ def get_time_position(self): backend = self._get_backend() if backend: return backend.playback.get_time_position().get() else: return 0 time_position = property(get_time_position) """Time position in milliseconds.""" def get_volume(self): if self.audio: return self.audio.get_volume().get() else: # For testing return self._volume def set_volume(self, volume): if self.audio: self.audio.set_volume(volume) else: # For testing self._volume = volume self._trigger_volume_changed(volume) volume = property(get_volume, set_volume) """Volume as int in range [0..100] or :class:`None`""" def get_mute(self): if self.audio: return self.audio.get_mute().get() else: # For testing return self._mute def set_mute(self, value): value = bool(value) if self.audio: self.audio.set_mute(value) else: # For testing self._mute = value self._trigger_mute_changed(value) mute = property(get_mute, set_mute) """Mute state as a :class:`True` if muted, :class:`False` otherwise""" ### Methods
[docs] def change_track(self, tl_track, on_error_step=1): """ Change to the given track, keeping the current playback state. :param tl_track: track to change to :type tl_track: :class:`mopidy.models.TlTrack` or :class:`None` :param on_error_step: direction to step at play error, 1 for next track (default), -1 for previous track :type on_error_step: int, -1 or 1 """ old_state = self.state self.stop() self.current_tl_track = tl_track if old_state == PlaybackState.PLAYING: self.play(on_error_step=on_error_step) elif old_state == PlaybackState.PAUSED: self.pause()
[docs] def on_end_of_track(self): """ Tell the playback controller that end of track is reached. Used by event handler in :class:`mopidy.core.Core`. """ if self.state == PlaybackState.STOPPED: return original_tl_track = self.current_tl_track next_tl_track = self.core.tracklist.eot_track(original_tl_track) if next_tl_track: self.change_track(next_tl_track) else: self.stop(clear_current_track=True) self.core.tracklist.mark_played(original_tl_track)
[docs] def on_tracklist_change(self): """ Tell the playback controller that the current playlist has changed. Used by :class:`mopidy.core.TracklistController`. """ if self.current_tl_track not in self.core.tracklist.tl_tracks: self.stop(clear_current_track=True)
[docs] def next(self): """ Change to the next track. The current playback state will be kept. If it was playing, playing will continue. If it was paused, it will still be paused, etc. """ tl_track = self.core.tracklist.next_track(self.current_tl_track) if tl_track: self.change_track(tl_track) else: self.stop(clear_current_track=True)
[docs] def pause(self): """Pause playback.""" backend = self._get_backend() if not backend or backend.playback.pause().get(): self.state = PlaybackState.PAUSED self._trigger_track_playback_paused()
[docs] def play(self, tl_track=None, on_error_step=1): """ Play the given track, or if the given track is :class:`None`, play the currently active track. :param tl_track: track to play :type tl_track: :class:`mopidy.models.TlTrack` or :class:`None` :param on_error_step: direction to step at play error, 1 for next track (default), -1 for previous track :type on_error_step: int, -1 or 1 """ assert on_error_step in (-1, 1) if tl_track is None: if self.state == PlaybackState.PAUSED: return self.resume() if self.current_tl_track is not None: tl_track = self.current_tl_track else: if on_error_step == 1: tl_track = self.core.tracklist.next_track(tl_track) elif on_error_step == -1: tl_track = self.core.tracklist.previous_track(tl_track) if tl_track is None: return assert tl_track in self.core.tracklist.tl_tracks if self.state == PlaybackState.PLAYING: self.stop() self.current_tl_track = tl_track self.state = PlaybackState.PLAYING backend = self._get_backend() success = backend and backend.playback.play(tl_track.track).get() if success: self.core.tracklist.mark_playing(tl_track) self._trigger_track_playback_started() else: self.core.tracklist.mark_unplayable(tl_track) if on_error_step == 1: # TODO: can cause an endless loop for single track repeat. self.next() elif on_error_step == -1: self.previous()
[docs] def previous(self): """ Change to the previous track. The current playback state will be kept. If it was playing, playing will continue. If it was paused, it will still be paused, etc. """ tl_track = self.current_tl_track self.change_track( self.core.tracklist.previous_track(tl_track), on_error_step=-1)
[docs] def resume(self): """If paused, resume playing the current track.""" if self.state != PlaybackState.PAUSED: return backend = self._get_backend() if backend and backend.playback.resume().get(): self.state = PlaybackState.PLAYING self._trigger_track_playback_resumed()
[docs] def seek(self, time_position): """ Seeks to time position given in milliseconds. :param time_position: time position in milliseconds :type time_position: int :rtype: :class:`True` if successful, else :class:`False` """ if not self.core.tracklist.tracks: return False if self.state == PlaybackState.STOPPED: self.play() elif self.state == PlaybackState.PAUSED: self.resume() if time_position < 0: time_position = 0 elif time_position > self.current_track.length: self.next() return True backend = self._get_backend() if not backend: return False success = backend.playback.seek(time_position).get() if success: self._trigger_seeked(time_position) return success
[docs] def stop(self, clear_current_track=False): """ Stop playing. :param clear_current_track: whether to clear the current track _after_ stopping :type clear_current_track: boolean """ if self.state != PlaybackState.STOPPED: backend = self._get_backend() time_position_before_stop = self.time_position if not backend or backend.playback.stop().get(): self.state = PlaybackState.STOPPED self._trigger_track_playback_ended(time_position_before_stop) if clear_current_track: self.current_tl_track = None
def _trigger_track_playback_paused(self): logger.debug('Triggering track playback paused event') if self.current_track is None: return listener.CoreListener.send( 'track_playback_paused', tl_track=self.current_tl_track, time_position=self.time_position) def _trigger_track_playback_resumed(self): logger.debug('Triggering track playback resumed event') if self.current_track is None: return listener.CoreListener.send( 'track_playback_resumed', tl_track=self.current_tl_track, time_position=self.time_position) def _trigger_track_playback_started(self): logger.debug('Triggering track playback started event') if self.current_tl_track is None: return listener.CoreListener.send( 'track_playback_started', tl_track=self.current_tl_track) def _trigger_track_playback_ended(self, time_position_before_stop): logger.debug('Triggering track playback ended event') if self.current_tl_track is None: return listener.CoreListener.send( 'track_playback_ended', tl_track=self.current_tl_track, time_position=time_position_before_stop) def _trigger_playback_state_changed(self, old_state, new_state): logger.debug('Triggering playback state change event') listener.CoreListener.send( 'playback_state_changed', old_state=old_state, new_state=new_state) def _trigger_volume_changed(self, volume): logger.debug('Triggering volume changed event') listener.CoreListener.send('volume_changed', volume=volume) def _trigger_mute_changed(self, mute): logger.debug('Triggering mute changed event') listener.CoreListener.send('mute_changed', mute=mute) def _trigger_seeked(self, time_position): logger.debug('Triggering seeked event') listener.CoreListener.send('seeked', time_position=time_position)