mirror of
https://github.imc.re/void-land/hyprland-void-dots
synced 2025-07-26 03:12:49 +02:00
refactor: eww python script
This commit is contained in:
hesam-init
parent
caad2ba8ef
commit
ee43261699
8 changed files with 33 additions and 266 deletions
|
|
@ -69,15 +69,6 @@
|
|||
(defpoll workspacesJson :initial "[]" :interval "60s" "./scripts/workspace/get-workspaces")
|
||||
(deflisten currentWorkspace :initial "1" "./scripts/workspace/get-active")
|
||||
|
||||
; Notifications
|
||||
(deflisten notifications :initial '{
|
||||
"count": 0,
|
||||
"dnd": false,
|
||||
"popups": []
|
||||
}'
|
||||
"./scripts/notification/notifs.py"
|
||||
)
|
||||
|
||||
; Playerctl
|
||||
(deflisten pstatus :initial "" "playerctl status -F")
|
||||
(deflisten psong :initial "" "playerctl metadata -F --format '{{ title }}'")
|
||||
|
|
@ -95,9 +86,25 @@
|
|||
; Controlpanel
|
||||
(defvar revealControlpanel false)
|
||||
(defvar revealWeather false)
|
||||
(deflisten timerdis "./scripts/timer.py subscribe")
|
||||
(defpoll hostname :initial "idk" :interval "24h" 'echo "$(whoami)"')
|
||||
(defpoll uptime :initial "idk" :interval "1m" "uptime -p")
|
||||
(defpoll datehour :initial "12" :interval "30m" "date +'%H'")
|
||||
(defpoll notesc :interval "2s" :run-while reveal4 "cat -s ~/Documents/fuck.txt")
|
||||
|
||||
; Python
|
||||
(defpoll quotejson :interval "1h" "./scripts/python/quote.py")
|
||||
|
||||
(deflisten timerdis "./scripts/python/timer.py subscribe")
|
||||
|
||||
(deflisten notifications :initial '{
|
||||
"count": 0,
|
||||
"dnd": false,
|
||||
"notifications": [],
|
||||
"popups": []
|
||||
}'
|
||||
"./scripts/python/notifications.py"
|
||||
)
|
||||
|
||||
(defpoll weatherjson
|
||||
:initial '{
|
||||
"FeelsLikeC": "0",
|
||||
|
|
@ -127,7 +134,4 @@
|
|||
}'
|
||||
:interval "1h"
|
||||
:run-while revealControlpanel
|
||||
"./scripts/weather.py")
|
||||
(defpoll datehour :initial "12" :interval "30m" "date +'%H'")
|
||||
(defpoll notesc :interval "2s" :run-while reveal4 "cat -s ~/Documents/fuck.txt")
|
||||
(defpoll quotejson :interval "1h" `./scripts/quote.py`)
|
||||
"./scripts/python/weather.py")
|
||||
|
|
@ -1,236 +0,0 @@
|
|||
#!/usr/bin/python
|
||||
import gi
|
||||
gi.require_version("GdkPixbuf", "2.0")
|
||||
gi.require_version("Gtk", "3.0")
|
||||
|
||||
import dbus
|
||||
import dbus.service
|
||||
from dbus.mainloop.glib import DBusGMainLoop
|
||||
from gi.repository import GLib
|
||||
import datetime
|
||||
import os
|
||||
import typing
|
||||
import sys
|
||||
import json
|
||||
from gi.repository import Gtk, GdkPixbuf
|
||||
import subprocess
|
||||
|
||||
cache_dir = f"{os.getenv('HOME')}/.cache/notify_img_data"
|
||||
log_file = f"{os.getenv('HOME')}/.cache/notifications.json"
|
||||
os.makedirs(cache_dir, exist_ok=True)
|
||||
active_popups = {}
|
||||
|
||||
# RECIEVE NOTIFICATIONS
|
||||
|
||||
class NotificationDaemon(dbus.service.Object):
|
||||
def __init__(self):
|
||||
bus_name = dbus.service.BusName("org.freedesktop.Notifications", dbus.SessionBus())
|
||||
dbus.service.Object.__init__(self, bus_name, "/org/freedesktop/Notifications")
|
||||
self.dnd = False
|
||||
|
||||
@dbus.service.method("org.freedesktop.Notifications", in_signature="susssasa{sv}i", out_signature="u")
|
||||
def Notify(self, app_name, replaces_id, app_icon, summary, body, actions, hints, timeout):
|
||||
replaces_id = int(replaces_id)
|
||||
actions = list(actions)
|
||||
app_icon = str(app_icon)
|
||||
app_name = str(app_name)
|
||||
summary = str(summary)
|
||||
body = str(body)
|
||||
|
||||
if replaces_id != 0:
|
||||
id = replaces_id
|
||||
else:
|
||||
log_file = self.read_log_file()
|
||||
if log_file['notifications'] != []:
|
||||
id = log_file['notifications'][0]['id'] + 1
|
||||
else:
|
||||
id = 1
|
||||
|
||||
acts = []
|
||||
for i in range(0, len(actions), 2):
|
||||
acts.append([str(actions[i]), str(actions[i + 1])])
|
||||
|
||||
details = {
|
||||
"id": id,
|
||||
"app": app_name,
|
||||
"summary": self.format_long_string(summary, 35),
|
||||
"body": self.format_long_string(body, 35),
|
||||
"time": datetime.datetime.now().strftime("%H:%M"),
|
||||
"urgency": hints["urgency"] if "urgency" in hints else 1,
|
||||
"actions": acts
|
||||
}
|
||||
|
||||
|
||||
if app_icon.strip():
|
||||
if os.path.isfile(app_icon) or app_icon.startswith("file://"):
|
||||
details["image"] = app_icon
|
||||
else:
|
||||
details["image"] = self.get_gtk_icon(app_icon)
|
||||
else:
|
||||
details["image"] = None
|
||||
|
||||
if "image-data" in hints:
|
||||
details["image"] = f"{cache_dir}/{details['id']}.png"
|
||||
self.save_img_byte(hints["image-data"], details["image"])
|
||||
|
||||
self.save_notifications(details)
|
||||
if not self.dnd:
|
||||
self.save_popup(details)
|
||||
return id
|
||||
|
||||
|
||||
|
||||
def format_long_string(self, long_string, interval):
|
||||
split_string = []
|
||||
max_length = 256
|
||||
|
||||
for i in range(0, len(long_string), interval):
|
||||
split_string.append(long_string[i:i+interval])
|
||||
|
||||
result = "-\n".join(split_string)
|
||||
|
||||
if len(result) > max_length:
|
||||
result = result[:max_length] + "..."
|
||||
|
||||
return result
|
||||
|
||||
@dbus.service.method("org.freedesktop.Notifications", in_signature="", out_signature="ssss")
|
||||
def GetServerInformation(self):
|
||||
return ("linkfrg's notification daemon", "linkfrg", "1.0", "1.2")
|
||||
|
||||
@dbus.service.method("org.freedesktop.Notifications", in_signature="", out_signature="as")
|
||||
def GetCapabilities(self):
|
||||
return ('actions', 'body', 'icon-static', 'persistence')
|
||||
|
||||
@dbus.service.signal("org.freedesktop.Notifications", signature="us")
|
||||
def ActionInvoked(self, id, action):
|
||||
return (id, action)
|
||||
|
||||
@dbus.service.method("org.freedesktop.Notifications", in_signature="us", out_signature="")
|
||||
def InvokeAction(self, id, action):
|
||||
self.ActionInvoked(id, action)
|
||||
|
||||
@dbus.service.signal("org.freedesktop.Notifications", signature="uu")
|
||||
def NotificationClosed(self, id, reason):
|
||||
return (id, reason)
|
||||
|
||||
@dbus.service.method("org.freedesktop.Notifications", in_signature="u", out_signature="")
|
||||
def CloseNotification(self, id):
|
||||
current = self.read_log_file()
|
||||
current["notifications"] = [n for n in current["notifications"] if n["id"] != id]
|
||||
current["count"] = len(current["notifications"])
|
||||
|
||||
self.write_log_file(current)
|
||||
self.NotificationClosed(id, 2)
|
||||
self.DismissPopup(id)
|
||||
|
||||
@dbus.service.method("org.freedesktop.Notifications", in_signature="", out_signature="")
|
||||
def ToggleDND(self):
|
||||
match self.dnd:
|
||||
case False:
|
||||
self.dnd = True
|
||||
case True:
|
||||
self.dnd = False
|
||||
|
||||
@dbus.service.method("org.freedesktop.Notifications", in_signature="", out_signature="")
|
||||
def GetDNDState(self):
|
||||
subprocess.run(["eww", "update", f"do-not-disturb={json.dumps(self.dnd)}"])
|
||||
|
||||
|
||||
def get_gtk_icon(self, icon_name):
|
||||
theme = Gtk.IconTheme.get_default()
|
||||
icon_info = theme.lookup_icon(icon_name, 128, 0)
|
||||
|
||||
if icon_info is not None:
|
||||
return icon_info.get_filename()
|
||||
|
||||
|
||||
def save_img_byte(self, px_args: typing.Iterable, save_path: str):
|
||||
GdkPixbuf.Pixbuf.new_from_bytes(
|
||||
width=px_args[0],
|
||||
height=px_args[1],
|
||||
has_alpha=px_args[3],
|
||||
data=GLib.Bytes(px_args[6]),
|
||||
colorspace=GdkPixbuf.Colorspace.RGB,
|
||||
rowstride=px_args[2],
|
||||
bits_per_sample=px_args[4],
|
||||
).savev(save_path, "png")
|
||||
|
||||
|
||||
def write_log_file(self, data):
|
||||
output_json = json.dumps(data, indent=2)
|
||||
subprocess.run(["eww", "update", f"notifications={output_json}"])
|
||||
with open(log_file, "w") as log:
|
||||
log.write(output_json)
|
||||
|
||||
def read_log_file(self):
|
||||
empty = {"count": 0, "notifications": [], "popups": []}
|
||||
try:
|
||||
with open(log_file, "r") as log:
|
||||
return json.load(log)
|
||||
except FileNotFoundError:
|
||||
with open(log_file, "w") as log:
|
||||
json.dump(empty, log)
|
||||
return empty
|
||||
|
||||
|
||||
def save_notifications(self, notification):
|
||||
current = self.read_log_file()
|
||||
current["notifications"].insert(0, notification)
|
||||
current["count"] = len(current["notifications"])
|
||||
|
||||
self.write_log_file(current)
|
||||
|
||||
@dbus.service.method("org.freedesktop.Notifications", in_signature="", out_signature="")
|
||||
def ClearAll(self):
|
||||
for notify in self.read_log_file()['notifications']:
|
||||
self.NotificationClosed(notify['id'], 2)
|
||||
data = {"count": 0, "notifications": [], "popups": []}
|
||||
|
||||
self.write_log_file(data)
|
||||
|
||||
|
||||
# OPERATIONS WITH POPUPS
|
||||
|
||||
def save_popup(self, notification):
|
||||
global active_popups
|
||||
|
||||
current = self.read_log_file()
|
||||
if len(current["popups"]) >= 3:
|
||||
oldest_popup = current["popups"].pop()
|
||||
self.DismissPopup(oldest_popup["id"])
|
||||
|
||||
current["popups"].append(notification)
|
||||
self.write_log_file(current)
|
||||
|
||||
popup_id = notification["id"]
|
||||
active_popups[popup_id] = GLib.timeout_add_seconds(5, self.DismissPopup, popup_id)
|
||||
|
||||
@dbus.service.method("org.freedesktop.Notifications", in_signature="u", out_signature="")
|
||||
def DismissPopup(self, id):
|
||||
global active_popups
|
||||
|
||||
current = self.read_log_file()
|
||||
current["popups"] = [n for n in current["popups"] if n["id"] != id]
|
||||
self.write_log_file(current)
|
||||
|
||||
active_popups.pop(id, None)
|
||||
|
||||
@dbus.service.method("org.freedesktop.Notifications", in_signature="", out_signature="")
|
||||
def GetCurrent(self):
|
||||
subprocess.run(["eww", "update", f"notifications={json.dumps(self.read_log_file())}"])
|
||||
|
||||
|
||||
# MAINLOOP
|
||||
|
||||
def main():
|
||||
DBusGMainLoop(set_as_default=True)
|
||||
loop = GLib.MainLoop()
|
||||
NotificationDaemon()
|
||||
try:
|
||||
loop.run()
|
||||
except KeyboardInterrupt:
|
||||
exit(0)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -7,9 +7,8 @@ try:
|
|||
res = json.loads(req)
|
||||
except:
|
||||
res = {
|
||||
"content": "I can't get quotes, you are not online right now",
|
||||
"content": "I can't get quotes, you are not online right now",
|
||||
"author": "quote widget"
|
||||
}
|
||||
# print(f"(box :valign 'fill' :vexpand true :orientation 'v' :space-evenly false (scroll :height 100 :width 300 :hscroll true :vscroll true (label :class 'quote' :text `\"{res['content']}\"` :wrap true :width 300)) (label :class 'quoteauthor' :text `- {res['author']}`))")
|
||||
print(json.dumps(res))
|
||||
|
||||
|
|
@ -1,4 +1,5 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
import time
|
||||
import os
|
||||
import sys
|
||||
|
|
@ -15,12 +16,10 @@ def startstop():
|
|||
f.write(str(time.time()))
|
||||
os.popen("notify-send 'timer start'")
|
||||
|
||||
|
||||
|
||||
# fuck it, I'm sure we don't need the loop function
|
||||
def readtime():
|
||||
if not os.path.exists("/tmp/timer"):
|
||||
with open("/tmp/timer", "x") as f:
|
||||
with open("/tmp/timer", "x") as f:
|
||||
f.write("25")
|
||||
|
||||
while True:
|
||||
|
|
@ -4,11 +4,11 @@ import requests
|
|||
import json
|
||||
import time
|
||||
|
||||
mytime = time.localtime()
|
||||
mytime = time.localtime()
|
||||
day = (6 < mytime.tm_hour < 20)
|
||||
|
||||
code_icon = {
|
||||
113: "clear",
|
||||
113: "clear",
|
||||
116: "few-clouds",
|
||||
119: "clouds",
|
||||
122: "clouds",
|
||||
|
|
@ -67,8 +67,10 @@ code_icon_n = {
|
|||
356: "rain",
|
||||
}
|
||||
|
||||
city = "Rasht"
|
||||
|
||||
try:
|
||||
req = requests.get(r"https://wttr.in/?format=j1", timeout=15).text
|
||||
req = requests.get(r"https://wttr.in/{city}?format=j1", timeout=15).text
|
||||
req = json.loads(req)
|
||||
res = req["current_condition"][0].copy()
|
||||
|
||||
|
|
@ -84,14 +86,14 @@ try:
|
|||
else:
|
||||
res["icon"] = "idk"
|
||||
|
||||
# hourly
|
||||
|
||||
it = 0
|
||||
|
||||
while it < 8 and int(req["weather"][0]["hourly"][it]["time"]) < mytime.tm_hour * 100:
|
||||
it += 1
|
||||
|
||||
res["hourly"] = []
|
||||
for i in range(8):
|
||||
|
||||
for i in range(8):
|
||||
res["hourly"].append(req["weather"][(it+i)//8]["hourly"][(it+i)%8].copy())
|
||||
|
||||
for hour in res["hourly"]:
|
||||
|
|
@ -121,7 +123,6 @@ try:
|
|||
print(json.dumps(res))
|
||||
|
||||
except Exception as e:
|
||||
# print(e)
|
||||
print("""
|
||||
{
|
||||
"FeelsLikeC": "0",
|
||||
|
|
@ -349,7 +349,7 @@
|
|||
)
|
||||
)
|
||||
|
||||
(defwidget weatherhour[hour]
|
||||
(defwidget weatherhour [hour]
|
||||
(box
|
||||
:class "smallentry"
|
||||
:orientation "h"
|
||||
|
|
@ -412,9 +412,9 @@
|
|||
:orientation "h"
|
||||
:class "timer_butt"
|
||||
:valign "end"
|
||||
(button :onclick "./scripts/timer.py timedec" (label :text "-"))
|
||||
(button :onclick "./scripts/timer.py toggle" (label :style "padding-right: 3px;" :text { !matches(timerdis, ":") ? "" : ""}))
|
||||
(button :onclick "./scripts/timer.py timeinc" (label :text "+")))))
|
||||
(button :onclick "./scripts/python/timer.py timedec" (label :text "-"))
|
||||
(button :onclick "./scripts/python/timer.py toggle" (label :style "padding-right: 3px;" :text { !matches(timerdis, ":") ? "" : ""}))
|
||||
(button :onclick "./scripts/python/timer.py timeinc" (label :text "+")))))
|
||||
|
||||
(defwidget bigslides []
|
||||
(box
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue