andpp/mopidy
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge branch 'feature/encapsulate-queues' into develop

Browse Source
This commit is contained in:
Stein Magnus Jodal 2010-08-24 01:13:42 +02:00
commit 0433dad077
14 changed files with 261 additions and 124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
mopidy/backends/base/__init__.py
View File

@ -23,17 +23,17 @@ class BaseBackend(object):
:param core_queue: a queue for sending messages to
:class:`mopidy.process.CoreProcess`
:type core_queue: :class:`multiprocessing.Queue`
:param output_queue: a queue for sending messages to the output process
:type output_queue: :class:`multiprocessing.Queue`
:param output: the audio output
:type output: :class:`mopidy.outputs.gstreamer.GStreamerOutput` or similar
:param mixer_class: either a mixer class, or :class:`None` to use the mixer
defined in settings
:type mixer_class: a subclass of :class:`mopidy.mixers.BaseMixer` or
:class:`None`
"""
def __init__(self, core_queue=None, output_queue=None, mixer_class=None):
def __init__(self, core_queue=None, output=None, mixer_class=None):
self.core_queue = core_queue
self.output_queue = output_queue
self.output = output
if mixer_class is None:
mixer_class = get_class(settings.MIXER)
self.mixer = mixer_class(self)

9
mopidy/backends/base/playback.py
View File

@ -442,8 +442,9 @@ class BasePlaybackController(object):
:type time_position: int
:rtype: :class:`True` if successful, else :class:`False`
"""
# FIXME I think return value is only really useful for internal
# testing, as such it should probably not be exposed in API.
if not self.backend.current_playlist.tracks:
return False
if self.state == self.STOPPED:
self.play()
elif self.state == self.PAUSED:
@ -451,9 +452,9 @@ class BasePlaybackController(object):
if time_position < 0:
time_position = 0
elif self.current_track and time_position > self.current_track.length:
elif time_position > self.current_track.length:
self.next()
return
return True
self._play_time_started = self._current_wall_time
self._play_time_accumulated = time_position

2
mopidy/backends/libspotify/__init__.py
View File

@ -55,6 +55,6 @@ class LibspotifyBackend(BaseBackend):
spotify = LibspotifySessionManager(
settings.SPOTIFY_USERNAME, settings.SPOTIFY_PASSWORD,
core_queue=self.core_queue,
output_queue=self.output_queue)
output=self.output)
spotify.start()
return spotify

25
mopidy/backends/libspotify/playback.py
View File

@ -1,30 +1,17 @@
import logging
import multiprocessing
from spotify import Link, SpotifyError
from mopidy.backends.base import BasePlaybackController
from mopidy.utils.process import pickle_connection
logger = logging.getLogger('mopidy.backends.libspotify.playback')
class LibspotifyPlaybackController(BasePlaybackController):
def _set_output_state(self, state_name):
logger.debug(u'Setting output state to %s ...', state_name)
(my_end, other_end) = multiprocessing.Pipe()
self.backend.output_queue.put({
'command': 'set_state',
'state': state_name,
'reply_to': pickle_connection(other_end),
})
my_end.poll(None)
return my_end.recv()
def _pause(self):
return self._set_output_state('PAUSED')
return self.backend.output.set_state('PAUSED')
def _play(self, track):
self._set_output_state('READY')
self.backend.output.set_state('READY')
if self.state == self.PLAYING:
self.backend.spotify.session.play(0)
if track.uri is None:
@ -33,7 +20,7 @@ class LibspotifyPlaybackController(BasePlaybackController):
self.backend.spotify.session.load(
Link.from_string(track.uri).as_track())
self.backend.spotify.session.play(1)
self._set_output_state('PLAYING')
self.backend.output.set_state('PLAYING')
return True
except SpotifyError as e:
logger.warning('Play %s failed: %s', track.uri, e)
@ -43,12 +30,12 @@ class LibspotifyPlaybackController(BasePlaybackController):
return self._seek(self.time_position)
def _seek(self, time_position):
self._set_output_state('READY')
self.backend.output.set_state('READY')
self.backend.spotify.session.seek(time_position)
self._set_output_state('PLAYING')
self.backend.output.set_state('PLAYING')
return True
def _stop(self):
result = self._set_output_state('READY')
result = self.backend.output.set_state('READY')
self.backend.spotify.session.play(0)
return result

