85 lines
3.3 KiB
Python
85 lines
3.3 KiB
Python
"""VolumeMeter — a single-channel volume control styled as a segmented meter
|
|
bar (a Gtk.Scale dressed up via CSS, see .sb-meter in style/style.css) rather
|
|
than a bare slider, since the user asked for "an audio meter" per channel.
|
|
|
|
Two guards keep it usable against a live, event-driven backend:
|
|
- commits are debounced (COMMIT_DEBOUNCE_MS) so dragging doesn't spawn a
|
|
`pactl set-*-volume` per pixel of motion:
|
|
- `set_value_quiet()` (called when a `pactl subscribe` refresh reports the
|
|
authoritative value) is a no-op while the user has the handle pressed, so
|
|
an in-flight drag never gets yanked back by its own not-yet-applied change.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Callable
|
|
|
|
import gi
|
|
|
|
gi.require_version("Gtk", "4.0")
|
|
from gi.repository import GLib, Gtk # noqa: E402
|
|
|
|
COMMIT_DEBOUNCE_MS = 80
|
|
|
|
|
|
class VolumeMeter(Gtk.Box):
|
|
def __init__(self, label: str, value: int, on_change: Callable[[int], None],
|
|
muted: bool = False) -> None:
|
|
super().__init__(orientation=Gtk.Orientation.HORIZONTAL, spacing=6)
|
|
self.add_css_class("sb-meter-row")
|
|
self._on_change = on_change
|
|
self._dragging = False
|
|
self._commit_id: int | None = None
|
|
self._pending: int | None = None
|
|
|
|
if label:
|
|
tag = Gtk.Label(label=label)
|
|
tag.add_css_class("sb-meter-label")
|
|
self.append(tag)
|
|
|
|
self.scale = Gtk.Scale.new_with_range(Gtk.Orientation.HORIZONTAL, 0, 150, 1)
|
|
self.scale.set_value(value)
|
|
self.scale.set_draw_value(True)
|
|
self.scale.set_value_pos(Gtk.PositionType.RIGHT)
|
|
self.scale.set_hexpand(True)
|
|
self.scale.set_size_request(120, -1)
|
|
self.scale.add_css_class("sb-meter")
|
|
self.scale.add_mark(100, Gtk.PositionType.BOTTOM, None)
|
|
self.scale.connect("value-changed", self._on_value_changed)
|
|
self.set_muted(muted)
|
|
|
|
click = Gtk.GestureClick()
|
|
click.set_propagation_phase(Gtk.PropagationPhase.CAPTURE)
|
|
click.connect("pressed", lambda *_a: setattr(self, "_dragging", True))
|
|
click.connect("released", lambda *_a: setattr(self, "_dragging", False))
|
|
self.scale.add_controller(click)
|
|
|
|
self.append(self.scale)
|
|
|
|
def _on_value_changed(self, _scale) -> None:
|
|
self._pending = int(self.scale.get_value())
|
|
if self._commit_id is not None:
|
|
GLib.source_remove(self._commit_id)
|
|
self._commit_id = GLib.timeout_add(COMMIT_DEBOUNCE_MS, self._commit)
|
|
|
|
def _commit(self) -> bool:
|
|
self._commit_id = None
|
|
if self._pending is not None:
|
|
self._on_change(self._pending)
|
|
return False
|
|
|
|
def set_value_quiet(self, value: int) -> None:
|
|
"""Reflect a backend-reported value without re-triggering on_change."""
|
|
if self._dragging or self._commit_id is not None:
|
|
return # user (or our own not-yet-applied commit) owns the value right now
|
|
if int(self.scale.get_value()) != value:
|
|
self.scale.handler_block_by_func(self._on_value_changed)
|
|
self.scale.set_value(value)
|
|
self.scale.handler_unblock_by_func(self._on_value_changed)
|
|
|
|
def set_muted(self, muted: bool) -> None:
|
|
if muted:
|
|
self.scale.add_css_class("sb-meter-muted")
|
|
else:
|
|
self.scale.remove_css_class("sb-meter-muted")
|