forked from Shinonome/caelestia-cli
@@ -1,18 +1,120 @@
|
|||||||
|
import json
|
||||||
|
import shutil
|
||||||
import subprocess
|
import subprocess
|
||||||
from argparse import Namespace
|
from argparse import Namespace
|
||||||
|
from collections import ChainMap
|
||||||
|
|
||||||
from caelestia.utils import hypr
|
from caelestia.utils import hypr
|
||||||
|
from caelestia.utils.paths import user_config_path
|
||||||
|
|
||||||
|
|
||||||
|
def is_subset(superset, subset):
|
||||||
|
for key, value in subset.items():
|
||||||
|
if key not in superset:
|
||||||
|
return False
|
||||||
|
|
||||||
|
if isinstance(value, dict):
|
||||||
|
if not is_subset(superset[key], value):
|
||||||
|
return False
|
||||||
|
|
||||||
|
elif isinstance(value, str):
|
||||||
|
if value not in superset[key]:
|
||||||
|
return False
|
||||||
|
|
||||||
|
elif isinstance(value, list):
|
||||||
|
if not set(value) <= set(superset[key]):
|
||||||
|
return False
|
||||||
|
elif isinstance(value, set):
|
||||||
|
if not value <= superset[key]:
|
||||||
|
return False
|
||||||
|
|
||||||
|
else:
|
||||||
|
if not value == superset[key]:
|
||||||
|
return False
|
||||||
|
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
class DeepChainMap(ChainMap):
|
||||||
|
def __getitem__(self, key):
|
||||||
|
values = (mapping[key] for mapping in self.maps if key in mapping)
|
||||||
|
try:
|
||||||
|
first = next(values)
|
||||||
|
except StopIteration:
|
||||||
|
return self.__missing__(key)
|
||||||
|
if isinstance(first, dict):
|
||||||
|
return self.__class__(first, *values)
|
||||||
|
return first
|
||||||
|
|
||||||
|
def __repr__(self):
|
||||||
|
return repr(dict(self))
|
||||||
|
|
||||||
|
|
||||||
class Command:
|
class Command:
|
||||||
args: Namespace
|
args: Namespace
|
||||||
|
cfg: dict[str, dict[str, dict[str, any]]] | DeepChainMap
|
||||||
clients: list[dict[str, any]] = None
|
clients: list[dict[str, any]] = None
|
||||||
|
|
||||||
def __init__(self, args: Namespace) -> None:
|
def __init__(self, args: Namespace) -> None:
|
||||||
self.args = args
|
self.args = args
|
||||||
|
|
||||||
|
self.cfg = {
|
||||||
|
"communication": {
|
||||||
|
"discord": {
|
||||||
|
"enable": True,
|
||||||
|
"match": [{"class": "discord"}],
|
||||||
|
"command": ["discord"],
|
||||||
|
"move": True,
|
||||||
|
},
|
||||||
|
"whatsapp": {
|
||||||
|
"enable": True,
|
||||||
|
"match": [{"class": "whatsapp"}],
|
||||||
|
"move": True,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
"music": {
|
||||||
|
"spotify": {
|
||||||
|
"enable": True,
|
||||||
|
"match": [{"class": "Spotify"}, {"initialTitle": "Spotify"}, {"initialTitle": "Spotify Free"}],
|
||||||
|
"command": ["spicetify", "watch", "-s"],
|
||||||
|
"move": True,
|
||||||
|
},
|
||||||
|
"feishin": {
|
||||||
|
"enable": True,
|
||||||
|
"match": [{"class": "feishin"}],
|
||||||
|
"move": True,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
"sysmon": {
|
||||||
|
"btop": {
|
||||||
|
"enable": True,
|
||||||
|
"match": [{"class": "btop", "title": "btop", "workspace": {"name": "special:sysmon"}}],
|
||||||
|
"command": ["foot", "-a", "btop", "-T", "btop", "fish", "-C", "exec btop"],
|
||||||
|
},
|
||||||
|
},
|
||||||
|
"todo": {
|
||||||
|
"todoist": {
|
||||||
|
"enable": True,
|
||||||
|
"match": [{"class": "Todoist"}],
|
||||||
|
"command": ["todoist"],
|
||||||
|
"move": True,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
try:
|
||||||
|
self.cfg = DeepChainMap(json.loads(user_config_path.read_text())["toggles"], self.cfg)
|
||||||
|
except (FileNotFoundError, json.JSONDecodeError, KeyError):
|
||||||
|
pass
|
||||||
|
|
||||||
def run(self) -> None:
|
def run(self) -> None:
|
||||||
getattr(self, self.args.workspace)()
|
if self.args.workspace == "specialws":
|
||||||
|
self.specialws()
|
||||||
|
return
|
||||||
|
|
||||||
|
for client in self.cfg[self.args.workspace].values():
|
||||||
|
if "enable" in client and client["enable"]:
|
||||||
|
self.handle_client_config(client)
|
||||||
|
hypr.dispatch("togglespecialworkspace", self.args.workspace)
|
||||||
|
|
||||||
def get_clients(self) -> list[dict[str, any]]:
|
def get_clients(self) -> list[dict[str, any]]:
|
||||||
if self.clients is None:
|
if self.clients is None:
|
||||||
@@ -22,45 +124,27 @@ class Command:
|
|||||||
|
|
||||||
def move_client(self, selector: callable, workspace: str) -> None:
|
def move_client(self, selector: callable, workspace: str) -> None:
|
||||||
for client in self.get_clients():
|
for client in self.get_clients():
|
||||||
if selector(client):
|
if selector(client) and client["workspace"]["name"] != f"special:{workspace}":
|
||||||
hypr.dispatch("movetoworkspacesilent", f"special:{workspace},address:{client['address']}")
|
hypr.dispatch("movetoworkspacesilent", f"special:{workspace},address:{client['address']}")
|
||||||
|
|
||||||
def spawn_client(self, selector: callable, spawn: list[str]) -> bool:
|
def spawn_client(self, selector: callable, spawn: list[str]) -> None:
|
||||||
exists = any(selector(client) for client in self.get_clients())
|
if (spawn[0].endswith(".desktop") or shutil.which(spawn[0])) and not any(
|
||||||
|
selector(client) for client in self.get_clients()
|
||||||
if not exists:
|
):
|
||||||
subprocess.Popen(["app2unit", "--", *spawn], start_new_session=True)
|
subprocess.Popen(["app2unit", "--", *spawn], start_new_session=True)
|
||||||
|
|
||||||
return not exists
|
def handle_client_config(self, client: dict[str, any]) -> None:
|
||||||
|
def selector(c: dict[str, any]) -> bool:
|
||||||
|
# Each match is or, inside matches is and
|
||||||
|
for match in client["match"]:
|
||||||
|
if is_subset(c, match):
|
||||||
|
return True
|
||||||
|
return False
|
||||||
|
|
||||||
def spawn_or_move(self, selector: callable, spawn: list[str], workspace: str) -> None:
|
if "command" in client and client["command"]:
|
||||||
if not self.spawn_client(selector, spawn):
|
self.spawn_client(selector, client["command"])
|
||||||
self.move_client(selector, workspace)
|
if "move" in client and client["move"]:
|
||||||
|
self.move_client(selector, self.args.workspace)
|
||||||
def communication(self) -> None:
|
|
||||||
self.spawn_or_move(lambda c: c["class"] == "discord", ["discord"], "communication")
|
|
||||||
self.move_client(lambda c: c["class"] == "whatsapp", "communication")
|
|
||||||
hypr.dispatch("togglespecialworkspace", "communication")
|
|
||||||
|
|
||||||
def music(self) -> None:
|
|
||||||
self.spawn_or_move(
|
|
||||||
lambda c: c["class"] == "Spotify" or c["initialTitle"] == "Spotify" or c["initialTitle"] == "Spotify Free",
|
|
||||||
["spicetify", "watch", "-s"],
|
|
||||||
"music",
|
|
||||||
)
|
|
||||||
self.move_client(lambda c: c["class"] == "feishin", "music")
|
|
||||||
hypr.dispatch("togglespecialworkspace", "music")
|
|
||||||
|
|
||||||
def sysmon(self) -> None:
|
|
||||||
self.spawn_client(
|
|
||||||
lambda c: c["class"] == "btop" and c["title"] == "btop" and c["workspace"]["name"] == "special:sysmon",
|
|
||||||
["foot", "-a", "btop", "-T", "btop", "fish", "-C", "exec btop"],
|
|
||||||
)
|
|
||||||
hypr.dispatch("togglespecialworkspace", "sysmon")
|
|
||||||
|
|
||||||
def todo(self) -> None:
|
|
||||||
self.spawn_or_move(lambda c: c["class"] == "Todoist", ["todoist"], "todo")
|
|
||||||
hypr.dispatch("togglespecialworkspace", "todo")
|
|
||||||
|
|
||||||
def specialws(self) -> None:
|
def specialws(self) -> None:
|
||||||
workspaces = hypr.message("workspaces")
|
workspaces = hypr.message("workspaces")
|
||||||
|
|||||||
Reference in New Issue
Block a user