frobar-ng: Rudimentary Pulseaudio

curacao/co2meter/default.nix

@@ -1,7 +1,6 @@
 {
   pkgs,
   lib,
-  config,
   ...
 }:
 let
@@ -12,7 +11,7 @@ let
     rev = "push-nurpouorqoyr"; # Humidity + availability support
     sha256 = "sha256-nOhyBAgvjeQh9ys3cBJOVR67SDs96zBzxIRGpaq4yoA=";
   };
-  zytemp_mqtt = pkgs.python3Packages.buildPythonPackage rec {
+  zytemp_mqtt = pkgs.python3Packages.buildPythonPackage {
     name = "zytemp_mqtt";
     src = zytemp_mqtt_src;
     propagatedBuildInputs = with pkgs.python3Packages; [
@@ -33,6 +32,7 @@ let
     installPhase = ''
       mkdir -p $out/lib/udev/rules.d
       cp udev/90-usb-zytemp-permissions.rules $out/lib/udev/rules.d/90-usb-zytemp.rules
+      sed -i 's|"usb"|"hidraw"|' $out/lib/udev/rules.d/90-usb-zytemp.rules
     '';
   };
   mqtt_host = "192.168.7.53"; # Ludwig
@@ -74,7 +74,7 @@ in
           SystemCallFilter = [
             "@system-service"
             "~@privileged"
-            "~@resouces"
+            "~@resources"
           ];
           UMask = "0077";
         };

curacao/desk/default.nix

@@ -1,7 +1,5 @@
 {
   pkgs,
-  lib,
-  config,
   ...
 }:
 let
@@ -53,7 +51,7 @@ in
           SystemCallFilter = [
             "@system-service"
             "~@privileged"
-            "~@resouces"
+            "~@resources"
           ];
           UMask = "0077";
         };

curacao/desk/desk_mqtt.py

@@ -32,8 +32,8 @@ class Desk:
     # CBD4+5 Configurator
     # https://www.linak.nl/technische-ondersteuning/#/cbd4-cbd6s-configurator
     # Connects, and settings can be changed.
-    # Don't  think there's much that would help with our problem.
-    # Decompiled with Ghidra (hard), didn't go super far
+    # Don't think there's much that would help with our problem.
+    # Tried to decompile with Ghidra (hard), didn't go super far
 
     VEND = 0x12D3
     PROD = 0x0002
@@ -265,7 +265,7 @@ class Desk:
 
 
 if __name__ == "__main__":
-    logging.basicConfig(level=logging.DEBUG)
+    logging.basicConfig()
     log = logging.getLogger(__name__)
 
     desk = Desk()
@@ -344,10 +344,17 @@ if __name__ == "__main__":
     )
     height_min = ha_mqtt_discoverable.sensors.Sensor(height_min_settings)
 
+    last_published_state = None
+
     def fetch_callback(desk: Desk) -> None:
-        log.debug("Received state, sending")
         hcur = desk.get_height()
         hmin, hmax = desk.get_height_bounds()
+        global last_published_state
+
+        state = hcur, hmin, hmax
+        if state == last_published_state:
+            return
+        last_published_state = state
 
         # If none this will set as unknown
         # Also readings can be a bit outside the boundaries,

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
   ];