♻️ refactor(dock): overhaul dock implementation for improved performance and maintainability
This commit is contained in:
sankalp committed
1 parent
8ba98e33d3
commit
4852fa01e2
22 files changed
+876
-2755
No files matched your search
@@ -1,98 +0,0 @@
|
||||
from xml import etree
|
||||
from xml.etree import ElementTree
|
||||
from fabric.utils import get_relative_path
|
||||
import json
|
||||
|
||||
|
||||
def parse(xml):
|
||||
with open(xml, "rb") as f:
|
||||
tree = ElementTree.parse(f)
|
||||
root = tree.getroot()
|
||||
return root
|
||||
|
||||
|
||||
def dxml(root, depth=0):
|
||||
val = []
|
||||
for child in root:
|
||||
attrstring = {}
|
||||
additional = {}
|
||||
if len(child.attrib) == 0:
|
||||
if child.tag == "BoxV":
|
||||
additional["orientation"] = "vertical"
|
||||
elif child.tag == "BoxH":
|
||||
additional["orientation"] = "horizontal"
|
||||
for k, v in child.attrib.items():
|
||||
if v == None:
|
||||
continue
|
||||
if child.tag == "Ref":
|
||||
continue
|
||||
if child.tag == "BoxV":
|
||||
additional["orientation"] = "vertical"
|
||||
elif child.tag == "BoxH":
|
||||
additional["orientation"] = "horizontal"
|
||||
if k == "path":
|
||||
# Handle path resolution more robustly
|
||||
import os
|
||||
|
||||
# Get the directory where the XML file is located
|
||||
current_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
# Go up to project root and then to the specified path
|
||||
project_root = os.path.dirname(current_dir) # Go up from utils/
|
||||
full_path = os.path.join(project_root, v)
|
||||
v = full_path
|
||||
k = "svg_file"
|
||||
if v == "true" or v == "false":
|
||||
v = v == "true"
|
||||
attrstring[k] = v
|
||||
for k, v in additional.items():
|
||||
attrstring[k] = v
|
||||
tag = child.tag
|
||||
if tag in ["BoxV", "BoxH"]:
|
||||
tag = "Box"
|
||||
if tag == "Ref":
|
||||
val.append(
|
||||
{
|
||||
"name": tag,
|
||||
"text": child.text,
|
||||
}
|
||||
)
|
||||
continue
|
||||
val.append(
|
||||
{
|
||||
"name": tag,
|
||||
"attributes": attrstring,
|
||||
"children": dxml(child, depth + 1),
|
||||
}
|
||||
)
|
||||
return val
|
||||
|
||||
|
||||
def parse_child(tags, refs, child, parent):
|
||||
for ch in child:
|
||||
ch = ch.copy()
|
||||
name = ch.get("name", None)
|
||||
attrs = ch.get("attributes", None)
|
||||
if attrs:
|
||||
for attr in attrs:
|
||||
if str(attrs[attr]).isnumeric():
|
||||
attrs[attr] = int(attrs[attr])
|
||||
children = ch.get("children", None)
|
||||
if children:
|
||||
ch.pop("children")
|
||||
if name == "Ref":
|
||||
text = ch.get("text", None)
|
||||
v = refs[text]
|
||||
parent.children = [*parent.children, v]
|
||||
else:
|
||||
v = tags[name](**attrs)
|
||||
if children:
|
||||
v.children = parse_child(tags, refs, children, v)
|
||||
parent.children = [*parent.children, v]
|
||||
return parent.children
|
||||
|
||||
|
||||
def exml(file: str, tags: dict, refs: dict, root):
|
||||
odi = dxml(parse(file))
|
||||
di = root(children=[])
|
||||
parse_child(tags, refs, odi, di)
|
||||
return di
|
||||
@@ -0,0 +1,51 @@
|
||||
import json
|
||||
import os
|
||||
import threading
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
from loguru import logger
|
||||
|
||||
# Threading helper functions
|
||||
|
||||
|
||||
def thread(target, *args, **kwargs) -> threading.Thread:
|
||||
"""
|
||||
Simply run the given function in a thread.
|
||||
The provided args and kwargs will be passed to the function.
|
||||
"""
|
||||
th = threading.Thread(target=target, args=args, kwargs=kwargs, daemon=True)
|
||||
th.start()
|
||||
return th
|
||||
|
||||
|
||||
def run_in_thread(func):
|
||||
"""
|
||||
Decorator to run the decorated function in a thread.
|
||||
"""
|
||||
|
||||
def wrapper(*args, **kwargs):
|
||||
return thread(func, *args, **kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
@run_in_thread
|
||||
def write_json_file(data: Dict, path: str):
|
||||
try:
|
||||
with open(path, "w") as f:
|
||||
json.dump(data, f, indent=4, ensure_ascii=False)
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to write json: {e}")
|
||||
|
||||
|
||||
def read_json_file(file_path: str) -> Optional[List]:
|
||||
if not os.path.exists(file_path):
|
||||
logger.error(f"JSON file {file_path} does not exist.")
|
||||
return None
|
||||
|
||||
with open(file_path, "r") as file:
|
||||
try:
|
||||
return json.load(file)
|
||||
except json.JSONDecodeError as e:
|
||||
logger.error(f"Failed to read JSON file {file_path}: {e}")
|
||||
return None
|
||||
-234
@@ -1,234 +0,0 @@
|
||||
# Parameters
|
||||
font_family: str = "tabler-icons"
|
||||
font_weight: str = "normal"
|
||||
|
||||
span: str = f"<span font-family='{font_family}' font-weight='{font_weight}'>"
|
||||
|
||||
# Panels
|
||||
apps: str = ""
|
||||
dashboard: str = ""
|
||||
chat: str = ""
|
||||
windows: str = ""
|
||||
|
||||
# Bar
|
||||
colorpicker: str = ""
|
||||
media: str = ""
|
||||
|
||||
# Toolbox
|
||||
|
||||
toolbox: str = ""
|
||||
ssfull: str = ""
|
||||
ssregion: str = ""
|
||||
sswindow: str = ""
|
||||
screenshots: str = ""
|
||||
screenrecord: str = ""
|
||||
recordings: str = ""
|
||||
ocr: str = "ﳃ"
|
||||
gamemode: str = ""
|
||||
gamemode_off: str = ""
|
||||
close: str = ""
|
||||
|
||||
# Circles
|
||||
temp: str = ""
|
||||
disk: str = ""
|
||||
battery: str = ""
|
||||
memory: str = "流"
|
||||
cpu: str = ""
|
||||
gpu: str = ""
|
||||
swap: str = ""
|
||||
|
||||
|
||||
# AIchat
|
||||
reload: str = ""
|
||||
detach: str = ""
|
||||
|
||||
# Wallpapers
|
||||
add: str = ""
|
||||
sort: str = ""
|
||||
circle: str = ""
|
||||
|
||||
# Chevrons
|
||||
chevron_up: str = ""
|
||||
chevron_down: str = ""
|
||||
chevron_left: str = ""
|
||||
chevron_right: str = ""
|
||||
|
||||
# Power
|
||||
lock: str = ""
|
||||
suspend: str = ""
|
||||
logout: str = ""
|
||||
reboot: str = ""
|
||||
shutdown: str = ""
|
||||
|
||||
# Power Manager
|
||||
power_saving: str = ""
|
||||
power_balanced: str = "勺"
|
||||
power_performance: str = ""
|
||||
charging: str = ""
|
||||
discharging: str = ""
|
||||
alert: str = ""
|
||||
bat_charging: str = ""
|
||||
bat_discharging: str = ""
|
||||
bat_low: str = "＝"
|
||||
bat_full: str = ""
|
||||
|
||||
|
||||
# Applets
|
||||
|
||||
# Applets
|
||||
wifi_0: str = ""
|
||||
wifi_1: str = ""
|
||||
wifi_2: str = ""
|
||||
wifi_3: str = ""
|
||||
world: str = ""
|
||||
world_off: str = ""
|
||||
bluetooth: str = ""
|
||||
night: str = ""
|
||||
coffee: str = ""
|
||||
notifications: str = ""
|
||||
notifications_dots: str = ""
|
||||
|
||||
wifi_off: str = ""
|
||||
bluetooth_off: str = ""
|
||||
night_off: str = ""
|
||||
notifications_off: str = ""
|
||||
|
||||
notifications_clear: str = ""
|
||||
|
||||
download: str = ""
|
||||
upload: str = ""
|
||||
|
||||
# Bluetooth
|
||||
bluetooth_connected: str = ""
|
||||
bluetooth_disconnected: str = ""
|
||||
|
||||
# Player
|
||||
pause: str = ""
|
||||
play: str = ""
|
||||
stop: str = ""
|
||||
skip_back: str = ""
|
||||
skip_forward: str = ""
|
||||
prev: str = ""
|
||||
next: str = ""
|
||||
shuffle: str = ""
|
||||
repeat: str = ""
|
||||
music: str = ""
|
||||
rewind_backward_5: str = "謹"
|
||||
rewind_forward_5: str = "難"
|
||||
|
||||
next_fill: str = chr(0xF04AD)
|
||||
previous_fill: str = chr(0xF04AE)
|
||||
|
||||
# Volume
|
||||
vol_off: str = ""
|
||||
vol_mute: str = ""
|
||||
vol_medium: str = ""
|
||||
vol_high: str = ""
|
||||
|
||||
mic: str = ""
|
||||
mic_mute: str = ""
|
||||
|
||||
speaker: str = "𐁅"
|
||||
headphones: str = "屮"
|
||||
mic_filled: str = "️"
|
||||
|
||||
# Overview
|
||||
circle_plus: str = ""
|
||||
|
||||
# Pins
|
||||
paperclip: str = ""
|
||||
|
||||
# Clipboard Manager
|
||||
clipboard: str = ""
|
||||
clip_text: str = ""
|
||||
|
||||
# Confirm
|
||||
accept: str = ""
|
||||
cancel: str = ""
|
||||
trash: str = ""
|
||||
|
||||
# Config
|
||||
config: str = ""
|
||||
|
||||
# Icons
|
||||
firefox: str = ""
|
||||
chromium: str = ""
|
||||
spotify: str = "ﺆ"
|
||||
disc: str = "𐀾"
|
||||
disc_off: str = ""
|
||||
|
||||
# Brightness
|
||||
brightness_low: str = ""
|
||||
brightness_medium: str = ""
|
||||
brightness_high: str = ""
|
||||
|
||||
brightness: str = ""
|
||||
|
||||
# Dashboard
|
||||
widgets: str = ""
|
||||
pins: str = ""
|
||||
kanban: str = ""
|
||||
wallpapers: str = ""
|
||||
sparkles: str = ""
|
||||
|
||||
# Misc
|
||||
dot: str = ""
|
||||
palette: str = ""
|
||||
cloud_off: str = ""
|
||||
loader: str = ""
|
||||
radar: str = ""
|
||||
emoji: str = ""
|
||||
keyboard: str = ""
|
||||
terminal: str = ""
|
||||
timer_off: str = ""
|
||||
timer_on: str = ""
|
||||
spy: str = ""
|
||||
calculator: str = ""
|
||||
file: str = ""
|
||||
calendar: str = ""
|
||||
connect: str = ""
|
||||
disconnect: str = "ﴖ"
|
||||
auth: str = ""
|
||||
bookmark: str = ""
|
||||
star: str = ""
|
||||
process: str = ""
|
||||
|
||||
|
||||
music: str = ""
|
||||
# otp
|
||||
check: str = ""
|
||||
bulb: str = ""
|
||||
key: str = ""
|
||||
magnifier: str = ""
|
||||
plus: str = ""
|
||||
info: str = ""
|
||||
|
||||
# Dice
|
||||
dice_1: str = ""
|
||||
dice_2: str = ""
|
||||
dice_3: str = ""
|
||||
dice_4: str = ""
|
||||
dice_5: str = ""
|
||||
dice_6: str = ""
|
||||
|
||||
# search engine
|
||||
reddit: str = ""
|
||||
youtube: str = ""
|
||||
duckduckgo: str = ""
|
||||
github: str = ""
|
||||
stackoverflow: str = ""
|
||||
google: str = ""
|
||||
linkedin: str = ""
|
||||
wikipedia: str = "啕"
|
||||
|
||||
exceptions: list[str] = ["font_family", "font_weight", "span"]
|
||||
|
||||
|
||||
def apply_span() -> None:
|
||||
global_dict = globals()
|
||||
for key in global_dict:
|
||||
if key not in exceptions and not key.startswith("__"):
|
||||
global_dict[key] = f"{span}{global_dict[key]}</span>"
|
||||
|
||||
|
||||
apply_span()
|
||||
@@ -0,0 +1,60 @@
|
||||
import json
|
||||
import warnings
|
||||
from typing import Dict
|
||||
import time
|
||||
from fabric.hyprland import Hyprland
|
||||
from gi.repository import Gdk
|
||||
from functools import lru_cache
|
||||
|
||||
warnings.filterwarnings("ignore", category=DeprecationWarning)
|
||||
|
||||
|
||||
def ttl_lru_cache(seconds_to_live: int, maxsize: int = 128):
|
||||
def wrapper(func):
|
||||
@lru_cache(maxsize)
|
||||
def inner(__ttl, *args, **kwargs):
|
||||
return func(*args, **kwargs)
|
||||
|
||||
return lambda *args, **kwargs: inner(
|
||||
time.time() // seconds_to_live, *args, **kwargs
|
||||
)
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
class HyprlandWithMonitors(Hyprland):
|
||||
"""A Hyprland class with additional monitor common."""
|
||||
|
||||
instance = None
|
||||
|
||||
@staticmethod
|
||||
def get_default():
|
||||
if HyprlandWithMonitors.instance is None:
|
||||
HyprlandWithMonitors.instance = HyprlandWithMonitors()
|
||||
|
||||
return HyprlandWithMonitors.instance
|
||||
|
||||
def __init__(self, commands_only: bool = False, **kwargs):
|
||||
self.display: Gdk.Display = Gdk.Display.get_default()
|
||||
super().__init__(commands_only, **kwargs)
|
||||
|
||||
@ttl_lru_cache(100, 5)
|
||||
def get_all_monitors(self) -> Dict:
|
||||
monitors = json.loads(self.send_command("j/monitors").reply)
|
||||
return {monitor["id"]: monitor["name"] for monitor in monitors}
|
||||
|
||||
def get_gdk_monitor_id_from_name(self, plug_name: str) -> int | None:
|
||||
for i in range(self.display.get_n_monitors()):
|
||||
if self.display.get_default_screen().get_monitor_plug_name(i) == plug_name:
|
||||
return i
|
||||
return None
|
||||
|
||||
def get_gdk_monitor_id(self, hyprland_id: int) -> int | None:
|
||||
monitors = self.get_all_monitors()
|
||||
if hyprland_id in monitors:
|
||||
return self.get_gdk_monitor_id_from_name(monitors[hyprland_id])
|
||||
return None
|
||||
|
||||
def get_current_gdk_monitor_id(self) -> int | None:
|
||||
active_workspace = json.loads(self.send_command("j/activeworkspace").reply)
|
||||
return self.get_gdk_monitor_id_from_name(active_workspace["monitor"])
|
||||
Reference in new issue
Block a user