14
mopidy/backends/libspotify/session_manager.py
View File

@ -16,14 +16,14 @@ class LibspotifySessionManager(SpotifySessionManager, threading.Thread):
appkey_file = os.path.join(os.path.dirname(__file__), 'spotify_appkey.key')
user_agent = 'Mopidy %s' % get_version()
def __init__(self, username, password, core_queue, output_queue):
def __init__(self, username, password, core_queue, output):
SpotifySessionManager.__init__(self, username, password)
threading.Thread.__init__(self, name='LibspotifySessionManagerThread')
# Run as a daemon thread, so Mopidy won't wait for this thread to exit
# before Mopidy exits.
self.daemon = True
self.core_queue = core_queue
self.output_queue = output_queue
self.output = output
self.connected = threading.Event()
self.session = None
@ -68,7 +68,7 @@ class LibspotifySessionManager(SpotifySessionManager, threading.Thread):
sample_type, sample_rate, channels):
"""Callback used by pyspotify"""
# TODO Base caps_string on arguments
caps_string = """
capabilites = """
audio/x-raw-int,
endianness=(int)1234,
channels=(int)2,
@ -77,11 +77,7 @@ class LibspotifySessionManager(SpotifySessionManager, threading.Thread):
signed=True,
rate=(int)44100
"""
self.output_queue.put({
'command': 'deliver_data',
'caps': caps_string,
'data': bytes(frames),
})
self.output.deliver_data(capabilites, bytes(frames))
def play_token_lost(self, session):
"""Callback used by pyspotify"""
@ -95,7 +91,7 @@ class LibspotifySessionManager(SpotifySessionManager, threading.Thread):
def end_of_track(self, session):
"""Callback used by pyspotify"""
logger.debug('End of data stream.')
self.output_queue.put({'command': 'end_of_data_stream'})
self.output.end_of_data_stream()
def search(self, query, connection):
"""Search method used by Mopidy backend"""

26
mopidy/backends/local/__init__.py
View File

@ -41,38 +41,24 @@ class LocalPlaybackController(BasePlaybackController):
super(LocalPlaybackController, self).__init__(backend)
self.stop()
def _send_recv(self, message):
(my_end, other_end) = multiprocessing.Pipe()
message.update({'reply_to': pickle_connection(other_end)})
self.backend.output_queue.put(message)
my_end.poll(None)
return my_end.recv()
def _send(self, message):
self.backend.output_queue.put(message)
def _set_state(self, state):
return self._send_recv({'command': 'set_state', 'state': state})
def _play(self, track):
return self._send_recv({'command': 'play_uri', 'uri': track.uri})
return self.backend.output.play_uri(track.uri)
def _stop(self):
return self._set_state('READY')
return self.backend.output.set_state('READY')
def _pause(self):
return self._set_state('PAUSED')
return self.backend.output.set_state('PAUSED')
def _resume(self):
return self._set_state('PLAYING')
return self.backend.output.set_state('PLAYING')
def _seek(self, time_position):
return self._send_recv({'command': 'set_position',
'position': time_position})
return self.backend.output.set_position(time_position)
@property
def time_position(self):
return self._send_recv({'command': 'get_position'})
return self.backend.output.get_position()
class LocalStoredPlaylistsController(BaseStoredPlaylistsController):

18
mopidy/core.py
View File

@ -16,7 +16,7 @@ class CoreProcess(BaseProcess):
super(CoreProcess, self).__init__(name='CoreProcess')
self.core_queue = multiprocessing.Queue()
self.options = self.parse_options()
self.output_queue = None
self.output = None
self.backend = None
self.frontend = None
@ -46,8 +46,8 @@ class CoreProcess(BaseProcess):
def setup(self):
self.setup_logging()
self.setup_settings()
self.output_queue = self.setup_output(self.core_queue)
self.backend = self.setup_backend(self.core_queue, self.output_queue)
self.output = self.setup_output(self.core_queue)
self.backend = self.setup_backend(self.core_queue, self.output)
self.frontend = self.setup_frontend(self.core_queue, self.backend)
def setup_logging(self):
@ -59,12 +59,12 @@ class CoreProcess(BaseProcess):
settings.validate()
def setup_output(self, core_queue):
output_queue = multiprocessing.Queue()
get_class(settings.OUTPUT)(core_queue, output_queue)
return output_queue
output = get_class(settings.OUTPUT)(core_queue)
output.start()
return output
def setup_backend(self, core_queue, output_queue):
return get_class(settings.BACKENDS[0])(core_queue, output_queue)
def setup_backend(self, core_queue, output):
return get_class(settings.BACKENDS[0])(core_queue, output)
def setup_frontend(self, core_queue, backend):
frontend = get_class(settings.FRONTENDS[0])(core_queue, backend)
@ -73,7 +73,7 @@ class CoreProcess(BaseProcess):
def process_message(self, message):
if message.get('to') == 'output':
self.output_queue.put(message)
self.output.process_message(message)
elif message.get('to') == 'frontend':
self.frontend.process_message(message)
elif message['command'] == 'end_of_track':

16
mopidy/mixers/gstreamer_software.py
View File

@ -1,7 +1,4 @@
import multiprocessing
from mopidy.mixers import BaseMixer
from mopidy.utils.process import pickle_connection
class GStreamerSoftwareMixer(BaseMixer):
"""Mixer which uses GStreamer to control volume in software."""
@ -10,16 +7,7 @@ class GStreamerSoftwareMixer(BaseMixer):
super(GStreamerSoftwareMixer, self).__init__(*args, **kwargs)
def _get_volume(self):
my_end, other_end = multiprocessing.Pipe()
self.backend.output_queue.put({
'command': 'get_volume',
'reply_to': pickle_connection(other_end),
})
my_end.poll(None)
return my_end.recv()
return self.backend.output.get_volume()
def _set_volume(self, volume):
self.backend.output_queue.put({
'command': 'set_volume',
'volume': volume,
})
self.backend.output.set_volume(volume)

72
mopidy/outputs/base.py
View File

@ -3,6 +3,9 @@ class BaseOutput(object):
Base class for audio outputs.
"""
def __init__(self, core_queue):
self.core_queue = core_queue
def start(self):
"""Start the output."""
pass
@ -14,3 +17,72 @@ class BaseOutput(object):
def process_message(self, message):
"""Process messages with the output as destination."""
raise NotImplementedError
def play_uri(self, uri):
"""
Play URI.
:param uri: the URI to play
:type uri: string
:rtype: :class:`True` if successful, else :class:`False`
"""
raise NotImplementedError
def deliver_data(self, capabilities, data):
"""
Deliver audio data to be played.
:param capabilities: a GStreamer capabilities string
:type capabilities: string
"""
raise NotImplementedError
def end_of_data_stream(self):
"""Signal that the last audio data has been delivered."""
raise NotImplementedError
def get_position(self):
"""
Get position in milliseconds.
:rtype: int
"""
raise NotImplementedError
def set_position(self, position):
"""
Set position in milliseconds.
:param position: the position in milliseconds
:type volume: int
:rtype: :class:`True` if successful, else :class:`False`
"""
raise NotImplementedError
def set_state(self, state):
"""
Set playback state.
:param state: the state
:type state: string
:rtype: :class:`True` if successful, else :class:`False`
"""
raise NotImplementedError
def get_volume(self):
"""
Get volume level for software mixer.
:rtype: int in range [0..100]
"""
raise NotImplementedError
def set_volume(self, volume):
"""
Set volume level for software mixer.
:param volume: the volume in the range [0..100]
:type volume: int
:rtype: :class:`True` if successful, else :class:`False`
"""
raise NotImplementedError

