132 lines
5.6 KiB
Python
132 lines
5.6 KiB
Python
"""ApplicationCard — one card in the Applications tab: everything MeterCard
|
|
gives every card (icon+name, live peak bar, volume meter/L-R pair, mute),
|
|
plus a dropdown to route this one app's playback stream to a different
|
|
output device (`pactl move-sink-input`), and — mirroring that — a second
|
|
dropdown to route the same app's *microphone* stream (if it has one) to a
|
|
different input device (`pactl move-source-output`). The meter/mute/peak
|
|
controls above stay tied to the app's playback stream (this tab is keyed by
|
|
sink-inputs); the input dropdown is purely a routing control for whichever
|
|
source-output shares this app's `application.process.binary` (falling back
|
|
to matching by name), disabled when the app isn't recording anything.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import gi
|
|
|
|
gi.require_version("Gtk", "4.0")
|
|
from gi.repository import Gtk # noqa: E402
|
|
|
|
from ui.card_base import MeterCard
|
|
|
|
|
|
class ApplicationCard(MeterCard):
|
|
def __init__(self, backend, peaks, item: dict, sinks: list[dict],
|
|
sources: list[dict], source_outputs: list[dict]) -> None:
|
|
self._sinks = sinks
|
|
self._sink_names: list[str] = []
|
|
self._sources = sources
|
|
self._source_names: list[str] = []
|
|
self._recording: dict | None = None
|
|
super().__init__(backend, peaks, "sink-input", item)
|
|
self._update_extra(item, sinks, sources, source_outputs)
|
|
|
|
def _build_extra(self, item: dict):
|
|
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8)
|
|
|
|
out_col = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2)
|
|
out_label = Gtk.Label(label="Output", xalign=0.0)
|
|
out_label.add_css_class("sb-field-label")
|
|
out_col.append(out_label)
|
|
self._sink_dropdown = Gtk.DropDown()
|
|
self._sink_dropdown.add_css_class("sb-device-dropdown")
|
|
self._sink_guard = False
|
|
self._sink_dropdown.connect("notify::selected", self._on_sink_selected)
|
|
out_col.append(self._sink_dropdown)
|
|
box.append(out_col)
|
|
|
|
in_col = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2)
|
|
in_label = Gtk.Label(label="Input", xalign=0.0)
|
|
in_label.add_css_class("sb-field-label")
|
|
in_col.append(in_label)
|
|
self._source_dropdown = Gtk.DropDown()
|
|
self._source_dropdown.add_css_class("sb-device-dropdown")
|
|
self._source_guard = False
|
|
self._source_dropdown.connect("notify::selected", self._on_source_selected)
|
|
in_col.append(self._source_dropdown)
|
|
box.append(in_col)
|
|
|
|
self._sync_sink_dropdown(item, self._sinks)
|
|
return box
|
|
|
|
def _update_extra(self, item: dict, sinks: list[dict], sources: list[dict],
|
|
source_outputs: list[dict]) -> None:
|
|
self._sinks = sinks
|
|
self._sources = sources
|
|
self._recording = self._find_recording(item, source_outputs)
|
|
self._sync_sink_dropdown(item, sinks)
|
|
self._sync_source_dropdown()
|
|
|
|
# -- output routing (playback stream -> sink) --------------------------------
|
|
def _sync_sink_dropdown(self, item: dict, sinks: list[dict]) -> None:
|
|
names = [s["name"] for s in sinks]
|
|
if names != self._sink_names:
|
|
self._sink_names = names
|
|
self._sink_guard = True
|
|
self._sink_dropdown.set_model(Gtk.StringList.new([s["description"] for s in sinks]))
|
|
self._sink_guard = False
|
|
current = item.get("device")
|
|
idx = next((i for i, s in enumerate(sinks) if s["id"] == current), None)
|
|
if idx is not None and self._sink_dropdown.get_selected() != idx:
|
|
self._sink_guard = True
|
|
self._sink_dropdown.set_selected(idx)
|
|
self._sink_guard = False
|
|
|
|
def _on_sink_selected(self, dropdown: Gtk.DropDown, _pspec) -> None:
|
|
if self._sink_guard:
|
|
return
|
|
idx = dropdown.get_selected()
|
|
if 0 <= idx < len(self._sink_names):
|
|
self._backend.move_sink_input(self._id, self._sink_names[idx])
|
|
|
|
# -- input routing (recording stream -> source), mirrors the above -----------
|
|
@staticmethod
|
|
def _find_recording(item: dict, source_outputs: list[dict]) -> dict | None:
|
|
binary = item.get("binary")
|
|
name = item.get("name")
|
|
for so in source_outputs:
|
|
if binary and so.get("binary") == binary:
|
|
return so
|
|
if not binary and so.get("name") == name:
|
|
return so
|
|
return None
|
|
|
|
def _sync_source_dropdown(self) -> None:
|
|
sources = self._sources
|
|
names = [s["name"] for s in sources]
|
|
if names != self._source_names:
|
|
self._source_names = names
|
|
self._source_guard = True
|
|
self._source_dropdown.set_model(Gtk.StringList.new([s["description"] for s in sources]))
|
|
self._source_guard = False
|
|
|
|
self._source_dropdown.set_sensitive(self._recording is not None)
|
|
if self._recording is None:
|
|
return
|
|
current = self._recording.get("device")
|
|
idx = next((i for i, s in enumerate(sources) if s["id"] == current), None)
|
|
if idx is not None and self._source_dropdown.get_selected() != idx:
|
|
self._source_guard = True
|
|
self._source_dropdown.set_selected(idx)
|
|
self._source_guard = False
|
|
|
|
def _on_source_selected(self, dropdown: Gtk.DropDown, _pspec) -> None:
|
|
if self._source_guard or self._recording is None:
|
|
return
|
|
idx = dropdown.get_selected()
|
|
if 0 <= idx < len(self._source_names):
|
|
self._backend.move_source_output(self._recording["id"], self._source_names[idx])
|
|
|
|
def _peak_target(self, item: dict) -> list[str] | None:
|
|
return [f"--monitor-stream={item['id']}"]
|