1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
|
#!/usr/bin/env python3
import asyncio
import os
import sys
import json
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.container is None or e.change == "close":
await self.dunst.call_rule_enable("transient_skip", 0)
else:
await self.dunst.call_rule_enable("transient_skip", e.container.fullscreen_mode)
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())
|
