Merge pull request #6 from caelestia-dots/python-rework

feat: rewrite in python

10 files changed, 472 insertions, 0 deletions

diff --git a/src/caelestia/subcommands/clipboard.py b/src/caelestia/subcommands/clipboard.py
new file mode 100644
index 0000000..c0eddb5
--- /dev/null
+++ b/src/caelestia/subcommands/clipboard.py
@@ -0,0 +1,25 @@
+import subprocess
+from argparse import Namespace
+
+
+class Command:
+    args: Namespace
+
+    def __init__(self, args: Namespace) -> None:
+        self.args = args
+
+    def run(self) -> None:
+        clip = subprocess.check_output(["cliphist", "list"])
+
+        if self.args.delete:
+            args = ["--prompt=del > ", "--placeholder=Delete from clipboard"]
+        else:
+            args = ["--placeholder=Type to search clipboard"]
+
+        chosen = subprocess.check_output(["fuzzel", "--dmenu", *args], input=clip)
+
+        if self.args.delete:
+            subprocess.run(["cliphist", "delete"], input=chosen)
+        else:
+            decoded = subprocess.check_output(["cliphist", "decode"], input=chosen)
+            subprocess.run(["wl-copy"], input=decoded)
diff --git a/src/caelestia/subcommands/emoji.py b/src/caelestia/subcommands/emoji.py
new file mode 100644
index 0000000..f04b502
--- /dev/null
+++ b/src/caelestia/subcommands/emoji.py
@@ -0,0 +1,18 @@
+import subprocess
+from argparse import Namespace
+
+from caelestia.utils.paths import cli_data_dir
+
+
+class Command:
+    args: Namespace
+
+    def __init__(self, args: Namespace) -> None:
+        self.args = args
+
+    def run(self) -> None:
+        emojis = (cli_data_dir / "emojis.txt").read_text()
+        chosen = subprocess.check_output(
+            ["fuzzel", "--dmenu", "--placeholder=Type to search emojis"], input=emojis, text=True
+        )
+        subprocess.run(["wl-copy"], input=chosen.split()[0], text=True)
diff --git a/src/caelestia/subcommands/pip.py b/src/caelestia/subcommands/pip.py
new file mode 100644
index 0000000..5f1b5fa
--- /dev/null
+++ b/src/caelestia/subcommands/pip.py
@@ -0,0 +1,44 @@
+import re
+import socket
+from argparse import Namespace
+
+from caelestia.utils import hypr
+
+
+class Command:
+    args: Namespace
+
+    def __init__(self, args: Namespace) -> None:
+        self.args = args
+
+    def run(self) -> None:
+        if self.args.daemon:
+            self.daemon()
+        else:
+            win = hypr.message("activewindow")
+            if win["floating"]:
+                self.handle_window(win["address"], win["workspace"]["name"])
+
+    def handle_window(self, address: str, ws: str) -> None:
+        mon_id = next(w for w in hypr.message("workspaces") if w["name"] == ws)["monitorID"]
+        mon = next(m for m in hypr.message("monitors") if m["id"] == mon_id)
+        width, height = next(c for c in hypr.message("clients") if c["address"] == address)["size"]
+
+        scale_factor = mon["height"] / 4 / height
+        scaled_win_size = f"{int(width * scale_factor)} {int(height * scale_factor)}"
+        off = min(mon["width"], mon["height"]) * 0.03
+        move_to = f"{int(mon['width'] - off - width * scale_factor)} {int(mon['height'] - off - height * scale_factor)}"
+
+        hypr.dispatch("resizewindowpixel", "exact", f"{scaled_win_size},address:{address}")
+        hypr.dispatch("movewindowpixel", "exact", f"{move_to},address:{address}")
+
+    def daemon(self) -> None:
+        with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
+            sock.connect(hypr.socket2_path)
+
+            while True:
+                data = sock.recv(4096).decode()
+                if data.startswith("openwindow>>"):
+                    address, ws, cls, title = data[12:].split(",")
+                    if re.match(r"^[Pp]icture(-| )in(-| )[Pp]icture$", title):
+                        self.handle_window(f"0x{address}", ws)
diff --git a/src/caelestia/subcommands/record.py b/src/caelestia/subcommands/record.py
new file mode 100644
index 0000000..a4fa51d
--- /dev/null
+++ b/src/caelestia/subcommands/record.py
@@ -0,0 +1,122 @@
+import subprocess
+import time
+from argparse import Namespace
+from datetime import datetime
+
+from caelestia.utils.paths import recording_notif_path, recording_path, recordings_dir
+
+
+class Command:
+    args: Namespace
+
+    def __init__(self, args: Namespace) -> None:
+        self.args = args
+
+    def run(self) -> None:
+        proc = subprocess.run(["pidof", "wl-screenrec"])
+        if proc.returncode == 0:
+            self.stop()
+        else:
+            self.start()
+
+    def start(self) -> None:
+        args = []
+
+        if self.args.region:
+            if self.args.region == "slurp":
+                region = subprocess.check_output(["slurp"], text=True)
+            else:
+                region = self.args.region
+            args += ["-g", region.strip()]
+
+        if self.args.sound:
+            sources = subprocess.check_output(["pactl", "list", "short", "sources"], text=True).splitlines()
+            for source in sources:
+                if "RUNNING" in source:
+                    args += ["--audio", "--audio-device", source.split()[1]]
+                    break
+            else:
+                raise ValueError("No audio source found")
+
+        proc = subprocess.Popen(
+            ["wl-screenrec", *args, "--codec", "hevc", "-f", recording_path],
+            stderr=subprocess.PIPE,
+            text=True,
+            start_new_session=True,
+        )
+
+        # Send notif if proc hasn't ended after a small delay
+        time.sleep(0.1)
+        if proc.poll() is None:
+            notif = subprocess.check_output(
+                ["notify-send", "-p", "-a", "caelestia-cli", "Recording started", "Recording..."], text=True
+            ).strip()
+            recording_notif_path.write_text(notif)
+        else:
+            subprocess.run(
+                [
+                    "notify-send",
+                    "-a",
+                    "caelestia-cli",
+                    "Recording failed",
+                    f"Recording failed to start: {proc.communicate()[1]}",
+                ]
+            )
+
+    def stop(self) -> None:
+        subprocess.run(["pkill", "wl-screenrec"])
+
+        # Move to recordings folder
+        new_path = recordings_dir / f"recording_{datetime.now().strftime('%Y%m%d_%H-%M-%S')}.mp4"
+        recording_path.rename(new_path)
+
+        # Close start notification
+        try:
+            notif = recording_notif_path.read_text()
+            subprocess.run(
+                [
+                    "gdbus",
+                    "call",
+                    "--session",
+                    "--dest=org.freedesktop.Notifications",
+                    "--object-path=/org/freedesktop/Notifications",
+                    "--method=org.freedesktop.Notifications.CloseNotification",
+                    notif,
+                ]
+            )
+        except IOError:
+            pass
+
+        action = subprocess.check_output(
+            [
+                "notify-send",
+                "-a",
+                "caelestia-cli",
+                "--action=watch=Watch",
+                "--action=open=Open",
+                "--action=delete=Delete",
+                "Recording stopped",
+                f"Recording saved in {new_path}",
+            ],
+            text=True,
+        ).strip()
+
+        if action == "watch":
+            subprocess.Popen(["app2unit", "-O", new_path], start_new_session=True)
+        elif action == "open":
+            p = subprocess.run(
+                [
+                    "dbus-send",
+                    "--session",
+                    "--dest=org.freedesktop.FileManager1",
+                    "--type=method_call",
+                    "/org/freedesktop/FileManager1",
+                    "org.freedesktop.FileManager1.ShowItems",
+                    f"array:string:file://{new_path}",
+                    "string:",
+                ]
+            )
+            if p.returncode != 0:
+                subprocess.Popen(["app2unit", "-O", new_path.parent], start_new_session=True)
+        elif action == "delete":
+            new_path.unlink()
diff --git a/src/caelestia/subcommands/scheme.py b/src/caelestia/subcommands/scheme.py
new file mode 100644
index 0000000..c95df96
--- /dev/null
+++ b/src/caelestia/subcommands/scheme.py
@@ -0,0 +1,30 @@
+from argparse import Namespace
+
+from caelestia.utils.scheme import get_scheme
+from caelestia.utils.theme import apply_colours
+
+
+class Command:
+    args: Namespace
+
+    def __init__(self, args: Namespace) -> None:
+        self.args = args
+
+    def run(self) -> None:
+        scheme = get_scheme()
+
+        if self.args.random:
+            scheme.set_random()
+            apply_colours(scheme.colours, scheme.mode)
+        elif self.args.name or self.args.flavour or self.args.mode or self.args.variant:
+            if self.args.name:
+                scheme.name = self.args.name
+            if self.args.flavour:
+                scheme.flavour = self.args.flavour
+            if self.args.mode:
+                scheme.mode = self.args.mode
+            if self.args.variant:
+                scheme.variant = self.args.variant
+            apply_colours(scheme.colours, scheme.mode)
+        else:
+            print(scheme)
diff --git a/src/caelestia/subcommands/screenshot.py b/src/caelestia/subcommands/screenshot.py
new file mode 100644
index 0000000..73d65f7
--- /dev/null
+++ b/src/caelestia/subcommands/screenshot.py
@@ -0,0 +1,78 @@
+import subprocess
+from argparse import Namespace
+from datetime import datetime
+
+from caelestia.utils import hypr
+from caelestia.utils.paths import screenshots_cache_dir, screenshots_dir
+
+
+class Command:
+    args: Namespace
+
+    def __init__(self, args: Namespace) -> None:
+        self.args = args
+
+    def run(self) -> None:
+        if self.args.region:
+            self.region()
+        else:
+            self.fullscreen()
+
+    def region(self) -> None:
+        freeze_proc = None
+
+        if self.args.freeze:
+            freeze_proc = subprocess.Popen(["wayfreeze", "--hide-cursor"])
+
+        if self.args.region == "slurp":
+            ws = hypr.message("activeworkspace")["id"]
+            geoms = [
+                f"{','.join(map(str, c['at']))} {'x'.join(map(str, c['size']))}"
+                for c in hypr.message("clients")
+                if c["workspace"]["id"] == ws
+            ]
+            region = subprocess.check_output(["slurp"], input="\n".join(geoms), text=True)
+        else:
+            region = self.args.region
+
+        sc_data = subprocess.check_output(["grim", "-l", "0", "-g", region.strip(), "-"])
+        swappy = subprocess.Popen(["swappy", "-f", "-"], stdin=subprocess.PIPE, start_new_session=True)
+        swappy.stdin.write(sc_data)
+        swappy.stdin.close()
+
+        if freeze_proc:
+            freeze_proc.kill()
+
+    def fullscreen(self) -> None:
+        sc_data = subprocess.check_output(["grim", "-"])
+
+        subprocess.run(["wl-copy"], input=sc_data)
+
+        dest = screenshots_cache_dir / datetime.now().strftime("%Y%m%d%H%M%S")
+        screenshots_cache_dir.mkdir(exist_ok=True, parents=True)
+        dest.write_bytes(sc_data)
+
+        action = subprocess.check_output(
+            [
+                "notify-send",
+                "-i",
+                "image-x-generic-symbolic",
+                "-h",
+                f"STRING:image-path:{dest}",
+                "-a",
+                "caelestia-cli",
+                "--action=open=Open",
+                "--action=save=Save",
+                "Screenshot taken",
+                f"Screenshot stored in {dest} and copied to clipboard",
+            ],
+            text=True,
+        ).strip()
+
+        if action == "open":
+            subprocess.Popen(["swappy", "-f", dest], start_new_session=True)
+        elif action == "save":
+            new_dest = (screenshots_dir / dest.name).with_suffix(".png")
+            new_dest.parent.mkdir(exist_ok=True, parents=True)
+            dest.rename(new_dest)
+            subprocess.run(["notify-send", "Screenshot saved", f"Saved to {new_dest}"])
diff --git a/src/caelestia/subcommands/shell.py b/src/caelestia/subcommands/shell.py
new file mode 100644
index 0000000..25a39d8
--- /dev/null
+++ b/src/caelestia/subcommands/shell.py
@@ -0,0 +1,41 @@
+import subprocess
+from argparse import Namespace
+
+from caelestia.utils import paths
+
+
+class Command:
+    args: Namespace
+
+    def __init__(self, args: Namespace) -> None:
+        self.args = args
+
+    def run(self) -> None:
+        if self.args.show:
+            # Print the ipc
+            self.print_ipc()
+        elif self.args.log:
+            # Print the log
+            self.print_log()
+        elif self.args.message:
+            # Send a message
+            self.message(*self.args.message)
+        else:
+            # Start the shell
+            self.shell()
+
+    def shell(self, *args: list[str]) -> str:
+        return subprocess.check_output(["qs", "-p", paths.c_data_dir / "shell", *args], text=True)
+
+    def print_ipc(self) -> None:
+        print(self.shell("ipc", "show"), end="")
+
+    def print_log(self) -> None:
+        log = self.shell("log")
+        # FIXME: remove when logging rules are added/warning is removed
+        for line in log.splitlines():
+            if "QProcess: Destroyed while process" not in line:
+                print(line)
+
+    def message(self, *args: list[str]) -> None:
+        print(self.shell("ipc", "call", *args), end="")
diff --git a/src/caelestia/subcommands/toggle.py b/src/caelestia/subcommands/toggle.py
new file mode 100644
index 0000000..b8ad11b
--- /dev/null
+++ b/src/caelestia/subcommands/toggle.py
@@ -0,0 +1,75 @@
+import subprocess
+from argparse import Namespace
+
+from caelestia.utils import hypr
+
+
+class Command:
+    args: Namespace
+    clients: list[dict[str, any]] = None
+
+    def __init__(self, args: Namespace) -> None:
+        self.args = args
+
+    def run(self) -> None:
+        getattr(self, self.args.workspace)()
+
+    def get_clients(self) -> list[dict[str, any]]:
+        if self.clients is None:
+            self.clients = hypr.message("clients")
+
+        return self.clients
+
+    def move_client(self, selector: callable, workspace: str) -> None:
+        for client in self.get_clients():
+            if selector(client):
+                hypr.dispatch("movetoworkspacesilent", f"special:{workspace},address:{client['address']}")
+
+    def spawn_client(self, selector: callable, spawn: list[str]) -> bool:
+        exists = any(selector(client) for client in self.get_clients())
+
+        if not exists:
+            subprocess.Popen(["app2unit", "--", *spawn], start_new_session=True)
+
+        return not exists
+
+    def spawn_or_move(self, selector: callable, spawn: list[str], workspace: str) -> None:
+        if not self.spawn_client(selector, spawn):
+            self.move_client(selector, 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", "--", "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:
+        workspaces = hypr.message("workspaces")
+        on_special_ws = any(ws["name"] == "special:special" for ws in workspaces)
+        toggle_ws = "special"
+
+        if not on_special_ws:
+            active_ws = hypr.message("activewindow")["workspace"]["name"]
+            if active_ws.startswith("special:"):
+                toggle_ws = active_ws[8:]
+
+        hypr.dispatch("togglespecialworkspace", toggle_ws)
diff --git a/src/caelestia/subcommands/wallpaper.py b/src/caelestia/subcommands/wallpaper.py
new file mode 100644
index 0000000..940dcb5
--- /dev/null
+++ b/src/caelestia/subcommands/wallpaper.py
@@ -0,0 +1,21 @@
+import json
+from argparse import Namespace
+
+from caelestia.utils.wallpaper import get_colours_for_wall, get_wallpaper, set_random, set_wallpaper
+
+
+class Command:
+    args: Namespace
+
+    def __init__(self, args: Namespace) -> None:
+        self.args = args
+
+    def run(self) -> None:
+        if self.args.print:
+            print(json.dumps(get_colours_for_wall(self.args.print, self.args.no_smart)))
+        elif self.args.file:
+            set_wallpaper(self.args.file, self.args.no_smart)
+        elif self.args.random:
+            set_random(self.args)
+        else:
+            print(get_wallpaper() or "No wallpaper set")
diff --git a/src/caelestia/subcommands/wsaction.py b/src/caelestia/subcommands/wsaction.py
new file mode 100644
index 0000000..d496381
--- /dev/null
+++ b/src/caelestia/subcommands/wsaction.py
@@ -0,0 +1,18 @@
+from argparse import Namespace
+
+from caelestia.utils import hypr
+
+
+class Command:
+    args: Namespace
+
+    def __init__(self, args: Namespace) -> None:
+        self.args = args
+
+    def run(self) -> None:
+        active_ws = hypr.message("activeworkspace")["id"]
+
+        if self.args.group:
+            hypr.dispatch(self.args.dispatcher, (self.args.workspace - 1) * 10 + active_ws % 10)
+        else:
+            hypr.dispatch(self.args.dispatcher, int((active_ws - 1) / 10) * 10 + self.args.workspace)