v 1.0

.pre-commit-config.yaml

@@ -11,7 +11,7 @@ repos:
   #    hooks:
   #      - id: prettier
   - repo: https://github.com/ambv/black
-    rev: "23.1.0"
+    rev: "23.3.0"
     hooks:
       - id: black
   - repo: https://github.com/lovesegfault/beautysh
@@ -19,7 +19,7 @@ repos:
     hooks:
       - id: beautysh
   - repo: https://github.com/adrienverge/yamllint
-    rev: "v1.29.0"
+    rev: "v1.31.0"
     hooks:
       - id: yamllint
 

README.md

@@ -0,0 +1,184 @@
+# Extensions & tweaks for hyprland
+
+Host process for multiple Hyprland plugins.
+A single config file `~/.config/hypr/pyprland.json` is used, using the following syntax:
+
+```json
+{
+  "pyprland": {
+    "plugins": ["plugin_name"]
+  },
+  "plugin_name": {
+    "plugin_option": 42
+  }
+}
+```
+
+Built-in plugins are:
+
+- `scratchpad` implements dropdowns & togglable poppups
+- `monitors` allows relative placement of monitors depending on the model
+- `workspaces_follow_focus` provides commands and handlers allowing a more flexible workspaces usage on multi-monitor setups
+
+## Installation
+
+```
+pip install pyprland
+```
+
+## Getting started
+
+Create a configuration file in `~/.config/hypr/pyprland.json` enabling a list of plugins, each plugin may have its own configuration needs, eg:
+
+```json
+{
+  "pyprland": {
+    "plugins": [
+      "scratchpads",
+      "monitors",
+      "workspaces_follow_focus"
+    ]
+  },
+  "scratchpads": {
+    "term": {
+      "command": "kitty --class kitty-dropterm",
+      "class": "kitty-dropterm",
+      "animation": "fromTop",
+      "unfocus": "hide"
+    },
+    "volume": {
+      "command": "pavucontrol",
+      "class": "pavucontrol",
+      "unfocus": "hide",
+      "animation": "fromRight"
+    }
+  },
+  "monitors": {
+    "placement": {
+      "BenQ PJ": {
+        "topOf": "eDP-1"
+      }
+    }
+  }
+}
+```
+
+# Configuring plugins
+
+## `monitors`
+
+Requires `wlr-randr`.
+
+Allows relative placement of monitors depending on the model ("description" returned by `hyprctl monitors`).
+
+### Configuration
+
+Supported placements are:
+
+- leftOf
+- topOf
+- rightOf
+- bottomOf
+
+## `workspaces_follow_focus`
+
+Make non-visible workspaces follow the focused monitor.
+Also provides commands to switch between workspaces wile preserving the current monitor assignments: 
+
+### Commands
+
+- `change_workspace` `<direction>`: changes the workspace of the focused monitor
+
+Example usage in `hyprland.conf`:
+
+```
+bind = $mainMod, K, exec, pypr change_workspace +1
+bind = $mainMod, J, exec, pypr change_workspace -1
+ ```
+
+### Configuration
+
+You can set the `max_workspaces` property, defaults to `10`.
+
+## `scratchpads`
+
+Check [hpr-scratcher](https://github.com/hyprland-community/hpr-scratcher), it's fully compatible, just put the configuration under "scratchpads".
+
+As an example, defining two scratchpads:
+
+- _term_ which would be a kitty terminal on upper part of the screen
+- _volume_ which would be a pavucontrol window on the right part of the screen
+
+In your `hyprland.conf` add something like this:
+
+```ini
+exec-once = hpr-scratcher
+
+# Repeat this for each scratchpad you need
+bind = $mainMod,V,exec,hpr-scratcher toggle volume
+windowrule = float,^(pavucontrol)$
+windowrule = workspace special silent,^(pavucontrol)$
+
+bind = $mainMod,A,exec,hpr-scratcher toggle term
+$dropterm  = ^(kitty-dropterm)$
+windowrule = float,$dropterm
+windowrule = workspace special silent,$dropterm
+windowrule = size 75% 60%,$dropterm
+```
+
+Then in the configuration file, add something like this:
+
+```json
+"scratchpads": {
+  "term": {
+    "command": "kitty --class kitty-dropterm",
+    "animation": "fromTop",
+    "margin": 50,
+    "unfocus": "hide"
+  },
+  "volume": {
+    "command": "pavucontrol",
+    "animation": "fromRight"
+  }
+}
+```
+
+And you'll be able to toggle pavucontrol with MOD + V.
+
+### Command-line options
+
+- `reload` : reloads the configuration file
+- `toggle <scratchpad name>` : toggle the given scratchpad
+- `show <scratchpad name>` : show the given scratchpad
+- `hide <scratchpad name>` : hide the given scratchpad
+
+Note: with no argument it runs the daemon (doesn't fork in the background)
+
+### Scratchpad Options
+
+#### command
+
+This is the command you wish to run in the scratchpad.
+For a nice startup you need to be able to identify this window in `hyprland.conf`, using `--class` is often a good idea.
+
+#### animation
+
+Type of animation to use
+
+- `null` / `""` / not defined
+- "fromTop"
+- "fromBottom"
+- "fromLeft"
+- "fromRight"
+
+#### offset (optional)
+
+number of pixels for the animation.
+
+#### unfocus (optional)
+
+allow to hide the window when the focus is lost when set to "hide"
+
+#### margin (optional)
+
+number of pixels separating the scratchpad from the screen border

pyprland/command.py

@@ -0,0 +1,167 @@
+#!/bin/env python
+import asyncio
+import json
+import sys
+import os
+import importlib
+import traceback
+
+
+from .ipc import get_event_stream
+from .common import DEBUG
+from .plugins.interface import Plugin
+
+CONTROL = f'/tmp/hypr/{ os.environ["HYPRLAND_INSTANCE_SIGNATURE"] }/.pyprland.sock'
+
+CONFIG_FILE = "~/.config/hypr/pyprland.json"
+
+
+class Pyprland:
+    server: asyncio.Server
+    event_reader: asyncio.StreamReader
+    stopped = False
+    name = "builtin"
+
+    def __init__(self):
+        self.plugins: dict[str, Plugin] = {}
+
+    async def load_config(self):
+        self.config = json.loads(
+            open(os.path.expanduser(CONFIG_FILE), encoding="utf-8").read()
+        )
+        for name in self.config["pyprland"]["plugins"]:
+            if name not in self.plugins:
+                modname = name if "." in name else f"pyprland.plugins.{name}"
+                try:
+                    plug = importlib.import_module(modname).Exported(name)
+                    await plug.init()
+                    self.plugins[name] = plug
+                except Exception as e:
+                    print(f"Error loading plugin {name}: {e}")
+                    if DEBUG:
+                        traceback.print_exc()
+            await self.plugins[name].load_config(self.config)
+
+    async def _callHandler(self, full_name, *params):
+        for plugin in [self] + list(self.plugins.values()):
+            if hasattr(plugin, full_name):
+                try:
+                    await getattr(plugin, full_name)(*params)
+                except Exception as e:
+                    print(f"{plugin.name}::{full_name}({params}) failed:")
+                    if DEBUG:
+                        traceback.print_exc()
+
+    async def read_events_loop(self):
+        while not self.stopped:
+            data = (await self.event_reader.readline()).decode()
+            if not data:
+                print("Reader starved")
+                return
+            cmd, params = data.split(">>")
+            full_name = f"event_{cmd}"
+
+            if DEBUG:
+                print(f"EVT {full_name}({params.strip()})")
+            await self._callHandler(full_name, params)
+
+    async def read_command(self, reader, writer):
+        data = (await reader.readline()).decode()
+        if not data:
+            print("Server starved")
+            return
+        if data == "exit\n":
+            self.stopped = True
+            writer.close()
+            await writer.wait_closed()
+            self.server.close()
+            return
+        args = data.split(None, 1)
+        if len(args) == 1:
+            cmd = args[0]
+            args = []
+        else:
+            cmd = args[0]
+            args = args[1:]
+
+        full_name = f"run_{cmd}"
+
+        if DEBUG:
+            print(f"CMD: {full_name}({args})")
+
+        await self._callHandler(full_name, *args)
+
+    async def serve(self):
+        try:
+            async with self.server:
+                await self.server.serve_forever()
+        finally:
+            for plugin in self.plugins.values():
+                await plugin.exit()
+
+    async def run(self):
+        await asyncio.gather(
+            asyncio.create_task(self.serve()),
+            asyncio.create_task(self.read_events_loop()),
+        )
+
+    run_reload = load_config
+
+
+async def run_daemon():
+    manager = Pyprland()
+    manager.server = await asyncio.start_unix_server(manager.read_command, CONTROL)
+    events_reader, events_writer = await get_event_stream()
+    manager.event_reader = events_reader
+
+    try:
+        await manager.load_config()  # ensure sockets are connected first
+    except FileNotFoundError:
+        print(
+            f"No config file found, create one at {CONFIG_FILE} with a valid pyprland.plugins list"
+        )
+        raise SystemExit(1)
+
+    try:
+        await manager.run()
+    except KeyboardInterrupt:
+        print("Interrupted")
+    except asyncio.CancelledError:
+        print("Bye!")
+    finally:
+        events_writer.close()
+        await events_writer.wait_closed()
+        manager.server.close()
+        await manager.server.wait_closed()
+
+
+async def run_client():
+    if sys.argv[1] == "--help":
+        print(
+            """Commands:
+  reload
+  show   <scratchpad name>
+  hide   <scratchpad name>
+  toggle <scratchpad name>
+
+If arguments are ommited, runs the daemon which will start every configured command.
+"""
+        )
+        return
+
+    _, writer = await asyncio.open_unix_connection(CONTROL)
+    writer.write((" ".join(sys.argv[1:])).encode())
+    await writer.drain()
+    writer.close()
+    await writer.wait_closed()
+
+
+def main():
+    try:
+        asyncio.run(run_daemon() if len(sys.argv) <= 1 else run_client())
+    except KeyboardInterrupt:
+        pass
+
+
+if __name__ == "__main__":
+    main()

pyprland/common.py

@@ -0,0 +1,4 @@
+import os
+
+DEBUG = os.environ.get("DEBUG", False)
+CONFIG_FILE = os.path.expanduser("~/.config/hypr/scratchpads.json")

pyprland/ipc.py

@@ -0,0 +1,58 @@
+#!/bin/env python
+import asyncio
+from typing import Any
+import json
+import os
+
+from .common import DEBUG
+
+
+HYPRCTL = f'/tmp/hypr/{ os.environ["HYPRLAND_INSTANCE_SIGNATURE"] }/.socket.sock'
+EVENTS = f'/tmp/hypr/{ os.environ["HYPRLAND_INSTANCE_SIGNATURE"] }/.socket2.sock'
+
+
+async def get_event_stream():
+    return await asyncio.open_unix_connection(EVENTS)
+
+
+async def hyprctlJSON(command) -> list[dict[str, Any]] | dict[str, Any]:
+    if DEBUG:
+        print("(JS)>>>", command)
+    ctl_reader, ctl_writer = await asyncio.open_unix_connection(HYPRCTL)
+    ctl_writer.write(f"-j/{command}".encode())
+    await ctl_writer.drain()
+    resp = await ctl_reader.read()
+    ctl_writer.close()
+    await ctl_writer.wait_closed()
+    return json.loads(resp)
+
+
+async def hyprctl(command):
+    if DEBUG:
+        print(">>>", command)
+    ctl_reader, ctl_writer = await asyncio.open_unix_connection(HYPRCTL)
+    ctl_writer.write(f"/dispatch {command}".encode())
+    await ctl_writer.drain()
+    resp = await ctl_reader.read(100)
+    ctl_writer.close()
+    await ctl_writer.wait_closed()
+    if DEBUG:
+        print("<<<", resp)
+    return resp == b"ok"
+
+
+async def get_workspaces() -> list[dict[str, Any]]:
+    return await hyprctlJSON("workspaces")
+
+
+async def get_focused_monitor_props():
+    for monitor in await hyprctlJSON("monitors"):
+        assert isinstance(monitor, dict)
+        if monitor.get("focused") == True:
+            return monitor
+
+
+async def get_client_props_by_pid(pid: int):
+    for client in await hyprctlJSON("clients"):
+        if client.get("pid") == pid:
+            return client

pyprland/plugins/experimental.py

@@ -0,0 +1,10 @@
+from .interface import Plugin
+
+from ..ipc import hyprctlJSON, hyprctl, get_workspaces
+
+
+class Experimental(Plugin):
+    pass
+
+
+Exported = Experimental

pyprland/plugins/interface.py

@@ -0,0 +1,18 @@
+from typing import Any
+
+
+class Plugin:
+    def __init__(self, name: str):
+        self.name = name
+
+    async def init(self):
+        pass
+
+    async def exit(self):
+        return
+
+    async def load_config(self, config: dict[str, Any]):
+        try:
+            self.config = config[self.name]
+        except KeyError:
+            self.config = {}

pyprland/plugins/monitors.py

@@ -0,0 +1,48 @@
+from typing import Any
+from .interface import Plugin
+import subprocess
+
+from ..ipc import hyprctlJSON
+
+
+class MonitorLayout(Plugin):
+    async def event_monitoradded(self, screenid):
+        screenid = screenid.strip()
+
+        monitors: list[dict[str, Any]] = await hyprctlJSON("monitors")
+        for mon in monitors:
+            if mon["name"].startswith(screenid):
+                mon_name = mon["description"]
+                break
+        else:
+            print(f"Monitor {screenid} not found")
+            return
+
+        mon_by_name = {m["name"]: m for m in monitors}
+
+        newmon = mon_by_name[screenid]
+
+        for mon_pattern, conf in self.config["placement"].items():
+            if mon_pattern in mon_name:
+                for placement, mon_name in conf.items():
+                    ref = mon_by_name[mon_name]
+                    if ref:
+                        place = placement.lower()
+                        if place == "topof":
+                            x: int = ref["x"]
+                            y: int = ref["y"] - newmon["height"]
+                        elif place == "bottomof":
+                            x: int = ref["x"]
+                            y: int = ref["y"] + ref["height"]
+                        elif place == "leftof":
+                            x: int = ref["x"] - newmon["width"]
+                            y: int = ref["y"]
+                        else:  # rightof
+                            x: int = ref["x"] + ref["width"]
+                            y: int = ref["y"]
+                        subprocess.call(
+                            ["wlr-randr", "--output", screenid, "--pos", f"{x},{y}"]
+                        )
+
+
+Exported = MonitorLayout

pyprland/plugins/scratchpads.py

@@ -0,0 +1,293 @@
+import subprocess
+import asyncio
+from ..ipc import (
+    hyprctl,
+    hyprctlJSON,
+    get_focused_monitor_props,
+    get_client_props_by_pid,
+)
+import os
+
+from .interface import Plugin
+
+DEFAULT_MARGIN = 60
+
+
+class Scratch:
+    def __init__(self, uid, opts):
+        self.uid = uid
+        self.pid = 0
+        self.conf = opts
+        self.visible = False
+        self.just_created = True
+        self.clientInfo = {}
+
+    def isAlive(self):
+        path = f"/proc/{self.pid}"
+        if os.path.exists(path):
+            for line in open(os.path.join(path, "status"), "r").readlines():
+                if line.startswith("State"):
+                    state = line.split()[1]
+                    return state in "RSDTt"  # not "Z (zombie)"or "X (dead)"
+        return False
+
+    def reset(self, pid: int):
+        self.pid = pid
+        self.visible = False
+        self.just_created = True
+        self.clientInfo = {}
+
+    @property
+    def address(self) -> str:
+        return str(self.clientInfo.get("address", ""))[2:]
+
+    async def updateClientInfo(self, clientInfo=None):
+        if clientInfo is None:
+            clientInfo = await get_client_props_by_pid(self.pid)
+        assert isinstance(clientInfo, dict)
+        self.clientInfo.update(clientInfo)
+
+
+class ScratchpadManager(Plugin):
+    async def init(self):
+        self.procs: dict[str, subprocess.Popen] = {}
+        self.scratches: dict[str, Scratch] = {}
+        self.transitioning_scratches: set[str] = set()
+        self._respawned_scratches: set[str] = set()
+        self.scratches_by_address: dict[str, Scratch] = {}
+        self.scratches_by_pid: dict[int, Scratch] = {}
+
+    async def exit(self):
+        async def die_in_piece(scratch: Scratch):
+            proc = self.procs[scratch.uid]
+            proc.terminate()
+            for n in range(10):
+                if not scratch.isAlive():
+                    break
+                await asyncio.sleep(0.1)
+            if scratch.isAlive():
+                proc.kill()
+            proc.wait()
+
+        await asyncio.gather(
+            *(die_in_piece(scratch) for scratch in self.scratches.values())
+        )
+
+    async def load_config(self, config):
+        config = config["scratchpads"]
+        scratches = {k: Scratch(k, v) for k, v in config.items()}
+
+        is_updating = bool(self.scratches)
+
+        for name in scratches:
+            if name not in self.scratches:
+                self.scratches[name] = scratches[name]
+            else:
+                self.scratches[name].conf = scratches[name].conf
+
+        if is_updating:
+            await self.exit()
+
+        # not known yet
+        for name in self.scratches:
+            self.start_scratch_command(name)
+
+    def start_scratch_command(self, name: str):
+        self._respawned_scratches.add(name)
+        scratch = self.scratches[name]
+        old_pid = self.procs[name].pid if name in self.procs else 0
+        self.procs[name] = subprocess.Popen(
+            scratch.conf["command"],
+            stdin=subprocess.DEVNULL,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            shell=True,
+        )
+        pid = self.procs[name].pid
+        self.scratches[name].reset(pid)
+        self.scratches_by_pid[self.procs[name].pid] = scratch
+        if old_pid:
+            del self.scratches_by_pid[old_pid]
+
+    # Events
+    async def event_activewindowv2(self, addr):
+        addr = addr.strip()
+        scratch = self.scratches_by_address.get(addr)
+        if scratch:
+            if scratch.just_created:
+                await self.run_hide(scratch.uid, force=True)
+                scratch.just_created = False
+        else:
+            for uid, scratch in self.scratches.items():
+                if scratch.clientInfo and scratch.address != addr:
+                    if (
+                        scratch.visible
+                        and scratch.conf.get("unfocus") == "hide"
+                        and scratch.uid not in self.transitioning_scratches
+                    ):
+                        await self.run_hide(uid)
+
+    async def event_openwindow(self, params):
+        addr, wrkspc, kls, title = params.split(",", 3)
+        if wrkspc.startswith("special"):
+            item = self.scratches_by_address.get(addr)
+            if not item and self._respawned_scratches:
+                await self.updateScratchInfo()
+                item = self.scratches_by_address.get(addr)
+            if item and item.just_created:
+                self._respawned_scratches.discard(item.uid)
+                await self.run_hide(item.uid, force=True)
+                item.just_created = False
+
+    async def run_toggle(self, uid: str):
+        uid = uid.strip()
+        item = self.scratches.get(uid)
+        if not item:
+            print(f"{uid} is not configured")
+            return
+        if item.visible:
+            await self.run_hide(uid)
+        else:
+            await self.run_show(uid)
+
+    async def updateScratchInfo(self, scratch: Scratch | None = None):
+        if scratch is None:
+            for client in await hyprctlJSON("clients"):
+                assert isinstance(client, dict)
+                pid = client["pid"]
+                assert isinstance(pid, int)
+                scratch = self.scratches_by_pid.get(pid)
+                if scratch:
+                    await scratch.updateClientInfo(client)
+                    self.scratches_by_address[
+                        scratch.clientInfo["address"][2:]
+                    ] = scratch
+        else:
+            add_to_address_book = ("address" not in scratch.clientInfo) or (
+                scratch.address not in self.scratches_by_address
+            )
+            await scratch.updateClientInfo()
+            if add_to_address_book:
+                self.scratches_by_address[scratch.clientInfo["address"][2:]] = scratch
+
+    async def run_hide(self, uid: str, force=False):
+        uid = uid.strip()
+        item = self.scratches.get(uid)
+        if not item:
+            print(f"{uid} is not configured")
+            return
+        if not item.visible and not force:
+            print(f"{uid} is already hidden")
+            return
+        item.visible = False
+        pid = "pid:%d" % item.pid
+        animation_type = item.conf.get("animation", "").lower()
+        if animation_type:
+            offset = item.conf.get("offset")
+            if offset is None:
+                if "size" not in item.clientInfo:
+                    await self.updateScratchInfo(item)
+
+                offset = int(1.3 * item.clientInfo["size"][1])
+
+            if animation_type == "fromtop":
+                await hyprctl(f"movewindowpixel 0 -{offset},{pid}")
+            elif animation_type == "frombottom":
+                await hyprctl(f"movewindowpixel 0 {offset},{pid}")
+            elif animation_type == "fromleft":
+                await hyprctl(f"movewindowpixel -{offset} 0,{pid}")
+            elif animation_type == "fromright":
+                await hyprctl(f"movewindowpixel {offset} 0,{pid}")
+
+            if uid in self.transitioning_scratches:
+                return  # abort sequence
+            await asyncio.sleep(0.2)  # await for animation to finish
+        if uid not in self.transitioning_scratches:
+            await hyprctl(f"movetoworkspacesilent special:scratch,{pid}")
+
+    async def _animation_fromtop(self, monitor, client, client_uid, margin):
+        mon_x = monitor["x"]
+        mon_y = monitor["y"]
+        mon_width = monitor["width"]
+
+        client_width = client["size"][0]
+        margin_x = int((mon_width - client_width) / 2) + mon_x
+        await hyprctl(f"movewindowpixel exact {margin_x} {mon_y + margin},{client_uid}")
+
+    async def _animation_frombottom(self, monitor, client, client_uid, margin):
+        mon_x = monitor["x"]
+        mon_y = monitor["y"]
+        mon_width = monitor["width"]
+        mon_height = monitor["height"]
+
+        client_width = client["size"][0]
+        client_height = client["size"][1]
+        margin_x = int((mon_width - client_width) / 2) + mon_x
+        await hyprctl(
+            f"movewindowpixel exact {margin_x} {mon_y + mon_height - client_height - margin},{client_uid}"
+        )
+
+    async def _animation_fromleft(self, monitor, client, client_uid, margin):
+        mon_y = monitor["y"]
+        mon_height = monitor["height"]
+
+        client_height = client["size"][1]
+        margin_y = int((mon_height - client_height) / 2) + mon_y
+
+        await hyprctl(f"movewindowpixel exact {margin} {margin_y},{client_uid}")
+
+    async def _animation_fromright(self, monitor, client, client_uid, margin):
+        mon_y = monitor["y"]
+        mon_width = monitor["width"]
+        mon_height = monitor["height"]
+
+        client_width = client["size"][0]
+        client_height = client["size"][1]
+        margin_y = int((mon_height - client_height) / 2) + mon_y
+        await hyprctl(
+            f"movewindowpixel exact {mon_width - client_width - margin} {margin_y},{client_uid}"
+        )
+
+    async def run_show(self, uid, force=False):
+        uid = uid.strip()
+        item = self.scratches.get(uid)
+
+        if not item:
+            print(f"{uid} is not configured")
+            return
+
+        if item.visible and not force:
+            print(f"{uid} is already visible")
+            return
+
+        if not item.isAlive():
+            print(f"{uid} is not running, restarting...")
+            self.procs[uid].kill()
+            self.start_scratch_command(uid)
+            while uid in self._respawned_scratches:
+                await asyncio.sleep(0.05)
+
+        item.visible = True
+        monitor = await get_focused_monitor_props()
+        assert monitor
+
+        await self.updateScratchInfo(item)
+
+        pid = "pid:%d" % item.pid
+
+        animation_type = item.conf.get("animation", "").lower()
+
+        wrkspc = monitor["activeWorkspace"]["id"]
+        self.transitioning_scratches.add(uid)
+        await hyprctl(f"movetoworkspacesilent {wrkspc},{pid}")
+        if animation_type:
+            margin = item.conf.get("margin", DEFAULT_MARGIN)
+            fn = getattr(self, "_animation_%s" % animation_type)
+            await fn(monitor, item.clientInfo, pid, margin)
+
+        await hyprctl(f"focuswindow {pid}")
+        await asyncio.sleep(0.2)  # ensure some time for events to propagate
+        self.transitioning_scratches.discard(uid)
+
+
+Exported = ScratchpadManager

pyprland/plugins/workspaces_follow_focus.py

@@ -0,0 +1,55 @@
+from .interface import Plugin
+
+from ..ipc import hyprctlJSON, hyprctl, get_workspaces
+
+
+class Extension(Plugin):
+    async def load_config(self, config):
+        await super().load_config(config)
+        self.workspace_list = list(range(1, self.config.get("max_workspaces", 10)))
+
+    async def event_focusedmon(self, screenid_index):
+        monitor_id, workspace_id = screenid_index.split(",")
+        workspace_id = int(workspace_id)
+        # move every free wokrspace to the currently focused desktop
+        busy_workspaces = set(
+            mon["activeWorkspace"]["id"]
+            for mon in await hyprctlJSON("monitors")
+            if mon["name"] != monitor_id
+        )
+
+        for n in self.workspace_list:
+            if n in busy_workspaces or n == workspace_id:
+                continue
+            await hyprctl(f"moveworkspacetomonitor {n} {monitor_id}")
+        await hyprctl(f"workspace {workspace_id}")
+
+    async def run_change_workspace(self, direction: str):
+        increment = int(direction)
+        # get focused screen info
+        monitors = await hyprctlJSON("monitors")
+        assert isinstance(monitors, list)
+        for monitor in monitors:
+            if monitor["focused"]:
+                break
+        assert isinstance(monitor, dict)
+        busy_workspaces = set(
+            m["activeWorkspace"]["id"] for m in monitors if m["id"] != monitor["id"]
+        )
+        # get workspaces info
+        workspaces = await get_workspaces()
+        assert isinstance(workspaces, list)
+        workspaces.sort(key=lambda x: x["id"])
+        cur_workspace = monitor["activeWorkspace"]["id"]
+        available_workspaces = [
+            i for i in self.workspace_list if i not in busy_workspaces
+        ]
+        idx = available_workspaces.index(cur_workspace)
+        next_workspace = available_workspaces[
+            (idx + increment) % len(available_workspaces)
+        ]
+        await hyprctl(f"moveworkspacetomonitor {next_workspace},{monitor['name']}")
+        await hyprctl(f"workspace {next_workspace}")
+
+
+Exported = Extension

pyproject.toml

@@ -0,0 +1,20 @@
+[tool.poetry]
+name = "pyprland"
+version = "1.0.0"
+description = "An hyperland plugin system"
+authors = ["fdev31 <fdev31@gmail.com>"]
+license = "MIT"
+readme = "README.md"
+packages = [{include = "pyprland"}]
+homepage = "https://github.com/fdev31/pyprland/"
+
+[tool.poetry.scripts]
+pypr = "pyprland.command:main"
+
+[tool.poetry.dependencies]
+python = "^3.10"
+
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"