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

Merge pull request #163 from adamcik/feature/switch-to-gst-mixers

Browse Source
This commit is contained in:
Stein Magnus Jodal 2012-09-05 16:40:44 -07:00
commit 1bb13b8c1b
46 changed files with 372 additions and 824 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
docs/api/backends/concepts.rst
View File

@ -27,4 +27,3 @@ Providers:
"Playback\ncontroller" -> "Playback\nproviders"
Backend -> "Stored\nplaylists\ncontroller"
"Stored\nplaylists\ncontroller" -> "Stored\nplaylist\nproviders"
Backend -> Mixer

10
docs/api/backends/controllers.rst
View File

@ -20,19 +20,13 @@ The backend
Playback controller
===================
Manages playback, with actions like play, pause, stop, next, previous, and
seek.
Manages playback, with actions like play, pause, stop, next, previous,
seek, and volume control.
.. autoclass:: mopidy.backends.base.PlaybackController
:members:
Mixer controller
================
Manages volume. See :class:`mopidy.mixers.base.BaseMixer`.
Current playlist controller
===========================

43
docs/api/mixers.rst
View File

@ -1,43 +0,0 @@
*********
Mixer API
*********
Mixers are responsible for controlling volume. Clients of the mixers will
simply instantiate a mixer and read/write to the ``volume`` attribute::
>>> from mopidy.mixers.alsa import AlsaMixer
>>> mixer = AlsaMixer()
>>> mixer.volume
100
>>> mixer.volume = 80
>>> mixer.volume
80
Most users will use one of the internal mixers which controls the volume on the
computer running Mopidy. If you do not specify which mixer you want to use in
the settings, Mopidy will choose one for you based upon what OS you run. See
:attr:`mopidy.settings.MIXER` for the defaults.
Mopidy also supports controlling volume on other hardware devices instead of on
the computer running Mopidy through the use of custom mixer implementations. To
enable one of the hardware device mixers, you must the set
:attr:`mopidy.settings.MIXER` setting to point to one of the classes found
below, and possibly add some extra settings required by the mixer you choose.
All mixers should subclass :class:`mopidy.mixers.base.BaseMixer` and override
methods as described below.
.. automodule:: mopidy.mixers.base
:synopsis: Mixer API
:members:
Mixer implementations
=====================
* :mod:`mopidy.mixers.alsa`
* :mod:`mopidy.mixers.denon`
* :mod:`mopidy.mixers.dummy`
* :mod:`mopidy.mixers.gstreamer_software`
* :mod:`mopidy.mixers.osa`
* :mod:`mopidy.mixers.nad`

8
docs/changes.rst
View File

@ -34,6 +34,14 @@ v0.8 (in development)
``autoaudiosink``. (Fixes: :issue:`81`, :issue:`115`, :issue:`121`,
:issue:`159`)
- Switch to pure GStreamer based mixing. This implies that users setup a
GStreamer bin with a mixer in it in :attr:`mopidy.setting.MIXER`. The default
value is ``autoaudiomixer``, a custom mixer that attempts to find a mixer that
will work on your system. If this picks the wrong mixer you can of course
override it. Setting the mixer to :class:`None` is also supported. MPD
protocol support for volume has also been updated to return -1 when we have
no mixer set.
v0.7.3 (2012-08-11)
===================

5
docs/clients/mpd.rst
View File

@ -181,11 +181,6 @@ they artist and album tabs do not hang. The folder tab still freezes when
``lsinfo`` returns a list of stored playlists, though zero files. Thus, we've
discovered a couple of bugs in Droid MPD Client.
The volume control is very slick, with a turn knob, just like on an amplifier.
It lends itself to showing off to friends when combined with Mopidy's external
amplifier mixers. Everybody loves turning a knob on a touch screen and see the
physical knob on the amplifier turn as well ;-)
Even though ``lsinfo`` returns the stored playlists for the folder tab, they
are not displayed anywhere. Thus, we had to select an album in the album tab to
complete the test procedure.

3
docs/installation/index.rst
View File

@ -46,9 +46,6 @@ dependencies installed.
sudo apt-get install python-dbus python-indicate
- Some custom mixers (but not the default one) require additional
dependencies. See the docs for each mixer.
Install latest stable release
=============================

7
docs/modules/mixers/alsa.rst
View File

@ -1,7 +0,0 @@
*************************************************
:mod:`mopidy.mixers.alsa` -- ALSA mixer for Linux
*************************************************
.. automodule:: mopidy.mixers.alsa
:synopsis: ALSA mixer for Linux
:members:

7
docs/modules/mixers/denon.rst
View File

@ -1,7 +0,0 @@
*****************************************************************
:mod:`mopidy.mixers.denon` -- Hardware mixer for Denon amplifiers
*****************************************************************
.. automodule:: mopidy.mixers.denon
:synopsis: Hardware mixer for Denon amplifiers
:members:

7
docs/modules/mixers/dummy.rst
View File

@ -1,7 +0,0 @@
*****************************************************
:mod:`mopidy.mixers.dummy` -- Dummy mixer for testing
*****************************************************
.. automodule:: mopidy.mixers.dummy
:synopsis: Dummy mixer for testing
:members:

7
docs/modules/mixers/gstreamer_software.rst
View File

@ -1,7 +0,0 @@
***************************************************************************
:mod:`mopidy.mixers.gstreamer_software` -- Software mixer for all platforms
***************************************************************************
.. automodule:: mopidy.mixers.gstreamer_software
:synopsis: Software mixer for all platforms
:members:

7
docs/modules/mixers/nad.rst
View File

@ -1,7 +0,0 @@
*************************************************************
:mod:`mopidy.mixers.nad` -- Hardware mixer for NAD amplifiers
*************************************************************
.. automodule:: mopidy.mixers.nad
:synopsis: Hardware mixer for NAD amplifiers
:members:

7
docs/modules/mixers/osa.rst
View File

@ -1,7 +0,0 @@
**********************************************
:mod:`mopidy.mixers.osa` -- Osa mixer for OS X
**********************************************
.. automodule:: mopidy.mixers.osa
:synopsis: Osa mixer for OS X
:members:

11
mopidy/__main__.py
View File

@ -52,7 +52,6 @@ def main():
check_old_folders()
setup_settings(options.interactive)
setup_gstreamer()
setup_mixer()
setup_backend()
setup_frontends()
loop.run()
@ -66,7 +65,6 @@ def main():
loop.quit()
stop_frontends()
stop_backend()
stop_mixer()
stop_gstreamer()
stop_remaining_actors()
@ -126,15 +124,6 @@ def setup_gstreamer():
def stop_gstreamer():
stop_actors_by_class(GStreamer)
def setup_mixer():
get_class(settings.MIXER).start()
def stop_mixer():
stop_actors_by_class(get_class(settings.MIXER))
def setup_backend():
get_class(settings.BACKENDS[0]).start()

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