52
mopidy/outputs/dummy.py
View File

@ -14,6 +14,29 @@ class DummyOutput(BaseOutput):
#: For testing. Contains all messages :meth:`process_message` has received.
messages = []
#: For testing. Contains the last URI passed to :meth:`play_uri`.
uri = None
#: For testing. Contains the last capabilities passed to
#: :meth:`deliver_data`.
capabilities = None
#: For testing. Contains the last data passed to :meth:`deliver_data`.
data = None
#: For testing. :class:`True` if :meth:`end_of_data_stream` has been
#: called.
end_of_data_stream_called = False
#: For testing. Contains the current position.
position = 0
#: For testing. Contains the current state.
state = 'NULL'
#: For testing. Contains the current volume.
volume = 100
def start(self):
self.start_called = True
@ -22,3 +45,32 @@ class DummyOutput(BaseOutput):
def process_message(self, message):
self.messages.append(message)
def play_uri(self, uri):
self.uri = uri
return True
def deliver_data(self, capabilities, data):
self.capabilities = capabilities
self.data = data
def end_of_data_stream(self):
self.end_of_data_stream_called = True
def get_position(self):
return self.position
def set_position(self, position):
self.position = position
return True
def set_state(self, state):
self.state = state
return True
def get_volume(self):
return self.volume
def set_volume(self, volume):
self.volume = volume
return True

