feat: add initial eww config
This commit is contained in:
Executable
+60
@@ -0,0 +1,60 @@
|
||||
import evdev
|
||||
import json
|
||||
import sys
|
||||
import select
|
||||
import os
|
||||
|
||||
class KeyboardLedMonitor:
|
||||
def __init__(self):
|
||||
self.last_state = {"caps": "0", "num": "0"}
|
||||
self.kbd_devices = []
|
||||
|
||||
# ищем все устройства, которые умеют в CapsLock
|
||||
devices = [evdev.InputDevice(path) for path in evdev.list_devices()]
|
||||
for dev in devices:
|
||||
capabilities = dev.capabilities().get(evdev.ecodes.EV_KEY, [])
|
||||
if evdev.ecodes.KEY_CAPSLOCK in capabilities:
|
||||
self.kbd_devices.append(dev)
|
||||
|
||||
if not self.kbd_devices:
|
||||
sys.exit(1)
|
||||
|
||||
# выводим начальное состояние
|
||||
self.push_update(self.kbd_devices[0])
|
||||
|
||||
def get_led_state(self, dev):
|
||||
# проверяем коды
|
||||
leds = dev.leds()
|
||||
return {
|
||||
"caps": "1" if 1 in leds else "0",
|
||||
"num": "1" if 0 in leds else "0"
|
||||
}
|
||||
|
||||
def push_update(self, dev):
|
||||
# отдаем json в stdout при изменении
|
||||
current_state = self.get_led_state(dev)
|
||||
if current_state != self.last_state:
|
||||
sys.stdout.write(json.dumps(current_state) + "\n")
|
||||
sys.stdout.flush()
|
||||
self.last_state = current_state
|
||||
|
||||
def run(self):
|
||||
# создаем мапу файловых дескрипторов для селектора
|
||||
devices_dict = {d.fd: d for d in self.kbd_devices}
|
||||
|
||||
try:
|
||||
while True:
|
||||
# ждем событий от любого из устройств
|
||||
r, _, _ = select.select(devices_dict, [], [])
|
||||
for fd in r:
|
||||
dev = devices_dict[fd]
|
||||
for event in dev.read():
|
||||
# если прилетело событие изменения светодиода
|
||||
if event.type == evdev.ecodes.EV_LED:
|
||||
self.push_update(dev)
|
||||
|
||||
except (PermissionError, KeyboardInterrupt):
|
||||
sys.exit(0)
|
||||
|
||||
if __name__ == "__main__":
|
||||
KeyboardLedMonitor().run()
|
||||
Executable
+46
@@ -0,0 +1,46 @@
|
||||
import dbus
|
||||
from dbus.mainloop.glib import DBusGMainLoop
|
||||
from gi.repository import GLib
|
||||
import sys
|
||||
|
||||
class TrayMonitor:
|
||||
def __init__(self):
|
||||
# настраиваем dbus и вешаем слушателей
|
||||
DBusGMainLoop(set_as_default=True)
|
||||
self.bus = dbus.SessionBus()
|
||||
self.interface = "org.kde.StatusNotifierWatcher"
|
||||
self.path = "/StatusNotifierWatcher"
|
||||
|
||||
# подписываемся на регистрацию и удаление иконок
|
||||
self.bus.add_signal_receiver(self.push_update, "StatusNotifierItemRegistered", self.interface)
|
||||
self.bus.add_signal_receiver(self.push_update, "StatusNotifierItemUnregistered", self.interface)
|
||||
|
||||
# выводим начальное состояние
|
||||
self.push_update()
|
||||
|
||||
def get_count(self):
|
||||
# опрашиваем шину на количество активных элементов
|
||||
try:
|
||||
obj = self.bus.get_object(self.interface, self.path)
|
||||
props = dbus.Interface(obj, "org.freedesktop.DBus.Properties")
|
||||
items = props.Get(self.interface, "RegisteredStatusNotifierItems")
|
||||
return len(items)
|
||||
except:
|
||||
return 0
|
||||
|
||||
def push_update(self, *args):
|
||||
# шлем цифру в stdout
|
||||
count = self.get_count()
|
||||
sys.stdout.write(f"{count}\n")
|
||||
sys.stdout.flush()
|
||||
|
||||
def run(self):
|
||||
# цикл glib, чтобы слушал сигналы
|
||||
self.loop = GLib.MainLoop()
|
||||
try:
|
||||
self.loop.run()
|
||||
except KeyboardInterrupt:
|
||||
sys.exit(0)
|
||||
|
||||
if __name__ == "__main__":
|
||||
TrayMonitor().run()
|
||||
Executable
+42
@@ -0,0 +1,42 @@
|
||||
import subprocess
|
||||
import sys
|
||||
import os
|
||||
import signal
|
||||
|
||||
class VolumeMonitor:
|
||||
def __init__(self):
|
||||
# отдаём текущую громкость при старте
|
||||
self.push_update()
|
||||
|
||||
def get_vol(self):
|
||||
# достаём цифры из pactl
|
||||
try:
|
||||
raw = subprocess.check_output(["pactl", "get-sink-volume", "@DEFAULT_SINK@"], stderr=subprocess.DEVNULL).decode()
|
||||
if "/" in raw:
|
||||
# забираем канал и чистим от процентов
|
||||
return raw.split("/")[1].strip().replace("%", "")
|
||||
except:
|
||||
return "0"
|
||||
return "0"
|
||||
|
||||
def push_update(self):
|
||||
# отдаём данные в stdout для eww
|
||||
vol = self.get_vol()
|
||||
sys.stdout.write(f"{vol}\n")
|
||||
sys.stdout.flush()
|
||||
|
||||
def run(self):
|
||||
# подписываемся на события аудио-сервера
|
||||
proc = subprocess.Popen(["pactl", "subscribe"], stdout=subprocess.PIPE, text=True)
|
||||
|
||||
try:
|
||||
for line in proc.stdout:
|
||||
# если изменился основной выход
|
||||
if "sink" in line:
|
||||
self.push_update()
|
||||
except KeyboardInterrupt:
|
||||
proc.terminate()
|
||||
sys.exit(0)
|
||||
|
||||
if __name__ == "__main__":
|
||||
VolumeMonitor().run()
|
||||
Executable
+213
@@ -0,0 +1,213 @@
|
||||
import json
|
||||
import subprocess
|
||||
import os
|
||||
import sys
|
||||
import socket
|
||||
import threading
|
||||
import time
|
||||
|
||||
class UnifiedWorkspaceManager:
|
||||
def __init__(self):
|
||||
self.last_sent_json = ""
|
||||
self.urgent_window_addresses = set()
|
||||
self.script_directory = os.path.dirname(os.path.realpath(__file__))
|
||||
|
||||
self.clients_cache = {}
|
||||
self.monitors_cache = {}
|
||||
|
||||
self.load_icon_configuration()
|
||||
self.clients_refresh()
|
||||
self.monitors_refresh()
|
||||
|
||||
def load_icon_configuration(self):
|
||||
# тянем иконки из конфига
|
||||
config_path = os.path.join(self.script_directory, "workspaces_map.json")
|
||||
try:
|
||||
with open(config_path, 'r') as file:
|
||||
self.icon_map = json.load(file)
|
||||
except:
|
||||
self.icon_map = {"default": "", "no-icon": ""}
|
||||
|
||||
def monitors_refresh(self):
|
||||
# обновляем инфу по мониторам
|
||||
try:
|
||||
raw = subprocess.check_output(["hyprctl", "-j", "monitors"], stderr=subprocess.DEVNULL)
|
||||
monitors = json.loads(raw)
|
||||
self.monitors_cache = {
|
||||
m['name']: {
|
||||
"id": m['id'],
|
||||
"active_workspace_id": m['activeWorkspace']['id'],
|
||||
"focused": m.get('focused', False)
|
||||
} for m in monitors
|
||||
}
|
||||
except:
|
||||
self.monitors_cache = {}
|
||||
|
||||
def clients_refresh(self):
|
||||
# полный сброс и пересборка кэша окон
|
||||
try:
|
||||
raw = subprocess.check_output(["hyprctl", "-j", "clients"], stderr=subprocess.DEVNULL)
|
||||
clients = json.loads(raw)
|
||||
self.clients_cache = {
|
||||
c['address'].strip().replace('0x', ''): {
|
||||
"class": c['class'],
|
||||
"workspace_id": c['workspace']['id'],
|
||||
"monitor_id": c['monitor']
|
||||
} for c in clients
|
||||
}
|
||||
except:
|
||||
self.clients_cache = {}
|
||||
|
||||
def get_current_state_json(self):
|
||||
# собираем финальный json
|
||||
try:
|
||||
output = {}
|
||||
|
||||
for mon_name, mon_data in self.monitors_cache.items():
|
||||
m_id = mon_data["id"]
|
||||
active_id = mon_data["active_workspace_id"]
|
||||
focused = mon_data["focused"]
|
||||
|
||||
# готовим 10 слотов
|
||||
workspaces = {
|
||||
str(i): {"exists": False, "active": False, "focused": False, "urgent": False, "icons": []}
|
||||
for i in range(1, 11)
|
||||
}
|
||||
|
||||
# раскидываем окна по воркспейсам
|
||||
for addr, client in self.clients_cache.items():
|
||||
if client['monitor_id'] == m_id:
|
||||
ws_id = str(client['workspace_id'])
|
||||
if ws_id in workspaces:
|
||||
slot = workspaces[ws_id]
|
||||
slot["exists"] = True
|
||||
if addr in self.urgent_window_addresses:
|
||||
slot["urgent"] = True
|
||||
|
||||
icon = self.icon_map.get(client['class'], self.icon_map.get("default", ""))
|
||||
if icon not in slot["icons"]:
|
||||
slot["icons"].append(icon)
|
||||
|
||||
# помечаем текущий активный воркспейс
|
||||
curr_ws = str(active_id)
|
||||
if curr_ws in workspaces:
|
||||
workspaces[curr_ws].update({"active": True, "focused": focused, "exists": True})
|
||||
|
||||
# рисуем no-icon
|
||||
for ws in workspaces.values():
|
||||
if ws["exists"] and not ws["icons"]:
|
||||
ws["icons"] = [self.icon_map.get("no-icon", "")]
|
||||
|
||||
output[mon_name.replace("-", "_")] = workspaces
|
||||
|
||||
return json.dumps(output)
|
||||
except:
|
||||
return "{}"
|
||||
|
||||
def push_update(self):
|
||||
# отдаём json в stdout при изменении
|
||||
current = self.get_current_state_json()
|
||||
if current != self.last_sent_json:
|
||||
self.last_sent_json = current
|
||||
sys.stdout.write(current + '\n')
|
||||
sys.stdout.flush()
|
||||
|
||||
def update_client(self, window_address):
|
||||
# обновляем окно
|
||||
try:
|
||||
clean_addr = window_address.replace('0x', '')
|
||||
search_addr = f"0x{clean_addr}"
|
||||
|
||||
raw = subprocess.check_output(["hyprctl", "-j", "clients"], stderr=subprocess.DEVNULL)
|
||||
clients = json.loads(raw)
|
||||
|
||||
target = next((c for c in clients if c['address'] == search_addr), None)
|
||||
|
||||
if target:
|
||||
self.clients_cache[clean_addr] = {
|
||||
"class": target.get('class', 'unknown'),
|
||||
"workspace_id": target['workspace']['id'],
|
||||
"monitor_id": target['monitor']
|
||||
}
|
||||
else:
|
||||
self.clients_cache.pop(clean_addr, None)
|
||||
|
||||
except:
|
||||
# если что-то отвалилось, то просто пересобираем всё
|
||||
self.clients_refresh()
|
||||
|
||||
def run(self):
|
||||
self.push_update()
|
||||
|
||||
def event_socket_listener():
|
||||
# слушаем сокет хайпера
|
||||
signature = os.environ.get('HYPRLAND_INSTANCE_SIGNATURE')
|
||||
socket_path = f"/run/user/{os.getuid()}/hypr/{signature}/.socket2.sock"
|
||||
|
||||
while True:
|
||||
try:
|
||||
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
|
||||
sock.connect(socket_path)
|
||||
buf = ""
|
||||
while True:
|
||||
data = sock.recv(4096).decode('utf-8', errors='ignore')
|
||||
if not data: break
|
||||
buf += data
|
||||
|
||||
if '\n' in buf:
|
||||
lines = buf.split('\n')
|
||||
buf = lines.pop()
|
||||
|
||||
trigger = False
|
||||
for line in lines:
|
||||
if ">>" not in line: continue
|
||||
event, args = line.split('>>')
|
||||
|
||||
# ивенты с окнами
|
||||
if event in ["openwindow", "movewindow"]:
|
||||
addr = args.split(',')[0].replace('0x', '')
|
||||
self.update_client(addr)
|
||||
trigger = True
|
||||
|
||||
elif event == "closewindow":
|
||||
addr = args.replace('0x', '')
|
||||
self.clients_cache.pop(addr, None)
|
||||
self.urgent_window_addresses.discard(addr)
|
||||
trigger = True
|
||||
|
||||
# ивенты с мониторами
|
||||
elif event in ["workspace", "workspacev2", "monitoradded", "monitorremoved"]:
|
||||
self.monitors_refresh()
|
||||
trigger = True
|
||||
|
||||
elif event in ["focusedmon", "focusedmonv2"]:
|
||||
mon_name = args.split(',')[0]
|
||||
for name in self.monitors_cache:
|
||||
self.monitors_cache[name]["focused"] = (name == mon_name)
|
||||
trigger = True
|
||||
|
||||
# алерты
|
||||
elif event == "urgent":
|
||||
self.urgent_window_addresses.add(args.replace('0x', ''))
|
||||
trigger = True
|
||||
|
||||
elif event in ["activewindow", "activewindowv2"]:
|
||||
addr = args.split(',')[0].replace('0x', '')
|
||||
if addr in self.urgent_window_addresses:
|
||||
self.urgent_window_addresses.remove(addr)
|
||||
trigger = True
|
||||
|
||||
if trigger:
|
||||
self.push_update()
|
||||
except:
|
||||
time.sleep(2) # реконнект, если хайпер отвалился
|
||||
|
||||
# слушаем поток в фоне
|
||||
threading.Thread(target=event_socket_listener, daemon=True).start()
|
||||
|
||||
try:
|
||||
while True: time.sleep(100)
|
||||
except KeyboardInterrupt: sys.exit(0)
|
||||
|
||||
if __name__ == "__main__":
|
||||
UnifiedWorkspaceManager().run()
|
||||
@@ -0,0 +1,13 @@
|
||||
{
|
||||
"rustdesk": "",
|
||||
"org.gajim.Gajim": "",
|
||||
"firefox": "",
|
||||
"code-oss": "",
|
||||
"org.telegram.desktop": "",
|
||||
"kitty": "",
|
||||
"steam": "",
|
||||
"spotify": "",
|
||||
"discord": "",
|
||||
"no-icon": "",
|
||||
"default": ""
|
||||
}
|
||||
Reference in New Issue
Block a user