Compare commits

..

No commits in common. "main" and "1.4.0" have entirely different histories.
main ... 1.4.0

19 changed files with 72 additions and 281 deletions

2
.gitignore vendored
View file

@ -1,2 +0,0 @@
# Nix
result

View file

@ -6,16 +6,7 @@ Host process for multiple Hyprland plugins.
Check the [wiki](https://github.com/hyprland-community/pyprland/wiki) for more information. Check the [wiki](https://github.com/hyprland-community/pyprland/wiki) for more information.
# 1.4.2 (WIP) # Changelog
- [two new options](https://github.com/hyprland-community/pyprland/wiki/Plugins#size-optional) for scratchpads: `position` and `size` - from @iliayar
- bugfixes
# 1.4.1
- minor bugfixes
# 1.4.0
- Add [expose](https://github.com/hyprland-community/pyprland/wiki/Plugins#expose) addon - Add [expose](https://github.com/hyprland-community/pyprland/wiki/Plugins#expose) addon
- scratchpad: add [lazy](https://github.com/hyprland-community/pyprland/wiki/Plugins#lazy-optional) option - scratchpad: add [lazy](https://github.com/hyprland-community/pyprland/wiki/Plugins#lazy-optional) option

61
flake.lock generated
View file

@ -1,61 +0,0 @@
{
"nodes": {
"flake-utils": {
"inputs": {
"systems": "systems"
},
"locked": {
"lastModified": 1689068808,
"narHash": "sha256-6ixXo3wt24N/melDWjq70UuHQLxGV8jZvooRanIHXw0=",
"owner": "numtide",
"repo": "flake-utils",
"rev": "919d646de7be200f3bf08cb76ae1f09402b6f9b4",
"type": "github"
},
"original": {
"owner": "numtide",
"repo": "flake-utils",
"type": "github"
}
},
"nixpkgs": {
"locked": {
"lastModified": 1695416179,
"narHash": "sha256-610o1+pwbSu+QuF3GE0NU5xQdTHM3t9wyYhB9l94Cd8=",
"owner": "nixos",
"repo": "nixpkgs",
"rev": "715d72e967ec1dd5ecc71290ee072bcaf5181ed6",
"type": "github"
},
"original": {
"owner": "nixos",
"ref": "nixos-23.05",
"repo": "nixpkgs",
"type": "github"
}
},
"root": {
"inputs": {
"flake-utils": "flake-utils",
"nixpkgs": "nixpkgs"
}
},
"systems": {
"locked": {
"lastModified": 1681028828,
"narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
"owner": "nix-systems",
"repo": "default",
"rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
"type": "github"
},
"original": {
"owner": "nix-systems",
"repo": "default",
"type": "github"
}
}
},
"root": "root",
"version": 7
}

View file

@ -1,27 +0,0 @@
{
description = "pyprland";
inputs = {
flake-utils.url = "github:numtide/flake-utils";
nixpkgs.url = "github:nixos/nixpkgs/nixos-23.05";
};
outputs = { self, nixpkgs, flake-utils, ... }:
flake-utils.lib.eachDefaultSystem
(system:
let
pkgs = import nixpkgs {
inherit system;
};
in
{
packages = rec {
pyprland = pkgs.poetry2nix.mkPoetryApplication {
projectDir = ./.;
python = pkgs.python310;
};
default = pyprland;
};
}
);
}

7
poetry.lock generated
View file

@ -1,7 +0,0 @@
# This file is automatically @generated by Poetry 1.4.2 and should not be changed by hand.
package = []
[metadata]
lock-version = "2.0"
python-versions = "^3.10"
content-hash = "53f2eabc9c26446fbcc00d348c47878e118afc2054778c3c803a0a8028af27d9"

View file

@ -1,16 +1,16 @@
#!/bin/env python #!/bin/env python
""" Pyprland - an Hyprland companion app (cli client & daemon) """ " Pyprland - an Hyprland companion app "
import asyncio import asyncio
from typing import cast
import json
import sys
import os
import importlib import importlib
import itertools import itertools
import json
import os
import sys
from typing import cast
from .common import PyprError, get_logger, init_logger
from .ipc import get_event_stream from .ipc import get_event_stream, init as ipc_init
from .ipc import init as ipc_init from .common import init_logger, get_logger, PyprError
from .plugins.interface import Plugin from .plugins.interface import Plugin
CONTROL = f'/tmp/hypr/{ os.environ["HYPRLAND_INSTANCE_SIGNATURE"] }/.pyprland.sock' CONTROL = f'/tmp/hypr/{ os.environ["HYPRLAND_INSTANCE_SIGNATURE"] }/.pyprland.sock'
@ -73,11 +73,6 @@ class Pyprland:
self.log.debug("%s.%s%s", plugin.name, full_name, params) self.log.debug("%s.%s%s", plugin.name, full_name, params)
try: try:
await getattr(plugin, full_name)(*params) await getattr(plugin, full_name)(*params)
except AssertionError as e:
self.log.error(
"Bug detected, please report on https://github.com/fdev31/pyprland/issues"
)
self.log.exception(e)
except Exception as e: # pylint: disable=W0718 except Exception as e: # pylint: disable=W0718
self.log.warning( self.log.warning(
"%s::%s(%s) failed:", plugin.name, full_name, params "%s::%s(%s) failed:", plugin.name, full_name, params
@ -87,15 +82,11 @@ class Pyprland:
async def read_events_loop(self): async def read_events_loop(self):
"Consumes the event loop and calls corresponding handlers" "Consumes the event loop and calls corresponding handlers"
while not self.stopped: while not self.stopped:
try: data = (await self.event_reader.readline()).decode()
data = (await self.event_reader.readline()).decode()
except UnicodeDecodeError:
self.log.error("Invalid unicode while reading events")
continue
if not data: if not data:
self.log.critical("Reader starved") self.log.critical("Reader starved")
return return
cmd, params = data.split(">>", 1) cmd, params = data.split(">>")
full_name = f"event_{cmd}" full_name = f"event_{cmd}"
await self._callHandler(full_name, params) await self._callHandler(full_name, params)

View file

@ -1,6 +1,6 @@
""" Shared utilities: logging """ """ Shared utilities: logging """
import logging
import os import os
import logging
__all__ = ["DEBUG", "get_logger", "init_logger"] __all__ = ["DEBUG", "get_logger", "init_logger"]

View file

@ -1,12 +1,12 @@
#!/bin/env python #!/bin/env python
""" Interact with hyprland using sockets """ """ Interact with hyprland using sockets """
import asyncio import asyncio
import json
import os
from logging import Logger from logging import Logger
from typing import Any from typing import Any
import json
import os
from .common import PyprError, get_logger from .common import get_logger, PyprError
log: Logger | None = None log: Logger | None = None

View file

@ -2,10 +2,10 @@
toggle_minimized allows having an "expose" like selection of minimized windows toggle_minimized allows having an "expose" like selection of minimized windows
""" """
from typing import Any, cast from typing import Any, cast
from ..ipc import hyprctl, hyprctlJSON
from .interface import Plugin from .interface import Plugin
from ..ipc import hyprctlJSON, hyprctl
class Extension(Plugin): # pylint: disable=missing-class-docstring class Extension(Plugin): # pylint: disable=missing-class-docstring
exposed: list[dict] = [] exposed: list[dict] = []
@ -42,7 +42,7 @@ class Extension(Plugin): # pylint: disable=missing-class-docstring
await hyprctl( await hyprctl(
f"movetoworkspacesilent {client['workspace']['id']},address:{client['address']}" f"movetoworkspacesilent {client['workspace']['id']},address:{client['address']}"
) )
# await hyprctl("togglespecialworkspace exposed") await hyprctl("togglespecialworkspace exposed")
await hyprctl(f"focuswindow address:{focused_addr}") await hyprctl(f"focuswindow address:{focused_addr}")
self.exposed = [] self.exposed = []
else: else:

View file

@ -1,6 +1,5 @@
" Common plugin interface " " Common plugin interface "
from typing import Any from typing import Any
from ..common import get_logger from ..common import get_logger

View file

@ -1,27 +0,0 @@
" Ironbar Plugin "
import os
import json
import asyncio
from .interface import Plugin
SOCKET = f"/run/user/{os.getuid()}/ironbar-ipc.sock"
async def ipcCall(**params):
ctl_reader, ctl_writer = await asyncio.open_unix_connection(SOCKET)
ctl_writer.write(json.dumps(params).encode("utf-8"))
await ctl_writer.drain()
ret = await ctl_reader.read()
ctl_writer.close()
await ctl_writer.wait_closed()
return json.loads(ret)
class Extension(Plugin):
"Toggles ironbar on/off"
is_visible = True
async def run_toggle_ironbar(self, bar_name: str):
self.is_visible = not self.is_visible
await ipcCall(type="set_visible", visible=self.is_visible, bar_name=bar_name)

View file

@ -1,9 +1,9 @@
" Moves unreachable client windows to the currently focused workspace" " Moves unreachable client windows to the currently focused workspace"
from typing import Any, cast from typing import Any, cast
from ..ipc import hyprctl, hyprctlJSON
from .interface import Plugin from .interface import Plugin
from ..ipc import hyprctlJSON, hyprctl
def contains(monitor, window): def contains(monitor, window):
"Tell if a window is visible in a monitor" "Tell if a window is visible in a monitor"

View file

@ -1,7 +1,8 @@
" Toggles workspace zooming " " Toggles workspace zooming "
from ..ipc import hyprctl
from .interface import Plugin from .interface import Plugin
from ..ipc import hyprctl
class Extension(Plugin): # pylint: disable=missing-class-docstring class Extension(Plugin): # pylint: disable=missing-class-docstring
zoomed = False zoomed = False

View file

@ -1,9 +1,9 @@
" The monitors plugin " " The monitors plugin "
import subprocess import subprocess
from typing import Any, cast from typing import Any, cast
from .interface import Plugin
from ..ipc import hyprctlJSON from ..ipc import hyprctlJSON
from .interface import Plugin
def configure_monitors(monitors, screenid: str, pos_x: int, pos_y: int) -> None: def configure_monitors(monitors, screenid: str, pos_x: int, pos_y: int) -> None:
@ -40,9 +40,6 @@ def configure_monitors(monitors, screenid: str, pos_x: int, pos_y: int) -> None:
class Extension(Plugin): # pylint: disable=missing-class-docstring class Extension(Plugin): # pylint: disable=missing-class-docstring
async def load_config(self, config) -> None: async def load_config(self, config) -> None:
await super().load_config(config) await super().load_config(config)
await self.run_relayout()
async def run_relayout(self):
monitors = cast(list[dict], await hyprctlJSON("monitors")) monitors = cast(list[dict], await hyprctlJSON("monitors"))
for monitor in monitors: for monitor in monitors:
await self.event_monitoradded( await self.event_monitoradded(
@ -85,10 +82,7 @@ class Extension(Plugin): # pylint: disable=missing-class-docstring
for mon_pattern, conf in self.config["placement"].items(): for mon_pattern, conf in self.config["placement"].items():
if mon_pattern in mon_description: if mon_pattern in mon_description:
for placement, other_mon_description in conf.items(): for placement, other_mon_description in conf.items():
try: ref = mon_by_name[other_mon_description]
ref = mon_by_name[other_mon_description]
except KeyError:
continue
if ref: if ref:
place = placement.lower() place = placement.lower()
x: int = 0 x: int = 0

View file

@ -1,12 +1,15 @@
" Scratchpads addon " " Scratchpads addon "
import asyncio
import os import os
from itertools import count import asyncio
import subprocess import subprocess
from typing import Any, cast from typing import Any, cast
import logging
from ..ipc import get_focused_monitor_props, hyprctl, hyprctlJSON from ..ipc import (
hyprctl,
hyprctlJSON,
get_focused_monitor_props,
)
from .interface import Plugin from .interface import Plugin
DEFAULT_MARGIN = 60 DEFAULT_MARGIN = 60
@ -14,7 +17,6 @@ DEFAULT_MARGIN = 60
async def get_client_props_by_address(addr: str): async def get_client_props_by_address(addr: str):
"Returns client properties given its address" "Returns client properties given its address"
assert len(addr) > 2, "Client address is invalid"
for client in await hyprctlJSON("clients"): for client in await hyprctlJSON("clients"):
assert isinstance(client, dict) assert isinstance(client, dict)
if client.get("address") == addr: if client.get("address") == addr:
@ -85,7 +87,6 @@ class Animations:
class Scratch: class Scratch:
"A scratchpad state including configuration & client state" "A scratchpad state including configuration & client state"
log = logging.getLogger("scratch")
def __init__(self, uid, opts): def __init__(self, uid, opts):
self.uid = uid self.uid = uid
@ -103,7 +104,7 @@ class Scratch:
for line in f.readlines(): for line in f.readlines():
if line.startswith("State"): if line.startswith("State"):
state = line.split()[1] state = line.split()[1]
return state not in "ZX" # not "Z (zombie)"or "X (dead)" return state in "RSDTt" # not "Z (zombie)"or "X (dead)"
return False return False
def reset(self, pid: int) -> None: def reset(self, pid: int) -> None:
@ -122,14 +123,7 @@ class Scratch:
"update the internal client info property, if not provided, refresh based on the current address" "update the internal client info property, if not provided, refresh based on the current address"
if client_info is None: if client_info is None:
client_info = await get_client_props_by_address("0x" + self.address) client_info = await get_client_props_by_address("0x" + self.address)
try: assert isinstance(client_info, dict)
assert isinstance(client_info, dict)
except AssertionError as e:
self.log.error(
f"client_info of {self.address} must be a dict: {client_info}"
)
raise AssertionError(e) from e
self.client_info.update(client_info) self.client_info.update(client_info)
def __str__(self): def __str__(self):
@ -200,8 +194,7 @@ class Extension(Plugin): # pylint: disable=missing-class-docstring
self.procs[name] = proc self.procs[name] = proc
pid = proc.pid pid = proc.pid
self.scratches[name].reset(pid) self.scratches[name].reset(pid)
self.scratches_by_pid[pid] = scratch self.scratches_by_pid[proc.pid] = scratch
self.log.info(f"scratch {scratch.uid} has pid {pid}")
if old_pid and old_pid in self.scratches_by_pid: if old_pid and old_pid in self.scratches_by_pid:
del self.scratches_by_pid[old_pid] del self.scratches_by_pid[old_pid]
@ -218,7 +211,6 @@ class Extension(Plugin): # pylint: disable=missing-class-docstring
scratch.just_created = False scratch.just_created = False
else: else:
for uid, scratch in self.scratches.items(): for uid, scratch in self.scratches.items():
self.log.info((scratch.address, addr))
if scratch.client_info and scratch.address != addr: if scratch.client_info and scratch.address != addr:
if ( if (
scratch.visible scratch.visible
@ -238,7 +230,6 @@ class Extension(Plugin): # pylint: disable=missing-class-docstring
if not class_lookup_hack: if not class_lookup_hack:
return False return False
self.log.debug("Lookup hack triggered") self.log.debug("Lookup hack triggered")
# hack to update the client info from the provided class
for client in await hyprctlJSON("clients"): for client in await hyprctlJSON("clients"):
assert isinstance(client, dict) assert isinstance(client, dict)
for pending_scratch in class_lookup_hack: for pending_scratch in class_lookup_hack:
@ -251,7 +242,7 @@ class Extension(Plugin): # pylint: disable=missing-class-docstring
async def event_openwindow(self, params) -> None: async def event_openwindow(self, params) -> None:
"open windows hook" "open windows hook"
addr, wrkspc, _kls, _title = params.split(",", 3) addr, wrkspc, _kls, _title = params.split(",", 3)
if self._respawned_scratches: if wrkspc.startswith("special"):
item = self.scratches_by_address.get(addr) item = self.scratches_by_address.get(addr)
if not item and self._respawned_scratches: if not item and self._respawned_scratches:
# hack for windows which aren't related to the process (see #8) # hack for windows which aren't related to the process (see #8)
@ -301,21 +292,26 @@ class Extension(Plugin): # pylint: disable=missing-class-docstring
return # abort sequence return # abort sequence
await asyncio.sleep(0.2) # await for animation to finish await asyncio.sleep(0.2) # await for animation to finish
async def updateScratchInfo(self, orig_scratch: Scratch | None = None) -> None: async def updateScratchInfo(self, scratch: Scratch | None = None) -> None:
"""Update every scratchpads information if no `scratch` given, """Update every scratchpads information if no `scratch` given,
else update a specific scratchpad info""" else update a specific scratchpad info"""
pid = orig_scratch.pid if orig_scratch else None if scratch is None:
for client in await hyprctlJSON("clients"): for client in await hyprctlJSON("clients"):
assert isinstance(client, dict) assert isinstance(client, dict)
if pid and pid != client["pid"]: scratch = self.scratches_by_address.get(client["address"][2:])
continue if not scratch:
scratch = self.scratches_by_address.get(client["address"][2:]) scratch = self.scratches_by_pid.get(client["pid"])
if not scratch: if scratch:
scratch = self.scratches_by_pid.get(client["pid"]) self.scratches_by_address[client["address"][2:]] = scratch
if scratch: if scratch:
self.scratches_by_address[client["address"][2:]] = scratch await scratch.updateClientInfo(client)
if scratch: else:
await scratch.updateClientInfo(client) add_to_address_book = ("address" not in scratch.client_info) or (
scratch.address not in self.scratches_by_address
)
await scratch.updateClientInfo()
if add_to_address_book:
self.scratches_by_address[scratch.client_info["address"][2:]] = scratch
async def run_hide(self, uid: str, force=False, autohide=False) -> None: async def run_hide(self, uid: str, force=False, autohide=False) -> None:
"""<name> hides scratchpad "name" """ """<name> hides scratchpad "name" """
@ -327,11 +323,8 @@ class Extension(Plugin): # pylint: disable=missing-class-docstring
if not scratch.visible and not force: if not scratch.visible and not force:
self.log.warning("%s is already hidden", uid) self.log.warning("%s is already hidden", uid)
return return
scratch.visible = False
if not scratch.isAlive():
await self.run_show(uid, force=True)
return
self.log.info("Hiding %s", uid) self.log.info("Hiding %s", uid)
scratch.visible = False
addr = "address:0x" + scratch.address addr = "address:0x" + scratch.address
animation_type: str = scratch.conf.get("animation", "").lower() animation_type: str = scratch.conf.get("animation", "").lower()
if animation_type: if animation_type:
@ -349,27 +342,6 @@ class Extension(Plugin): # pylint: disable=missing-class-docstring
) )
del self.focused_window_tracking[uid] del self.focused_window_tracking[uid]
async def ensure_alive(self, uid, item=None):
if item is None:
item = self.scratches.get(uid)
if not item.isAlive():
self.log.info("%s is not running, restarting...", uid)
if uid in self.procs:
self.procs[uid].kill()
if item.pid in self.scratches_by_pid:
del self.scratches_by_pid[item.pid]
if item.address in self.scratches_by_address:
del self.scratches_by_address[item.address]
self.log.info(f"starting {uid}")
await self.start_scratch_command(uid)
self.log.info(f"{uid} started")
self.log.info("==> Wait for spawning")
loop_count = count()
while uid in self._respawned_scratches and next(loop_count) < 10:
await asyncio.sleep(0.05)
self.log.info(f"=> spawned {uid} as proc {item.pid}")
async def run_show(self, uid, force=False) -> None: async def run_show(self, uid, force=False) -> None:
"""<name> shows scratchpad "name" """ """<name> shows scratchpad "name" """
uid = uid.strip() uid = uid.strip()
@ -388,7 +360,18 @@ class Extension(Plugin): # pylint: disable=missing-class-docstring
return return
self.log.info("Showing %s", uid) self.log.info("Showing %s", uid)
await self.ensure_alive(uid, item)
if not item.isAlive():
self.log.info("%s is not running, restarting...", uid)
if uid in self.procs:
self.procs[uid].kill()
if item.pid in self.scratches_by_pid:
del self.scratches_by_pid[item.pid]
if item.address in self.scratches_by_address:
del self.scratches_by_address[item.address]
await self.start_scratch_command(uid)
while uid in self._respawned_scratches:
await asyncio.sleep(0.05)
item.visible = True item.visible = True
monitor = await get_focused_monitor_props() monitor = await get_focused_monitor_props()
@ -396,8 +379,6 @@ class Extension(Plugin): # pylint: disable=missing-class-docstring
await self.updateScratchInfo(item) await self.updateScratchInfo(item)
assert item.address, "No address !"
addr = "address:0x" + item.address addr = "address:0x" + item.address
animation_type = item.conf.get("animation", "").lower() animation_type = item.conf.get("animation", "").lower()
@ -413,47 +394,5 @@ class Extension(Plugin): # pylint: disable=missing-class-docstring
await fn(monitor, item.client_info, addr, margin) await fn(monitor, item.client_info, addr, margin)
await hyprctl(f"focuswindow {addr}") await hyprctl(f"focuswindow {addr}")
size = item.conf.get("size")
if size:
x_size, y_size = self._convert_coords(size, monitor)
await hyprctl(f"resizewindowpixel exact {x_size} {y_size},{addr}")
position = item.conf.get("position")
if position:
x_pos, y_pos = self._convert_coords(position, monitor)
x_pos_abs, y_pos_abs = x_pos + monitor["x"], y_pos + monitor["y"]
await hyprctl(f"movewindowpixel exact {x_pos_abs} {y_pos_abs},{addr}")
await asyncio.sleep(0.2) # ensure some time for events to propagate await asyncio.sleep(0.2) # ensure some time for events to propagate
self.transitioning_scratches.discard(uid) self.transitioning_scratches.discard(uid)
def _convert_coords(self, coords, monitor):
"""
Converts a string like "X Y" to coordinates relative to monitor
Supported formats for X, Y:
- Percentage: "V%". V in [0; 100]
Example:
"10% 20%", monitor 800x600 => 80, 120
"""
assert coords, "coords must be non null"
def convert(s, dim):
if s[-1] == "%":
p = int(s[:-1])
if p < 0 or p > 100:
raise Exception(f"Percentage must be in range [0; 100], got {p}")
scale = float(monitor["scale"])
return int(monitor[dim] / scale * p / 100)
else:
raise Exception(f"Unsupported format for dimension {dim} size, got {s}")
try:
x_str, y_str = coords.split()
return convert(x_str, "width"), convert(y_str, "height")
except Exception as e:
self.log.error(f"Failed to read coordinates: {e}")
raise e

View file

@ -1,9 +1,9 @@
" shift workspaces across monitors " " shift workspaces across monitors "
from typing import cast from typing import cast
from ..ipc import hyprctl, hyprctlJSON
from .interface import Plugin from .interface import Plugin
from ..ipc import hyprctlJSON, hyprctl
class Extension(Plugin): # pylint: disable=missing-class-docstring class Extension(Plugin): # pylint: disable=missing-class-docstring
monitors: list[str] = [] monitors: list[str] = []

View file

@ -1,9 +1,9 @@
" Toggle monitors on or off " " Toggle monitors on or off "
from typing import Any, cast from typing import Any, cast
from ..ipc import hyprctl, hyprctlJSON
from .interface import Plugin from .interface import Plugin
from ..ipc import hyprctlJSON, hyprctl
class Extension(Plugin): # pylint: disable=missing-class-docstring class Extension(Plugin): # pylint: disable=missing-class-docstring
async def run_toggle_dpms(self): async def run_toggle_dpms(self):

View file

@ -1,9 +1,9 @@
""" Force workspaces to follow the focus / mouse """ """ Force workspaces to follow the focus / mouse """
from typing import cast from typing import cast
from ..ipc import hyprctl, hyprctlJSON
from .interface import Plugin from .interface import Plugin
from ..ipc import hyprctlJSON, hyprctl
class Extension(Plugin): # pylint: disable=missing-class-docstring class Extension(Plugin): # pylint: disable=missing-class-docstring
workspace_list: list[int] = [] workspace_list: list[int] = []

View file

@ -1,6 +1,6 @@
[tool.poetry] [tool.poetry]
name = "pyprland" name = "pyprland"
version = "1.4.1" version = "1.3.1"
description = "An hyperland plugin system" description = "An hyperland plugin system"
authors = ["fdev31 <fdev31@gmail.com>"] authors = ["fdev31 <fdev31@gmail.com>"]
license = "MIT" license = "MIT"