@ -319,6 +319,14 @@ class PlaybackController(object):
def _current_wall_time(self):
return int(time.time() * 1000)
@property
def volume(self):
return self.provider.get_volume()
@volume.setter
def volume(self, volume):
self.provider.set_volume(volume)
def change_track(self, cp_track, on_error_step=1):
"""
Change to the given track, keeping the current playback state.
@ -601,3 +609,24 @@ class BasePlaybackProvider(object):
:rtype: :class:`True` if successful, else :class:`False`
"""
raise NotImplementedError
def get_volume(self):
"""
Get current volume
*MUST be implemented by subclass.*
:rtype: int [0..100] or :class:`None`
"""
raise NotImplementedError
def set_volume(self, volume):
"""
Get current volume
*MUST be implemented by subclass.*
:param: volume
:type volume: int [0..100]
"""
raise NotImplementedError

10
mopidy/backends/dummy/__init__.py
View File

@ -56,6 +56,10 @@ class DummyLibraryProvider(BaseLibraryProvider):
class DummyPlaybackProvider(BasePlaybackProvider):
def __init__(self, *args, **kwargs):
super(DummyPlaybackProvider, self).__init__(*args, **kwargs)
self._volume = None
def pause(self):
return True
@ -72,6 +76,12 @@ class DummyPlaybackProvider(BasePlaybackProvider):
def stop(self):
return True
def get_volume(self):
return self._volume
def set_volume(self, volume):
self._volume = volume
class DummyStoredPlaylistsProvider(BaseStoredPlaylistsProvider):
def create(self, name):

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

@ -102,6 +102,12 @@ class LocalPlaybackProvider(BasePlaybackProvider):
def stop(self):
return self.backend.gstreamer.stop_playback().get()
def get_volume(self):
return self.backend.gstreamer.get_volume().get()
def set_volume(self, volume):
self.backend.gstreamer.set_volume(volume).get()
class LocalStoredPlaylistsProvider(BaseStoredPlaylistsProvider):
def __init__(self, *args, **kwargs):

6
mopidy/backends/spotify/playback.py
View File

@ -41,3 +41,9 @@ class SpotifyPlaybackProvider(BasePlaybackProvider):
result = self.backend.gstreamer.stop_playback()
self.backend.spotify.session.play(0)
return result
def get_volume(self):
return self.backend.gstreamer.get_volume().get()
def set_volume(self, volume):
self.backend.gstreamer.set_volume(volume)

13
mopidy/frontends/mpd/dispatcher.py
View File

@ -15,7 +15,6 @@ from mopidy.frontends.mpd.protocol import (audio_output, command_list,
connection, current_playlist, empty, music_db, playback, reflection,
status, stickers, stored_playlists)
# pylint: enable = W0611
from mopidy.mixers.base import BaseMixer
from mopidy.utils import flatten
logger = logging.getLogger('mopidy.frontends.mpd.dispatcher')
@ -235,7 +234,6 @@ class MpdContext(object):
self.events = set()
self.subscriptions = set()
self._backend = None
self._mixer = None
@property
def backend(self):
@ -248,14 +246,3 @@ class MpdContext(object):
'Expected exactly one running backend.'
self._backend = backend_refs[0].proxy()
return self._backend
@property
def mixer(self):
"""
The mixer. An instance of :class:`mopidy.mixers.base.BaseMixer`.
"""
if self._mixer is None:
mixer_refs = ActorRegistry.get_by_class(BaseMixer)
assert len(mixer_refs) == 1, 'Expected exactly one running mixer.'
self._mixer = mixer_refs[0].proxy()
return self._mixer

2
mopidy/frontends/mpd/protocol/playback.py
View File

@ -353,7 +353,7 @@ def setvol(context, volume):
volume = 0
if volume > 100:
volume = 100
context.mixer.volume = volume
context.backend.playback.volume = volume
@handle_request(r'^single (?P<state>[01])$')
@handle_request(r'^single "(?P<state>[01])"$')

8
mopidy/frontends/mpd/protocol/status.py
View File

