diff options
Diffstat (limited to 'i3blocks/.local/bin/i3blocks-dunst')
-rwxr-xr-x | i3blocks/.local/bin/i3blocks-dunst | 146 |
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()) |