-
Notifications
You must be signed in to change notification settings - Fork 686
/
mixer.py
134 lines (100 loc) · 3.76 KB
/
mixer.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
from __future__ import annotations
import contextlib
import logging
from collections.abc import Generator, Iterable
from typing import TYPE_CHECKING, Any
from pykka.typing import proxy_method
from mopidy import exceptions
from mopidy.internal import validation
from mopidy.internal.models import MixerState
if TYPE_CHECKING:
from mopidy.mixer import MixerProxy
from mopidy.types import Percentage
logger = logging.getLogger(__name__)
@contextlib.contextmanager
def _mixer_error_handling(mixer: MixerProxy) -> Generator[None, Any, None]:
try:
yield
except exceptions.ValidationError as e:
logger.error(
"%s mixer returned bad data: %s",
mixer.actor_ref.actor_class.__name__,
e,
)
except Exception:
logger.exception(
"%s mixer caused an exception.",
mixer.actor_ref.actor_class.__name__,
)
class MixerController:
def __init__(self, mixer: MixerProxy | None) -> None:
self._mixer = mixer
def get_volume(self) -> Percentage | None:
"""Get the volume.
Integer in range [0..100] or :class:`None` if unknown.
The volume scale is linear.
"""
if self._mixer is None:
return None
with _mixer_error_handling(self._mixer):
volume = self._mixer.get_volume().get()
if volume is not None:
validation.check_integer(volume, min=0, max=100)
return volume
return None
def set_volume(self, volume: Percentage) -> bool:
"""Set the volume.
The volume is defined as an integer in range [0..100].
The volume scale is linear.
Returns :class:`True` if call is successful, otherwise :class:`False`.
"""
validation.check_integer(volume, min=0, max=100)
if self._mixer is None:
return False # TODO: 2.0 return None
with _mixer_error_handling(self._mixer):
result = self._mixer.set_volume(volume).get()
validation.check_instance(result, bool)
return result
return False
def get_mute(self) -> bool | None:
"""Get mute state.
:class:`True` if muted, :class:`False` unmuted, :class:`None` if
unknown.
"""
if self._mixer is None:
return None
with _mixer_error_handling(self._mixer):
mute = self._mixer.get_mute().get()
if mute is not None:
validation.check_instance(mute, bool)
return mute
return None
def set_mute(self, mute: bool) -> bool:
"""Set mute state.
:class:`True` to mute, :class:`False` to unmute.
Returns :class:`True` if call is successful, otherwise :class:`False`.
"""
validation.check_boolean(mute)
if self._mixer is None:
return False # TODO: 2.0 return None
with _mixer_error_handling(self._mixer):
result = self._mixer.set_mute(bool(mute)).get()
validation.check_instance(result, bool)
return result
return False
def _save_state(self) -> MixerState:
return MixerState(
volume=self.get_volume(),
mute=self.get_mute(),
)
def _load_state(self, state: MixerState, coverage: Iterable[str]) -> None:
if state and "mixer" in coverage:
if state.mute is not None:
self.set_mute(state.mute)
if state.volume is not None:
self.set_volume(state.volume)
class MixerControllerProxy:
get_volume = proxy_method(MixerController.get_volume)
set_volume = proxy_method(MixerController.set_volume)
get_mute = proxy_method(MixerController.get_mute)
set_mute = proxy_method(MixerController.set_mute)