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
scroll_controller = Gtk.EventControllerScroll.new(Gtk.EventControllerScrollFlags.VERTICAL)
63
scroll_controller.set_propagation_phase(Gtk.PropagationPhase.CAPTURE)
64
self.volume_popover.add_controller(scroll_controller)
65
scroll_controller.connect("scroll", self.on_popover_scroll)
66
scroll_controller = Gtk.EventControllerScroll.new(Gtk.EventControllerScrollFlags.VERTICAL)
67
scroll_controller.set_propagation_phase(Gtk.PropagationPhase.CAPTURE)
68
self.add_controller(scroll_controller)
69
scroll_controller.connect("scroll", self.on_button_scroll)
70
self.popdown_timeout = None
71
self.set_button_icon()
72
gesture_click = Gtk.GestureClick()
73
gesture_click.set_button(Gdk.BUTTON_MIDDLE)
74
gesture_click.connect("released", self.on_middle_click_released)
75
self.add_controller(gesture_click)
76
self.append(self.menu_button)
77
loop = asyncio.get_event_loop()
79
loop.create_task(self.listen())
80
def on_middle_click_released(self, gesture, n_press, x, y):
82
loop = asyncio.get_event_loop()
83
loop.create_task(self.idle_on_middle_click_released())
84
async def idle_on_middle_click_released(self):
86
if self.default_sink.mute:
87
await self.pulse.sink_mute(self.default_sink.index, False)
88
self.default_sink.mute = False
89
else:
90
await self.pulse.sink_mute(self.default_sink.index, True)
91
self.default_sink.mute = True
92
self.set_button_icon()
93
def on_popover_show(self, popover):
95
if self.popdown_timeout is not None:
96
GLib.source_remove(self.popdown_timeout)
97
self.popdown_timeout = GLib.timeout_add(2000, self.close_scale_popover)
98
def set_button_icon(self):
100
volume = self.current_volume
101
threshold = 1.0 / 3.0
102
if volume == 0.0 or self.default_sink.mute:
103
self.menu_button.set_icon_name("audio-volume-muted-symbolic")
104
elif volume < threshold:
105
self.menu_button.set_icon_name("audio-volume-low-symbolic")
106
elif volume < threshold * 2:
107
self.menu_button.set_icon_name("audio-volume-medium-symbolic")
108
else:
109
self.menu_button.set_icon_name("audio-volume-high-symbolic")
110
def on_button_scroll(self, controller, dx, dy):
112
if dy < 0 or dy > 0:
113
self.menu_button.popup()
114
return Gdk.EVENT_STOP
115
def on_popover_scroll(self, controller, dx, dy):
117
if dy > 0:
118
new_volume = self.current_volume - self.volume_step
119
elif dy < 0:
120
new_volume = self.current_volume + self.volume_step
121
if new_volume < 0:
122
new_volume = 0.0
123
elif new_volume > 1:
124
new_volume = 1.0
125
if self.current_volume == new_volume:
126
return Gdk.EVENT_STOP
127
self.volume_scale.set_value(new_volume)
128
self.current_volume = new_volume
129
self.set_button_icon()
130
if self.popdown_timeout is not None:
131
GLib.source_remove(self.popdown_timeout)
132
self.popdown_timeout = GLib.timeout_add(2000, self.close_scale_popover)
133
return Gdk.EVENT_STOP
134
async def idle_volume_update(self):
136
new_volume = self.volume_scale.get_value()
137
await self.pulse.volume_set_all_chans(self.default_sink, new_volume)
138
self.current_volume = new_volume
139
self.set_button_icon()
140
if self.popdown_timeout is not None:
141
GLib.source_remove(self.popdown_timeout)
142
self.popdown_timeout = GLib.timeout_add(2000, self.close_scale_popover)
143
return False
144
def on_scale_changed(self, scale, *args):
146
loop = asyncio.get_event_loop()
147
loop.create_task(self.idle_volume_update())
148
async def idle_update_scale(self):
150
server_info = await self.pulse.server_info()
151
self.default_sink = await self.pulse.get_sink_by_name(server_info.default_sink_name)
152
new_volume = self.default_sink.volume.value_flat
153
if self.current_volume == new_volume:
154
self.set_button_icon()
155
return False
156
self.current_volume = new_volume
157
self.volume_scale.set_value(self.current_volume)
158
self.set_button_icon()
159
self.menu_button.popup()
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