129 lines
4.1 KiB
Python
Executable File
129 lines
4.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
from gi import require_version
|
|
|
|
require_version("AstalWp", "0.1")
|
|
from gi.repository import AstalWp as Wp, Gio, GObject
|
|
from gi.events import GLibEventLoopPolicy
|
|
import json
|
|
import asyncio
|
|
import sys
|
|
from typing import Any, Literal
|
|
from enum import IntEnum
|
|
import signal
|
|
import math
|
|
|
|
|
|
class EndpointBinding:
|
|
"""Collection of connection IDs for an audio endpoint."""
|
|
|
|
def __init__(self, endpoint: Wp.Endpoint, callback):
|
|
self.callback = callback
|
|
self.volume_binding = endpoint.connect_after("notify::volume", callback)
|
|
self.mute_binding = endpoint.connect_after("notify::mute", callback)
|
|
self.id = endpoint.props.id
|
|
self.endpoint = endpoint
|
|
|
|
def hash(self) -> int:
|
|
return self.volume_binding + (self.mute_binding << 32)
|
|
|
|
def __del__(self):
|
|
"""Object cleanup."""
|
|
self.endpoint.disconnect_by_func(self.callback)
|
|
|
|
|
|
class AudioController(Gio.Application):
|
|
"""Audio controller application."""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.wp = Wp.get_default()
|
|
self.bindings = dict[int, EndpointBinding]()
|
|
|
|
def do_activate(self):
|
|
self.wp.connect("ready", self._on_ready)
|
|
self.hold()
|
|
loop = asyncio.get_running_loop()
|
|
for sig in (signal.SIGTERM, signal.SIGHUP, signal.SIGINT):
|
|
loop.add_signal_handler(sig, self.release)
|
|
|
|
def _on_ready(self, wp: Wp.Wp):
|
|
self.audio = self.wp.get_audio()
|
|
if len(sys.argv) > 1:
|
|
# this is a set-default call, so lets do that.
|
|
self.set_default(sys.argv[1], " ".join(sys.argv[2:]))
|
|
else:
|
|
self.audio.connect("microphone-added", self._device_added)
|
|
self.audio.connect("speaker-added", self._device_added)
|
|
self.audio.connect("microphone-removed", self._device_removed)
|
|
self.audio.connect("speaker-removed", self._device_removed)
|
|
|
|
for mic in self.audio.props.microphones:
|
|
self.bindings[mic.props.id] = EndpointBinding(mic, self._dump_info)
|
|
for sink in self.audio.props.speakers:
|
|
self.bindings[sink.props.id] = EndpointBinding(sink, self._dump_info)
|
|
|
|
self._dump_info()
|
|
|
|
def set_default(self, dir: Literal["in"] | Literal["out"], description: str):
|
|
"""Set the default source or sink."""
|
|
if dir == "in":
|
|
arr: list[Wp.Endpoint] = self.audio.get_microphones()
|
|
else:
|
|
arr = self.audio.get_speakers()
|
|
|
|
for dev in arr:
|
|
if dev.get_description() == description:
|
|
dev.set_is_default(True)
|
|
break
|
|
|
|
self.release()
|
|
|
|
def _device_added(self, audio: Wp.Audio, device: Wp.Endpoint):
|
|
self.bindings[device.props.id] = EndpointBinding(device, self._dump_info)
|
|
self._dump_info()
|
|
|
|
def _device_removed(self, audio: Wp.Audio, device: Wp.Endpoint):
|
|
try:
|
|
del self.bindings[device.props.id]
|
|
except KeyError:
|
|
pass
|
|
self._dump_info()
|
|
|
|
def _dump_info(self, *args):
|
|
asyncio.ensure_future(self._do_dump())
|
|
|
|
async def _do_dump(self):
|
|
|
|
json.dump(
|
|
{
|
|
"inputs": [
|
|
mic.get_description() for mic in self.audio.get_microphones()
|
|
],
|
|
"outputs": [
|
|
sink.get_description() for sink in self.audio.get_speakers()
|
|
],
|
|
"input": {
|
|
"volume": round(
|
|
self.audio.get_default_microphone().get_volume() * 100
|
|
),
|
|
"mute": self.audio.get_default_microphone().get_mute(),
|
|
},
|
|
"output": {
|
|
"volume": round(
|
|
self.audio.get_default_speaker().get_volume() * 100
|
|
),
|
|
"mute": self.audio.get_default_speaker().get_mute(),
|
|
},
|
|
},
|
|
sys.stdout,
|
|
)
|
|
print()
|
|
sys.stdout.flush()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.set_event_loop_policy(GLibEventLoopPolicy())
|
|
app = AudioController()
|
|
app.run()
|