feat(hyprdrive,hyprlua): mirror output-device dropdown with input-device on Applications tab

Each app card in the mixer panel's Applications tab now also shows an Input
dropdown next to Output, letting you re-route that app's own microphone
stream (pactl move-source-output) the same way Output already re-routes its
playback stream. Matches an app's sink-input to its source-output via
application.process.binary (falling back to name) — the same client tags
both when an app plays and records, e.g. Discord. Apps with no matching
recording stream get the dropdown disabled rather than hidden, so card
layout stays consistent across the row.

hyprlua's audio-panel picks this up via regen-audio-panel.sh, unchanged.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TkkH9eiCBfpWysUXAD9bDG
main
Amir Alexander Abdelbaki 2026-07-29 16:42:07 +02:00
parent ad386f3c89
commit ed43cd2ac7
4 changed files with 192 additions and 36 deletions

View File

@ -1,7 +1,14 @@
"""ApplicationCard — one card in the Applications tab: everything MeterCard """ApplicationCard — one card in the Applications tab: everything MeterCard
gives every card (icon+name, live peak bar, volume meter/L-R pair, mute), 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 plus a dropdown to route this one app's playback stream to a different
output device (`pactl move-sink-input`).""" 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 from __future__ import annotations
@ -14,48 +21,111 @@ from ui.card_base import MeterCard
class ApplicationCard(MeterCard): class ApplicationCard(MeterCard):
def __init__(self, backend, peaks, item: dict, sinks: list[dict]) -> None: def __init__(self, backend, peaks, item: dict, sinks: list[dict],
sources: list[dict], source_outputs: list[dict]) -> None:
self._sinks = sinks self._sinks = sinks
self._sink_names: list[str] = [] 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) super().__init__(backend, peaks, "sink-input", item)
self._update_extra(item, sinks, sources, source_outputs)
def _build_extra(self, item: dict): def _build_extra(self, item: dict):
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2) box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8)
label = Gtk.Label(label="Output", xalign=0.0)
label.add_css_class("sb-field-label") out_col = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2)
box.append(label) 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 = Gtk.DropDown()
self._sink_dropdown.add_css_class("sb-device-dropdown") self._sink_dropdown.add_css_class("sb-device-dropdown")
self._dropdown_guard = False self._sink_guard = False
self._sink_dropdown.connect("notify::selected", self._on_sink_selected) self._sink_dropdown.connect("notify::selected", self._on_sink_selected)
box.append(self._sink_dropdown) 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) self._sync_sink_dropdown(item, self._sinks)
return box return box
def _update_extra(self, item: dict, sinks: list[dict]) -> None: def _update_extra(self, item: dict, sinks: list[dict], sources: list[dict],
source_outputs: list[dict]) -> None:
self._sinks = sinks self._sinks = sinks
self._sources = sources
self._recording = self._find_recording(item, source_outputs)
self._sync_sink_dropdown(item, sinks) 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: def _sync_sink_dropdown(self, item: dict, sinks: list[dict]) -> None:
names = [s["name"] for s in sinks] names = [s["name"] for s in sinks]
if names != self._sink_names: if names != self._sink_names:
self._sink_names = names self._sink_names = names
self._dropdown_guard = True self._sink_guard = True
self._sink_dropdown.set_model(Gtk.StringList.new([s["description"] for s in sinks])) self._sink_dropdown.set_model(Gtk.StringList.new([s["description"] for s in sinks]))
self._dropdown_guard = False self._sink_guard = False
current = item.get("device") current = item.get("device")
idx = next((i for i, s in enumerate(sinks) if s["id"] == current), None) 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: if idx is not None and self._sink_dropdown.get_selected() != idx:
self._dropdown_guard = True self._sink_guard = True
self._sink_dropdown.set_selected(idx) self._sink_dropdown.set_selected(idx)
self._dropdown_guard = False self._sink_guard = False
def _on_sink_selected(self, dropdown: Gtk.DropDown, _pspec) -> None: def _on_sink_selected(self, dropdown: Gtk.DropDown, _pspec) -> None:
if self._dropdown_guard: if self._sink_guard:
return return
idx = dropdown.get_selected() idx = dropdown.get_selected()
if 0 <= idx < len(self._sink_names): if 0 <= idx < len(self._sink_names):
self._backend.move_sink_input(self._id, self._sink_names[idx]) 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: def _peak_target(self, item: dict) -> list[str] | None:
return [f"--monitor-stream={item['id']}"] return [f"--monitor-stream={item['id']}"]

View File

@ -1,6 +1,9 @@
"""Applications tab: horizontally-scrolling row of every app currently """Applications tab: horizontally-scrolling row of every app currently
playing audio (pactl sink-inputs), each with its own volume meter (or playing audio (pactl sink-inputs), each with its own volume meter (or
independent L/R pair), mute, live peak bar, and an output-device dropdown.""" independent L/R pair), mute, live peak bar, an output-device dropdown, and
for whichever of those apps also happen to be recording (matched against
pactl source-outputs by application binary/name) an input-device dropdown
mirroring the output one."""
from __future__ import annotations from __future__ import annotations
@ -22,13 +25,18 @@ class ApplicationsTab(Gtk.Box):
self._row = ScrollRow(self._make_card, empty_text="Nothing is playing audio right now") self._row = ScrollRow(self._make_card, empty_text="Nothing is playing audio right now")
self.append(self._row) self.append(self._row)
def _make_card(self, item: dict, sinks: list[dict]) -> ApplicationCard: def _make_card(self, item: dict, sinks: list[dict], sources: list[dict],
return ApplicationCard(self._backend, self._peaks, item, sinks) source_outputs: list[dict]) -> ApplicationCard:
return ApplicationCard(self._backend, self._peaks, item, sinks, sources, source_outputs)
def refresh(self) -> None: def refresh(self) -> None:
def got_sinks(sinks: list[dict]) -> None: def got_sinks(sinks: list[dict]) -> None:
def got_inputs(inputs: list[dict]) -> None: def got_inputs(inputs: list[dict]) -> None:
self._row.sync(inputs, sinks) def got_sources(sources: list[dict]) -> None:
def got_source_outputs(source_outputs: list[dict]) -> None:
self._row.sync(inputs, sinks, sources, source_outputs)
self._backend.list_source_outputs(got_source_outputs)
self._backend.list_sources(got_sources)
self._backend.list_sink_inputs(got_inputs) self._backend.list_sink_inputs(got_inputs)
self._backend.list_sinks(got_sinks) self._backend.list_sinks(got_sinks)

View File

@ -1,7 +1,14 @@
"""ApplicationCard — one card in the Applications tab: everything MeterCard """ApplicationCard — one card in the Applications tab: everything MeterCard
gives every card (icon+name, live peak bar, volume meter/L-R pair, mute), 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 plus a dropdown to route this one app's playback stream to a different
output device (`pactl move-sink-input`).""" 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 from __future__ import annotations
@ -14,48 +21,111 @@ from ui.card_base import MeterCard
class ApplicationCard(MeterCard): class ApplicationCard(MeterCard):
def __init__(self, backend, peaks, item: dict, sinks: list[dict]) -> None: def __init__(self, backend, peaks, item: dict, sinks: list[dict],
sources: list[dict], source_outputs: list[dict]) -> None:
self._sinks = sinks self._sinks = sinks
self._sink_names: list[str] = [] 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) super().__init__(backend, peaks, "sink-input", item)
self._update_extra(item, sinks, sources, source_outputs)
def _build_extra(self, item: dict): def _build_extra(self, item: dict):
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2) box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8)
label = Gtk.Label(label="Output", xalign=0.0)
label.add_css_class("sb-field-label") out_col = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2)
box.append(label) 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 = Gtk.DropDown()
self._sink_dropdown.add_css_class("sb-device-dropdown") self._sink_dropdown.add_css_class("sb-device-dropdown")
self._dropdown_guard = False self._sink_guard = False
self._sink_dropdown.connect("notify::selected", self._on_sink_selected) self._sink_dropdown.connect("notify::selected", self._on_sink_selected)
box.append(self._sink_dropdown) 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) self._sync_sink_dropdown(item, self._sinks)
return box return box
def _update_extra(self, item: dict, sinks: list[dict]) -> None: def _update_extra(self, item: dict, sinks: list[dict], sources: list[dict],
source_outputs: list[dict]) -> None:
self._sinks = sinks self._sinks = sinks
self._sources = sources
self._recording = self._find_recording(item, source_outputs)
self._sync_sink_dropdown(item, sinks) 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: def _sync_sink_dropdown(self, item: dict, sinks: list[dict]) -> None:
names = [s["name"] for s in sinks] names = [s["name"] for s in sinks]
if names != self._sink_names: if names != self._sink_names:
self._sink_names = names self._sink_names = names
self._dropdown_guard = True self._sink_guard = True
self._sink_dropdown.set_model(Gtk.StringList.new([s["description"] for s in sinks])) self._sink_dropdown.set_model(Gtk.StringList.new([s["description"] for s in sinks]))
self._dropdown_guard = False self._sink_guard = False
current = item.get("device") current = item.get("device")
idx = next((i for i, s in enumerate(sinks) if s["id"] == current), None) 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: if idx is not None and self._sink_dropdown.get_selected() != idx:
self._dropdown_guard = True self._sink_guard = True
self._sink_dropdown.set_selected(idx) self._sink_dropdown.set_selected(idx)
self._dropdown_guard = False self._sink_guard = False
def _on_sink_selected(self, dropdown: Gtk.DropDown, _pspec) -> None: def _on_sink_selected(self, dropdown: Gtk.DropDown, _pspec) -> None:
if self._dropdown_guard: if self._sink_guard:
return return
idx = dropdown.get_selected() idx = dropdown.get_selected()
if 0 <= idx < len(self._sink_names): if 0 <= idx < len(self._sink_names):
self._backend.move_sink_input(self._id, self._sink_names[idx]) 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: def _peak_target(self, item: dict) -> list[str] | None:
return [f"--monitor-stream={item['id']}"] return [f"--monitor-stream={item['id']}"]