@ -137,7 +137,7 @@ def status(context):
Reports the current status of the player and the volume level.
- ``volume``: 0-100
- ``volume``: 0-100 or -1
- ``repeat``: 0 or 1
- ``single``: 0 or 1
- ``consume``: 0 or 1
@ -168,7 +168,7 @@ def status(context):
futures = {
'current_playlist.length': context.backend.current_playlist.length,
'current_playlist.version': context.backend.current_playlist.version,
'mixer.volume': context.mixer.volume,
'playback.volume': context.backend.playback.volume,
'playback.consume': context.backend.playback.consume,
'playback.random': context.backend.playback.random,
'playback.repeat': context.backend.playback.repeat,
@ -263,11 +263,11 @@ def _status_time_total(futures):
return current_cp_track.track.length
def _status_volume(futures):
volume = futures['mixer.volume'].get()
volume = futures['playback.volume'].get()
if volume is not None:
return volume
else:
return 0
return -1
def _status_xfade(futures):
return 0 # Not supported

18
mopidy/frontends/mpris/objects.py
View File

@ -17,7 +17,6 @@ from pykka.registry import ActorRegistry
from mopidy import settings
from mopidy.backends.base import Backend
from mopidy.backends.base.playback import PlaybackController
from mopidy.mixers.base import BaseMixer
from mopidy.utils.process import exit_process
# Must be done before dbus.SessionBus() is called
@ -37,7 +36,6 @@ class MprisObject(dbus.service.Object):
def __init__(self):
self._backend = None
self._mixer = None
self.properties = {
ROOT_IFACE: self._get_root_iface_properties(),
PLAYER_IFACE: self._get_player_iface_properties(),
@ -95,14 +93,6 @@ class MprisObject(dbus.service.Object):
self._backend = backend_refs[0].proxy()
return self._backend
@property
def mixer(self):
if self._mixer is None:
mixer_refs = ActorRegistry.get_by_class(BaseMixer)
assert len(mixer_refs) == 1, 'Expected exactly one running mixer.'
self._mixer = mixer_refs[0].proxy()
return self._mixer
def _get_track_id(self, cp_track):
return '/com/mopidy/track/%d' % cp_track.cpid
@ -380,7 +370,7 @@ class MprisObject(dbus.service.Object):
return dbus.Dictionary(metadata, signature='sv')
def get_Volume(self):
volume = self.mixer.volume.get()
volume = self.backend.playback.volume.get()
if volume is not None:
return volume / 100.0
@ -391,11 +381,11 @@ class MprisObject(dbus.service.Object):
if value is None:
return
elif value < 0:
self.mixer.volume = 0
self.backend.playback.volume = 0
elif value > 1:
self.mixer.volume = 100
self.backend.playback.volume = 100
elif 0 <= value <= 1:
self.mixer.volume = int(value * 100)
self.backend.playback.volume = int(value * 100)
def get_Position(self):
return self.backend.playback.time_position.get() * 1000

103
mopidy/gstreamer.py
View File

@ -1,6 +1,5 @@
import pygst
pygst.require('0.10')
import gobject
import gst
import logging
@ -8,9 +7,9 @@ import logging
from pykka.actor import ThreadingActor
from pykka.registry import ActorRegistry
from mopidy import settings
from mopidy import settings, utils
from mopidy.backends.base import Backend
from mopidy import mixers # Trigger install of gst mixer plugins.
logger = logging.getLogger('mopidy.gstreamer')
@ -22,6 +21,8 @@ class GStreamer(ThreadingActor):
**Settings:**
- :attr:`mopidy.settings.OUTPUT`
- :attr:`mopidy.settings.MIXER`
- :attr:`mopidy.settings.MIXER_TRACK`
"""
@ -38,39 +39,79 @@ class GStreamer(ThreadingActor):
self._pipeline = None
self._source = None
self._uridecodebin = None
self._volume = None
self._output = None
self._mixer = None
self._setup_pipeline()
self._setup_output()
self._setup_mixer()
self._setup_message_processor()
def _setup_pipeline(self):
# TODO: replace with and input bin so we simply have an input bin we
# connect to an output bin with a mixer on the side. set_uri on bin?
description = ' ! '.join([
'uridecodebin name=uri',
'audioconvert name=convert',
'audioresample name=resample',
'queue name=queue',
'volume name=volume'])
'queue name=queue'])
logger.debug(u'Setting up base GStreamer pipeline: %s', description)
self._pipeline = gst.parse_launch(description)
self._volume = self._pipeline.get_by_name('volume')
self._uridecodebin = self._pipeline.get_by_name('uri')
self._uridecodebin.connect('notify::source', self._on_new_source)
self._uridecodebin.connect('pad-added', self._on_new_pad,
self._pipeline.get_by_name('convert').get_pad('sink'))
self._pipeline.get_by_name('queue').get_pad('sink'))
def _setup_output(self):
# This will raise a gobject.GError if the description is bad.
self._output = gst.parse_bin_from_description(settings.OUTPUT,
ghost_unconnected_pads=True)
self._output = gst.parse_bin_from_description(
settings.OUTPUT, ghost_unconnected_pads=True)
self._pipeline.add(self._output)
gst.element_link_many(self._volume, self._output)
logger.debug('Output set to %s', settings.OUTPUT)
gst.element_link_many(self._pipeline.get_by_name('queue'),
self._output)
logger.info('Output set to %s', settings.OUTPUT)
def _setup_mixer(self):
if not settings.MIXER:
logger.info('Not setting up mixer.')
return
# This will raise a gobject.GError if the description is bad.
mixerbin = gst.parse_bin_from_description(settings.MIXER, False)
# We assume that the bin will contain a single mixer.
mixer = mixerbin.get_by_interface('GstMixer')
if not mixer:
logger.warning('Did not find any mixers in %r', settings.MIXER)
return
if mixerbin.set_state(gst.STATE_READY) != gst.STATE_CHANGE_SUCCESS:
logger.warning('Setting mixer %r to READY failed.', settings.MIXER)
return
track = self._select_mixer_track(mixer, settings.MIXER_TRACK)
if not track:
logger.warning('Could not find usable mixer track.')
return
self._mixer = (mixer, track)
logger.info('Mixer set to %s using track called %s',
mixer.get_factory().get_name(), track.label)
def _select_mixer_track(self, mixer, track_label):
# Look for track with label == MIXER_TRACK, otherwise fallback to
# master track which is also an output.
for track in mixer.list_tracks():
if track_label:
if track.label == track_label:
return track
elif track.flags & (gst.interfaces.MIXER_TRACK_MASTER |
gst.interfaces.MIXER_TRACK_OUTPUT):
return track
def _setup_message_processor(self):
bus = self._pipeline.get_bus()
@ -244,22 +285,48 @@ class GStreamer(ThreadingActor):
def get_volume(self):
"""
Get volume level of the GStreamer software mixer.
Get volume level of the installed mixer.
:rtype: int in range [0..100]
0 == muted.
100 == max volume for given system.
None == no mixer present, i.e. volume unknown.
:rtype: int in range [0..100] or :class:`None`
"""
return int(self._volume.get_property('volume') * 100)
if self._mixer is None:
return None
mixer, track = self._mixer
volumes = mixer.get_volume(track)
avg_volume = float(sum(volumes)) / len(volumes)
new_scale = (0, 100)
old_scale = (track.min_volume, track.max_volume)
return utils.rescale(avg_volume, old=old_scale, new=new_scale)
def set_volume(self, volume):
"""
Set volume level of the GStreamer software mixer.
Set volume level of the installed mixer.
:param volume: the volume in the range [0..100]
:type volume: int
:rtype: :class:`True` if successful, else :class:`False`
"""
self._volume.set_property('volume', volume / 100.0)
return True
if self._mixer is None:
return False
mixer, track = self._mixer
old_scale = (0, 100)
new_scale = (track.min_volume, track.max_volume)
volume = utils.rescale(volume, old=old_scale, new=new_scale)
volumes = (volume,) * track.num_channels
mixer.set_volume(track, volumes)
return mixer.get_volume(track) == volumes
def set_metadata(self, track):
"""

2
mopidy/mixers/__init__.py
View File

@ -0,0 +1,2 @@
from mopidy.mixers.auto import AutoAudioMixer
from mopidy.mixers.fake import FakeMixer

60
mopidy/mixers/alsa.py
View File

@ -1,60 +0,0 @@
import alsaaudio
import logging
from pykka.actor import ThreadingActor
from mopidy import settings
from mopidy.mixers.base import BaseMixer
logger = logging.getLogger('mopidy.mixers.alsa')
class AlsaMixer(ThreadingActor, BaseMixer):
"""
Mixer which uses the Advanced Linux Sound Architecture (ALSA) to control
volume.
**Dependencies:**
- pyalsaaudio >= 0.2 (python-alsaaudio on Debian/Ubuntu)
**Settings:**
- :attr:`mopidy.settings.MIXER_ALSA_CONTROL`
"""
def __init__(self):
super(AlsaMixer, self).__init__()
self._mixer = None
def on_start(self):
self._mixer = alsaaudio.Mixer(self._get_mixer_control())
assert self._mixer is not None
def _get_mixer_control(self):
"""Returns the first mixer control candidate that is known to ALSA"""
candidates = self._get_mixer_control_candidates()
for control in candidates:
if control in alsaaudio.mixers():
logger.info(u'Mixer control in use: %s', control)
return control
else:
logger.debug(u'Mixer control not found, skipping: %s', control)
logger.warning(u'No working mixer controls found. Tried: %s',
candidates)
def _get_mixer_control_candidates(self):
"""
A mixer named 'Master' does not always exist, so we fall back to using
'PCM'. If this does not work for you, you may set
:attr:`mopidy.settings.MIXER_ALSA_CONTROL`.
"""
if settings.MIXER_ALSA_CONTROL:
return [settings.MIXER_ALSA_CONTROL]
return [u'Master', u'PCM']
def get_volume(self):
# FIXME does not seem to see external volume changes.
return self._mixer.getvolume()[0]
def set_volume(self, volume):
self._mixer.setvolume(volume)

72
mopidy/mixers/auto.py Normal file
View File

@ -0,0 +1,72 @@
import pygst
pygst.require('0.10')
import gobject
import gst
import logging
logger = logging.getLogger('mopidy.mixers.auto')
# TODO: we might want to add some ranking to the mixers we know about?
class AutoAudioMixer(gst.Bin):
__gstdetails__ = ('AutoAudioMixer',
'Mixer',
'Element automatically selects a mixer.',
'Thomas Adamcik')
def __init__(self):
gst.Bin.__init__(self)
mixer = self._find_mixer()
if mixer:
self.add(mixer)
logger.debug('AutoAudioMixer chose: %s', mixer.get_name())
else:
logger.debug('AutoAudioMixer did not find any usable mixers')
def _find_mixer(self):
registry = gst.registry_get_default()
factories = registry.get_feature_list(gst.TYPE_ELEMENT_FACTORY)
factories.sort(key=lambda f: (-f.get_rank(), f.get_name()))
for factory in factories:
# Avoid sink/srcs that implement mixing.
if factory.get_klass() != 'Generic/Audio':
continue
# Avoid anything that doesn't implement mixing.
elif not factory.has_interface('GstMixer'):
continue
if self._test_mixer(factory):
return factory.create()
return None
def _test_mixer(self, factory):
element = factory.create()
if not element:
return False
try:
result = element.set_state(gst.STATE_READY)
if result != gst.STATE_CHANGE_SUCCESS:
return False
# Trust that the default device is sane and just check tracks.
return self._test_tracks(element)
finally:
element.set_state(gst.STATE_NULL)
def _test_tracks(self, element):
# Only allow elements that have a least one output track.
flags = gst.interfaces.MIXER_TRACK_OUTPUT
for track in element.list_tracks():
if track.flags & flags:
return True
return False
gobject.type_register(AutoAudioMixer)
gst.element_register(AutoAudioMixer, 'autoaudiomixer', gst.RANK_MARGINAL)

68
mopidy/mixers/base.py
View File

@ -1,68 +0,0 @@
import logging
from mopidy import listeners, settings
logger = logging.getLogger('mopidy.mixers')
class BaseMixer(object):
"""
**Settings:**
- :attr:`mopidy.settings.MIXER_MAX_VOLUME`
"""
amplification_factor = settings.MIXER_MAX_VOLUME / 100.0
@property
def volume(self):
"""
The audio volume
Integer in range [0, 100]. :class:`None` if unknown. Values below 0 is
equal to 0. Values above 100 is equal to 100.
"""
if not hasattr(self, '_user_volume'):
self._user_volume = 0
volume = self.get_volume()
if volume is None or not self.amplification_factor < 1:
return volume
else:
user_volume = int(volume / self.amplification_factor)
if (user_volume - 1) <= self._user_volume <= (user_volume + 1):
return self._user_volume
else:
return user_volume
@volume.setter
def volume(self, volume):
if not hasattr(self, '_user_volume'):
self._user_volume = 0
volume = int(volume)
if volume < 0:
volume = 0
elif volume > 100:
volume = 100
self._user_volume = volume
real_volume = int(volume * self.amplification_factor)
self.set_volume(real_volume)
self._trigger_volume_changed()
def get_volume(self):
"""
Return volume as integer in range [0, 100]. :class:`None` if unknown.
*MUST be implemented by subclass.*
"""
raise NotImplementedError
def set_volume(self, volume):
"""
Set volume as integer in range [0, 100].
*MUST be implemented by subclass.*
"""
raise NotImplementedError
def _trigger_volume_changed(self):
logger.debug(u'Triggering volume changed event')
listeners.BackendListener.send('volume_changed')

58
mopidy/mixers/denon.py
View File

@ -1,58 +0,0 @@
import logging
from pykka.actor import ThreadingActor
from mopidy import settings
from mopidy.mixers.base import BaseMixer
logger = logging.getLogger(u'mopidy.mixers.denon')
class DenonMixer(ThreadingActor, BaseMixer):
"""
Mixer for controlling Denon amplifiers and receivers using the RS-232
protocol.
The external mixer is the authoritative source for the current volume.
This allows the user to use his remote control the volume without Mopidy
cancelling the volume setting.
**Dependencies**
- pyserial (python-serial on Debian/Ubuntu)
**Settings**
- :attr:`mopidy.settings.MIXER_EXT_PORT` -- Example: ``/dev/ttyUSB0``
"""
def __init__(self, device=None):
super(DenonMixer, self).__init__()
self._device = device
self._levels = ['99'] + ["%(#)02d" % {'#': v} for v in range(0, 99)]
self._volume = 0
def on_start(self):
if self._device is None:
from serial import Serial
self._device = Serial(port=settings.MIXER_EXT_PORT, timeout=0.2)
def get_volume(self):
self._ensure_open_device()
self._device.write('MV?\r')
vol = str(self._device.readline()[2:4])
logger.debug(u'_get_volume() = %s' % vol)
return self._levels.index(vol)
def set_volume(self, volume):
# Clamp according to Denon-spec
if volume > 99:
volume = 99
self._ensure_open_device()
self._device.write('MV%s\r'% self._levels[volume])
vol = self._device.readline()[2:4]
self._volume = self._levels.index(vol)
def _ensure_open_device(self):
if not self._device.isOpen():
logger.debug(u'(re)connecting to Denon device')
self._device.open()

16
mopidy/mixers/dummy.py
View File

@ -1,16 +0,0 @@
from pykka.actor import ThreadingActor
from mopidy.mixers.base import BaseMixer
class DummyMixer(ThreadingActor, BaseMixer):
"""Mixer which just stores and reports the chosen volume."""
def __init__(self):
super(DummyMixer, self).__init__()
self._volume = None
def get_volume(self):
return self._volume
def set_volume(self, volume):
self._volume = volume

80
mopidy/mixers/fake.py Normal file
View File

@ -0,0 +1,80 @@
import pygst
pygst.require('0.10')
import gobject
import gst
def create_fake_track(label, intial_volume, min_volume, max_volume,
num_channels, flags):
class Track(gst.interfaces.MixerTrack):
def __init__(self):
super(Track, self).__init__()
self.volumes = (intial_volume,) * self.num_channels
@gobject.property
def label(self):
return label
@gobject.property
def min_volume(self):
return min_volume
@gobject.property
def max_volume(self):
return max_volume
@gobject.property
def num_channels(self):
return num_channels
@gobject.property
def flags(self):
return flags
return Track()
class FakeMixer(gst.Element, gst.ImplementsInterface, gst.interfaces.Mixer):
__gstdetails__ = ('FakeMixer',
'Mixer',
'Fake mixer for use in tests.',
'Thomas Adamcik')
track_label = gobject.property(type=str, default='Master')
track_initial_volume = gobject.property(type=int, default=0)
track_min_volume = gobject.property(type=int, default=0)
track_max_volume = gobject.property(type=int, default=100)
track_num_channels = gobject.property(type=int, default=2)
track_flags = gobject.property(type=int,
default=(gst.interfaces.MIXER_TRACK_MASTER |
gst.interfaces.MIXER_TRACK_OUTPUT))
def __init__(self):
gst.Element.__init__(self)
def list_tracks(self):
track = create_fake_track(self.track_label,
self.track_initial_volume,
self.track_min_volume,
self.track_max_volume,
self.track_num_channels,
self.track_flags)
return [track]
def get_volume(self, track):
return track.volumes
def set_volume(self, track, volumes):
track.volumes = volumes
def set_record(self, track, record):
pass
gobject.type_register(FakeMixer)
gst.element_register(FakeMixer, 'fakemixer', gst.RANK_MARGINAL)

23
mopidy/mixers/gstreamer_software.py
View File

@ -1,23 +0,0 @@
from pykka.actor import ThreadingActor
from pykka.registry import ActorRegistry
from mopidy.mixers.base import BaseMixer
from mopidy.gstreamer import GStreamer
class GStreamerSoftwareMixer(ThreadingActor, BaseMixer):
"""Mixer which uses GStreamer to control volume in software."""
def __init__(self):
super(GStreamerSoftwareMixer, self).__init__()
self.output = None
def on_start(self):
output_refs = ActorRegistry.get_by_class(GStreamer)
assert len(output_refs) == 1, 'Expected exactly one running output.'
self.output = output_refs[0].proxy()
def get_volume(self):
return self.output.get_volume().get()
def set_volume(self, volume):
self.output.set_volume(volume).get()

198
mopidy/mixers/nad.py
View File

@ -1,198 +0,0 @@
import logging
import serial
from pykka.actor import ThreadingActor
from mopidy import settings
from mopidy.mixers.base import BaseMixer
logger = logging.getLogger('mopidy.mixers.nad')
class NadMixer(ThreadingActor, BaseMixer):
"""
Mixer for controlling NAD amplifiers and receivers using the NAD RS-232
protocol.
The NAD mixer was created using a NAD C 355BEE amplifier, but should also
work with other NAD amplifiers supporting the same RS-232 protocol (v2.x).
The C 355BEE does not give you access to the current volume. It only
supports increasing or decreasing the volume one step at the time. Other
NAD amplifiers may support more advanced volume adjustment than what is
currently used by this mixer.
Sadly, this means that if you use the remote control to change the volume
on the amplifier, Mopidy will no longer report the correct volume.
**Dependencies**
- pyserial (python-serial on Debian/Ubuntu)
**Settings**
- :attr:`mopidy.settings.MIXER_EXT_PORT` -- Example: ``/dev/ttyUSB0``
- :attr:`mopidy.settings.MIXER_EXT_SOURCE` -- Example: ``Aux``
- :attr:`mopidy.settings.MIXER_EXT_SPEAKERS_A` -- Example: ``On``
- :attr:`mopidy.settings.MIXER_EXT_SPEAKERS_B` -- Example: ``Off``
"""
def __init__(self):
super(NadMixer, self).__init__()
self._volume_cache = None
self._nad_talker = NadTalker.start().proxy()
def get_volume(self):
return self._volume_cache
def set_volume(self, volume):
self._volume_cache = volume
self._nad_talker.set_volume(volume)
class NadTalker(ThreadingActor):
"""
Independent process which does the communication with the NAD device.
Since the communication is done in an independent process, Mopidy won't
block other requests while doing rather time consuming work like
calibrating the NAD device's volume.
"""
# Timeout in seconds used for read/write operations.
# If you set the timeout too low, the reads will never get complete
# confirmations and calibration will decrease volume forever. If you set
# the timeout too high, stuff takes more time. 0.2s seems like a good value
# for NAD C 355BEE.
TIMEOUT = 0.2
# Number of volume levels the device supports. 40 for NAD C 355BEE.
VOLUME_LEVELS = 40
# Volume in range 0..VOLUME_LEVELS. :class:`None` before calibration.
_nad_volume = None
def __init__(self):
super(NadTalker, self).__init__()
self._device = None
def on_start(self):
self._open_connection()
self._set_device_to_known_state()
def _open_connection(self):
# Opens serial connection to the device.
# Communication settings: 115200 bps 8N1
logger.info(u'Connecting to serial device "%s"',
settings.MIXER_EXT_PORT)
self._device = serial.Serial(port=settings.MIXER_EXT_PORT,
baudrate=115200, timeout=self.TIMEOUT)
self._get_device_model()
def _set_device_to_known_state(self):
self._power_device_on()
self._select_speakers()
self._select_input_source()
self._unmute()
self._calibrate_volume()
def _get_device_model(self):
model = self._ask_device('Main.Model')
logger.info(u'Connected to device of model "%s"', model)
return model
def _power_device_on(self):
while self._ask_device('Main.Power') != 'On':
logger.info(u'Powering device on')
self._command_device('Main.Power', 'On')
def _select_speakers(self):
if settings.MIXER_EXT_SPEAKERS_A is not None:
while (self._ask_device('Main.SpeakerA')
!= settings.MIXER_EXT_SPEAKERS_A):
logger.info(u'Setting speakers A "%s"',
settings.MIXER_EXT_SPEAKERS_A)
self._command_device('Main.SpeakerA',
settings.MIXER_EXT_SPEAKERS_A)
if settings.MIXER_EXT_SPEAKERS_B is not None:
while (self._ask_device('Main.SpeakerB') !=
settings.MIXER_EXT_SPEAKERS_B):
logger.info(u'Setting speakers B "%s"',
settings.MIXER_EXT_SPEAKERS_B)
self._command_device('Main.SpeakerB',
settings.MIXER_EXT_SPEAKERS_B)
def _select_input_source(self):
if settings.MIXER_EXT_SOURCE is not None:
while self._ask_device('Main.Source') != settings.MIXER_EXT_SOURCE:
logger.info(u'Selecting input source "%s"',
settings.MIXER_EXT_SOURCE)
self._command_device('Main.Source', settings.MIXER_EXT_SOURCE)
def _unmute(self):
while self._ask_device('Main.Mute') != 'Off':
logger.info(u'Unmuting device')
self._command_device('Main.Mute', 'Off')
def _ask_device(self, key):
self._write('%s?' % key)
return self._readline().replace('%s=' % key, '')
def _command_device(self, key, value):
if type(value) == unicode:
value = value.encode('utf-8')
self._write('%s=%s' % (key, value))
self._readline()
def _calibrate_volume(self):
# The NAD C 355BEE amplifier has 40 different volume levels. We have no
# way of asking on which level we are. Thus, we must calibrate the
# mixer by decreasing the volume 39 times.
logger.info(u'Calibrating NAD amplifier')
steps_left = self.VOLUME_LEVELS - 1
while steps_left:
if self._decrease_volume():
steps_left -= 1
self._nad_volume = 0
logger.info(u'Done calibrating NAD amplifier')
def set_volume(self, volume):
# Increase or decrease the amplifier volume until it matches the given
# target volume.
logger.debug(u'Setting volume to %d' % volume)
target_nad_volume = int(round(volume * self.VOLUME_LEVELS / 100.0))
if self._nad_volume is None:
return # Calibration needed
while target_nad_volume > self._nad_volume:
if self._increase_volume():
self._nad_volume += 1
while target_nad_volume < self._nad_volume:
if self._decrease_volume():
self._nad_volume -= 1
def _increase_volume(self):
# Increase volume. Returns :class:`True` if confirmed by device.
self._write('Main.Volume+')
return self._readline() == 'Main.Volume+'
def _decrease_volume(self):
# Decrease volume. Returns :class:`True` if confirmed by device.
self._write('Main.Volume-')
return self._readline() == 'Main.Volume-'
def _write(self, data):
# Write data to device. Prepends and appends a newline to the data, as
# recommended by the NAD documentation.
if not self._device.isOpen():
self._device.open()
self._device.write('\n%s\n' % data)
logger.debug('Write: %s', data)
def _readline(self):
# Read line from device. The result is stripped for leading and
# trailing whitespace.
if not self._device.isOpen():
self._device.open()
result = self._device.readline().strip()
if result:
logger.debug('Read: %s', result)
return result

46
mopidy/mixers/osa.py
View File

@ -1,46 +0,0 @@
from subprocess import Popen, PIPE
import time
from pykka.actor import ThreadingActor
from mopidy.mixers.base import BaseMixer
class OsaMixer(ThreadingActor, BaseMixer):
"""
Mixer which uses ``osascript`` on OS X to control volume.
**Dependencies:**
- None
**Settings:**
- None
"""
CACHE_TTL = 30
_cache = None
_last_update = None
def _valid_cache(self):
return (self._cache is not None
and self._last_update is not None
and (int(time.time() - self._last_update) < self.CACHE_TTL))
def get_volume(self):
if not self._valid_cache():
try:
self._cache = int(Popen(
['osascript', '-e',
'output volume of (get volume settings)'],
stdout=PIPE).communicate()[0])
except ValueError:
self._cache = None
self._last_update = int(time.time())
return self._cache
def set_volume(self, volume):
Popen(['osascript', '-e', 'set volume output volume %d' % volume])
self._cache = volume
self._last_update = int(time.time())

50
mopidy/settings.py
View File

@ -103,50 +103,28 @@ LOCAL_PLAYLIST_PATH = None
#: LOCAL_TAG_CACHE_FILE = None # Implies $XDG_DATA_DIR/mopidy/tag_cache
LOCAL_TAG_CACHE_FILE = None
#: Sound mixer to use. See :mod:`mopidy.mixers` for all available mixers.
#: Sound mixer to use.
#:
#: Expects a GStreamer mixer to use, typical values are:
#: alsamixer, pulsemixer, oss4mixer, ossmixer.
#:
#: Setting this to ``None`` means no volume control.
#:
#: Default::
#:
#: MIXER = u'mopidy.mixers.gstreamer_software.GStreamerSoftwareMixer'
MIXER = u'mopidy.mixers.gstreamer_software.GStreamerSoftwareMixer'
#: MIXER = u'autoaudiomixer'
MIXER = u'autoaudiomixer'
#: ALSA mixer only. What mixer control to use. If set to :class:`False`, first
#: ``Master`` and then ``PCM`` will be tried.
#: Sound mixer track to use.
#:
#: Example: ``Master Front``. Default: :class:`False`
MIXER_ALSA_CONTROL = False
#: External mixers only. Which port the mixer is connected to.
#:
#: This must point to the device port like ``/dev/ttyUSB0``.
#:
#: Default: :class:`None`
MIXER_EXT_PORT = None
#: External mixers only. What input source the external mixer should use.
#:
#: Example: ``Aux``. Default: :class:`None`
MIXER_EXT_SOURCE = None
#: External mixers only. What state Speakers A should be in.
#:
#: Default: :class:`None`.
MIXER_EXT_SPEAKERS_A = None
#: External mixers only. What state Speakers B should be in.
#:
#: Default: :class:`None`.
MIXER_EXT_SPEAKERS_B = None
#: The maximum volume. Integer in the range 0 to 100.
#:
#: If this settings is set to 80, the mixer will set the actual volume to 80
#: when asked to set it to 100.
#: Name of the mixer track to use. If this is not set we will try to find the
#: output track with master set. As an example, using ``alsamixer`` you would
#: typically set this to ``Master`` or ``PCM``.
#:
#: Default::
#:
#: MIXER_MAX_VOLUME = 100
MIXER_MAX_VOLUME = 100
#: MIXER_TRACK = None
MIXER_TRACK = None
#: Which address Mopidy's MPD server should bind to.
#:

12
mopidy/utils/__init__.py
View File

@ -1,3 +1,5 @@
from __future__ import division
import locale
import logging
import os
@ -6,7 +8,7 @@ import sys
logger = logging.getLogger('mopidy.utils')
# TODO: user itertools.chain.from_iterable(the_list)?
# TODO: use itertools.chain.from_iterable(the_list)?
def flatten(the_list):
result = []
for element in the_list:
@ -17,6 +19,14 @@ def flatten(the_list):
return result
def rescale(v, old=None, new=None):
"""Convert value between scales."""
new_min, new_max = new
old_min, old_max = old
scaling = float(new_max - new_min) / (old_max - old_min)
return round(scaling * (v - old_min) + new_min)
def import_module(name):
__import__(name)
return sys.modules[name]

7
mopidy/utils/settings.py
View File

@ -122,6 +122,11 @@ def validate_settings(defaults, settings):
'LOCAL_OUTPUT_OVERRIDE': 'OUTPUT',
'LOCAL_PLAYLIST_FOLDER': 'LOCAL_PLAYLIST_PATH',
'LOCAL_TAG_CACHE': 'LOCAL_TAG_CACHE_FILE',
'MIXER_ALSA_CONTROL': None,
'MIXER_EXT_PORT': None,
'MIXER_EXT_SPEAKERS_A': None,
'MIXER_EXT_SPEAKERS_B': None,
'MIXER_MAX_VOLUME': None,
'SERVER': None,
'SERVER_HOSTNAME': 'MPD_SERVER_HOSTNAME',
'SERVER_PORT': 'MPD_SERVER_PORT',
@ -147,7 +152,7 @@ def validate_settings(defaults, settings):
elif setting == 'OUTPUTS':
errors[setting] = (
u'Deprecated setting, please change to OUTPUT. OUTPUT expectes '
u'Deprecated setting, please change to OUTPUT. OUTPUT expects '
u'a GStreamer bin description string for your desired output.')
elif setting == 'SPOTIFY_BITRATE':

3
tests/frontends/mpd/dispatcher_test.py
View File

@ -2,7 +2,6 @@ from mopidy.backends.dummy import DummyBackend
from mopidy.frontends.mpd.dispatcher import MpdDispatcher
from mopidy.frontends.mpd.exceptions import MpdAckError
from mopidy.frontends.mpd.protocol import request_handlers, handle_request
from mopidy.mixers.dummy import DummyMixer
from tests import unittest
@ -10,12 +9,10 @@ from tests import unittest
class MpdDispatcherTest(unittest.TestCase):
def setUp(self):
self.backend = DummyBackend.start().proxy()
self.mixer = DummyMixer.start().proxy()
self.dispatcher = MpdDispatcher()
def tearDown(self):
self.backend.stop().get()
self.mixer.stop().get()
def test_register_same_pattern_twice_fails(self):
func = lambda: None

3
tests/frontends/mpd/protocol/__init__.py
View File

@ -3,7 +3,6 @@ import mock
from mopidy import settings
from mopidy.backends import dummy as backend
from mopidy.frontends import mpd
from mopidy.mixers import dummy as mixer
from tests import unittest
@ -23,7 +22,6 @@ class MockConnection(mock.Mock):
class BaseTestCase(unittest.TestCase):
def setUp(self):
self.backend = backend.DummyBackend.start().proxy()
self.mixer = mixer.DummyMixer.start().proxy()
self.connection = MockConnection()
self.session = mpd.MpdSession(self.connection)
@ -32,7 +30,6 @@ class BaseTestCase(unittest.TestCase):
def tearDown(self):
self.backend.stop().get()
self.mixer.stop().get()
settings.runtime.clear()
def sendRequest(self, request):

14
tests/frontends/mpd/protocol/playback_test.py
View File

@ -76,37 +76,37 @@ class PlaybackOptionsHandlerTest(protocol.BaseTestCase):
def test_setvol_below_min(self):
self.sendRequest(u'setvol "-10"')
self.assertEqual(0, self.mixer.volume.get())
self.assertEqual(0, self.backend.playback.volume.get())
self.assertInResponse(u'OK')
def test_setvol_min(self):
self.sendRequest(u'setvol "0"')
self.assertEqual(0, self.mixer.volume.get())
self.assertEqual(0, self.backend.playback.volume.get())
self.assertInResponse(u'OK')
def test_setvol_middle(self):
self.sendRequest(u'setvol "50"')
self.assertEqual(50, self.mixer.volume.get())
self.assertEqual(50, self.backend.playback.volume.get())
self.assertInResponse(u'OK')
def test_setvol_max(self):
self.sendRequest(u'setvol "100"')
self.assertEqual(100, self.mixer.volume.get())
self.assertEqual(100, self.backend.playback.volume.get())
self.assertInResponse(u'OK')
def test_setvol_above_max(self):
self.sendRequest(u'setvol "110"')
self.assertEqual(100, self.mixer.volume.get())
self.assertEqual(100, self.backend.playback.volume.get())
self.assertInResponse(u'OK')
def test_setvol_plus_is_ignored(self):
self.sendRequest(u'setvol "+10"')
self.assertEqual(10, self.mixer.volume.get())
self.assertEqual(10, self.backend.playback.volume.get())
self.assertInResponse(u'OK')
def test_setvol_without_quotes(self):
self.sendRequest(u'setvol 50')
self.assertEqual(50, self.mixer.volume.get())
self.assertEqual(50, self.backend.playback.volume.get())
self.assertInResponse(u'OK')
def test_single_off(self):

9
tests/frontends/mpd/status_test.py
View File

@ -1,7 +1,6 @@
from mopidy.backends import dummy as backend
from mopidy.frontends.mpd import dispatcher
from mopidy.frontends.mpd.protocol import status
from mopidy.mixers import dummy as mixer
from mopidy.models import Track
from tests import unittest
@ -17,13 +16,11 @@ STOPPED = backend.PlaybackController.STOPPED
class StatusHandlerTest(unittest.TestCase):
def setUp(self):
self.backend = backend.DummyBackend.start().proxy()
self.mixer = mixer.DummyMixer.start().proxy()
self.dispatcher = dispatcher.MpdDispatcher()
self.context = self.dispatcher.context
def tearDown(self):
self.backend.stop().get()
self.mixer.stop().get()
def test_stats_method(self):
result = status.stats(self.context)
@ -42,13 +39,13 @@ class StatusHandlerTest(unittest.TestCase):
self.assert_('playtime' in result)
self.assert_(int(result['playtime']) >= 0)
def test_status_method_contains_volume_which_defaults_to_0(self):
def test_status_method_contains_volume_with_na_value(self):
result = dict(status.status(self.context))
self.assert_('volume' in result)
self.assertEqual(int(result['volume']), 0)
self.assertEqual(int(result['volume']), -1)
def test_status_method_contains_volume(self):
self.mixer.volume = 17
self.backend.playback.volume = 17
result = dict(status.status(self.context))
self.assert_('volume' in result)
self.assertEqual(int(result['volume']), 17)

21
tests/frontends/mpris/player_interface_test.py
View File

@ -5,7 +5,6 @@ import mock
from mopidy import OptionalDependencyError
from mopidy.backends.dummy import DummyBackend
from mopidy.backends.base.playback import PlaybackController
from mopidy.mixers.dummy import DummyMixer
from mopidy.models import Album, Artist, Track
try:
@ -24,14 +23,12 @@ STOPPED = PlaybackController.STOPPED
class PlayerInterfaceTest(unittest.TestCase):
def setUp(self):
objects.MprisObject._connect_to_dbus = mock.Mock()
self.mixer = DummyMixer.start().proxy()
self.backend = DummyBackend.start().proxy()
self.mpris = objects.MprisObject()
self.mpris._backend = self.backend
def tearDown(self):
self.backend.stop()
self.mixer.stop()
def test_get_playback_status_is_playing_when_playing(self):
self.backend.playback.state = PLAYING
@ -208,36 +205,36 @@ class PlayerInterfaceTest(unittest.TestCase):
self.assertEquals(result['xesam:trackNumber'], 7)
def test_get_volume_should_return_volume_between_zero_and_one(self):
self.mixer.volume = 0
self.backend.playback.volume = 0
result = self.mpris.Get(objects.PLAYER_IFACE, 'Volume')
self.assertEquals(result, 0)
self.mixer.volume = 50
self.backend.playback.volume = 50
result = self.mpris.Get(objects.PLAYER_IFACE, 'Volume')
self.assertEquals(result, 0.5)
self.mixer.volume = 100
self.backend.playback.volume = 100
result = self.mpris.Get(objects.PLAYER_IFACE, 'Volume')
self.assertEquals(result, 1)
def test_set_volume_is_ignored_if_can_control_is_false(self):
self.mpris.get_CanControl = lambda *_: False
self.mixer.volume = 0
self.backend.playback.volume = 0
self.mpris.Set(objects.PLAYER_IFACE, 'Volume', 1.0)
self.assertEquals(self.mixer.volume.get(), 0)
self.assertEquals(self.backend.playback.volume.get(), 0)
def test_set_volume_to_one_should_set_mixer_volume_to_100(self):
self.mpris.Set(objects.PLAYER_IFACE, 'Volume', 1.0)
self.assertEquals(self.mixer.volume.get(), 100)
self.assertEquals(self.backend.playback.volume.get(), 100)
def test_set_volume_to_anything_above_one_should_set_mixer_volume_to_100(self):
self.mpris.Set(objects.PLAYER_IFACE, 'Volume', 2.0)
self.assertEquals(self.mixer.volume.get(), 100)
self.assertEquals(self.backend.playback.volume.get(), 100)
def test_set_volume_to_anything_not_a_number_does_not_change_volume(self):
self.mixer.volume = 10
self.backend.playback.volume = 10
self.mpris.Set(objects.PLAYER_IFACE, 'Volume', None)
self.assertEquals(self.mixer.volume.get(), 10)
self.assertEquals(self.backend.playback.volume.get(), 10)
def test_get_position_returns_time_position_in_microseconds(self):
self.backend.current_playlist.append([Track(uri='a', length=40000)])

18
tests/gstreamer_test.py
View File

@ -11,7 +11,7 @@ from tests import unittest, path_to_data_dir
'Our Windows build server does not support GStreamer yet')
class GStreamerTest(unittest.TestCase):
def setUp(self):
settings.BACKENDS = ('mopidy.backends.local.LocalBackend',)
settings.MIXER = 'fakemixer track_max_volume=65536'
settings.OUTPUT = 'fakesink'
self.song_uri = path_to_uri(path_to_data_dir('song1.wav'))
self.gstreamer = GStreamer()
@ -52,20 +52,10 @@ class GStreamerTest(unittest.TestCase):
def test_end_of_data_stream(self):
pass # TODO
def test_default_get_volume_result(self):
self.assertEqual(100, self.gstreamer.get_volume())
def test_set_volume(self):
self.assertTrue(self.gstreamer.set_volume(50))
self.assertEqual(50, self.gstreamer.get_volume())
def test_set_volume_to_zero(self):
self.assertTrue(self.gstreamer.set_volume(0))
self.assertEqual(0, self.gstreamer.get_volume())
def test_set_volume_to_one_hundred(self):
self.assertTrue(self.gstreamer.set_volume(100))
self.assertEqual(100, self.gstreamer.get_volume())
for value in range(0, 101):
self.assertTrue(self.gstreamer.set_volume(value))
self.assertEqual(value, self.gstreamer.get_volume())
@unittest.SkipTest
def test_set_state_encapsulation(self):

0
tests/mixers/__init__.py
View File

38
tests/mixers/base_test.py
View File

@ -1,38 +0,0 @@
class BaseMixerTest(object):
MIN = 0
MAX = 100
ACTUAL_MIN = MIN
ACTUAL_MAX = MAX
INITIAL = None
mixer_class = None
def setUp(self):
assert self.mixer_class is not None, \
"mixer_class must be set in subclass"
# pylint: disable = E1102
self.mixer = self.mixer_class()
# pylint: enable = E1102
def test_initial_volume(self):
self.assertEqual(self.mixer.volume, self.INITIAL)
def test_volume_set_to_min(self):
self.mixer.volume = self.MIN
self.assertEqual(self.mixer.volume, self.ACTUAL_MIN)
def test_volume_set_to_max(self):
self.mixer.volume = self.MAX
self.assertEqual(self.mixer.volume, self.ACTUAL_MAX)
def test_volume_set_to_below_min_results_in_min(self):
self.mixer.volume = -10
self.assertEqual(self.mixer.volume, self.ACTUAL_MIN)
def test_volume_set_to_above_max_results_in_max(self):
self.mixer.volume = self.MAX + 10
self.assertEqual(self.mixer.volume, self.ACTUAL_MAX)
def test_volume_is_not_float(self):
self.mixer.volume = 1.0 / 3 * 100
self.assertEqual(self.mixer.volume, 33)

42
tests/mixers/denon_test.py
View File

@ -1,42 +0,0 @@
from mopidy.mixers.denon import DenonMixer
from tests.mixers.base_test import BaseMixerTest
from tests import unittest
class DenonMixerDeviceMock(object):
def __init__(self):
self._open = True
self.ret_val = bytes('MV00\r')
def write(self, x):
if x[2] != '?':
self.ret_val = bytes(x)
def read(self, x):
return self.ret_val
def readline(self):
return self.ret_val
def isOpen(self):
return self._open
def open(self):
self._open = True
class DenonMixerTest(BaseMixerTest, unittest.TestCase):
ACTUAL_MAX = 99
INITIAL = 1
mixer_class = DenonMixer
def setUp(self):
self.device = DenonMixerDeviceMock()
self.mixer = DenonMixer(device=self.device)
def test_reopen_device(self):
self.device._open = False
self.mixer.volume = 10
self.assertTrue(self.device.isOpen())

23
tests/mixers/dummy_test.py
View File

@ -1,23 +0,0 @@
from mopidy.mixers.dummy import DummyMixer
from tests import unittest
from tests.mixers.base_test import BaseMixerTest
class DummyMixerTest(BaseMixerTest, unittest.TestCase):
mixer_class = DummyMixer
def test_set_volume_is_capped(self):
self.mixer.amplification_factor = 0.5
self.mixer.volume = 100
self.assertEquals(self.mixer._volume, 50)
def test_get_volume_does_not_show_that_the_volume_is_capped(self):
self.mixer.amplification_factor = 0.5
self.mixer._volume = 50
self.assertEquals(self.mixer.volume, 100)
def test_get_volume_get_the_same_number_as_was_set(self):
self.mixer.amplification_factor = 0.5
self.mixer.volume = 13
self.assertEquals(self.mixer.volume, 13)

15
tests/utils/init_test.py
View File

@ -1,24 +1,27 @@
from mopidy.utils import get_class
from mopidy import utils
from tests import unittest
class GetClassTest(unittest.TestCase):
def test_loading_module_that_does_not_exist(self):
self.assertRaises(ImportError, get_class, 'foo.bar.Baz')
with self.assertRaises(ImportError):
utils.get_class('foo.bar.Baz')
def test_loading_class_that_does_not_exist(self):
self.assertRaises(ImportError, get_class, 'unittest.FooBarBaz')
with self.assertRaises(ImportError):
utils.get_class('unittest.FooBarBaz')
def test_loading_incorrect_class_path(self):
self.assertRaises(ImportError, get_class, 'foobarbaz')
with self.assertRaises(ImportError):
utils.get_class('foobarbaz')
def test_import_error_message_contains_complete_class_path(self):
try:
get_class('foo.bar.Baz')
utils.get_class('foo.bar.Baz')
except ImportError as e:
self.assert_('foo.bar.Baz' in str(e))
def test_loading_existing_class(self):
cls = get_class('unittest.TestCase')
cls = utils.get_class('unittest.TestCase')
self.assertEqual(cls.__name__, 'TestCase')