1 files changed, 146 insertions, 0 deletions
diff --git a/i3blocks/.local/bin/i3blocks-dunst b/i3blocks/.local/bin/i3blocks-dunst
new file mode 100755
index 0000000..df3d9d4
--- /dev/null
+++ b/i3blocks/.local/bin/i3blocks-dunst
@@ -0,0 +1,146 @@
+#!/usr/bin/env python3
+
+import asyncio
+import os
+import sys
+import json
+import pprint
+
+from dbus_next.aio import MessageBus
+import i3ipc
+from i3ipc.aio import Connection
+
+
+BUS_NAME = "org.freedesktop.Notifications"
+OBJECT_PATH = "/org/freedesktop/Notifications"
+DUNST_INTERFACE = "org.dunstproject.cmd0"
+PROPERTIES_INTERFACE = "org.freedesktop.DBus.Properties"
+I3_EVENTS = [
+    i3ipc.Event.WINDOW_FOCUS,
+    i3ipc.Event.WINDOW_FULLSCREEN_MODE,
+    i3ipc.Event.WINDOW_CLOSE,
+    i3ipc.Event.WORKSPACE_FOCUS,
+]
+
+
+class Dunst:
+    @classmethod
+    async def connect(cls):
+        self = cls()
+        self.bus = await MessageBus().connect()
+
+        self.obj = self.bus.get_proxy_object(
+            BUS_NAME,
+            OBJECT_PATH,
+            await self.bus.introspect(BUS_NAME, OBJECT_PATH),
+        )
+
+        self.dunst = self.obj.get_interface(DUNST_INTERFACE)
+        self.paused = await self.dunst.get_paused()
+        self.displayed = await self.dunst.get_displayed_length()
+        self.waiting = await self.dunst.get_waiting_length()
+
+        self.i3 = await Connection().connect()
+
+        for event in I3_EVENTS:
+            self.i3.on(event, self.handle_i3_event)
+
+        return self
+
+    async def handle_i3_event(self, conn, e):
+        if e.change == "close":
+            enable = 0
+        elif isinstance(e, i3ipc.WorkspaceEvent):
+            enable = e.current.fullscreen_mode
+        else:
+            enable = e.container.fullscreen_mode
+
+        await self.dunst.call_rule_enable("transient_skip", enable)
+
+
+    def print_status(self):
+        if self.paused:
+            icon = "\U000f009b"  # 󰂛
+        else:
+            icon = "\U000f009a"  # 󰂚
+
+        if self.waiting > 0:
+            count = f" {self.waiting}"
+        else:
+            count = ""
+
+        output = {"full_text": f" {icon}{count} "}
+
+        if self.displayed + self.waiting > 0:
+            output["color"] = f"#{os.environ.get("BASE16_COLOR_00_HEX")}"
+            output["background"] = f"#{os.environ.get("BASE16_COLOR_0C_HEX")}"
+        elif self.paused:
+            output["color"] = f"#{os.environ.get("BASE16_COLOR_00_HEX")}"
+            output["background"] = f"#{os.environ.get("BASE16_COLOR_0A_HEX")}"
+
+        print(json.dumps(output, ensure_ascii=False), flush=True)
+
+    def on_properties_changed(self, _name, property, invalidated):
+        update = False
+        for property, variant in property.items():
+            match property:
+                case "paused":
+                    self.paused = variant.value
+                case "displayedLength":
+                    self.displayed = variant.value
+                case "waitingLength":
+                    self.waiting = variant.value
+                case _:
+                    continue
+            update = True
+
+        if update:
+            self.print_status()
+
+    async def listener(self):
+        properties = self.obj.get_interface(PROPERTIES_INTERFACE)
+        properties.on_properties_changed(self.on_properties_changed)
+
+        await self.bus.wait_for_disconnect()
+
+    async def button_handler(self):
+        loop = asyncio.get_event_loop()
+        reader = asyncio.StreamReader()
+        protocol = asyncio.StreamReaderProtocol(reader)
+        await loop.connect_read_pipe(lambda: protocol, sys.stdin)
+
+        while True:
+            line = await reader.readline()
+
+            if not line:
+                await asyncio.sleep(1)
+                continue
+
+            match json.loads(line):
+                case {"button": 1}:
+                    await self.dunst.call_notification_show()
+                case {"button": 2}:
+                    pass
+                case {"button": 3}:
+                    await self.dunst.set_paused(not self.paused)
+                case {"button": 4}:
+                    pass
+                case {"button": 5}:
+                    pass
+
+
+async def main():
+    dunst = await Dunst.connect()
+    dunst.print_status()
+
+    try:
+        async with asyncio.TaskGroup() as task_group:
+            task_group.create_task(dunst.i3.main())
+            task_group.create_task(dunst.listener())
+            task_group.create_task(dunst.button_handler())
+    except asyncio.CancelledError:
+        return
+
+
+if __name__ == "__main__":
+    asyncio.run(main())