-
-
Notifications
You must be signed in to change notification settings - Fork 620
/
soundSplit.py
310 lines (272 loc) · 10.6 KB
/
soundSplit.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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
# A part of NonVisual Desktop Access (NVDA)
# Copyright (C) 2024 NV Access Limited
# This file is covered by the GNU General Public License.
# See the file COPYING for more details.
import atexit
import config
from enum import IntEnum, unique
import globalVars
from logHandler import log
import nvwave
from pycaw.api.audiopolicy import IAudioSessionManager2
from pycaw.callbacks import AudioSessionNotification, AudioSessionEvents
from pycaw.utils import AudioSession, AudioUtilities
import ui
from utils.displayString import DisplayStringIntEnum
from dataclasses import dataclass
import os
from comtypes import COMError
from threading import Lock
import core
VolumeTupleT = tuple[float, float]
@unique
class SoundSplitState(DisplayStringIntEnum):
OFF = 0
NVDA_BOTH_APPS_BOTH = 1
NVDA_LEFT_APPS_RIGHT = 2
NVDA_LEFT_APPS_BOTH = 3
NVDA_RIGHT_APPS_LEFT = 4
NVDA_RIGHT_APPS_BOTH = 5
NVDA_BOTH_APPS_LEFT = 6
NVDA_BOTH_APPS_RIGHT = 7
@property
def _displayStringLabels(self) -> dict[IntEnum, str]:
return {
# Translators: Sound split state
SoundSplitState.OFF: pgettext("SoundSplit", "Sound split disabled"),
SoundSplitState.NVDA_BOTH_APPS_BOTH: pgettext(
"SoundSplit",
# Translators: Sound split state
"NVDA in both channels and applications in both channels",
),
# Translators: Sound split state
SoundSplitState.NVDA_LEFT_APPS_RIGHT: _("NVDA on the left and applications on the right"),
# Translators: Sound split state
SoundSplitState.NVDA_LEFT_APPS_BOTH: _("NVDA on the left and applications in both channels"),
# Translators: Sound split state
SoundSplitState.NVDA_RIGHT_APPS_LEFT: _("NVDA on the right and applications on the left"),
# Translators: Sound split state
SoundSplitState.NVDA_RIGHT_APPS_BOTH: _("NVDA on the right and applications in both channels"),
# Translators: Sound split state
SoundSplitState.NVDA_BOTH_APPS_LEFT: _("NVDA in both channels and applications on the left"),
# Translators: Sound split state
SoundSplitState.NVDA_BOTH_APPS_RIGHT: _("NVDA in both channels and applications on the right"),
}
def getAppVolume(self) -> VolumeTupleT:
match self:
case (
SoundSplitState.NVDA_BOTH_APPS_BOTH
| SoundSplitState.NVDA_LEFT_APPS_BOTH
| SoundSplitState.NVDA_RIGHT_APPS_BOTH
):
return (1.0, 1.0)
case SoundSplitState.NVDA_RIGHT_APPS_LEFT | SoundSplitState.NVDA_BOTH_APPS_LEFT:
return (1.0, 0.0)
case SoundSplitState.NVDA_LEFT_APPS_RIGHT | SoundSplitState.NVDA_BOTH_APPS_RIGHT:
return (0.0, 1.0)
case _:
raise RuntimeError(f"Unexpected or unknown state {self=}")
def getNVDAVolume(self) -> VolumeTupleT:
match self:
case (
SoundSplitState.NVDA_BOTH_APPS_BOTH
| SoundSplitState.NVDA_BOTH_APPS_LEFT
| SoundSplitState.NVDA_BOTH_APPS_RIGHT
):
return (1.0, 1.0)
case SoundSplitState.NVDA_LEFT_APPS_RIGHT | SoundSplitState.NVDA_LEFT_APPS_BOTH:
return (1.0, 0.0)
case SoundSplitState.NVDA_RIGHT_APPS_LEFT | SoundSplitState.NVDA_RIGHT_APPS_BOTH:
return (0.0, 1.0)
case _:
raise RuntimeError(f"Unexpected or unknown state {self=}")
audioSessionManager: IAudioSessionManager2 | None = None
activeCallback: AudioSessionNotification | None = None
def initialize() -> None:
if nvwave.usingWasapiWavePlayer():
global audioSessionManager
try:
audioSessionManager = AudioUtilities.GetAudioSessionManager()
except COMError:
log.exception("Could not initialize audio session manager")
return
state = SoundSplitState(config.conf["audio"]["soundSplitState"])
config.conf["audio"]["applicationsMuted"] = False
setSoundSplitState(state, initial=True)
else:
log.debug("Cannot initialize sound split as WASAPI is disabled")
@atexit.register
def terminate():
if nvwave.usingWasapiWavePlayer():
state = SoundSplitState(config.conf["audio"]["soundSplitState"])
if state != SoundSplitState.OFF:
setSoundSplitState(SoundSplitState.OFF, appsVolume=1.0)
unregisterCallback()
else:
log.debug("Skipping terminating sound split as WASAPI is disabled.")
@dataclass(unsafe_hash=True)
class AudioSessionNotificationWrapper(AudioSessionNotification):
listener: AudioSessionNotification
def on_session_created(self, new_session: AudioSession):
pid = new_session.ProcessId
with applicationExitCallbacksLock:
if pid not in applicationExitCallbacks:
volumeRestorer = VolumeRestorer(pid, new_session)
new_session.register_notification(volumeRestorer)
applicationExitCallbacks[pid] = volumeRestorer
self.listener.on_session_created(new_session)
def applyToAllAudioSessions(
callback: AudioSessionNotification,
applyToFuture: bool = True,
) -> None:
"""
Executes provided callback function on all active audio sessions.
Additionally, if applyToFuture is True, then it will register a notification with audio session manager,
which will execute the same callback for all future sessions as they are created.
That notification will be active until next invokation of this function,
or until unregisterCallback() is called.
"""
unregisterCallback()
callback = AudioSessionNotificationWrapper(callback)
if applyToFuture:
audioSessionManager.RegisterSessionNotification(callback)
# The following call is required to make callback to work:
audioSessionManager.GetSessionEnumerator()
global activeCallback
activeCallback = callback
sessions: list[AudioSession] = AudioUtilities.GetAllSessions()
for session in sessions:
callback.on_session_created(session)
def unregisterCallback() -> None:
global activeCallback
if activeCallback is not None:
audioSessionManager.UnregisterSessionNotification(activeCallback)
activeCallback = None
@dataclass(unsafe_hash=True)
class VolumeSetter(AudioSessionNotification):
leftVolume: float
rightVolume: float
leftNVDAVolume: float
rightNVDAVolume: float
foundSessionWithNot2Channels: bool = False
def on_session_created(self, new_session: AudioSession):
pid = new_session.ProcessId
process = new_session.Process
if process is not None:
exe = os.path.basename(process.exe())
isNvda = exe.lower() == "nvda.exe"
else:
isNvda = False
channelVolume = new_session.channelAudioVolume()
channelCount = channelVolume.GetChannelCount()
if channelCount != 2:
log.warning(f"Audio session for pid {pid} has {channelCount} channels instead of 2 - cannot set volume!")
self.foundSessionWithNot2Channels = True
return
if pid == globalVars.appPid:
channelVolume.SetChannelVolume(0, self.leftNVDAVolume, None)
channelVolume.SetChannelVolume(1, self.rightNVDAVolume, None)
elif isNvda:
# This might be NVDA running on secure screen; don't adjust its volume
pass
else:
channelVolume.SetChannelVolume(0, self.leftVolume, None)
channelVolume.SetChannelVolume(1, self.rightVolume, None)
def setSoundSplitState(
state: SoundSplitState,
appsVolume: float | None = None,
initial: bool = False
) -> dict:
applyToFuture = True
if state == SoundSplitState.OFF:
if initial:
return {}
else:
# Disabling sound split via command or via settings
# We need to restore volume of all applications, but then don't set up callback for future audio sessions
state = SoundSplitState.NVDA_BOTH_APPS_BOTH
applyToFuture = False
leftVolume, rightVolume = state.getAppVolume()
if appsVolume is None:
appsVolume = (
config.conf["audio"]["applicationsSoundVolume"] / 100
* (1 - int(config.conf["audio"]["applicationsMuted"]))
)
leftVolume *= appsVolume
rightVolume *= appsVolume
leftNVDAVolume, rightNVDAVolume = state.getNVDAVolume()
volumeSetter = VolumeSetter(leftVolume, rightVolume, leftNVDAVolume, rightNVDAVolume)
applyToAllAudioSessions(volumeSetter, applyToFuture=applyToFuture)
return {
"foundSessionWithNot2Channels": volumeSetter.foundSessionWithNot2Channels,
}
def updateSoundSplitState(increment: int | None = None) -> None:
if not nvwave.usingWasapiWavePlayer():
message = _(
# Translators: error message when wasapi is turned off.
"Sound split cannot be used. "
"Please enable WASAPI in the Advanced category in NVDA Settings to use it."
)
ui.message(message)
return
state = SoundSplitState(config.conf["audio"]["soundSplitState"])
if increment is None:
newState = state
else:
allowedStates: list[int] = config.conf["audio"]["includedSoundSplitModes"]
try:
i = allowedStates.index(state)
except ValueError:
# State not found, resetting to default (OFF)
i = -1
i = (i + increment) % len(allowedStates)
newState = SoundSplitState(allowedStates[i])
result = setSoundSplitState(newState)
config.conf["audio"]["soundSplitState"] = newState.value
if increment is not None:
ui.message(newState.displayString)
if result["foundSessionWithNot2Channels"]:
msg = _(
# Translators: warning message when sound split trigger wasn't successful due to one of audio sessions
# had number of channels other than 2 .
"Warning: couldn't set volumes for sound split: "
"one of audio sessions is either mono, or has more than 2 audio channels."
)
ui.message(msg)
@dataclass(unsafe_hash=True)
class VolumeRestorer(AudioSessionEvents):
pid: int
audioSession: AudioSession
def on_state_changed(self, new_state: str, new_state_id: int):
if new_state == "Expired":
# For some reason restoring volume doesn't work in this thread, so scheduling in the main thread.
core.callLater(0, self.restoreVolume)
def restoreVolume(self):
# Application connected to this audio session is terminating. Restore its volume.
try:
channelVolume = self.audioSession.channelAudioVolume()
channelCount = channelVolume.GetChannelCount()
if channelCount != 2:
log.warning(
f"Audio session for pid {self.pid} has {channelCount} channels instead of 2 - cannot set volume!"
)
return
channelVolume.SetChannelVolume(0, 1.0, None)
channelVolume.SetChannelVolume(1, 1.0, None)
except Exception:
log.exception(f"Could not restore volume of process {self.pid} upon exit.")
self.unregister()
def unregister(self):
with applicationExitCallbacksLock:
try:
del applicationExitCallbacks[self.pid]
except KeyError:
pass
try:
self.audioSession.unregister_notification()
except Exception:
log.exception(f"Cannot unregister audio session for process {self.pid}")
applicationExitCallbacksLock = Lock()
applicationExitCallbacks: dict[int, VolumeRestorer] = {}
def toggleSoundSplitState() -> None:
updateSoundSplitState(increment=1)