Shinonome/caelestia-cli
1
0
Fork 1
mirror of https://github.com/caelestia-dots/cli.git synced 2026-06-05 23:09:27 -05:00
Code Issues Packages Projects Releases Wiki Activity

theme: continue execution after failure for one theme (#50)

Browse Source
This commit is contained in:
sweenu
2025-09-09 05:59:04 +02:00
committed by GitHub
parent c20bc567a4
commit 3319d2ca19
3 changed files with 57 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/caelestia/subcommands/resizer.py
+23 -26
View File
@@ -7,6 +7,7 @@ from pathlib import Path
from typing import Any, Dict, Optional
from caelestia.utils import hypr
from caelestia.utils.logging import log_message
from caelestia.utils.paths import user_config_path
@@ -49,16 +50,12 @@ class Command:
)
return rules
except (json.JSONDecodeError, KeyError):
self._log_message("ERROR: invalid config")
log_message("ERROR: invalid config")
except FileNotFoundError:
pass
return default_rules
def _log_message(self, message: str) -> None:
timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
print(f"[{timestamp}] {message}")
def _is_rate_limited(self, key: str) -> bool:
current_time = time.time()
last_time = self.timeout_tracker.get(key, 0)
@@ -170,12 +167,12 @@ class Command:
command2 = f"dispatch movewindowpixel exact {int(move_x)} {int(move_y)},address:{address}"
hypr.batch(command1, command2)
self._log_message(
log_message(
f"Applied PiP action to window {address}: {scaled_width}x{scaled_height} at ({move_x}, {move_y})"
)
except Exception as e:
self._log_message(f"ERROR: Failed to apply PiP action to window 0x{window_id}: {e}")
log_message(f"ERROR: Failed to apply PiP action to window 0x{window_id}: {e}")
def _apply_window_actions(self, window_id: str, width: str, height: str, actions: list[str]) -> bool:
dispatch_commands = []
@@ -196,10 +193,10 @@ class Command:
try:
hypr.batch(*dispatch_commands)
self._log_message(f"Applied actions to window 0x{window_id}: {width} x {height} ({', '.join(actions)})")
log_message(f"Applied actions to window 0x{window_id}: {width} x {height} ({', '.join(actions)})")
return True
except Exception as e:
self._log_message(f"ERROR: Failed to apply window actions for window 0x{window_id}: {e}")
log_message(f"ERROR: Failed to apply window actions for window 0x{window_id}: {e}")
return False
def _match_window_rule(self, window_title: str, initial_title: str) -> WindowRule | None:
@@ -218,7 +215,7 @@ class Command:
if re.search(rule.name, window_title):
return rule
except re.error:
self._log_message(f"ERROR: Invalid regex pattern in rule '{rule.name}'")
log_message(f"ERROR: Invalid regex pattern in rule '{rule.name}'")
return None
@@ -240,7 +237,7 @@ class Command:
window_id = window_id.lstrip(">")
if not all(c in "0123456789abcdefABCDEF" for c in window_id):
self._log_message(f"ERROR: Invalid window ID format: {window_id}")
log_message(f"ERROR: Invalid window ID format: {window_id}")
return
window_info = self._get_window_info(window_id)
@@ -250,19 +247,19 @@ class Command:
window_title = window_info.get("title", "")
initial_title = window_info.get("initialTitle", "")
self._log_message(f"DEBUG: Window 0x{window_id} - Title: '{window_title}' | Initial: '{initial_title}'")
log_message(f"DEBUG: Window 0x{window_id} - Title: '{window_title}' | Initial: '{initial_title}'")
rule = self._match_window_rule(window_title, initial_title)
if rule:
if self._is_rate_limited(window_id):
self._log_message(f"Rate limited: skipping window 0x{window_id}")
log_message(f"Rate limited: skipping window 0x{window_id}")
return
self._log_message(f"Matched rule '{rule.name}' for window 0x{window_id}")
log_message(f"Matched rule '{rule.name}' for window 0x{window_id}")
self._apply_window_actions(window_id, rule.width, rule.height, rule.actions)
except (IndexError, ValueError) as e:
self._log_message(f"ERROR: Failed to parse window title event: {e}")
log_message(f"ERROR: Failed to parse window title event: {e}")
def _handle_open_event(self, event: str) -> None:
try:
@@ -278,22 +275,22 @@ class Command:
window_id = window_id.lstrip(">")
if not all(c in "0123456789abcdefABCDEF" for c in window_id):
self._log_message(f"ERROR: Invalid window ID format: {window_id}")
log_message(f"ERROR: Invalid window ID format: {window_id}")
return
self._log_message(f"DEBUG: New window 0x{window_id} - Title: '{title}' | Class: '{window_class}'")
log_message(f"DEBUG: New window 0x{window_id} - Title: '{title}' | Class: '{window_class}'")
rule = self._match_window_rule(title, title)
if rule:
if self._is_rate_limited(window_id):
self._log_message(f"Rate limited: skipping window 0x{window_id}")
log_message(f"Rate limited: skipping window 0x{window_id}")
return
self._log_message(f"Matched rule '{rule.name}' for new window 0x{window_id}")
log_message(f"Matched rule '{rule.name}' for new window 0x{window_id}")
self._apply_window_actions(window_id, rule.width, rule.height, rule.actions)
except (IndexError, ValueError) as e:
self._log_message(f"ERROR: Failed to parse window open event: {e}")
log_message(f"ERROR: Failed to parse window open event: {e}")
def run(self) -> None:
if self.args.daemon:
@@ -440,19 +437,19 @@ class Command:
return []
def _run_daemon(self) -> None:
self._log_message("Hyprland window resizer started")
self._log_message(f"Loaded {len(self.window_rules)} window rules")
log_message("Hyprland window resizer started")
log_message(f"Loaded {len(self.window_rules)} window rules")
socket_path = Path(hypr.socket2_path)
if not socket_path.exists():
self._log_message(f"ERROR: Hyprland socket not found at {socket_path}")
log_message(f"ERROR: Hyprland socket not found at {socket_path}")
return
try:
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
sock.connect(hypr.socket2_path)
self._log_message("Connected to Hyprland socket, listening for events...")
log_message("Connected to Hyprland socket, listening for events...")
while True:
data = sock.recv(4096).decode()
@@ -462,6 +459,6 @@ class Command:
self._handle_window_event(line)
except KeyboardInterrupt:
self._log_message("Resizer daemon stopped")
log_message("Resizer daemon stopped")
except Exception as e:
self._log_message(f"ERROR: {e}")
log_message(f"ERROR: {e}")
src/caelestia/utils/logging.py
+20
View File
@@ -0,0 +1,20 @@
from time import strftime
def log_message(message: str) -> None:
timestamp = strftime("%Y-%m-%d %H:%M:%S")
print(f"[{timestamp}] {message}")
def log_exception(func):
"""Log exceptions to stdout instead of raising
Used by the `apply_()` functions so that an exception, when applying
a theme, does not prevent the other themes from being applied.
"""
def wrapper(*args, **kwargs):
try:
func(*args, **kwargs)
except Exception as e:
log_message(f'Error during execution of "{func.__name__}()": {str(e)}')
return wrapper
src/caelestia/utils/theme.py
+14
View File
@@ -4,6 +4,7 @@ import subprocess
from pathlib import Path
from caelestia.utils.colour import get_dynamic_colours
from caelestia.utils.logging import log_exception
from caelestia.utils.paths import (
c_state_dir,
config_dir,
@@ -103,6 +104,7 @@ def write_file(path: Path, content: str) -> None:
path.write_text(content)
@log_exception
def apply_terms(sequences: str) -> None:
state = c_state_dir / "sequences.txt"
state.parent.mkdir(parents=True, exist_ok=True)
@@ -118,10 +120,12 @@ def apply_terms(sequences: str) -> None:
pass
@log_exception
def apply_hypr(conf: str) -> None:
write_file(config_dir / "hypr/scheme/current.conf", conf)
@log_exception
def apply_discord(scss: str) -> None:
import tempfile
@@ -133,33 +137,39 @@ def apply_discord(scss: str) -> None:
write_file(config_dir / client / "themes/caelestia.theme.css", conf)
@log_exception
def apply_spicetify(colours: dict[str, str], mode: str) -> None:
template = gen_replace(colours, templates_dir / f"spicetify-{mode}.ini")
write_file(config_dir / "spicetify/Themes/caelestia/color.ini", template)
@log_exception
def apply_fuzzel(colours: dict[str, str]) -> None:
template = gen_replace(colours, templates_dir / "fuzzel.ini")
write_file(config_dir / "fuzzel/fuzzel.ini", template)
@log_exception
def apply_btop(colours: dict[str, str]) -> None:
template = gen_replace(colours, templates_dir / "btop.theme", hash=True)
write_file(config_dir / "btop/themes/caelestia.theme", template)
subprocess.run(["killall", "-USR2", "btop"], stderr=subprocess.DEVNULL)
@log_exception
def apply_nvtop(colours: dict[str, str]) -> None:
template = gen_replace(colours, templates_dir / "nvtop.colors", hash=True)
write_file(config_dir / "nvtop/nvtop.colors", template)
@log_exception
def apply_htop(colours: dict[str, str]) -> None:
template = gen_replace(colours, templates_dir / "htop.theme", hash=True)
write_file(config_dir / "htop/htoprc", template)
subprocess.run(["killall", "-USR2", "htop"], stderr=subprocess.DEVNULL)
@log_exception
def apply_gtk(colours: dict[str, str], mode: str) -> None:
template = gen_replace(colours, templates_dir / "gtk.css", hash=True)
write_file(config_dir / "gtk-3.0/gtk.css", template)
@@ -170,6 +180,7 @@ def apply_gtk(colours: dict[str, str], mode: str) -> None:
subprocess.run(["dconf", "write", "/org/gnome/desktop/interface/icon-theme", f"'Papirus-{mode.capitalize()}'"])
@log_exception
def apply_qt(colours: dict[str, str], mode: str) -> None:
template = gen_replace(colours, templates_dir / f"qt{mode}.colors", hash=True)
write_file(config_dir / "qt5ct/colors/caelestia.colors", template)
@@ -196,6 +207,7 @@ general="Sans Serif,12,-1,5,400,0,0,0,0,0,0,0,0,0,0,1"
write_file(config_dir / f"qt{ver}ct/qt{ver}ct.conf", conf)
@log_exception
def apply_warp(colours: dict[str, str], mode: str) -> None:
warp_mode = "darker" if mode == "dark" else "lighter"
@@ -204,12 +216,14 @@ def apply_warp(colours: dict[str, str], mode: str) -> None:
write_file(data_dir / "warp-terminal/themes/caelestia.yaml", template)
@log_exception
def apply_cava(colours: dict[str, str]) -> None:
template = gen_replace(colours, templates_dir / "cava.conf", hash=True)
write_file(config_dir / "cava/config", template)
subprocess.run(["killall", "-USR2", "cava"], stderr=subprocess.DEVNULL)
@log_exception
def apply_user_templates(colours: dict[str, str]) -> None:
if not user_templates_dir.is_dir():
return