Python script, ASCII text executable
1
"""
2
The MIT License (MIT)
3
Copyright (c) 2025 Scott Moreau <oreaus@gmail.com>
5
Permission is hereby granted, free of charge, to any person obtaining a copy
7
of this software and associated documentation files (the "Software"), to deal
8
in the Software without restriction, including without limitation the rights
9
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
10
copies of the Software, and to permit persons to whom the Software is
11
furnished to do so, subject to the following conditions:
12
The above copyright notice and this permission notice shall be included in
14
all copies or substantial portions of the Software.
15
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
19
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
21
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
22
SOFTWARE.
23
"""
24
import panorama_panel
26
import gi
28
gi.require_version("Gtk", "4.0")
29
from gi.repository import Gtk, Gdk, GLib
31
import asyncio
33
import pulsectl
34
import pulsectl_asyncio
35
class Volume(panorama_panel.Applet):
37
name = "Volume"
38
description = "Volume applet"
39
def __init__(self, orientation=Gtk.Orientation.HORIZONTAL, config=None):
41
super().__init__()
42
asyncio.create_task(self.main())
43
async def main(self):
45
self.menu_button = Gtk.MenuButton()
46
self.volume_step = 0.05
47
self.pulse = pulsectl_asyncio.PulseAsync("panorama-panel-volume-manager")
48
await self.pulse.connect()
49
server_info = await self.pulse.server_info()
50
self.default_sink = await self.pulse.get_sink_by_name(server_info.default_sink_name)
51
self.current_volume = self.default_sink.volume.value_flat
52
self.volume_adjustment = Gtk.Adjustment(value=0.5, lower=0.0, upper=1.0, step_increment=0.01, page_increment=0.1)
53
self.volume_scale = Gtk.Scale.new(Gtk.Orientation.HORIZONTAL, self.volume_adjustment)
54
self.volume_scale.set_hexpand(True)
55
self.volume_scale.set_value(self.current_volume)
56
self.volume_scale.set_size_request(200, -1)
57
self.volume_scale_changed_handler = self.volume_scale.connect("value-changed", self.on_scale_changed)
58
self.volume_popover = Gtk.Popover()
59
self.volume_popover.set_child(self.volume_scale)
60
self.volume_popover.connect("show", self.on_popover_show)
61
self.menu_button.set_popover(self.volume_popover)
62
self.menu_button.set_has_frame(False)
63
scroll_controller = Gtk.EventControllerScroll.new(Gtk.EventControllerScrollFlags.VERTICAL)
64
scroll_controller.set_propagation_phase(Gtk.PropagationPhase.CAPTURE)
65
self.volume_popover.add_controller(scroll_controller)
66
scroll_controller.connect("scroll", self.on_popover_scroll)
67
scroll_controller = Gtk.EventControllerScroll.new(Gtk.EventControllerScrollFlags.VERTICAL)
68
scroll_controller.set_propagation_phase(Gtk.PropagationPhase.CAPTURE)
69
self.add_controller(scroll_controller)
70
scroll_controller.connect("scroll", self.on_button_scroll)
71
self.popdown_timeout = None
72
self.set_button_icon()
73
gesture_click = Gtk.GestureClick()
74
gesture_click.set_button(Gdk.BUTTON_MIDDLE)
75
gesture_click.connect("released", self.on_middle_click_released)
76
self.add_controller(gesture_click)
77
self.append(self.menu_button)
78
loop = asyncio.get_event_loop()
80
loop.create_task(self.listen())
81
def on_middle_click_released(self, gesture, n_press, x, y):
83
loop = asyncio.get_event_loop()
84
loop.create_task(self.idle_on_middle_click_released())
85
async def idle_on_middle_click_released(self):
87
if self.default_sink.mute:
88
await self.pulse.sink_mute(self.default_sink.index, False)
89
self.default_sink.mute = False
90
else:
91
await self.pulse.sink_mute(self.default_sink.index, True)
92
self.default_sink.mute = True
93
self.set_button_icon()
94
def on_popover_show(self, popover):
96
if self.popdown_timeout is not None:
97
GLib.source_remove(self.popdown_timeout)
98
self.popdown_timeout = GLib.timeout_add(2000, self.close_scale_popover)
99
def set_button_icon(self):
101
volume = self.current_volume
102
threshold = 1.0 / 3.0
103
if volume == 0.0 or self.default_sink.mute:
104
self.menu_button.set_icon_name("audio-volume-muted-symbolic")
105
elif volume < threshold:
106
self.menu_button.set_icon_name("audio-volume-low-symbolic")
107
elif volume < threshold * 2:
108
self.menu_button.set_icon_name("audio-volume-medium-symbolic")
109
else:
110
self.menu_button.set_icon_name("audio-volume-high-symbolic")
111
def on_button_scroll(self, controller, dx, dy):
113
if dy < 0 or dy > 0:
114
self.menu_button.popup()
115
return Gdk.EVENT_STOP
116
def on_popover_scroll(self, controller, dx, dy):
118
if dy > 0:
119
new_volume = self.current_volume - self.volume_step
120
elif dy < 0:
121
new_volume = self.current_volume + self.volume_step
122
if new_volume < 0:
123
new_volume = 0.0
124
elif new_volume > 1:
125
new_volume = 1.0
126
if self.current_volume == new_volume:
127
return Gdk.EVENT_STOP
128
self.volume_scale.set_value(new_volume)
129
self.current_volume = new_volume
130
self.set_button_icon()
131
if self.popdown_timeout is not None:
132
GLib.source_remove(self.popdown_timeout)
133
self.popdown_timeout = GLib.timeout_add(2000, self.close_scale_popover)
134
return Gdk.EVENT_STOP
135
async def idle_volume_update(self):
137
new_volume = self.volume_scale.get_value()
138
await self.pulse.volume_set_all_chans(self.default_sink, new_volume)
139
self.current_volume = new_volume
140
self.set_button_icon()
141
if self.popdown_timeout is not None:
142
GLib.source_remove(self.popdown_timeout)
143
self.popdown_timeout = GLib.timeout_add(2000, self.close_scale_popover)
144
return False
145
def on_scale_changed(self, scale, *args):
147
loop = asyncio.get_event_loop()
148
loop.create_task(self.idle_volume_update())
149
async def idle_update_scale(self):
151
server_info = await self.pulse.server_info()
152
self.default_sink = await self.pulse.get_sink_by_name(server_info.default_sink_name)
153
new_volume = self.default_sink.volume.value_flat
154
if self.current_volume == new_volume:
155
self.set_button_icon()
156
return False
157
self.current_volume = new_volume
158
self.volume_scale.set_value(self.current_volume)
159
self.set_button_icon()
160
if self.popdown_timeout is not None:
161
GLib.source_remove(self.popdown_timeout)
162
self.popdown_timeout = GLib.timeout_add(2000, self.close_scale_popover)
163
return False
164
def close_scale_popover(self):
166
self.popdown_timeout = None
167
self.menu_button.popdown()
168
return False
169
async def listen(self):
171
async for event in self.pulse.subscribe_events("all"):
172
if event.facility == pulsectl.PulseEventFacilityEnum.sink and \
173
event.t == pulsectl.PulseEventTypeEnum.change:
174
try:
175
sink = await self.pulse.sink_info(event.index)
176
loop = asyncio.get_event_loop()
177
loop.create_task(self.idle_update_scale())
178
except pulsectl_asyncio.PulseOperationFailed:
179
print(f"Could not retrieve info for sink index {event.index}")
180