View File

@ -1,6 +1,9 @@
"""Applications tab: horizontally-scrolling row of every app currently """Applications tab: horizontally-scrolling row of every app currently
playing audio (pactl sink-inputs), each with its own volume meter (or playing audio (pactl sink-inputs), each with its own volume meter (or
independent L/R pair), mute, live peak bar, and an output-device dropdown.""" independent L/R pair), mute, live peak bar, an output-device dropdown, and
for whichever of those apps also happen to be recording (matched against
pactl source-outputs by application binary/name) an input-device dropdown
mirroring the output one."""
from __future__ import annotations from __future__ import annotations
@ -22,13 +25,18 @@ class ApplicationsTab(Gtk.Box):
self._row = ScrollRow(self._make_card, empty_text="Nothing is playing audio right now") self._row = ScrollRow(self._make_card, empty_text="Nothing is playing audio right now")
self.append(self._row) self.append(self._row)
def _make_card(self, item: dict, sinks: list[dict]) -> ApplicationCard: def _make_card(self, item: dict, sinks: list[dict], sources: list[dict],
return ApplicationCard(self._backend, self._peaks, item, sinks) source_outputs: list[dict]) -> ApplicationCard:
return ApplicationCard(self._backend, self._peaks, item, sinks, sources, source_outputs)
def refresh(self) -> None: def refresh(self) -> None:
def got_sinks(sinks: list[dict]) -> None: def got_sinks(sinks: list[dict]) -> None:
def got_inputs(inputs: list[dict]) -> None: def got_inputs(inputs: list[dict]) -> None:
self._row.sync(inputs, sinks) def got_sources(sources: list[dict]) -> None:
def got_source_outputs(source_outputs: list[dict]) -> None:
self._row.sync(inputs, sinks, sources, source_outputs)
self._backend.list_source_outputs(got_source_outputs)
self._backend.list_sources(got_sources)
self._backend.list_sink_inputs(got_inputs) self._backend.list_sink_inputs(got_inputs)
self._backend.list_sinks(got_sinks) self._backend.list_sinks(got_sinks)