67
mopidy/outputs/gstreamer.py
View File

@ -6,14 +6,17 @@ pygst.require('0.10')
import gst
import logging
import multiprocessing
import threading
from mopidy import settings
from mopidy.utils.process import BaseProcess, unpickle_connection
from mopidy.outputs.base import BaseOutput
from mopidy.utils.process import (BaseProcess, pickle_connection,
unpickle_connection)
logger = logging.getLogger('mopidy.outputs.gstreamer')
class GStreamerOutput(object):
class GStreamerOutput(BaseOutput):
"""
Audio output through GStreamer.
@ -24,17 +27,68 @@ class GStreamerOutput(object):
- :attr:`mopidy.settings.GSTREAMER_AUDIO_SINK`
"""
def __init__(self, core_queue, output_queue):
self.process = GStreamerProcess(core_queue, output_queue)
def __init__(self, core_queue):
super(GStreamerOutput, self).__init__(core_queue)
self.output_queue = multiprocessing.Queue()
self.process = GStreamerProcess(core_queue, self.output_queue)
def start(self):
self.process.start()
def destroy(self):
self.process.terminate()
def process_message(self, message):
assert message['to'] == 'output', \
u'Message recipient must be "output".'
self.output_queue.put(message)
def _send_recv(self, message):
(my_end, other_end) = multiprocessing.Pipe()
message['to'] = 'output'
message['reply_to'] = pickle_connection(other_end)
self.process_message(message)
my_end.poll(None)
return my_end.recv()
def _send(self, message):
message['to'] = 'output'
self.process_message(message)
def play_uri(self, uri):
return self._send_recv({'command': 'play_uri', 'uri': uri})
def deliver_data(self, capabilities, data):
return self._send({
'command': 'deliver_data',
'caps': capabilities,
'data': data,
})
def end_of_data_stream(self):
return self._send({'command': 'end_of_data_stream'})
def get_position(self):
return self._send_recv({'command': 'get_position'})
def set_position(self, position):
return self._send_recv({'command': 'set_position'})
def set_state(self, state):
return self._send_recv({'command': 'set_state', 'state': state})
def get_volume(self):
return self._send_recv({'command': 'get_volume'})
def set_volume(self, volume):
return self._send_recv({'command': 'set_volume', 'volume': volume})
class GStreamerMessagesThread(threading.Thread):
def run(self):
gobject.MainLoop().run()
class GStreamerProcess(BaseProcess):
"""
A process for all work related to GStreamer.
@ -111,7 +165,9 @@ class GStreamerProcess(BaseProcess):
connection = unpickle_connection(message['reply_to'])
connection.send(volume)
elif message['command'] == 'set_volume':
self.set_volume(message['volume'])
response = self.set_volume(message['volume'])
connection = unpickle_connection(message['reply_to'])
connection.send(response)
elif message['command'] == 'set_position':
response = self.set_position(message['position'])
connection = unpickle_connection(message['reply_to'])
@ -195,6 +251,7 @@ class GStreamerProcess(BaseProcess):
"""Set volume in range [0..100]"""
gst_volume = self.gst_pipeline.get_by_name('volume')
gst_volume.set_property('volume', volume / 100.0)
return True
def set_position(self, position):
self.gst_pipeline.get_state() # block until state changes are done

7
tests/backends/base/current_playlist.py
View File

@ -4,6 +4,7 @@ import random
from mopidy import settings
from mopidy.mixers.dummy import DummyMixer
from mopidy.models import Playlist, Track
from mopidy.outputs.dummy import DummyOutput
from mopidy.utils import get_class
from tests.backends.base import populate_playlist
@ -12,12 +13,10 @@ class BaseCurrentPlaylistControllerTest(object):
tracks = []
def setUp(self):
self.output_queue = multiprocessing.Queue()
self.core_queue = multiprocessing.Queue()
self.output = get_class(settings.OUTPUT)(
self.core_queue, self.output_queue)
self.output = DummyOutput(self.core_queue)
self.backend = self.backend_class(
self.core_queue, self.output_queue, DummyMixer)
self.core_queue, self.output, DummyMixer)
self.controller = self.backend.current_playlist
self.playback = self.backend.playback

20
tests/backends/base/playback.py
View File

@ -5,6 +5,7 @@ import time
from mopidy import settings
from mopidy.mixers.dummy import DummyMixer
from mopidy.models import Track
from mopidy.outputs.dummy import DummyOutput
from mopidy.utils import get_class
from tests import SkipTest
@ -16,12 +17,10 @@ class BasePlaybackControllerTest(object):
tracks = []
def setUp(self):
self.output_queue = multiprocessing.Queue()
self.core_queue = multiprocessing.Queue()
self.output = get_class(settings.OUTPUT)(
self.core_queue, self.output_queue)
self.output = DummyOutput(self.core_queue)
self.backend = self.backend_class(
self.core_queue, self.output_queue, DummyMixer)
self.core_queue, self.output, DummyMixer)
self.playback = self.backend.playback
self.current_playlist = self.backend.current_playlist
@ -529,12 +528,13 @@ class BasePlaybackControllerTest(object):
self.assert_(wrapper.called)
@SkipTest # Blocks for 10ms and does not work with DummyOutput
@populate_playlist
def test_end_of_track_callback_gets_called(self):
self.playback.play()
result = self.playback.seek(self.tracks[0].length - 10)
self.assert_(result, 'Seek failed')
message = self.core_queue.get()
self.assertTrue(result, 'Seek failed')
message = self.core_queue.get(True, 1)
self.assertEqual('end_of_track', message['command'])
@populate_playlist
@ -608,6 +608,7 @@ class BasePlaybackControllerTest(object):
self.playback.pause()
self.assertEqual(self.playback.resume(), None)
@SkipTest # Uses sleep and does not work with DummyOutput+LocalBackend
@populate_playlist
def test_resume_continues_from_right_position(self):
self.playback.play()
@ -628,8 +629,7 @@ class BasePlaybackControllerTest(object):
self.assert_(position >= 990, position)
def test_seek_on_empty_playlist(self):
result = self.playback.seek(0)
self.assert_(not result, 'Seek return value was %s' % result)
self.assertFalse(self.playback.seek(0))
def test_seek_on_empty_playlist_updates_position(self):
self.playback.seek(0)
@ -740,15 +740,16 @@ class BasePlaybackControllerTest(object):
def test_time_position_when_stopped_with_playlist(self):
self.assertEqual(self.playback.time_position, 0)
@SkipTest # Uses sleep and does not work with LocalBackend+DummyOutput
@populate_playlist
def test_time_position_when_playing(self):
self.playback.play()
first = self.playback.time_position
time.sleep(1)
second = self.playback.time_position
self.assert_(second > first, '%s - %s' % (first, second))
@SkipTest # Uses sleep
@populate_playlist
def test_time_position_when_paused(self):
self.playback.play()
@ -757,7 +758,6 @@ class BasePlaybackControllerTest(object):
time.sleep(0.2)
first = self.playback.time_position
second = self.playback.time_position
self.assertEqual(first, second)
@populate_playlist

49
tests/outputs/gstreamer_test.py
View File

@ -19,48 +19,47 @@ class GStreamerOutputTest(unittest.TestCase):
def setUp(self):
settings.BACKENDS = ('mopidy.backends.local.LocalBackend',)
self.song_uri = path_to_uri(data_folder('song1.wav'))
self.output_queue = multiprocessing.Queue()
self.core_queue = multiprocessing.Queue()
self.output = GStreamerOutput(self.core_queue, self.output_queue)
self.output = GStreamerOutput(self.core_queue)
self.output.start()
def tearDown(self):
self.output.destroy()
settings.runtime.clear()
def send_recv(self, message):
(my_end, other_end) = multiprocessing.Pipe()
message.update({'reply_to': pickle_connection(other_end)})
self.output_queue.put(message)
my_end.poll(None)
return my_end.recv()
def send(self, message):
self.output_queue.put(message)
def test_play_uri_existing_file(self):
message = {'command': 'play_uri', 'uri': self.song_uri}
self.assertEqual(True, self.send_recv(message))
self.assertTrue(self.output.play_uri(self.song_uri))
def test_play_uri_non_existing_file(self):
message = {'command': 'play_uri', 'uri': self.song_uri + 'bogus'}
self.assertEqual(False, self.send_recv(message))
self.assertFalse(self.output.play_uri(self.song_uri + 'bogus'))
@SkipTest
def test_deliver_data(self):
pass # TODO
@SkipTest
def test_end_of_data_stream(self):
pass # TODO
def test_default_get_volume_result(self):
message = {'command': 'get_volume'}
self.assertEqual(100, self.send_recv(message))
self.assertEqual(100, self.output.get_volume())
def test_set_volume(self):
self.send({'command': 'set_volume', 'volume': 50})
self.assertEqual(50, self.send_recv({'command': 'get_volume'}))
self.assertTrue(self.output.set_volume(50))
self.assertEqual(50, self.output.get_volume())
def test_set_volume_to_zero(self):
self.send({'command': 'set_volume', 'volume': 0})
self.assertEqual(0, self.send_recv({'command': 'get_volume'}))
self.assertTrue(self.output.set_volume(0))
self.assertEqual(0, self.output.get_volume())
def test_set_volume_to_one_hundred(self):
self.send({'command': 'set_volume', 'volume': 100})
self.assertEqual(100, self.send_recv({'command': 'get_volume'}))
self.assertTrue(self.output.set_volume(100))
self.assertEqual(100, self.output.get_volume())
@SkipTest
def test_set_state(self):
raise NotImplementedError
pass # TODO
@SkipTest
def test_set_position(self):
pass # TODO