pyrschtjag/dotfiles-hypr
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test_version

Browse Source
This commit is contained in:
pyrschtjag
2026-07-11 16:41:17 +04:00
parent f0106699b9
commit d9c9bf546b
5 changed files with 120 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files
eww/scripts/workspaces.py
+114 -118
View File
@@ -1,10 +1,9 @@
#!/usr/bin/env python3
import asyncio
import json
import subprocess
import os
import sys
import socket
import threading
import time
class UnifiedWorkspaceManager:
def __init__(self):
@@ -16,11 +15,8 @@ class UnifiedWorkspaceManager:
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:
@@ -28,53 +24,59 @@ class UnifiedWorkspaceManager:
except:
self.icon_map = {"default": "", "no-icon": ""}
def monitors_refresh(self):
# обновляем инфу по мониторам
async 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
}
proc = await asyncio.create_subprocess_exec(
"hyprctl", "-j", "monitors",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.DEVNULL
)
stdout, _ = await proc.communicate()
if proc.returncode == 0:
monitors = json.loads(stdout.decode())
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):
# полный сброс и пересборка кэша окон
async 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
}
proc = await asyncio.create_subprocess_exec(
"hyprctl", "-j", "clients",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.DEVNULL
)
stdout, _ = await proc.communicate()
if proc.returncode == 0:
clients = json.loads(stdout.decode())
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'])
@@ -88,12 +90,10 @@ class UnifiedWorkspaceManager:
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", "")]
@@ -105,109 +105,105 @@ class UnifiedWorkspaceManager:
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):
# обновляем окно
async 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)
proc = await asyncio.create_subprocess_exec(
"hyprctl", "-j", "clients",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.DEVNULL
)
stdout, _ = await proc.communicate()
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)
if proc.returncode == 0:
clients = json.loads(stdout.decode())
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']
}
return
self.clients_cache.pop(clean_addr, None)
except:
# если что-то отвалилось, то просто пересобираем всё
self.clients_refresh()
await self.clients_refresh()
def run(self):
async def run(self):
await self.clients_refresh()
await self.monitors_refresh()
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"
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
while True:
try:
reader, writer = await asyncio.open_unix_connection(socket_path)
while True:
line_bytes = await reader.readline()
if not line_bytes:
break
line = line_bytes.decode('utf-8', errors='ignore').strip()
if ">>" not in line:
continue
event, args = line.split('>>')
trigger = False
if event in ["openwindow", "movewindow"]:
addr = args.split(',')[0].replace('0x', '')
await 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 ["workspace", "workspacev2", "monitoradded", "monitorremoved"]:
await 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 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
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 trigger:
self.push_update()
except (asyncio.CancelledError, KeyboardInterrupt):
break
except Exception as e:
await asyncio.sleep(2)
if __name__ == "__main__":
UnifiedWorkspaceManager().run()
try:
asyncio.run(UnifiedWorkspaceManager().run())
except KeyboardInterrupt:
sys.exit(0)
eww/scripts/workspaces_map.json
+1 -1
View File
@@ -4,7 +4,7 @@
"firefox": "",
"code-oss": "󰨞",
"org.telegram.desktop": "",
"kitty": "",
"foot": "",
"steam": "",
"Spotify": "",
"discord": "󰙯",
hypr/config/exec.conf
+1
View File
@@ -6,3 +6,4 @@ exec-once = gsettings set org.gnome.desktop.interface gtk-theme "Adwaita-dark"
exec-once = gsettings set org.gnome.desktop.interface color-scheme "prefer-dark"
exec-once = eww daemon && sleep 1 && eww open-many bar_0 bar_1 wallpaper_0 wallpaper_1
exec-once = hypridle --grace 15
exec-once = python3 /home/pyrschtjag/Documents/.ntfy-notify/ntfy-notify.py
hypr/config/programs.conf
+2 -2
View File
@@ -1,4 +1,4 @@
$mainMod = SUPER
$terminal = kitty
$fileManager = kitty mc
$terminal = foot
//$fileManager = kitty mc
$menu = rofi -show drun
hypr/hyprland.conf
+2 -2
View File
@@ -6,9 +6,9 @@ source = ~/.config/hypr/config/env.conf
source = ~/.config/hypr/config/general.conf
source = ~/.config/hypr/config/decoration.conf
source = ~/.config/hypr/config/animations.conf
source = ~/.config/hypr/config/dwindle.conf
//source = ~/.config/hypr/config/dwindle.conf
source = ~/.config/hypr/config/master.conf
source = ~/.config/hypr/config/misc.conf
source = ~/.config/hypr/config/input.conf
source = ~/.config/hypr/config/bind.conf
source = ~/.config/hypr/config/windowrule.conf
source = ~/.config/hypr/config/windowrule.conf