frobar-ng: Rudimentary Pulseaudio

hm/desktop/frobar/.dev/new.py

@@ -13,6 +13,8 @@ import typing
 import i3ipc
 import i3ipc.aio
 import psutil
+import pulsectl
+import pulsectl_asyncio
 import rich.color
 import rich.logging
 import rich.terminal_theme
@@ -48,13 +50,11 @@ def humanSize(numi: int) -> str:
     return f"{numi:d}YiB"
 
 
-def ramp(p: float, ramp: str = " ▁▂▃▄▅▆▇█") -> str:
-    if p > 1:
-        return ramp[-1]
-    elif p < 0:
-        return ramp[0]
-    else:
-        return ramp[round(p * (len(ramp) - 1))]
+def ramp(p: float, states: str = " ▁▂▃▄▅▆▇█") -> str:
+    if p < 0:
+        return ""
+    d, m = divmod(p, 1.0)
+    return states[-1] * int(d) + states[round(m * (len(states) - 1))]
 
 
 class ComposableText(typing.Generic[P, C]):
@@ -718,6 +718,7 @@ class TemperatureProvider(AlertingProvider, PeriodicStatefulProvider):
     RAMP = ""
     MAIN_TEMPS = ["coretemp", "amdgpu", "cpu_thermal"]
     # For Intel, AMD and ARM respectively.
+    # FIXME Threshold doesn't seem to match old version
 
     main: str
 
@@ -780,6 +781,42 @@ class BatteryProvider(AlertingProvider, PeriodicStatefulProvider):
         self.section.setText(text)
 
 
+class PulseaudioProvider(SingleSectionProvider):
+    async def update(self) -> None:
+        async with pulsectl_asyncio.PulseAsync("frobar-updater") as pulse:
+            text = ""
+            # TODO Sections
+            for sink in await pulse.sink_list():
+                log.debug(f"{sink}")
+                if (
+                    sink.port_active.name == "analog-output-headphones"
+                    or sink.port_active.description == "Headphones"
+                ):
+                    icon = ""
+                elif (
+                    sink.port_active.name == "analog-output-speaker"
+                    or sink.port_active.description == "Speaker"
+                ):
+                    icon = "" if sink.mute else ""
+                elif sink.port_active.name in ("headset-output", "headphone-output"):
+                    icon = ""
+                else:
+                    icon = "?"
+                vol = await pulse.volume_get_all_chans(sink)
+                fg = (sink.mute and "#333333") or (vol > 1 and "#FF0000") or None
+                # TODO Show which is default
+
+                text += f" {icon} {vol:.0%}"
+            self.section.setText(text)
+
+    async def run(self) -> None:
+        await super().run()
+        await self.update()
+        async with pulsectl_asyncio.PulseAsync("frobar-events") as pulse:
+            async for event in pulse.subscribe_events(pulsectl.PulseEventMaskEnum.sink):
+                await self.update()
+
+
 class NetworkProviderSection(StatefulSection):
     def __init__(
         self,
@@ -808,6 +845,7 @@ class NetworkProviderSection(StatefulSection):
             iface.startswith("tun") or iface.startswith("tap") or iface.startswith("wg")
         ):
             self.icon = ""
+
         elif iface.startswith("docker"):
             self.icon = ""
         elif iface.startswith("veth"):
@@ -972,6 +1010,8 @@ async def main() -> None:
 
     bar = Bar(theme=theme)
     dualScreen = len(bar.children) > 1
+    leftPreferred = 0 if dualScreen else None
+    rightPreferred = 1 if dualScreen else None
 
     color = rich.color.Color.parse
 
@@ -979,27 +1019,35 @@ async def main() -> None:
     bar.addProvider(I3WorkspacesProvider(), alignment=Alignment.LEFT)
     if dualScreen:
         bar.addProvider(
-            I3WindowTitleProvider(), screenNum=0, alignment=Alignment.CENTER
+            I3WindowTitleProvider(color=color("white")),
+            screenNum=0,
+            alignment=Alignment.CENTER,
         )
     bar.addProvider(
         StaticProvider(text="mpris", color=color("bright_white")),
-        screenNum=1 if dualScreen else None,
+        screenNum=rightPreferred,
         alignment=Alignment.CENTER,
     )
 
-    bar.addProvider(CpuProvider(), alignment=Alignment.RIGHT)
-    bar.addProvider(LoadProvider(), alignment=Alignment.RIGHT)
-    bar.addProvider(RamProvider(), alignment=Alignment.RIGHT)
-    bar.addProvider(TemperatureProvider(), alignment=Alignment.RIGHT)
-    bar.addProvider(BatteryProvider(), alignment=Alignment.RIGHT)
+    bar.addProvider(CpuProvider(), screenNum=leftPreferred, alignment=Alignment.RIGHT)
+    bar.addProvider(LoadProvider(), screenNum=leftPreferred, alignment=Alignment.RIGHT)
+    bar.addProvider(RamProvider(), screenNum=leftPreferred, alignment=Alignment.RIGHT)
     bar.addProvider(
-        StaticProvider("pulse", color=color("magenta")),
-        screenNum=1 if dualScreen else None,
+        TemperatureProvider(),
+        screenNum=leftPreferred,
+        alignment=Alignment.RIGHT,
+    )
+    bar.addProvider(
+        BatteryProvider(), screenNum=leftPreferred, alignment=Alignment.RIGHT
+    )
+    bar.addProvider(
+        PulseaudioProvider(color=color("magenta")),
+        screenNum=rightPreferred,
         alignment=Alignment.RIGHT,
     )
     bar.addProvider(
         NetworkProvider(color=color("blue")),
-        screenNum=0 if dualScreen else None,
+        screenNum=leftPreferred,
         alignment=Alignment.RIGHT,
     )
     bar.addProvider(TimeProvider(color=color("cyan")), alignment=Alignment.RIGHT)

hm/desktop/frobar/default.nix

@@ -30,7 +30,8 @@ pkgs.python3Packages.buildPythonApplication rec {
     mpd2
     notmuch
     psutil
-    pulsectl
+    pulsectl # old only
+    pulsectl-asyncio
     pyinotify
     rich
   ];